feat: 完整的 R2 文件管理系统初始化

- 实现 11 个 API 端点：上传、删除、重命名、复制、移动、创建文件夹等
- 构建存储抽象层，支持多后端（R2、S3、GitHub）
- 添加完整的前端界面：列表/网格视图、文件操作、批量管理
- 添加深色/浅色主题支持
- 迁移至 Jinja2 模板继承架构
- 添加完整的 API 文档和项目文档
- 优化 Vercel 部署配置

app.py

@@ -1,24 +1,24 @@
-import hashlib
 import os
-from datetime import datetime
-from io import BytesIO
-from pathlib import Path
-from typing import Any, Dict, List
 
-import boto3
-from botocore.config import Config
-from dotenv import load_dotenv
-from flask import Flask, Response, abort, render_template, request, send_file
-from PIL import Image
+import dotenv
+from flask import Flask
 
-# 加载环境变量
-load_dotenv()
+from handlers.routes import main_route
+from storages.factory import StorageFactory
 
-app = Flask(__name__, static_url_path="/static", static_folder="static")
+dotenv.load_dotenv()
 
+app = Flask(__name__)
+
+
+# 注册蓝图
+app.register_blueprint(main_route)
+
+# 初始化存储（使用工厂模式）
+storage = StorageFactory.get_storage()
 
 # 缩略图默认 TTL（秒），可通过环境变量覆盖
-THUMB_TTL = int(os.getenv("THUMB_TTL_SECONDS", "3600"))
+THUMB_TTL = int(os.environ.get("THUMB_TTL_SECONDS", "3600"))
 
 
 # 注册一个安全的 filesizeformat 过滤器，处理 None 和非数字值
@@ -86,258 +86,25 @@ def fileicon_filter(filename):
     return "fas fa-file"
 
 
-def get_s3_client():
-    """
-    创建并返回配置好的 S3 客户端，用于访问 R2 存储
-    """
-    endpoint = os.getenv("R2_ENDPOINT_URL")
-    if not endpoint:
-        # 更明确的错误，便于调试环境变量问题
-        raise RuntimeError("R2_ENDPOINT_URL environment variable is not set")
-
-    # 支持常见 AWS 环境变量名：AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY
-    access_key = os.getenv("ACCESS_KEY_ID") or os.getenv("ACCESS_KEY_ID")
-    secret_key = os.getenv("SECRET_ACCESS_KEY") or os.getenv("SECRET_ACCESS_KEY")
-
-    if not access_key or not secret_key:
-        raise RuntimeError("ACCESS_KEY_ID and SECRET_ACCESS_KEY must be set")
-
-    return boto3.client(
-        "s3",
-        endpoint_url=endpoint,
-        aws_access_key_id=access_key,
-        aws_secret_access_key=secret_key,
-        config=Config(signature_version="s3v4"),
-        region_name=os.getenv("R2_REGION", "auto"),
-    )
-
-
 def get_public_url(key: str) -> str:
     """
     生成对象的公共访问 URL
     """
-    base_url = os.getenv("R2_PUBLIC_URL")
-    if not base_url:
-        return None
-    return f"{base_url.rstrip('/')}/{key}"
+    return storage.get_public_url(key)
 
 
 def format_timestamp(timestamp) -> str:
     """
     格式化时间戳为人类可读的格式
     """
-    if isinstance(timestamp, datetime):
-        return timestamp.strftime("%Y-%m-%d %H:%M:%S")
-    return str(timestamp)
+    return storage.format_timestamp(timestamp)
 
 
 def generate_presigned_url(
     s3_client, bucket_name: str, key: str, expires: int = None
 ) -> str:
     """为指定对象生成 presigned URL（GET）。"""
