xsl
a1d0171fc5
CI / lint / typecheck / import-linter / tests / secrets (push) Canceled after 0s
- POST /uploads/direct：原始字节 body + Content-Type 申报 + scene query（默认 report_photo），场景/类型/大小三重校验复用 storage 既有规则，落盘 var/uploads/<scene>/<user_id>/<uuid>.<ext>（config 新增 upload_dir），key 布局与 COS 同构；不加 python-multipart 依赖 - storage/local.py：save_direct_upload + require_local_upload（resolve 归并 + is_relative_to 归属收敛 + 存在性校验，越界/模拟/不存在一律 404） - report/service.create_report：资格校验后、配额预扣前插入照片存在性校验（模拟 key 被拦且不消耗额度；regenerate 复用已验 photo_meta 不复核） - 测试：test_upload_direct 14 单测 + test_upload_direct_flow 4 集成（真 key 全链路/假 key 404 零流水/401/40010） 偏差说明：test_report_flow._create_report 改走 /uploads/direct 拿真 key（存在性校验必然封死假 key），40920 与幂等重放两条断言语义未动。
50 lines
2.1 KiB
Python
50 lines
2.1 KiB
Python
"""``POST /uploads/presign`` — COS 直传预签名（FR-SYS-03 / T-P1-27）。
|
||
|
||
图片不经应用服务器：本接口只做场景/类型/大小校验并签发 PUT 预签名 URL，
|
||
字节流由客户端直传 COS。真实联调需在 `.env` 配 COS 四件套
|
||
（COS_REGION / COS_SECRET_ID / COS_SECRET_KEY + 场景桶配置
|
||
UPLOAD_SCENES_JSON）；本地单测注入 FakeSigner，不依赖任何真实 COS。
|
||
|
||
``POST /uploads/direct`` — 本地磁盘直传（无 COS 四件套的降级路线，docs
|
||
提交 9228f67 决策）：body 即图片原始字节，Content-Type 头申报类型，
|
||
scene 走 query（默认 report_photo）。校验口径与 presign 完全一致，
|
||
落盘 ``settings.upload_dir``，返回的 object_key 供建报告等接口回传。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from fastapi import APIRouter, Query, Request
|
||
|
||
from apps.api.deps import CurrentUserId, SettingsDep, StorageServiceDep
|
||
from packages.shared.response import ok
|
||
from packages.storage.local import save_direct_upload
|
||
from packages.storage.schemas import PresignRequest
|
||
|
||
router = APIRouter(prefix="/uploads", tags=["uploads"])
|
||
|
||
|
||
@router.post("/presign", summary="签发图片直传 COS 的 PUT 预签名 URL")
|
||
async def presign_upload(
|
||
user_id: CurrentUserId,
|
||
storage: StorageServiceDep,
|
||
request: PresignRequest,
|
||
) -> dict[str, object]:
|
||
presigned = storage.build_presign(user_id=user_id, request=request)
|
||
return ok(presigned.model_dump())
|
||
|
||
|
||
@router.post("/direct", summary="本地磁盘直传图片（无 COS 降级路线）：body 为原始字节")
|
||
async def direct_upload(
|
||
user_id: CurrentUserId,
|
||
settings: SettingsDep,
|
||
request: Request,
|
||
scene: str = Query(default="report_photo", min_length=1, max_length=64),
|
||
) -> dict[str, object]:
|
||
data = await request.body()
|
||
# Content-Type 头可能带参数（如 "image/jpeg; boundary=…"），只取主类型。
|
||
content_type = request.headers.get("content-type", "").split(";")[0].strip().lower()
|
||
result = save_direct_upload(
|
||
settings, user_id=user_id, scene=scene, content_type=content_type, data=data
|
||
)
|
||
return ok(result.model_dump())
|