feat(storage): COS presigned PUT URL endpoint for direct upload

POST /uploads/presign：场景/类型/大小三重校验 + 可审计 object_key 布局 +
SDK 隔离在 CosSigner Protocol 后（FR-SYS-03，T-P1-27）。

实现：packages/storage/schemas.py（PresignRequest/PresignResponse）、
packages/storage/service.py（CosSigner Protocol / QcloudCosSigner /
StorageService.build_presign，bucket 四别名映射到 cos_bucket_*，
object_key 写死 {scene}/{user_id}/{yyyyMMdd}/{uuid4hex}.{ext}）、
apps/api/routers/uploads.py（POST /presign，CurrentUserId 认证 + ok 信封）。

测试：tests/unit/test_upload_presign.py 16 例全部经 FakeSigner 注入，
零网络、零 SDK client 构造；覆盖 key 布局逐段断言、三重校验统一 40010、
bucket 别名映射、白名单字符串解析（空串/空格容错）、配置错误分支、
schema 层 size>=1。门禁全绿：ruff check/format、mypy（宽松 48 文件 +
domain strict 7 文件）、lint-imports 4 契约、pytest 185 passed
（基线 169 + 新增 16）。

偏差说明：
1) SDK 实际无 qcloud_cos.method 模块，Method 按官方文档传字符串 "PUT"；
   get_presigned_url 参数名为 Expired（非任务文档写的 Expire）；配置类
   导出名为 CosConfig（非 CosS3Config）。按文档 §3 "以实际 SDK 签名为准"
   处理，已用假密钥端到端冒烟验证（纯本地签名计算，零网络）。
2) build_presign 增加可选 keyword-only 参数 now（默认 None=当前 UTC），
   供单测注入固定时钟（对齐 DailyBudgetGuard 的 clock 注入惯例），
   不改变文档签名的调用兼容性。
3) 两处防御分支文档未规定，均配了单测：场景 bucket 别名非法报
   CONFIG_ERROR 而非 40010（配置错误不应让客户端背 4xx）；COS 密钥为空
   在 SDK 构造前报 CONFIG_ERROR，避免未处理的裸 500。

apps/api/routers/uploads.py

@@ -1,10 +1,27 @@
-"""``POST /uploads/presign`` — COS 直传预签名（骨架，T5 填充实现）。
+"""``POST /uploads/presign`` — COS 直传预签名（FR-SYS-03 / T-P1-27）。
 