-    if expires is None:
-        try:
-            expires = int(os.getenv("R2_PRESIGN_EXPIRES", "3600"))
-        except Exception:
-            expires = 3600
-
-    try:
-        url = s3_client.generate_presigned_url(
-            "get_object", Params={"Bucket": bucket_name, "Key": key}, ExpiresIn=expires
-        )
-        return url
-    except Exception:
-        app.logger.exception("Failed to generate presigned URL for %s", key)
-        return None
-
-
-@app.route("/")
-def index():
-    """
-    返回 R2 存储桶中的文件和目录列表的 HTML 页面。
-    """
-    try:
-        s3_client = get_s3_client()
-        bucket_name = os.getenv("R2_BUCKET_NAME")
-
-        # 支持 prefix 查询参数（用于浏览子目录）
-        prefix = request.args.get("prefix", "") or ""
-        if prefix and not prefix.endswith("/"):
-            prefix = prefix + "/"
-
-        list_kwargs = {"Bucket": bucket_name, "Delimiter": "/"}
-        if prefix:
-            list_kwargs["Prefix"] = prefix
-
-        # 列出指定前缀（根或子目录）下的对象
-        response = s3_client.list_objects_v2(**list_kwargs)
-
-        entries: List[Dict[str, Any]] = []
-        if "Contents" in response:
-            for obj in response["Contents"]:
-                # 跳过等于 prefix 的条目（有时存在）
-                key = obj.get("Key", "")
-                if prefix and key == prefix:
-                    continue
-                if key.endswith("/"):
-                    continue
-
-                # 显示相对名称（去掉当前 prefix）
-                rel_name = key[len(prefix) :] if prefix else key
-
-                entry = {
-                    "name": rel_name,
-                    "key": key,
-                    "size": obj.get("Size"),
-                    "last_modified": format_timestamp(obj.get("LastModified")),
-                    "is_dir": False,
-                }
-
-                # 添加公共访问 URL（如果配置了）
-                public_url = get_public_url(key)
-                if public_url:
-                    entry["public_url"] = public_url
-
-                # 添加 presigned URL（优先于后端代理预览）
-                presigned = generate_presigned_url(s3_client, bucket_name, key)
-                if presigned:
-                    entry["presigned_url"] = presigned
-
-                # 通过服务器访问的文件 URL（用于预览和缩略图）
-                entry["file_url"] = get_file_url(key)
-                entries.append(entry)
-
-        # 添加当前前缀下的文件夹（CommonPrefixes）
-        if "CommonPrefixes" in response:
-            for p in response["CommonPrefixes"]:
-                pref = p.get("Prefix")
-                # 相对文件夹名
-                rel = pref[len(prefix) :].rstrip("/") if prefix else pref.rstrip("/")
-                entries.append({"name": rel, "key": pref, "is_dir": True})
-
-        # 按照类型（目录在前）和名称排序
-        entries.sort(key=lambda x: (not x.get("is_dir", False), x["name"]))
-
-        # 构造面包屑导航
-        crumbs = []
-        if prefix:
-            segs = prefix.rstrip("/").split("/")
-            acc = ""
-            for seg in segs:
-                acc = acc + seg + "/"
-                crumbs.append({"name": seg, "prefix": acc})
-
-        return render_template(
-            "index.html",
-            entries=entries,
-            current_prefix=prefix,
-            crumbs=crumbs,
-            current_year=datetime.now().year,
-        )
-    except Exception as e:
-        app.logger.exception("Error listing R2 bucket")
-        return render_template(
-            "index.html", error=str(e), current_year=datetime.now().year
-        )
-
-
-@app.route("/<path:prefix_path>")
-def browse(prefix_path):
-    """漂亮的目录路由。将 URL /a/b 映射为 prefix 'a/b/' 并重用 index 的逻辑。"""
-    # delegate to index-like logic but with provided prefix
-    try:
-        s3_client = get_s3_client()
-        bucket_name = os.getenv("R2_BUCKET_NAME")
-
-        prefix = prefix_path or ""
-        if prefix and not prefix.endswith("/"):
-            prefix = prefix + "/"
-
-        list_kwargs = {"Bucket": bucket_name, "Delimiter": "/", "Prefix": prefix}
-        response = s3_client.list_objects_v2(**list_kwargs)
-
-        entries: List[Dict[str, Any]] = []
-        if "Contents" in response:
-            for obj in response["Contents"]:
-                key = obj.get("Key", "")
-                if prefix and key == prefix:
-                    continue
-                if key.endswith("/"):
-                    continue
-                rel_name = key[len(prefix) :] if prefix else key
-                entry = {
-                    "name": rel_name,
-                    "key": key,
-                    "size": obj.get("Size"),
-                    "last_modified": format_timestamp(obj.get("LastModified")),
-                    "is_dir": False,
-                    "file_url": get_file_url(key),
-                }
-                entries.append(entry)
-
-        if "CommonPrefixes" in response:
-            for p in response["CommonPrefixes"]:
-                pref = p.get("Prefix")
-                rel = pref[len(prefix) :].rstrip("/") if prefix else pref.rstrip("/")
-                entries.append({"name": rel, "key": pref, "is_dir": True})
-
-        entries.sort(key=lambda x: (not x.get("is_dir", False), x["name"]))
-
-        crumbs = []
-        if prefix:
-            segs = prefix.rstrip("/").split("/")
-            acc = ""
-            for seg in segs:
-                acc = acc + seg + "/"
-                crumbs.append({"name": seg, "prefix": acc})
-
-        return render_template(
-            "index.html",
-            entries=entries,
-            current_prefix=prefix,
-            crumbs=crumbs,
-            current_year=datetime.now().year,
-        )
-    except Exception as e:
-        app.logger.exception("Error browsing R2 bucket")
-        return render_template(
-            "index.html", error=str(e), current_year=datetime.now().year
-        )
-
-
-@app.route("/file/<path:file_path>")
-def serve_file(file_path):
-    """通过服务器提供文件访问"""
-    try:
-        s3_client = get_s3_client()
-        bucket_name = os.getenv("R2_BUCKET_NAME")
-
-        # 获取文件的基本信息
-        try:
-            response = s3_client.head_object(Bucket=bucket_name, Key=file_path)
-        except s3_client.exceptions.ClientError as e:
-            if e.response["Error"]["Code"] == "404":
-                abort(404)
-            else:
-                abort(500)
-
-        # 获取文件对象
-        file_obj = s3_client.get_object(Bucket=bucket_name, Key=file_path)
-
-        # 设置响应头
-        headers = {
-            "Content-Type": file_obj["ContentType"],
-            "Content-Length": str(file_obj["ContentLength"]),
-        }
-
-        # 使用 Response 流式传输文件内容
-        return Response(
-            file_obj["Body"].iter_chunks(), headers=headers, direct_passthrough=True
-        )
-
-    except Exception as e:
-        app.logger.exception("Error serving file")
-        abort(500)
+    return storage.generate_presigned_url(key, expires)
 
 
 def get_file_url(key: str) -> str:
