这个Python脚本是一个用于操作阿里云OSS（对象存储服务）的工具。它允许用户上传文件和文件夹，列出文件（包括指定前缀的文件），删除文件（包括指定前缀的文件），以及创建目录。用户可以通过命令行交互选择功能，并输入相应的路径或前缀。该脚本主要用于方便地将图片或文件上传到OSS，生成URL，以便用于AI或图床等场景。

import os
import oss2
from dotenv import load_dotenv
from pathlib import Path
import logging
from rich.console import Console
from rich.prompt import Prompt, IntPrompt
from rich.table import Table

# 初始化日志记录
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
console = Console()

# 加载环境变量,需在.env文件中配置OSS_ENDPOINT,OSS_ACCESS_KEY_ID,OSS_ACCESS_KEY_SECRET,OSS_BUCKET_NAME
load_dotenv()

class OSSManager:
    def __init__(self, OSS_ENDPOINT, OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_BUCKET_NAME):
        """
        初始化OSSManager，设置认证和存储桶
        """
        self.auth = oss2.Auth(OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET)
        self.bucket = oss2.Bucket(self.auth, OSS_ENDPOINT, OSS_BUCKET_NAME)
        self.OSS_ENDPOINT = OSS_ENDPOINT
        self.OSS_ACCESS_KEY_ID = OSS_ACCESS_KEY_ID
        self.OSS_ACCESS_KEY_SECRET = OSS_ACCESS_KEY_SECRET
        self.OSS_BUCKET_NAME = OSS_BUCKET_NAME
        logging.info("OSSManager 初始化成功")

    def upload_file(self, local_path: str, upload_path: str = "uploads", object_name: str = None):
        """
        上传文件到阿里云OSS并返回文件URL
        :param local_path: 本地文件路径
        :param upload_path: 上传到OSS后的目录
        :param object_name: 上传到OSS后的文件名
        :return: 文件名和URL
        """
        local_file = Path(local_path)
        if not local_file.exists():
            logging.error(f"文件不存在: {local_path}")
            return None, None

        upload_dir = Path(upload_path) if upload_path else Path("uploads")
        self.create_dir(str(upload_dir))

        object_name = object_name or local_file.name
        object_key = f"{upload_dir}/{object_name}"

        try:
            self.bucket.put_object_from_file(str(object_key), str(local_file))
            logging.info(f"文件上传成功: {object_key}")
        except oss2.exceptions.OssError as e:
            logging.error(f"上传失败: {e}")
            return None, None

        url = f"https://{self.OSS_BUCKET_NAME}.{self.OSS_ENDPOINT}/{object_key}"
        return object_key, url

    def upload_dir(self, local_dir: str, upload_path: str = None):
        """
        上传目录到阿里云OSS并返回上传的目录路径
        :param local_dir: 本地目录路径
        :param upload_path: 上传到OSS后的目录
        :return: 上传后的目录路径
        """
        local_directory = Path(local_dir)
        if not local_directory.exists() or not local_directory.is_dir():
            logging.error(f"目录不存在或不是目录: {local_dir}")
            return None

        upload_dir = upload_path or local_directory.name
        self.create_dir(upload_dir)

        for item in local_directory.iterdir():
            if item.is_file():
                self.upload_file(str(item), upload_dir)
            elif item.is_dir():
                self.upload_dir(str(item), f"{upload_dir}/{item.name}")

        logging.info(f"目录上传完成: {upload_dir}")
        return upload_dir

    def delete_file(self, file_path: str):
        """
        删除文件
        :param file_path: 文件路径
        """
        try:
            self.bucket.delete_object(file_path)
            logging.info(f"文件已删除: {file_path}")
        except oss2.exceptions.OssError as e:
            logging.error(f"删除失败: {e}")

    def delete_files_by_prefix(self, prefix: str):
        """
        删除指定前缀的文件
        :param prefix: 前缀
        """
        try:
            for obj in oss2.ObjectIterator(self.bucket, prefix=prefix):
                self.bucket.delete_object(obj.key)
                logging.info(f"文件已删除: {obj.key}")
        except oss2.exceptions.OssError as e:
            logging.error(f"删除失败: {e}")

    def list_files(self, prefix: str = ""):
        """
        列举指定前缀的文件
        :param prefix: 前缀
        """
        try:
            table = Table(title="OSS 文件列表")
            table.add_column("序号", justify="right", style="cyan", no_wrap=True)
            table.add_column("文件名", style="magenta")
            table.add_column("URL", style="green")

            for idx, obj in enumerate(oss2.ObjectIterator(self.bucket, prefix=prefix)):
                url = f"https://{self.OSS_BUCKET_NAME}.{self.OSS_ENDPOINT}/{obj.key}"
                table.add_row(str(idx), obj.key, url)
            console.print(table)
        except oss2.exceptions.OssError as e:
            logging.error(f"列举文件失败: {e}")
        return oss2.ObjectIterator(self.bucket, prefix=prefix)

    def create_dir(self, dir_path: str):
        """
        创建目录
        :param dir_path: 目录路径
        """
        try:
            self.bucket.put_object(f"{dir_path}/", '')
            logging.info(f"目录已创建: {dir_path}")
        except oss2.exceptions.OssError as e:
            logging.error(f"创建目录失败: {e}")