-FR-SYS-03：图片不经应用服务器；实现契约见 docs/开发任务包/T5-上传预签名.md。
+图片不经应用服务器：本接口只做场景/类型/大小校验并签发 PUT 预签名 URL，
+字节流由客户端直传 COS。真实联调需在 `.env` 配 COS 四件套
+（COS_REGION / COS_SECRET_ID / COS_SECRET_KEY + 场景桶配置
+UPLOAD_SCENES_JSON）；本地单测注入 FakeSigner，不依赖任何真实 COS。
 """
 
 from __future__ import annotations
 
 from fastapi import APIRouter
 
+from apps.api.deps import CurrentUserId, StorageServiceDep
+from packages.shared.response import ok
+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())

packages/storage/schemas.py

@@ -0,0 +1,31 @@
+"""COS 直传预签名的请求/响应契约（FR-SYS-03 / T-P1-27）。
+
+放在 packages/storage 而非 apps/api/schemas：object_key 布局与场景语义属于
+存储域，报告、形象等业务域复用直传时不应该反向依赖 API 层。
+"""
+
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class PresignRequest(BaseModel):
+    """``POST /uploads/presign`` 请求体：客户端上传前的申报信息。
+
+    size_bytes 只是客户端自报值，仅用于服务端提前拒绝明显超限的请求；
+    真正的强制约束靠预签名 URL 签入的 Content-Type 与 COS 桶侧限制兜底。
+    """
+
+    scene: str = Field(min_length=1, max_length=64, description="上传场景，如 report_photo/avatar")
+    content_type: str = Field(min_length=1, max_length=128, description="MIME 类型，如 image/jpeg")
+    size_bytes: int = Field(ge=1, description="文件字节数")
+
+
+class PresignResponse(BaseModel):
+    """签发结果：客户端用 PUT 直传 upload_url，body 与 Content-Type 必须与申报一致。"""
+
+    upload_url: str = Field(description="带签名的 COS PUT URL，过期前有效")
+    object_key: str = Field(description="COS 对象键（可审计布局：scene/user/日期/uuid）")
+    expires_in: int = Field(description="URL 有效期（秒），与签发时配置一致")
+    max_bytes: int = Field(description="服务端允许的最大字节数，客户端用于前置校验")
+    content_type: str = Field(description="回显申报的 Content-Type，PUT 时必须携带同值")

packages/storage/service.py

@@ -1,11 +1,171 @@
-"""storage 域 service — 骨架（T5 任务填充实现）。"""
+"""storage 域 service — COS 直传 PUT 预签名（FR-SYS-03 / T-P1-27）。
+
+只做「签发」，不接收图片字节（图片经客户端直传 COS，应用服务器不中转）；
+上传后的内容校验属于报告链路（T-P3），不在本模块职责内。
+
+COS SDK（qcloud_cos）被隔离在 :class:`CosSigner` Protocol 后面：业务逻辑与
+单测都只面向 Protocol，:class:`QcloudCosSigner` 是全仓库唯一接触 SDK 的地方，
+测试注入 FakeSigner 即可零网络验证全部校验与布局规则。
+"""
 
 from __future__ import annotations
 
-from typing import Any
+from datetime import UTC, datetime
+from typing import Protocol
+from uuid import uuid4
+
+from qcloud_cos import CosConfig, CosS3Client
+
+from packages.shared.config import Settings
+from packages.shared.errors import AppError, ErrorCode
+from packages.shared.logging import get_logger
+from packages.storage.schemas import PresignRequest, PresignResponse
+
+logger = get_logger(__name__)
+
+# MIME → object_key 扩展名。新增可上传类型时，需同步 upload_allowed_types
+# 默认值与错误码 UPLOAD_REJECTED 的 hint 文案。
+_CONTENT_TYPE_EXT: dict[str, str] = {
+    "image/jpeg": "jpg",
+    "image/png": "png",
+    "image/webp": "webp",
+}
+
+# 场景配置里的 bucket 别名 → Settings 桶字段名。临时桶 2 小时生命周期（CR-02），
+# 正式素材由异步链路搬运确认后才落长期桶，所以 V1 两个场景默认都指向 temp。
+_BUCKET_FIELDS: dict[str, str] = {
+    "temp": "cos_bucket_temp",
+    "user": "cos_bucket_user",
+    "kb": "cos_bucket_kb",
+    "report": "cos_bucket_report",
+}
+
+
+class CosSigner(Protocol):
+    """预签名能力的最小抽象：业务只依赖这个接口，不依赖具体云厂商。"""
+
+    def presign_put(self, bucket: str, key: str, expires: int, content_type: str) -> str:
+        """返回带签名的 PUT URL；content_type 签入签名以锁定上传类型。"""
+        ...
+
+
+class QcloudCosSigner:
+    """:class:`CosSigner` 的腾讯云 COS 实现。
+
+    SDK 未提供独立的 Method 枚举模块，按其文档以字符串 ``"PUT"`` 传参；
+    ``Expired``/``Headers`` 是 ``get_presigned_url`` 的真实参数名。
+    """
+
+    def __init__(self, *, region: str, secret_id: str, secret_key: str) -> None:
+        # SDK 在密钥为空时抛的是裸 CosClientError（会以未处理异常变成 500 Internal），
+        # 提前映射为 CONFIG_ERROR，让本地未配 COS 四件套时得到结构化的配置错误。
+        if not secret_id or not secret_key:
+            raise AppError(ErrorCode.CONFIG_ERROR, message="COS 密钥未配置（COS_SECRET_ID/KEY）")
+        self._client = CosS3Client(
+            CosConfig(Region=region, SecretId=secret_id, SecretKey=secret_key)
+        )
+
+    def presign_put(self, bucket: str, key: str, expires: int, content_type: str) -> str:
+        # 签名是本地计算（不发网络请求）；Content-Type 签入后，
+        # 客户端用其它类型 PUT 该 URL 会被 COS 直接拒绝。
+        # SDK 无类型标注，str() 只为向 mypy 收窄返回类型。
+        return str(
+            self._client.get_presigned_url(
+                Method="PUT",
+                Bucket=bucket,
+                Key=key,
+                Expired=expires,
+                Headers={"Content-Type": content_type},
+            )
+        )
+
+
+def _parse_allowed_types(raw: str) -> frozenset[str]:
+    """解析逗号分隔的 MIME 白名单，容忍空串与多余空格（配置是人工维护的）。"""
+    return frozenset(part.strip() for part in raw.split(",") if part.strip())
 
 
 class StorageService:
-    def __init__(self, settings: Any, signer: Any | None = None) -> None:
+    """签发 COS 直传预签名 URL，并守住场景/类型/大小三重服务端校验。"""
+
+    def __init__(self, settings: Settings, signer: CosSigner | None = None) -> None:
         self._settings = settings
         self._signer = signer
+
+    @property
+    def _active_signer(self) -> CosSigner:
+        # 惰性构造：只有真实签发请求才创建 COS client，进程启动/测试路径零开销；
+        # 测试必须注入 FakeSigner，避免任何 SDK 构造。
+        if self._signer is None:
+            self._signer = QcloudCosSigner(
+                region=self._settings.cos_region,
+                secret_id=self._settings.cos_secret_id,
+                secret_key=self._settings.cos_secret_key,
+            )
+        return self._signer
+
+    def build_presign(
+        self,
+        user_id: int,
+        request: PresignRequest,
+        *,
+        now: datetime | None = None,
+    ) -> PresignResponse:
+        """三重校验通过后签发 PUT 预签名 URL。
+
+        ``now`` 仅用于测试注入固定时钟；生产调用不传，按当前 UTC 时间分桶。
+        """
+        scene_config = self._settings.upload_scenes.get(request.scene)
+        if scene_config is None:
+            raise AppError(ErrorCode.UPLOAD_REJECTED, message="不支持的上传场景")
+
+        # 类型校验：白名单之外一律拒绝（与大小、场景共用错误码 40010）。
+        if request.content_type not in _parse_allowed_types(self._settings.upload_allowed_types):
+            raise AppError(ErrorCode.UPLOAD_REJECTED)
+        # 扩展名映射兜底：白名单被运维改成映射表之外的类型时按客户端错误拒绝，
+        # 而不是在拼 object_key 时抛 KeyError 变成 5xx。
+        ext = _CONTENT_TYPE_EXT.get(request.content_type)
+        if ext is None:
+            raise AppError(ErrorCode.UPLOAD_REJECTED)
+
+        # 大小校验：客户端自报值超限直接拒绝，省一次无谓的签名计算。
+        if request.size_bytes > self._settings.upload_max_bytes:
+            raise AppError(ErrorCode.UPLOAD_REJECTED)
+
+        # bucket 别名 → 四桶配置。别名写错属于部署配置问题，报 CONFIG_ERROR
+        # 而不是让客户端背 4xx。
+        bucket_alias = scene_config.get("bucket")
+        bucket_field = _BUCKET_FIELDS.get(bucket_alias) if isinstance(bucket_alias, str) else None
+        if bucket_field is None:
+            raise AppError(
+                ErrorCode.CONFIG_ERROR,
+                message=f"上传场景 {request.scene} 的 bucket 配置无效",
+            )
+        bucket: str = getattr(self._settings, bucket_field)
+
+        # object_key 布局写死为 {scene}/{user_id}/{yyyyMMdd}/{uuid4hex}.{ext}：
+        # 按场景分目录、按用户隔离、按日分组、uuid 防猜——这是 COS 侧做
+        # 前缀审计与生命周期规则（如临时桶 2h 清理）的基础，改动需过安全评审。
+        # 日期取 UTC，保证多实例写入的分组一致。
+        day = (now or datetime.now(UTC)).strftime("%Y%m%d")
+        object_key = f"{request.scene}/{user_id}/{day}/{uuid4().hex}.{ext}"
+
+        expires = self._settings.cos_presign_expire_seconds
+        url = self._active_signer.presign_put(
+            bucket=bucket, key=object_key, expires=expires, content_type=request.content_type
+        )
+        logger.info(
+            "presign_issued",
+            scene=request.scene,
+            user_id=user_id,
+            bucket=bucket,
+            object_key=object_key,
+            size_bytes=request.size_bytes,
+        )
+        return PresignResponse(
+            upload_url=url,
+            object_key=object_key,
+            expires_in=expires,
+            max_bytes=self._settings.upload_max_bytes,
+            content_type=request.content_type,
+        )

tests/unit/test_upload_presign.py

@@ -0,0 +1,228 @@
+"""COS 直传预签名（FR-SYS-03 / T-P1-27）。
+
+全部用 FakeSigner 注入，零网络、零 SDK client 构造。object_key 布局
+（场景/用户/日期/uuid）是 COS 侧前缀审计与生命周期规则的基础，逐段精确
+断言；场景/类型/大小三重校验共用 40010，逐条锁死。
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from typing import Any
+
+import pytest
+from pydantic import ValidationError
+
+from packages.shared.config import Settings
+from packages.shared.errors import AppError, ErrorCode
+from packages.storage.schemas import PresignRequest
+from packages.storage.service import StorageService
+
+FIXED_NOW = datetime(2026, 8, 14, 1, 2, 3, tzinfo=UTC)
+
+
+@dataclass(frozen=True, slots=True)
+class FakeSigner:
+    """捕获签发参数并回放可预测 URL，替代 QcloudCosSigner（零网络）。"""
+
+    calls: list[dict[str, Any]] = field(default_factory=list)
+
+    def presign_put(self, bucket: str, key: str, expires: int, content_type: str) -> str:
+        self.calls.append(
+            {"bucket": bucket, "key": key, "expires": expires, "content_type": content_type}
+        )
+        return f"https://cos.fake/{bucket}/{key}?sign=stub"
+
+
+def make_settings(**overrides: Any) -> Settings:
+    """显式 kwargs 构造，测试不受 .env 与宿主环境变量影响。
+
+    upload_scenes 定义了 alias 且未开启 populate_by_name，必须按
+    ``UPLOAD_SCENES_JSON`` 传入——按字段名传会被静默忽略。
+    """
+    options: dict[str, Any] = {
+        "_env_file": None,
+        "cos_bucket_temp": "meida-temp-test",
+        "cos_bucket_user": "meida-user-test",
+        "cos_bucket_kb": "meida-kb-test",
+        "cos_bucket_report": "meida-report-test",
+        "cos_presign_expire_seconds": 321,
+        "upload_max_bytes": 1024,
+        "upload_allowed_types": "image/jpeg,image/png,image/webp",
+        "UPLOAD_SCENES_JSON": {"report_photo": {"bucket": "temp"}, "avatar": {"bucket": "temp"}},
+    }
+    options.update(overrides)
+    return Settings(**options)
+
+
+def make_service(**overrides: Any) -> tuple[StorageService, FakeSigner]:
+    signer = FakeSigner()
+    return StorageService(make_settings(**overrides), signer), signer
+
+
+def test_valid_request_builds_auditable_key_layout() -> None:
+    service, signer = make_service()
+    response = service.build_presign(
+        7,
+        PresignRequest(scene="report_photo", content_type="image/jpeg", size_bytes=100),
+        now=FIXED_NOW,
+    )
+
+    # key 布局：{scene}/{user_id}/{yyyyMMdd}/{uuid4hex}.{ext}（写死，改动需过安全评审）
+    prefix = "report_photo/7/20260814/"
+    assert response.object_key.startswith(prefix)
+    filename = response.object_key.removeprefix(prefix)
+    stem, ext = filename.split(".")
+    assert ext == "jpg"
+    assert len(stem) == 32
+    assert all(char in "0123456789abcdef" for char in stem)
+
+    # 回显与签发参数
+    assert (
+        response.upload_url == f"https://cos.fake/meida-temp-test/{response.object_key}?sign=stub"
+    )
+    assert response.expires_in == 321
+    assert response.max_bytes == 1024
+    assert response.content_type == "image/jpeg"
+    assert signer.calls == [
+        {
+            "bucket": "meida-temp-test",
+            "key": response.object_key,
+            "expires": 321,
+            "content_type": "image/jpeg",
+        }
+    ]
+
+
+def test_two_presigns_for_the_same_request_never_share_a_key() -> None:
+    service, _ = make_service()
+    request = PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1)
+    first = service.build_presign(1, request, now=FIXED_NOW)
+    second = service.build_presign(1, request, now=FIXED_NOW)
+    assert first.object_key != second.object_key  # uuid 防猜
+
+
+@pytest.mark.parametrize(
+    ("body", "message"),
+    [
+        (
+            PresignRequest(scene="evil", content_type="image/jpeg", size_bytes=1),
+            "不支持的上传场景",
+        ),
+        (PresignRequest(scene="report_photo", content_type="application/pdf", size_bytes=1), None),
+        (PresignRequest(scene="report_photo", content_type="image/jpeg", size_bytes=1025), None),
+    ],
+)
+def test_scene_type_size_rejections_share_error_code_40010(
+    body: PresignRequest, message: str | None
+) -> None:
+    service, signer = make_service()
+    with pytest.raises(AppError) as excinfo:
+        service.build_presign(1, body, now=FIXED_NOW)
+    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
+    if message is not None:
+        assert excinfo.value.message == message
+    assert signer.calls == []  # 拒绝路径绝不触发签名
+
+
+def test_size_equal_to_limit_is_allowed() -> None:
+    service, signer = make_service()
+    service.build_presign(
+        1,
+        PresignRequest(scene="avatar", content_type="image/webp", size_bytes=1024),
+        now=FIXED_NOW,
+    )
+    assert signer.calls[0]["key"].endswith(".webp")
+
+
+@pytest.mark.parametrize(
+    ("alias", "expected_bucket"),
+    [
+        ("temp", "meida-temp-test"),
+        ("user", "meida-user-test"),
+        ("kb", "meida-kb-test"),
+        ("report", "meida-report-test"),
+    ],
+)
+def test_bucket_alias_maps_to_the_matching_settings_field(alias: str, expected_bucket: str) -> None:
+    service, signer = make_service(UPLOAD_SCENES_JSON={"report_photo": {"bucket": alias}})
+    service.build_presign(
+        7,
+        PresignRequest(scene="report_photo", content_type="image/png", size_bytes=1),
+        now=FIXED_NOW,
+    )
+    assert signer.calls[0]["bucket"] == expected_bucket
+
+
+def test_allowed_types_string_tolerates_whitespace() -> None:
+    service, _ = make_service(upload_allowed_types=" image/jpeg , image/png ")
+    response = service.build_presign(
+        1,
+        PresignRequest(scene="avatar", content_type="image/png", size_bytes=1),
+        now=FIXED_NOW,
+    )
+    assert response.object_key.endswith(".png")
+    with pytest.raises(AppError) as excinfo:
+        service.build_presign(
+            1,
+            PresignRequest(scene="avatar", content_type="image/webp", size_bytes=1),
+            now=FIXED_NOW,
+        )
+    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
+
+
+def test_empty_allowed_types_rejects_everything() -> None:
+    service, signer = make_service(upload_allowed_types="")
+    with pytest.raises(AppError) as excinfo:
+        service.build_presign(
+            1,
+            PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1),
+            now=FIXED_NOW,
+        )
+    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
+    assert signer.calls == []
+
+
+def test_whitelisted_type_without_extension_mapping_is_rejected() -> None:
+    # 白名单被扩到扩展名映射表之外的类型（此处模拟 image/gif）时，
+    # 宁可按 40010 拒绝，也不能在拼 object_key 时抛 KeyError 变成 5xx。
+    service, signer = make_service(upload_allowed_types="image/gif")
+    with pytest.raises(AppError) as excinfo:
+        service.build_presign(
+            1,
+            PresignRequest(scene="avatar", content_type="image/gif", size_bytes=1),
+            now=FIXED_NOW,
+        )
+    assert excinfo.value.code is ErrorCode.UPLOAD_REJECTED
+    assert signer.calls == []
+
+
+def test_unknown_bucket_alias_is_a_config_error_not_a_client_error() -> None:
+    service, signer = make_service(UPLOAD_SCENES_JSON={"avatar": {"bucket": "ghost"}})
+    with pytest.raises(AppError) as excinfo:
+        service.build_presign(
+            1,
+            PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1),
+            now=FIXED_NOW,
+        )
+    assert excinfo.value.code is ErrorCode.CONFIG_ERROR
+    assert signer.calls == []
+
+
+def test_missing_cos_credentials_is_a_config_error_before_sdk_construction() -> None:
+    # 未注入 signer 且密钥为空：守卫须在 SDK 构造前抛出（零 SDK 接触、零网络），
+    # 本地未配 COS 四件套时得到结构化 CONFIG_ERROR 而非未处理 500。
+    service = StorageService(make_settings(), None)
+    with pytest.raises(AppError) as excinfo:
+        service.build_presign(
+            1,
+            PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=1),
+            now=FIXED_NOW,
+        )
+    assert excinfo.value.code is ErrorCode.CONFIG_ERROR
+
+
+def test_request_schema_rejects_non_positive_size() -> None:
+    with pytest.raises(ValidationError):
+        PresignRequest(scene="avatar", content_type="image/jpeg", size_bytes=0)