@@ -345,75 +112,6 @@ def get_file_url(key: str) -> str:
     return f"/file/{key}"
 
 
-@app.route("/thumb/<path:file_path>")
-def thumb(file_path):
-    """返回图片的缩略图，使用 Vercel Cache Headers 避免重复从 R2 拉取"""
-    bucket_name = os.getenv("R2_BUCKET_NAME")
-
-    # 设置更长的缓存控制头以支持浏览器本地缓存
-    cache_headers = {
-        "Cache-Control": f"public, max-age={THUMB_TTL}",
-        "ETag": f'W/"{hashlib.md5(file_path.encode("utf-8")).hexdigest()}"',
-    }
-
-    # 先检查客户端是否已经有缓存版本
-    etag = request.headers.get("If-None-Match")
-    if etag and etag == cache_headers["ETag"]:
-        return Response(status=304, headers=cache_headers)
-
-    # 从 R2 获取原始对象并生成缩略图
-    try:
-        s3 = get_s3_client()
-        try:
-            obj = s3.get_object(Bucket=bucket_name, Key=file_path)
-            data = obj["Body"].read()
-        except Exception:
-            app.logger.exception("Failed to fetch object for thumb: %s", file_path)
-            response = send_file(
-                os.path.join(app.static_folder, "thumb_placeholder.svg"),
-                mimetype="image/svg+xml",
-            )
-            response.headers.update(cache_headers)
-            return response
-
-        try:
-            img = Image.open(BytesIO(data))
-            img = img.convert("RGB")
-            img.thumbnail((320, 320))
-            buf = BytesIO()
-            img.save(buf, "JPEG", quality=80, optimize=True)
-            buf.seek(0)
-            thumb_bytes = buf.getvalue()
-
-            response = Response(thumb_bytes, mimetype="image/jpeg")
-            response.headers.update(cache_headers)
-            return response
-        except Exception:
-            app.logger.exception("Failed to generate thumbnail for %s", file_path)
-            response = send_file(
-                os.path.join(app.static_folder, "thumb_placeholder.svg"),
-                mimetype="image/svg+xml",
-            )
-            response.headers.update(cache_headers)
-            return response
-    except Exception:
-        app.logger.exception("Unexpected error in thumb endpoint")
-        response = send_file(
-            os.path.join(app.static_folder, "thumb_placeholder.svg"),
-            mimetype="image/svg+xml",
-        )
-        response.headers.update(cache_headers)
-        return response
-
-
-# 添加路由以提供Service Worker文件
-@app.route("/static/sw.js")
-def sw():
-    return send_file(
-        os.path.join(app.static_folder, "sw.js"), mimetype="application/javascript"
-    )
-
-
 if __name__ == "__main__":
     port = int(os.environ.get("PORT", 5000))
     host = os.environ.get("HOST", "0.0.0.0")