RhenCloud/Cloud-Index
1
0
Fork 0
mirror of https://github.com/RhenCloud/Cloud-Index.git synced 2025-12-06 07:06:41 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 重构配置管理，集中管理环境变量并优化存储后端实现

Browse Source
This commit is contained in:
RhenCloud
2025-11-14 23:26:21 +08:00
parent 07a8fafff2
commit 33fe06c59e
10 changed files with 347 additions and 193 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
.env.example
View File

@@ -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://<your-account-id>.r2.cloudflarestorage.com
# R2 区域 (默认: auto)
R2_REGION=auto
# R2 公共访问 URL (可选，例如: https://pub-<bucket-name>.r2.dev)
R2_PUBLIC_URL=https://pub-<bucket-name>.r2.dev
# R2 预签名 URL 过期时间（秒，默认: 3600）
R2_PRESIGN_EXPIRES=3600
# R2 公共访问域名 (可选，例如: https://pub-<bucket-name>.r2.dev)
R2_PUBLIC_DOMAIN=https://pub-<bucket-name>.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

42
README.md
View File

@@ -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 依赖
```

76
app.py
View File

@@ -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)

83
config.py Normal file
View File

@@ -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 {}

37
handlers/routes.py
View File

@@ -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}/"

2
pyproject.toml
View File

@@ -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"

20
storages/factory.py
View File

@@ -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

39
storages/github.py
View File

@@ -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",
}

38
storages/r2.py
View File

@@ -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:
"""

160
utils.py Normal file
View File

@@ -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