def main():
    # 从环境变量中获取配置信息
    OSS_ENDPOINT = os.getenv('OSS_ENDPOINT')
    OSS_ACCESS_KEY_ID = os.getenv('OSS_ACCESS_KEY_ID')
    OSS_ACCESS_KEY_SECRET = os.getenv('OSS_ACCESS_KEY_SECRET')
    OSS_BUCKET_NAME = os.getenv('OSS_BUCKET_NAME')

    if not all([OSS_ENDPOINT, OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_BUCKET_NAME]):
        console.print("[red]缺少必要的环境变量，请检查 .env 文件。[/red]")
        return

    oss_manager = OSSManager(OSS_ENDPOINT, OSS_ACCESS_KEY_ID, OSS_ACCESS_KEY_SECRET, OSS_BUCKET_NAME)

    while True:
        console.print("\n[bold blue]请选择功能：[/bold blue]")
        console.print("1. 上传文件")
        console.print("2. 上传文件夹")
        console.print("3. 列举所有文件")
        console.print("4. 列举指定前缀的文件")
        console.print("5. 删除文件")
        console.print("6. 删除指定前缀的文件")
        console.print("7. 创建目录")
        console.print("8. 退出")

        choice = Prompt.ask("请输入选择", choices=[str(i) for i in range(1, 9)])

        if choice == "1":
            local_path = Prompt.ask("请输入本地文件路径")
            upload_path = Prompt.ask("请输入上传到OSS后的目录（默认为 uploads）", default="uploads")
            object_name = Prompt.ask("请输入上传到OSS后的文件名（回车使用文件的原名）", default="", show_default=False)
            object_name = object_name if object_name else None
            oss_manager.upload_file(local_path, upload_path, object_name)
        elif choice == "2":
            folder_path = Prompt.ask("请输入本地文件夹路径")
            upload_path = Prompt.ask("请输入上传到OSS后的目录（默认为同名目录）", default="", show_default=False)
            upload_path = upload_path if upload_path else None
            oss_manager.upload_dir(folder_path, upload_path)
        elif choice == "3":
            oss_manager.list_files()
        elif choice == "4":
            prefix = Prompt.ask("请输入前缀", default="")
            oss_manager.list_files(prefix)
        elif choice == "5":
            file_path = Prompt.ask("请输入文件路径")
            oss_manager.delete_file(file_path)
        elif choice == "6":
            prefix = Prompt.ask("请输入前缀")
            oss_manager.delete_files_by_prefix(prefix)
        elif choice == "7":
            dir_path = Prompt.ask("请输入目录路径")
            oss_manager.create_dir(dir_path)
        elif choice == "8":
            console.print("退出程序。")
            break
        else:
            console.print("[red]无效选择，请重新输入。[/red]")

if __name__ == "__main__":
    main()