""" 文件上传路由。 接口（统一响应格式 {success, data, message}）： - POST /api/upload/avatar 头像上传（登录用户；jpg/png/webp，最大 2MB；成功后自动保存到头像字段） - POST /api/upload/article 文章图片上传（仅博主；jpg/png/webp，最大 6MB） - POST /api/upload/project 项目文件上传（仅博主；zip，最大 50MB） - POST /api/upload/doc 文章文档上传（仅博主；doc/docx，默认最大 20MB；仅用于生活/学习分区） 安全规则： - 随机文件名（uuid），不使用用户原始文件名，杜绝路径穿越 - 校验扩展名 + 文件魔数（内容真实性），防止伪装类型 - 文件仅作为静态资源由 Nginx 通过 /uploads/ 访问，禁止执行 """ import os import uuid from datetime import timedelta from pathlib import Path from fastapi import APIRouter, Depends, Form, HTTPException, UploadFile, status from sqlalchemy.orm import Session from ..auth import get_current_user from ..database import PROJECT_ROOT, get_db from ..models import ( ARTICLE_CATEGORY_LIFE, ARTICLE_CATEGORY_STUDY, ROLE_BLOGGER, ROLE_FRIEND, User, utcnow, ) from ..schemas import UnifiedResponse router = APIRouter(prefix="/api/upload", tags=["upload"]) # 上传根目录（可通过 .env 的 UPLOADS_DIR 覆盖；部署时 Nginx 将 /uploads/ 静态映射到该目录） UPLOADS_ROOT = Path(os.getenv("UPLOADS_DIR") or str(PROJECT_ROOT / "uploads")) # 允许的扩展名（值仅用于展示） ALLOWED_IMAGES = {"jpg", "jpeg", "png", "webp"} ALLOWED_VIDEO = {"mp4", "webm"} ALLOWED_PROJECT = {"zip"} ALLOWED_DOC = {"doc", "docx"} # 大小上限（字节）：头像上限可通过 .env 的 AVATAR_MAX_SIZE_MB 调整（默认 4MB） MAX_AVATAR_SIZE = int(os.getenv("AVATAR_MAX_SIZE_MB", "4")) * 1024 * 1024 # 文章图片（封面/正文插图）大小上限：可通过 .env 的 ARTICLE_IMAGE_MAX_SIZE_MB 调整（默认 6MB） MAX_IMAGE_SIZE = int(os.getenv("ARTICLE_IMAGE_MAX_SIZE_MB", "6")) * 1024 * 1024 MAX_VIDEO_SIZE = 100 * 1024 * 1024 # 文章视频 100MB MAX_PROJECT_SIZE = 50 * 1024 * 1024 # 项目文件 50MB # 文章文档（doc/docx）大小上限：可通过 .env 的 DOC_MAX_SIZE_MB 调整（默认 20MB） MAX_DOC_SIZE = int(os.getenv("DOC_MAX_SIZE_MB", "20")) * 1024 * 1024 def check_magic_bytes(content: bytes, ext: str) -> bool: """校验文件魔数，防止伪造扩展名。""" if ext in ("jpg", "jpeg"): return content[:3] == b"\xff\xd8\xff" if ext == "png": return content[:8] == b"\x89PNG\r\n\x1a\n" if ext == "webp": return content[:4] == b"RIFF" and content[8:12] == b"WEBP" if ext == "mp4": return content[4:8] == b"ftyp" if ext == "webm": return content[:4] == b"\x1aE\xdf\xa3" if ext == "zip": return content[:4] in (b"PK\x03\x04", b"PK\x05\x06", b"PK\x07\x08") if ext == "doc": # Word 97-2003：OLE 复合文档魔数 return content[:8] == b"\xd0\xcf\x11\xe0\xa1\xb1\x1a\xe1" if ext == "docx": # Word 2007+：本质是 zip 压缩包 return content[:4] in (b"PK\x03\x04", b"PK\x05\x06", b"PK\x07\x08") return False def save_upload(file: UploadFile, subdir: str, allowed: set, max_size: int) -> dict: """校验并保存上传文件，返回 URL 等元信息。""" original_name = file.filename or "" ext = original_name.rsplit(".", 1)[-1].lower() if "." in original_name else "" if ext not in allowed: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"不支持的文件类型（允许：{' / '.join(sorted(allowed))}）", ) # 一次读入，超出上限直接拒绝（+1 用于判断是否超限） content = file.file.read(max_size + 1) if len(content) > max_size: raise HTTPException(status_code=status.HTTP_413_CONTENT_TOO_LARGE, detail="文件大小超出限制") if not check_magic_bytes(content, ext): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="文件内容与扩展名不符") # 随机文件名，避免覆盖与路径穿越 saved_name = f"{uuid.uuid4().hex}.{ext}" target_dir = UPLOADS_ROOT / subdir target_dir.mkdir(parents=True, exist_ok=True) (target_dir / saved_name).write_bytes(content) return { "url": f"/uploads/{subdir}/{saved_name}", "filename": original_name, "size": len(content), } @router.post("/avatar", response_model=UnifiedResponse) def upload_avatar( file: UploadFile, current_user: User = Depends(get_current_user), db: Session = Depends(get_db), ) -> UnifiedResponse: """头像上传：仅好友或博主可上传（访客除注册外禁止一切上传）；成功后自动更新头像字段。 更换频率限制：同一账号两次更换之间至少间隔 AVATAR_CHANGE_INTERVAL_HOURS 小时（默认 24）。 """ if current_user.role not in (ROLE_FRIEND, ROLE_BLOGGER): raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="访客不能上传头像") # 更换冷却：距离上次更换不足 N 小时则拒绝，并提示剩余等待时间 interval_hours = int(os.getenv("AVATAR_CHANGE_INTERVAL_HOURS", "24")) if current_user.avatar_updated_time is not None: elapsed = utcnow() - current_user.avatar_updated_time if elapsed < timedelta(hours=interval_hours): remaining = timedelta(hours=interval_hours) - elapsed hours = int(remaining.total_seconds() // 3600) minutes = int((remaining.total_seconds() % 3600) // 60) raise HTTPException( status_code=status.HTTP_429_TOO_MANY_REQUESTS, detail=f"头像更换过于频繁，请约 {hours} 小时 {minutes} 分钟后再试", ) data = save_upload(file, "avatar", ALLOWED_IMAGES, MAX_AVATAR_SIZE) current_user.avatar = data["url"] current_user.avatar_updated_time = utcnow() db.commit() return UnifiedResponse(success=True, data=data, message="头像上传成功") @router.post("/article", response_model=UnifiedResponse) def upload_article_image( file: UploadFile, current_user: User = Depends(get_current_user), ) -> UnifiedResponse: """文章图片上传：仅博主可上传（用于封面与正文插图）。""" if current_user.role != ROLE_BLOGGER: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有博主可以上传文章图片") data = save_upload(file, "article", ALLOWED_IMAGES, MAX_IMAGE_SIZE) return UnifiedResponse(success=True, data=data, message="图片上传成功") @router.post("/video", response_model=UnifiedResponse) def upload_article_video( file: UploadFile, current_user: User = Depends(get_current_user), ) -> UnifiedResponse: """文章视频上传：仅博主可上传（mp4 / webm，最大 100MB），供正文插入视频。""" if current_user.role != ROLE_BLOGGER: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有博主可以上传文章视频") data = save_upload(file, "article", ALLOWED_VIDEO, MAX_VIDEO_SIZE) return UnifiedResponse(success=True, data=data, message="视频上传成功") @router.post("/project", response_model=UnifiedResponse) def upload_project_file( file: UploadFile, current_user: User = Depends(get_current_user), ) -> UnifiedResponse: """项目文件上传：仅博主可上传，支持 zip 压缩包（供下载项目使用）。""" if current_user.role != ROLE_BLOGGER: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有博主可以上传项目文件") data = save_upload(file, "project", ALLOWED_PROJECT, MAX_PROJECT_SIZE) return UnifiedResponse(success=True, data=data, message="项目文件上传成功") @router.post("/doc", response_model=UnifiedResponse) def upload_article_doc( file: UploadFile, category: str = Form(...), current_user: User = Depends(get_current_user), ) -> UnifiedResponse: """文章文档上传（doc / docx）：仅博主可上传，且只能用于“我的生活 / 我的学习”分区的文章。""" if current_user.role != ROLE_BLOGGER: raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="只有博主可以上传文档") if category not in (ARTICLE_CATEGORY_LIFE, ARTICLE_CATEGORY_STUDY): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="文档只能用于生活或学习分区的文章", ) data = save_upload(file, "article", ALLOWED_DOC, MAX_DOC_SIZE) return UnifiedResponse(success=True, data=data, message="文档上传成功")