From 33fe06c59e57790b8c82cdfca1f4c14b2e7a7c03 Mon Sep 17 00:00:00 2001 From: RhenCloud Date: Fri, 14 Nov 2025 23:26:21 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=87=8D=E6=9E=84=E9=85=8D=E7=BD=AE?= =?UTF-8?q?=E7=AE=A1=E7=90=86=EF=BC=8C=E9=9B=86=E4=B8=AD=E7=AE=A1=E7=90=86?= =?UTF-8?q?=E7=8E=AF=E5=A2=83=E5=8F=98=E9=87=8F=E5=B9=B6=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E5=AD=98=E5=82=A8=E5=90=8E=E7=AB=AF=E5=AE=9E=E7=8E=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 43 ++++++------ README.md | 42 +++++------- app.py | 76 +++------------------ config.py | 83 +++++++++++++++++++++++ handlers/routes.py | 37 +++------- pyproject.toml | 2 +- storages/factory.py | 20 ++---- storages/github.py | 39 ++++++----- storages/r2.py | 38 ++++++----- utils.py | 160 ++++++++++++++++++++++++++++++++++++++++++++ 10 files changed, 347 insertions(+), 193 deletions(-) create mode 100644 config.py create mode 100644 utils.py diff --git a/.env.example b/.env.example index 240da0b..e4aee31 100644 --- a/.env.example +++ b/.env.example @@ -9,43 +9,38 @@ STORAGE_TYPE=r2 # ==================== Cloudflare R2 配置 ==================== +# R2 账户 ID +R2_ACCOUNT_ID=your-account-id + # R2 访问凭证 -ACCESS_KEY_ID=YOUR_R2_ACCESS_KEY_ID -SECRET_ACCESS_KEY=YOUR_R2_SECRET_ACCESS_KEY +R2_ACCESS_KEY_ID=YOUR_R2_ACCESS_KEY_ID +R2_SECRET_ACCESS_KEY=YOUR_R2_SECRET_ACCESS_KEY # R2 存储桶名称 R2_BUCKET_NAME=drive -# R2 Endpoint URL -R2_ENDPOINT_URL=https://.r2.cloudflarestorage.com - -# R2 区域 (默认: auto) -R2_REGION=auto - -# R2 公共访问 URL (可选，例如: https://pub-.r2.dev) -R2_PUBLIC_URL=https://pub-.r2.dev - -# R2 预签名 URL 过期时间（秒，默认: 3600） -R2_PRESIGN_EXPIRES=3600 +# R2 公共访问域名 (可选，例如: https://pub-.r2.dev) +R2_PUBLIC_DOMAIN=https://pub-.r2.dev # ==================== GitHub 存储配置 ==================== -# GitHub 仓库所有者 -GITHUB_REPO_OWNER=your-username - -# GitHub 仓库名称 -GITHUB_REPO_NAME=your-repo +# GitHub 仓库 (格式: owner/repo) +GITHUB_REPO=your-username/your-repo # GitHub 访问令牌 (需要 repo 权限) -GITHUB_ACCESS_TOKEN=your-access-token +GITHUB_TOKEN=your-access-token # GitHub 分支名称 (默认: main) GITHUB_BRANCH=main -# GitHub Raw 文件反向代理 URL (可选，用于加速访问) -# 例如: https://raw.fastgit.org 或 https://ghproxy.com/https://raw.githubusercontent.com -# 留空则使用官方 raw.githubusercontent.com -GITHUB_RAW_PROXY_URL= - # ==================== 应用配置 ==================== +# 服务器配置 +HOST=0.0.0.0 +PORT=5000 +DEBUG=false + # 缩略图缓存时间（秒，默认: 3600） THUMB_TTL_SECONDS=3600 + +# 预签名 URL 过期时间（秒，默认: 3600） +PRESIGNED_URL_EXPIRES=3600 +THUMB_TTL_SECONDS=3600 diff --git a/README.md b/README.md index 5a81529..b9d74e3 100644 --- a/README.md +++ b/README.md @@ -96,20 +96,18 @@ python app.py ```env STORAGE_TYPE=r2 +# R2 账户 ID +R2_ACCOUNT_ID=your-account-id + # R2 访问凭证 -ACCESS_KEY_ID=your_access_key_id -SECRET_ACCESS_KEY=your_secret_access_key +R2_ACCESS_KEY_ID=your_access_key_id +R2_SECRET_ACCESS_KEY=your_secret_access_key # R2 存储桶配置 R2_BUCKET_NAME=your_bucket_name -R2_ENDPOINT_URL=https://your-account-id.r2.cloudflarestorage.com -R2_REGION=auto -# 可选：公共访问 URL -R2_PUBLIC_URL=https://pub-your-bucket.r2.dev - -# 可选：预签名 URL 过期时间（秒） -R2_PRESIGN_EXPIRES=3600 +# 可选：公共访问域名 +R2_PUBLIC_DOMAIN=https://pub-your-bucket.r2.dev ``` ### GitHub Repository 配置 @@ -117,26 +115,15 @@ R2_PRESIGN_EXPIRES=3600 ```env STORAGE_TYPE=github -# GitHub 仓库所有者（用户名或组织） -GITHUB_REPO_OWNER=your-username - -# GitHub 仓库名称 -GITHUB_REPO_NAME=your-repo +# GitHub 仓库 (格式: owner/repo) +GITHUB_REPO=your-username/your-repo # GitHub 个人访问令牌（需要 repo 权限） # 获取方式：https://github.com/settings/tokens -GITHUB_ACCESS_TOKEN=ghp_your_token_here +GITHUB_TOKEN=ghp_your_token_here # GitHub 分支名称（可选，默认: main） GITHUB_BRANCH=main - -# GitHub Raw 文件反向代理 URL（可选，用于加速访问） -# 常用反向代理： -# - https://raw.fastgit.org （推荐，速度快） -# - https://ghproxy.com -# - https://raw.kgithub.com -# 留空则使用官方 raw.githubusercontent.com（国内可能较慢） -GITHUB_RAW_PROXY_URL=https://raw.fastgit.org ``` ## 项目结构 @@ -144,6 +131,8 @@ GITHUB_RAW_PROXY_URL=https://raw.fastgit.org ```bash cloud-index/ ├── app.py # Flask 应用主入口 +├── config.py # 统一配置管理 +├── utils.py # 工具函数模块 ├── handlers/ │ └── routes.py # 路由处理器 ├── storages/ # 存储后端实现 @@ -153,11 +142,16 @@ cloud-index/ │ ├── r2.py # Cloudflare R2 实现 │ └── github.py # GitHub Repository 实现 ├── templates/ # HTML 模板 +│ ├── base.html │ ├── index.html │ └── footer.html ├── static/ # 静态资源 -│ └── thumbs/ +│ ├── css/ +│ │ └── main.css +│ └── js/ +│ └── main.js ├── .env.example # 环境变量示例 +├── pyproject.toml # 项目配置和依赖 └── requirements.txt # Python 依赖 ``` diff --git a/app.py b/app.py index fa1f4ab..ecf2398 100644 --- a/app.py +++ b/app.py @@ -1,14 +1,15 @@ -import os import tomllib from pathlib import Path -import dotenv from flask import Flask +import utils +from config import Config from handlers.routes import main_route from storages.factory import StorageFactory -dotenv.load_dotenv() +# 验证配置 +Config.validate() # 从 pyproject.toml 读取版本号 @@ -34,73 +35,18 @@ app.register_blueprint(main_route) # 初始化存储（使用工厂模式） storage = StorageFactory.get_storage() -# 缩略图默认 TTL（秒），可通过环境变量覆盖 -THUMB_TTL = int(os.environ.get("THUMB_TTL_SECONDS", "3600")) - -# 注册一个安全的 filesizeformat 过滤器，处理 None 和非数字值 +# 注册模板过滤器 @app.template_filter("filesizeformat") def filesizeformat_filter(value): - try: - if value is None: - return "-" - num = float(value) # 使用 float 而不是 int 以保持精度 - except Exception: - return "-" - - for unit in ["B", "KB", "MB", "GB", "TB"]: - if num < 1024: - # 对于字节，显示整数 - if unit == "B": - return f"{int(num)}{unit}" - # 其他单位保留两位小数 - return f"{num:.2f}{unit}" - num = num / 1024.0 - return f"{num:.2f}PB" + """格式化文件大小""" + return utils.format_file_size(value) -# 注册一个文件图标过滤器 @app.template_filter("fileicon") def fileicon_filter(filename): - if not filename: - return "fas fa-file" - - ext = filename.lower().split(".")[-1] if "." in filename else "" - - # 图片文件 - if ext in ["jpg", "jpeg", "png", "gif", "bmp", "webp", "svg"]: - return "fas fa-image" - - # 音频文件 - if ext in ["mp3", "wav", "ogg", "flac", "m4a", "aac"]: - return "fas fa-music" - - # 视频文件 - if ext in ["mp4", "webm", "avi", "mov", "wmv", "flv", "mkv"]: - return "fas fa-video" - - # 文档文件 - if ext in ["pdf", "doc", "docx", "txt", "md", "rtf"]: - return "fas fa-file-alt" - - # 压缩文件 - if ext in ["zip", "rar", "7z", "tar", "gz"]: - return "fas fa-file-archive" - - # 代码文件 - if ext in ["py", "js", "html", "css", "java", "cpp", "c", "php"]: - return "fas fa-file-code" - - # 表格文件 - if ext in ["xls", "xlsx", "csv"]: - return "fas fa-file-excel" - - # 演示文件 - if ext in ["ppt", "pptx"]: - return "fas fa-file-powerpoint" - - # 默认文件图标 - return "fas fa-file" + """获取文件图标""" + return utils.get_file_icon(filename) def get_public_url(key: str) -> str: @@ -135,6 +81,4 @@ def inject_version(): if __name__ == "__main__": - port = int(os.environ.get("PORT", 5000)) - host = os.environ.get("HOST", "0.0.0.0") - app.run(host=host, port=port, debug=True) + app.run(host=Config.HOST, port=Config.PORT, debug=Config.DEBUG) diff --git a/config.py b/config.py new file mode 100644 index 0000000..1be3829 --- /dev/null +++ b/config.py @@ -0,0 +1,83 @@ +""" +配置管理模块 +集中管理所有环境变量和应用配置 +""" + +import os +from typing import Optional + +import dotenv + +# 加载环境变量 +dotenv.load_dotenv() + + +class Config: + """应用配置类""" + + # 存储配置 + STORAGE_TYPE: str = os.getenv("STORAGE_TYPE", "").lower() + + # R2 配置 + R2_ACCOUNT_ID: Optional[str] = os.getenv("R2_ACCOUNT_ID") + R2_ACCESS_KEY_ID: Optional[str] = os.getenv("R2_ACCESS_KEY_ID") + R2_SECRET_ACCESS_KEY: Optional[str] = os.getenv("R2_SECRET_ACCESS_KEY") + R2_BUCKET_NAME: Optional[str] = os.getenv("R2_BUCKET_NAME") + R2_PUBLIC_DOMAIN: Optional[str] = os.getenv("R2_PUBLIC_DOMAIN") + + # GitHub 配置 + GITHUB_TOKEN: Optional[str] = os.getenv("GITHUB_TOKEN") + GITHUB_REPO: Optional[str] = os.getenv("GITHUB_REPO") # 格式: owner/repo + GITHUB_BRANCH: str = os.getenv("GITHUB_BRANCH", "main") + + # 应用配置 + HOST: str = os.getenv("HOST", "0.0.0.0") + PORT: int = int(os.getenv("PORT", "5000")) + DEBUG: bool = os.getenv("DEBUG", "false").lower() == "true" + + # 缩略图配置 + THUMB_TTL_SECONDS: int = int(os.getenv("THUMB_TTL_SECONDS", "3600")) + THUMB_SIZE: tuple[int, int] = (300, 300) # 缩略图尺寸 + + # URL过期时间配置 + PRESIGNED_URL_EXPIRES: int = int(os.getenv("PRESIGNED_URL_EXPIRES", "3600")) + + @classmethod + def validate(cls) -> None: + """验证必需的配置项是否已设置""" + if not cls.STORAGE_TYPE: + raise ValueError("STORAGE_TYPE environment variable is not set. Supported types: r2, github") + + if cls.STORAGE_TYPE == "r2": + required = ["R2_ACCOUNT_ID", "R2_ACCESS_KEY_ID", "R2_SECRET_ACCESS_KEY", "R2_BUCKET_NAME"] + missing = [key for key in required if not getattr(cls, key)] + if missing: + raise ValueError(f"Missing required R2 configuration: {', '.join(missing)}") + + elif cls.STORAGE_TYPE == "github": + required = ["GITHUB_TOKEN", "GITHUB_REPO"] + missing = [key for key in required if not getattr(cls, key)] + if missing: + raise ValueError(f"Missing required GitHub configuration: {', '.join(missing)}") + + elif cls.STORAGE_TYPE not in ["r2", "github"]: + raise ValueError(f"Unsupported storage type: {cls.STORAGE_TYPE}. Supported types: r2, github") + + @classmethod + def get_storage_config(cls) -> dict: + """获取当前存储类型的配置字典""" + if cls.STORAGE_TYPE == "r2": + return { + "account_id": cls.R2_ACCOUNT_ID, + "access_key_id": cls.R2_ACCESS_KEY_ID, + "secret_access_key": cls.R2_SECRET_ACCESS_KEY, + "bucket_name": cls.R2_BUCKET_NAME, + "public_domain": cls.R2_PUBLIC_DOMAIN, + } + elif cls.STORAGE_TYPE == "github": + return { + "token": cls.GITHUB_TOKEN, + "repo": cls.GITHUB_REPO, + "branch": cls.GITHUB_BRANCH, + } + return {} diff --git a/handlers/routes.py b/handlers/routes.py index 7542ea5..31e613f 100644 --- a/handlers/routes.py +++ b/handlers/routes.py @@ -1,10 +1,10 @@ import hashlib -import os from datetime import datetime from typing import Any, Dict, List from flask import Blueprint, Response, abort, jsonify, redirect, render_template, request +from config import Config from storages.factory import StorageFactory main_route = Blueprint("main", __name__) @@ -12,23 +12,6 @@ main_route = Blueprint("main", __name__) # 初始化存储（使用工厂模式） storage = StorageFactory.get_storage() -# 缩略图默认 TTL（秒），可通过环境变量覆盖 -THUMB_TTL = int(os.environ.get("THUMB_TTL_SECONDS", "3600")) - - -def format_timestamp(timestamp) -> str: - """ - 格式化时间戳为人类可读的格式 - """ - return storage.format_timestamp(timestamp) - - -def get_public_url(key: str) -> str: - """ - 生成对象的公共访问 URL - """ - return storage.get_public_url(key) - def get_file_url(key: str) -> str: """生成通过服务器访问文件的 URL""" @@ -53,12 +36,12 @@ def build_file_entry(obj: Dict[str, Any], prefix: str) -> Dict[str, Any] | None: "name": rel_name, "key": key, "size": obj.get("Size"), - "last_modified": format_timestamp(obj.get("LastModified")), + "last_modified": storage.format_timestamp(obj.get("LastModified")), "is_dir": False, "file_url": get_file_url(key), } - public_url = get_public_url(key) + public_url = storage.get_public_url(key) if public_url: entry["public_url"] = public_url @@ -220,7 +203,7 @@ def thumb(file_path): """返回图片的缩略图，使用 Vercel Cache Headers 避免重复从 R2 拉取""" # 设置更长的缓存控制头以支持浏览器本地缓存 cache_headers = { - "Cache-Control": f"public, max-age={THUMB_TTL}", + "Cache-Control": f"public, max-age={Config.THUMB_TTL_SECONDS}", "ETag": f'W/"{hashlib.md5(file_path.encode("utf-8")).hexdigest()}"', } @@ -326,9 +309,9 @@ def rename(old_key): return jsonify({"success": False, "error": "New name not provided"}), 400 # 构建新的文件路径 - prefix = os.path.dirname(old_key) - if prefix: - new_key = f"{prefix}/{new_name}" + old_key_parts = old_key.rsplit("/", 1) + if len(old_key_parts) > 1: + new_key = f"{old_key_parts[0]}/{new_name}" else: new_key = new_name @@ -379,9 +362,9 @@ def rename_folder_route(old_prefix): old_prefix += "/" # 构建新的文件夹路径 - parent_prefix = os.path.dirname(os.path.dirname(old_prefix)) - if parent_prefix: - new_prefix = f"{parent_prefix}/{new_name}/" + prefix_parts = old_prefix.rstrip("/").rsplit("/", 1) + if len(prefix_parts) > 1: + new_prefix = f"{prefix_parts[0]}/{new_name}/" else: new_prefix = f"{new_name}/" diff --git a/pyproject.toml b/pyproject.toml index 3e2f1dd..9006c47 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "Cloud-Index" -version = "0.9.0" +version = "0.10.0" description = "A cloud storage index system based on a number of cloud storage platforms" authors = [{ name = "RhenCloud", email = "i@rhen.cloud" }] readme = "README.md" diff --git a/storages/factory.py b/storages/factory.py index eef33d5..0d8408e 100644 --- a/storages/factory.py +++ b/storages/factory.py @@ -1,14 +1,11 @@ -import os from typing import Optional -import dotenv +from config import Config from .base import BaseStorage from .github import GitHubStorage from .r2 import R2Storage -dotenv.load_dotenv() - class StorageFactory: """存储工厂类，根据配置创建对应的存储实例""" @@ -29,24 +26,19 @@ class StorageFactory: if cls._instance is not None: return cls._instance - storage_type = os.getenv("STORAGE_TYPE") + storage_type = Config.STORAGE_TYPE if not storage_type: - raise RuntimeError( - "STORAGE_TYPE environment variable is not set. " - "Supported types: r2, github" - ) - - storage_type = storage_type.lower() + raise RuntimeError("STORAGE_TYPE environment variable is not set. Supported types: r2, github") if storage_type == "r2": cls._instance = R2Storage() elif storage_type == "github": cls._instance = GitHubStorage() else: - raise RuntimeError( - f"Unsupported storage type: {storage_type}. Supported types: r2, github" - ) + raise RuntimeError(f"Unsupported storage type: {storage_type}. Supported types: r2, github") + + return cls._instance return cls._instance diff --git a/storages/github.py b/storages/github.py index fdc17d6..20de8cb 100644 --- a/storages/github.py +++ b/storages/github.py @@ -1,16 +1,14 @@ import base64 -import os from datetime import datetime from io import BytesIO from typing import Any, Dict -import dotenv import requests from PIL import Image -from .base import BaseStorage +from config import Config -dotenv.load_dotenv() +from .base import BaseStorage class StreamWrapper: @@ -50,29 +48,30 @@ class GitHubStorage(BaseStorage): """基于 GitHub 仓库的存储实现""" def __init__(self): - self.repo_owner = os.getenv("GITHUB_REPO_OWNER") - self.repo_name = os.getenv("GITHUB_REPO_NAME") - self.access_token = os.getenv("GITHUB_ACCESS_TOKEN") - self.branch = os.getenv("GITHUB_BRANCH", "main") - # 反向代理 URL，用于加速 GitHub raw 文件访问 - # 例如：https://raw. githubusercontent.com/ 的反向代理 URL - self.raw_proxy_url = os.getenv("GITHUB_RAW_PROXY_URL", "").rstrip("/") + """初始化 GitHub 存储客户端""" + self.token = Config.GITHUB_TOKEN + repo_full = Config.GITHUB_REPO # 格式: owner/repo + self.branch = Config.GITHUB_BRANCH - if not all([self.repo_owner, self.repo_name, self.access_token]): - raise RuntimeError("GITHUB_REPO_OWNER, GITHUB_REPO_NAME, GITHUB_ACCESS_TOKEN must be set") + if not self.token or not repo_full: + raise RuntimeError("GITHUB_TOKEN and GITHUB_REPO must be set") + + # 解析 owner/repo + repo_parts = repo_full.split("/") + if len(repo_parts) != 2: + raise RuntimeError(f"GITHUB_REPO must be in format 'owner/repo', got: {repo_full}") + + self.repo_owner = repo_parts[0] + self.repo_name = repo_parts[1] + self.repo = repo_full self.api_base_url = f"https://api.github.com/repos/{self.repo_owner}/{self.repo_name}" - - # 如果配置了代理 URL，则使用代理 URL；否则使用官方 raw.githubusercontent.com - if self.raw_proxy_url: - self.raw_content_url = f"{self.raw_proxy_url}/https://raw.githubusercontent.com/{self.repo_owner}/{self.repo_name}/{self.branch}" - else: - self.raw_content_url = f"https://raw.githubusercontent.com/{self.repo_owner}/{self.repo_name}/{self.branch}" + self.raw_content_url = f"https://raw.githubusercontent.com/{self.repo_owner}/{self.repo_name}/{self.branch}" def _headers(self) -> Dict[str, str]: """返回 API 请求的公共头部信息""" return { - "Authorization": f"token {self.access_token}", + "Authorization": f"token {self.token}", "Accept": "application/vnd.github.v3+json", } diff --git a/storages/r2.py b/storages/r2.py index ad9e298..89ad268 100644 --- a/storages/r2.py +++ b/storages/r2.py @@ -3,29 +3,34 @@ from io import BytesIO from typing import Any, Dict import boto3 -import dotenv -from botocore.config import Config +from botocore.config import Config as BotocoreConfig from PIL import Image +from config import Config + from .base import BaseStorage -dotenv.load_dotenv() - class R2Storage(BaseStorage): - def __init__(self): - self.endpoint = os.getenv("R2_ENDPOINT_URL") - if not self.endpoint: - raise RuntimeError("R2_ENDPOINT_URL environment variable is not set") + """Cloudflare R2 存储后端实现""" - self.access_key = os.getenv("ACCESS_KEY_ID") or os.getenv("ACCESS_KEY_ID") - self.secret_key = os.getenv("SECRET_ACCESS_KEY") or os.getenv("SECRET_ACCESS_KEY") + def __init__(self): + """初始化 R2 存储客户端""" + # 从统一配置中读取 + account_id = Config.R2_ACCOUNT_ID + if not account_id: + raise RuntimeError("R2_ACCOUNT_ID environment variable is not set") + + self.endpoint = f"https://{account_id}.r2.cloudflarestorage.com" + self.access_key = Config.R2_ACCESS_KEY_ID + self.secret_key = Config.R2_SECRET_ACCESS_KEY if not self.access_key or not self.secret_key: - raise RuntimeError("ACCESS_KEY_ID and SECRET_ACCESS_KEY must be set") + raise RuntimeError("R2_ACCESS_KEY_ID and R2_SECRET_ACCESS_KEY must be set") - self.region_name = os.getenv("R2_REGION", "auto") - self.bucket_name = os.getenv("R2_BUCKET_NAME") + self.region_name = "auto" + self.bucket_name = Config.R2_BUCKET_NAME + self.public_domain = Config.R2_PUBLIC_DOMAIN def get_s3_client(self): """ @@ -36,7 +41,7 @@ class R2Storage(BaseStorage): endpoint_url=self.endpoint, aws_access_key_id=self.access_key, aws_secret_access_key=self.secret_key, - config=Config(signature_version="s3v4"), + config=BotocoreConfig(signature_version="s3v4"), region_name=self.region_name, ) @@ -93,10 +98,9 @@ class R2Storage(BaseStorage): """ 生成对象的公共访问 URL """ - base_url = os.getenv("R2_PUBLIC_URL") - if not base_url: + if not self.public_domain: return None - return f"{base_url.rstrip('/')}/{key}" + return f"{self.public_domain.rstrip('/')}/{key}" def generate_thumbnail(self, file_path: str) -> bytes: """ diff --git a/utils.py b/utils.py new file mode 100644 index 0000000..0d579e9 --- /dev/null +++ b/utils.py @@ -0,0 +1,160 @@ +""" +工具函数模块 +集中管理常用的辅助函数 +""" + +from datetime import datetime +from typing import Optional + + +def format_timestamp(timestamp) -> str: + """ + 格式化时间戳为人类可读的格式 + + Args: + timestamp: 时间戳对象（datetime 或其他） + + Returns: + 格式化后的时间字符串 + """ + if isinstance(timestamp, datetime): + return timestamp.strftime("%Y-%m-%d %H:%M:%S") + return str(timestamp) + + +def format_file_size(size_bytes: Optional[int]) -> str: + """ + 格式化文件大小为人类可读的格式 + + Args: + size_bytes: 文件大小（字节） + + Returns: + 格式化后的大小字符串（如 "1.23MB"） + """ + try: + if size_bytes is None: + return "-" + num = float(size_bytes) + except (ValueError, TypeError): + return "-" + + for unit in ["B", "KB", "MB", "GB", "TB"]: + if num < 1024: + # 对于字节，显示整数 + if unit == "B": + return f"{int(num)}{unit}" + # 其他单位保留两位小数 + return f"{num:.2f}{unit}" + num = num / 1024.0 + return f"{num:.2f}PB" + + +def get_file_icon(filename: Optional[str]) -> str: + """ + 根据文件名返回对应的 Font Awesome 图标类名 + + Args: + filename: 文件名 + + Returns: + Font Awesome 图标类名 + """ + if not filename: + return "fas fa-file" + + ext = filename.lower().split(".")[-1] if "." in filename else "" + + # 定义文件类型映射 + icon_map = { + "fas fa-image": ["jpg", "jpeg", "png", "gif", "bmp", "webp", "svg"], + "fas fa-music": ["mp3", "wav", "ogg", "flac", "m4a", "aac"], + "fas fa-video": ["mp4", "webm", "avi", "mov", "wmv", "flv", "mkv"], + "fas fa-file-alt": ["pdf", "doc", "docx", "txt", "md", "rtf"], + "fas fa-file-archive": ["zip", "rar", "7z", "tar", "gz"], + "fas fa-file-code": ["py", "js", "html", "css", "java", "cpp", "c", "php"], + "fas fa-file-excel": ["xls", "xlsx", "csv"], + "fas fa-file-powerpoint": ["ppt", "pptx"], + } + + for icon, extensions in icon_map.items(): + if ext in extensions: + return icon + + return "fas fa-file" + + +def normalize_path(path: str, is_folder: bool = False) -> str: + """ + 规范化路径格式 + + Args: + path: 原始路径 + is_folder: 是否为文件夹 + + Returns: + 规范化后的路径（文件夹以 / 结尾） + """ + path = path.strip() + if is_folder and not path.endswith("/"): + return path + "/" + if not is_folder and path.endswith("/"): + return path.rstrip("/") + return path + + +def get_file_extension(filename: str) -> str: + """ + 获取文件扩展名 + + Args: + filename: 文件名 + + Returns: + 小写的文件扩展名（不含点） + """ + if not filename or "." not in filename: + return "" + return filename.lower().split(".")[-1] + + +def is_image_file(filename: str) -> bool: + """ + 判断文件是否为图片 + + Args: + filename: 文件名 + + Returns: + 如果是图片文件返回 True + """ + image_extensions = ["jpg", "jpeg", "png", "gif", "bmp", "webp", "svg", "ico"] + return get_file_extension(filename) in image_extensions + + +def is_video_file(filename: str) -> bool: + """ + 判断文件是否为视频 + + Args: + filename: 文件名 + + Returns: + 如果是视频文件返回 True + """ + video_extensions = ["mp4", "webm", "ogg", "mov", "avi", "mkv", "m4v"] + return get_file_extension(filename) in video_extensions + + +def is_audio_file(filename: str) -> bool: + """ + 判断文件是否为音频 + + Args: + filename: 文件名 + + Returns: + 如果是音频文件返回 True + """ + audio_extensions = ["mp3", "wav", "ogg", "m4a", "aac", "flac", "opus", "weba"] + return get_file_extension(filename) in audio_extensions