src/main.py

from fastapi import FastAPI, HTTPException, UploadFile
from pydantic import BaseModel, ConfigDict, field_validator, model_validator
from PIL import Image

# PIL decompression bomb cap——與 paste_parser 一致 40M pixels（~200MP 才算 bomb）。
# 主入口 `/embed` + `/ocr/label` 都走 `Image.open(file.file)` 接 user upload，沒 cap
# 的話 4 KB deflated PNG 可宣告 50000×50000 RGB → PIL 嘗試 allocate ~7.5 GB 記憶體
# 把 ml-service 炸到 OOM。caller 有 INTERNAL_SECRET 但 upload 內容本質上是 user-
# supplied（hub-service 代 proxy user image）。與 paste_parser 同值，避免 class
# attr 在模組間互相覆蓋時誤差。memory: feedback_pil_decompression_bomb_cap.md
Image.MAX_IMAGE_PIXELS = 40_000_000

from src.models.fashion_clip import embed_image, get_embedding_dim
from src.services.email_parser import ParsedOrderData, parse_order_email
from src.services.paste_parser import PasteParseResult, parse_paste
from src.services.tryon import TryOnResult, run_tryon
from src.services.stylist import ItemInfo, StylistResponse, recommend_outfits
from src.services.stylist_training import launch_training_job
from src.services.gap_analysis import GapAnalysisResponse, analyze_wardrobe_gaps
from src.models.xgboost_valuation import (
    ValuationInput,
    ValuationResult,
    predict_with_fallback as xgb_predict_with_fallback,
)
from src.services.trend_predictor import TrendPrediction, predict_trend
from src.services.size_recommender import (
    build_fit_verdict,
    FitVerdict,
    CrowdInput,
    UserProfile,
)
from src.services.qa_answerer import QaInput, QaOutput, answer_question
from src.services.translator import TranslateInput, TranslateOutput, translate_text
from src.services.chat import ChatRequest, ChatResponse, chat_respond
from src.services.description_generator import DescriptionInput
from src.services.newsletter_parser import (
    NewsletterParseResult,
    parse_newsletter,
)

import asyncio
import hmac
import logging
import os
from contextlib import asynccontextmanager
from collections.abc import AsyncIterator

from starlette.requests import Request

_logger = logging.getLogger(__name__)


@asynccontextmanager
async def lifespan(_app: FastAPI) -> AsyncIterator[None]:
    """啟動時初始化 ModelRegistry + 共用 DB pool + 品牌正規化快取 + 背景預載。"""

    # 0. 預先初始化 DB pool（F4 v2 SR-6 closer）— 在 brand_normalizer.load()
    #    之前，因為 normalizer 第一個 acquire pool。pool 連 DB 失敗 (Neon 暫時
    #    503) 不阻塞啟動：lazy retry，warn-only。
    from src.db import get_pool

    try:
        get_pool()
    except Exception as exc:
        _logger.warning("DB pool eager init failed (will lazy-retry): %s", exc)

    # 1. 初始化全域 ModelRegistry（所有 service 共用）
    from src.registry import init_registry

    init_registry()

    # 2. Pre-load brand normalizer cache
    try:
        from src.services.brand_normalizer import get_normalizer

        get_normalizer().load()
    except Exception:
        pass  # Non-fatal: will lazy-load on first request

    # 3. 背景預載 capability，不阻塞 server 啟動
    #    載入完成前 health endpoint 仍可回應，model 相關 endpoint 會等 lock 釋放
    #
    #    預設 lazy（不預載）— Gemma 4 ~7GB 與 FashionSigLIP 共住在 Apple
    #    Silicon 17GB unified memory 會觸發 Metal kernel hang（/verify/local
    #    HTTP path 死鎖）。ModelRegistry 已 per-capability lazy load，第一次
    #    /chat 等請求才拉權重；只 SigLIP 工作流（backfill / paste / embed）
    #    永遠不會觸發 cohabitation。
    #
    #    Production 預載（HF Spaces L40S）：
    #    - ML_PRELOAD_CHAT=1     → 預載 chat (qwen3-vl-8b-instruct)，避免第一個聊天請求
    #                              ~30s 冷啟。Sprint D：chat 從 Gemma 4 切到 Qwen3-VL，
    #                              此 env 預載的是 Qwen3-VL instance（不再連帶 warm Gemma）。
    #    - ML_PRELOAD_TIMESFM=1  → 預載 trend_predict (TimesFM 2.5)，避免第一個趨勢請求
    #                              ~10s 冷啟 + ForecastConfig.compile() 開銷
    #    - ML_PRELOAD_OCR=1      → 預載 ocr (DeepSeek-OCR-2 specialist) + ocr_reasoning
    #                              (gemma-4-e4b)。Sprint D 之前 chat 與 ocr_reasoning 共用
    #                              gemma-4-e4b instance，ML_PRELOAD_CHAT=1 自動連帶 warm
    #                              Gemma；Sprint D 後 chat 切 Qwen3-VL，這個 sharing 斷了，
    #                              所以 ML_PRELOAD_OCR=1 需顯式預載 ocr_reasoning 才能完整
    #                              warm 兩階段 OCR pipeline。
    _preload_caps: list[str] = []
    if os.environ.get("ML_PRELOAD_CHAT") == "1":
        _preload_caps.append("chat")
    if os.environ.get("ML_PRELOAD_TIMESFM") == "1":
        _preload_caps.append("trend_predict")
    if os.environ.get("ML_PRELOAD_OCR") == "1":
        # Sprint D：顯式預載 ocr (DeepSeek Stage 1) + ocr_reasoning (Gemma 4 Stage 2)。
        # 兩階段 OCR pipeline 完整 warm 不再透過 chat capability sharing 連帶（chat 已切 Qwen3-VL）。
        _preload_caps.extend(["ocr", "ocr_reasoning"])

    if _preload_caps:

        async def _preload_critical_models() -> None:
            from src.registry import get_registry

            reg = get_registry()
            for cap in _preload_caps:
                try:
                    await reg.get(cap)
                    # WARN level so HF Spaces / Render container logs surface
                    # this — root logger defaults to WARNING and uvicorn does
                    # not configure src.* handlers, so INFO is silently dropped.
                    _logger.warning(
                        "ml-service startup: pre-loaded capability '%s'", cap
                    )
                except Exception as e:
                    _logger.warning("Pre-load '%s' failed (will lazy-load): %s", cap, e)

        # v17 P2 — fire-and-forget needs a done-callback so escaping
        # exceptions surface to structured logs. Inner coroutine wraps each
        # capability load in try/except, but `from src.registry import ...`
        # at line 74 + the for-loop itself can raise outside that guard;
        # without the callback those exceptions only land on stderr (not
        # Sentry / pino aggregation in HF Spaces) and `/health` keeps
        # returning OK while every model call later 500s.
        _preload_task = asyncio.create_task(_preload_critical_models())

        def _on_preload_done(t: "asyncio.Task[None]") -> None:
            if t.cancelled():
                return
            exc = t.exception()
            if exc is not None:
                # Local logger 仍寫 unredacted exc — operator 從 Render / HF
                # Spaces stdout 撈 trace。Sentry 只見 scrubbed label。
                _logger.error("Preload task escaped guard: %s", exc, exc_info=exc)
                # F3 deferred sweep（2026-05-03）—— 直接 capture_exception(exc)
                # 會把 exc.args（psycopg OperationalError 的 Neon DSN /
                # huggingface_hub 401 的 hf_ token）原樣寫進 event 的
                # exception.values[].value 欄位。Sentry-only reader（compromised
                # integration token 或 read-grant）拿到 plaintext credentials。
                # 改走 capture_message + scrub_exc_for_audit() opaque label。
                # Sibling site: lib/background.py:_log_unhandled_exception。
                # memory: feedback_python_sentry_capture_needs_scrub。
                #
                # AsyncioIntegration 因 t.exception() 已 consume 例外而失活，
                # 仍需顯式 capture 才能進告警鏈（ml-service.error-rate-spike /
                # .5xx-rate）。memory: feedback_sentry_done_callback_consumes_asyncio_capture。
                try:
                    import sentry_sdk

                    from src.lib.error_scrub import scrub_exc_for_audit

                    # F2/F5 closer (c-series wave-2, hardened in wave-3)：
                    # 必須用 isolation_scope，不是 new_scope。sentry-sdk Python
                    # 2.x 把 scope 拆成 global / isolation / current 三層，
                    # **breadcrumb 寫在 isolation scope**——LoggingIntegration
                    # level=INFO 把上面 _logger.error 寫的 raw exc 字串塞進
                    # ambient isolation scope 的 breadcrumb buffer。
                    # `new_scope()` 只 fork current scope，clear_breadcrumbs()
                    # 對 isolation scope 沒效，event 仍夾 ambient breadcrumb；
                    # `isolation_scope()` fork isolation scope，clear_breadcrumbs()
                    # 才真清。empirically verified on sentry-sdk 2.58.0。
                    # 鏡像 TS `Sentry.withScope`（TS scope model 不分層，
                    # withScope 等同 isolation_scope）。
                    # memory: feedback_sentry_breadcrumb_bypasses_message_scrub。
                    with sentry_sdk.isolation_scope() as scope:
                        scope.clear_breadcrumbs()
                        sentry_sdk.capture_message(
                            f"preload-failed: {scrub_exc_for_audit(exc)}",
                            level="error",
                        )
                except Exception:
                    pass

        _preload_task.add_done_callback(_on_preload_done)
    else:
        # WARN level — see Pre-loaded note above re: root logger default.
        _logger.warning(
            "ml-service startup: no ML_PRELOAD_* set — all capabilities lazy-load on first request"
        )

    yield  # 應用程式運行中

    # Shutdown
    # — drain DB pool；torch 模型仍由 process exit 釋放（沒明確 unload API）
    from src.db import close_pool

    close_pool()


# ── Sentry SDK 初始化 ───────────────────────────────────────────────────────
# 必須在 FastAPI 物件建立前呼叫，FastApiIntegration 才能 hook 進 ASGI middleware
# 鏈。詳見 src/sentry_init.py docstring。
#
# Fail-loud sentinel: 若 SENTRY_DSN 設了但 SENTRY_ENVIRONMENT 沒設，
# init_sentry() 會 raise RuntimeError，避免 dev 事件灌進 production project。
from src.sentry_init import init_sentry as _init_sentry  # noqa: E402

_init_sentry()


app = FastAPI(title="Wardrobe OS ML Service", version="0.0.1", lifespan=lifespan)

# ── Internal secret 驗證（防止公開存取 ML service） ─────────────────────────
INTERNAL_SECRET = os.environ.get("INTERNAL_SECRET")


@app.middleware("http")
async def verify_internal_secret(request: Request, call_next):
    """驗證 INTERNAL_SECRET header — 只有內部服務可呼叫 ML API。

    豁免：/health、/docs、/openapi.json（部署平台健康檢查）

    Fail-closed: 缺 INTERNAL_SECRET env 一律 503，鏡像 Node 端 receiver pattern。
    （v16 P0-1：v1~v15 之前是 `if INTERNAL_SECRET and ...` — 缺 env → 跳過驗證 → ML 全公開。）
    """
    from starlette.responses import JSONResponse

    path = request.url.path
    if path in ("/health", "/docs", "/openapi.json"):
        return await call_next(request)

    if not INTERNAL_SECRET:
        return JSONResponse(
            status_code=503,
            content={"detail": "ML service misconfigured: INTERNAL_SECRET not set"},
        )

    # 用 hmac.compare_digest 做 constant-time 比對，避免 plain `!=` 在第一個
    # 不同 byte 短路導致的 timing oracle（co-located attacker 可用 sub-ms RTT
    # variance brute-force secret prefix）。對應 Node 端 crypto.timingSafeEqual。
    # memory: feedback_timing_safe_secrets.md
    secret = request.headers.get("x-internal-secret") or ""
    if not hmac.compare_digest(secret, INTERNAL_SECRET):
        return JSONResponse(
            status_code=403,
            content={"detail": "Forbidden: invalid internal secret"},
        )
    return await call_next(request)


# 允許的圖片 MIME 類型（僅接受 JPEG、PNG、WebP，防止惡意檔案上傳）
ALLOWED_IMAGE_TYPES = {"image/jpeg", "image/png", "image/webp"}


def _validate_image_content_type(file: UploadFile) -> None:
    """驗證上傳檔案的 MIME 類型是否為允許的圖片格式。

    若 content_type 不在白名單內，回傳 HTTP 415 Unsupported Media Type。
    """
    if file.content_type not in ALLOWED_IMAGE_TYPES:
        raise HTTPException(
            status_code=415,
            detail=f"Unsupported image format: {file.content_type}. Allowed: JPEG, PNG, WebP",
        )


async def _decode_upload_image(file: UploadFile) -> "Image.Image":
    """讀取 upload bytes → 在 thread pool 解碼成 RGB PIL Image。

    兩個重點（對齊 src/ssrf.py._decode_image 的既有 discipline）：
    1. decode（`.convert("RGB")` 會 force full decode）放 `asyncio.to_thread`，
       不阻塞 event loop——大圖解碼期間不會餓死其他 coroutine。
    2. content-type header 只擋宣告的 MIME，truncated/garbage bytes（header 合法但
       內容不是圖）會在 PIL 丟 UnidentifiedImageError/OSError；宣告 pixel-bomb 會丟
       DecompressionBombError。包成 HTTP 422 而非讓未捕捉例外變成 FastAPI 預設 500。
    """
    import io

    raw = await file.read()

    def _decode() -> "Image.Image":
        # 顯式設 bomb cap 在解碼函式內，不依賴 module-level 賦值（mutable class
        # attr 可能被其他 code path reset 成 None → 靜默關閉防線）。對齊
        # paste_parser._decode_image_base64。memory: pil_bomb_cap_no_import_sideeffect
        Image.MAX_IMAGE_PIXELS = 40_000_000
        return Image.open(io.BytesIO(raw)).convert("RGB")

    try:
        return await asyncio.to_thread(_decode)
    except (
        Image.UnidentifiedImageError,
        Image.DecompressionBombError,
        OSError,
        ValueError,
    ) as exc:
        raise HTTPException(
            status_code=422,
            detail=f"Could not decode image: {type(exc).__name__}",
        ) from exc


class EmbeddingResponse(BaseModel):
    embedding: list[float]
    dimension: int


class ModelInfoResponse(BaseModel):
    model: str
    embedding_dim: int
    device: str


class LabelOcrResponse(BaseModel):
    brand_raw: str | None = None
    brand_normalized: str | None = None
    product_name: str | None = None
    product_code: str | None = None
    size: str | None = None
    material: str | None = None
    care_instructions: str | None = None
    country_of_origin: str | None = None


@app.get("/health")
async def health() -> dict[str, str]:
    return {"status": "ok", "service": "ml-service"}


@app.post("/api/v1/ml/embed", response_model=EmbeddingResponse)
async def embed(file: UploadFile) -> EmbeddingResponse:
    """Generate a FashionSigLIP embedding vector from an uploaded image."""
    _validate_image_content_type(file)
    image = await _decode_upload_image(file)
    vector = await asyncio.to_thread(embed_image, image)
    return EmbeddingResponse(embedding=vector, dimension=len(vector))


@app.get("/api/v1/ml/model-info", response_model=ModelInfoResponse)
async def model_info() -> ModelInfoResponse:
    """Return loaded model metadata."""
    # fashion_clip exports the repo string as MODEL_REPO (no MODEL_NAME symbol);
    # this lazy import referenced a non-existent name → ImportError → 500 on every
    # /model-info call. Pre-existing bug, fixed drive-by while in this file.
    from src.models.fashion_clip import DEVICE, MODEL_REPO as MODEL_NAME

    return ModelInfoResponse(
        model=MODEL_NAME,
        embedding_dim=get_embedding_dim(),
        device=DEVICE,
    )


@app.get("/api/v1/ml/registry/health")
async def registry_health() -> dict:
    """回傳所有 capability 的 ModelRegistry 載入狀態。

    【回傳格式】
    {
      "models": {
        "<capability>": {"status": "loaded" | "not_loaded", "model": "<model_key>"},
        ...
      }
    }
    """
    from src.registry import get_registry

    # get_registry() 在 startup 後一定存在；若測試環境尚未初始化則回空 dict
    try:
        reg = get_registry()
    except RuntimeError:
        return {"models": {}}

    return {"models": reg.health()}


@app.post("/api/v1/ml/ocr/label", response_model=LabelOcrResponse)
async def ocr_label(file: UploadFile) -> LabelOcrResponse:
    """從洗標照片提取商品資訊（Gemma 4 E4B 單步視覺 OCR），並正規化品牌名稱。"""
    from src.services.label_ocr import extract_label
    from src.services.brand_normalizer import normalize_brand

    _validate_image_content_type(file)
    image = await _decode_upload_image(file)
    # extract_label 已改為 async（透過 ModelRegistry 取得兩個模型），直接 await
    result = await extract_label(image)

    return LabelOcrResponse(
        brand_raw=result.brand_raw,
        brand_normalized=await asyncio.to_thread(normalize_brand, result.brand_raw),
        product_name=result.product_name,
        product_code=result.product_code,
        size=result.size,
        material=result.material,
        care_instructions=result.care_instructions,
        country_of_origin=result.country_of_origin,
    )


class EmailParseRequest(BaseModel):
    subject: str
    body: str


@app.post("/api/v1/ml/parse/email", response_model=ParsedOrderData)
async def parse_email(request: EmailParseRequest) -> ParsedOrderData:
    """解析訂單確認 email，透過 ModelRegistry 載入 Qwen2.5-0.5B + LoRA 進行推論。"""
    # parse_order_email 已改為 async（透過 ModelRegistry 取得模型），直接 await
    return await parse_order_email(request.subject, request.body)


class PasteParseRequest(BaseModel):
    # v17 P2 — extra="forbid": typo 在 mode 欄位該 loud fail，不是靜默走 text path
    model_config = ConfigDict(extra="forbid")

    # "text" | "html" | "image_base64" — 詳見 paste_parser.Mode
    mode: str
    # 內容；text 為 plain，html 為 outerHTML，image_base64 為 base64 或 data URI
    content: str


@app.post("/api/v1/ml/parse/paste", response_model=PasteParseResult)
async def parse_paste_endpoint(request: PasteParseRequest) -> PasteParseResult:
    """批次匯入的剪貼簿解析器 — Gemma 4 吃下 pasted profile，回傳多筆 item。"""
    if request.mode not in ("text", "html", "image_base64"):
        raise HTTPException(
            status_code=422,
            detail="mode must be 'text', 'html', or 'image_base64'",
        )
    return await parse_paste(request.mode, request.content)  # type: ignore[arg-type]


class TryOnRequest(BaseModel):
    # v17 P2 — extra="forbid": URL fields hit external GPU provider; typo'd
    # field name would 500 mid-async rather than 422.
    model_config = ConfigDict(extra="forbid")

    # v25 wire-contract fix — hub tryon worker + ml-client tryon capability 送
    # `person_image_url`（不是 `user_photo_url`）。原本 required `user_photo_url`
    # 缺欄位 + `person_image_url` 是 forbidden extra → 每次 try-on 雙重 422。
    # 對齊 caller 欄位名（extra="forbid" 保留，別退回 ignore）。
    person_image_url: str
    garment_image_url: str
    # capability 簽名帶 optional `category`；worker 目前不送，但接受以免未來加上時
    # 撞 forbidden-extra 422。目前不進 VTON prompt（provider 只吃兩張圖）。
    category: str | None = None


@app.post("/api/v1/ml/tryon", response_model=TryOnResult)
async def tryon(request: TryOnRequest) -> TryOnResult:
    """Generate a virtual try-on image using Replicate IDM-VTON."""
    return await run_tryon(request.person_image_url, request.garment_image_url)


class StylistRequest(BaseModel):
    # v17 P2 — extra="forbid": caller typo (e.g. `adapter_hf_repo` vs
    # `adapter_hf_repo_id`) returns 422 instead of silently ignoring → caller
    # would otherwise see fallback recommendations with no signal that the
    # adapter request was dropped.
    model_config = ConfigDict(extra="forbid")

    items: list[ItemInfo]
    occasion: str | None = None
    weather: dict | None = None  # S14-4：天氣資訊（temperature, weatherCode, humidity）
    location: dict | None = None  # S14-4：地點資訊（city, country）
    adapter_id: str | None = (
        None  # Sprint 16：LoRA adapter name（使用者 UUID 或 "default"）
    )
    adapter_hf_repo_id: str | None = None  # Sprint 16：adapter 的 HF Hub 位址


@app.post("/api/v1/ml/stylist/recommend", response_model=StylistResponse)
async def stylist_recommend(request: StylistRequest) -> StylistResponse:
    """Generate outfit recommendations from user's wardrobe items (weather-aware, adapter-aware)."""
    return await recommend_outfits(
        request.items,
        request.occasion,
        request.weather,
        request.location,
        adapter_id=request.adapter_id,
        adapter_hf_repo_id=request.adapter_hf_repo_id,
    )


class GapAnalysisRequest(BaseModel):
    items: list[ItemInfo]


@app.post("/api/v1/ml/stylist/gaps", response_model=GapAnalysisResponse)
async def stylist_gaps(request: GapAnalysisRequest) -> GapAnalysisResponse:
    """Analyze wardrobe gaps and suggest missing pieces."""
    return await analyze_wardrobe_gaps(request.items)


class LaunchTrainingRequest(BaseModel):
    user_id: str


class LaunchTrainingResponse(BaseModel):
    available: bool
    job_id: str | None


@app.post("/api/v1/ml/stylist/launch-training", response_model=LaunchTrainingResponse)
async def stylist_launch_training(
    request: LaunchTrainingRequest,
) -> LaunchTrainingResponse:
    """B1 — 啟動 per-user LoRA 訓練 Job（run_uv_job）。

    HF launch 失敗 → 回 available=False 讓 worker 釋鎖（不捏造已訓練 adapter）。
    Sentry capture 走 scrub + isolation_scope clear_breadcrumbs（HF_TOKEN/secret
    永不進 event；mirror main.py preload site）。
    """
    try:
        out = launch_training_job(request.user_id)
        return LaunchTrainingResponse(**out)
    except Exception as exc:  # noqa: BLE001 — fail-soft + scrubbed capture
        import sentry_sdk

        from src.lib.error_scrub import scrub_exc_for_audit

        with sentry_sdk.isolation_scope() as scope:
            scope.clear_breadcrumbs()
            sentry_sdk.capture_message(
                f"stylist-launch-training-failed: {scrub_exc_for_audit(exc)}",
                level="error",
            )
        # HF launch failed → tell the worker it's unavailable so it releases the lock
        return LaunchTrainingResponse(available=False, job_id=None)


# Cap on outstanding fire-and-forget observation tasks. record_observation
# does ~3s SSRF/download + torch embed + psycopg INSERT with no back-pressure
# on the request path; without a cap an attacker spamming /predict with
# image_urls (mlRateLimiterCheap 30/min/user × N accounts) could grow
# _BACKGROUND_TASKS unbounded and exhaust asyncio's to_thread worker pool,
# blocking other ml-service endpoints. 50 is sized so that at sustained
# traffic the backlog stays well below default ThreadPoolExecutor capacity
# (min(32, cpu+4)) — drops are logged for ops visibility but don't break the
# prediction SLA. Post-review SR-P1-1 (security-review conf 80).
_MAX_PENDING_OBSERVATIONS = 50


@app.post("/api/v1/ml/valuation/predict", response_model=ValuationResult)
async def valuation_predict(request: ValuationInput) -> ValuationResult:
    """XGBoost 估值推論（透過 ModelRegistry，fallback 規則系統）。

    【流程】
    1. predict_with_fallback 根據 n_samples 決定 rule / blended / xgboost 策略
       - n < 30：純規則（weight_xgb=0，不呼叫 XGBoost）
       - 30 ≤ n ≤ 50：線性 ramp 混合 rule 和 xgboost
       - n > 50：純 xgboost（若 model 可用，否則 fallback rule）
    2. 將 CPU 密集推論放在 asyncio.to_thread 避免阻塞 event loop
    3. F4 v2 (A) — 若 caller 帶 image_url：FashionSigLIP 嵌入 + INSERT 到
       valuation_observations 表，累積未來 image-feature 訓練資料。
       fire_and_forget 排程 record_observation：observation 失敗不影響
       prediction，且 ~3s 的 SSRF/download/embed/INSERT 不卡 critical path。
       worker SIGKILL 會丟 in-flight task — 跟原 await 版的代價相同（兩者
       都非 durable），但換來每 call ~3s 的 latency 改善。
       無 image_url 時跳過排程（避免 N requests 都 allocate 一個 no-op task
       打到 _BACKGROUND_TASKS — post-review CR-1 conf 88）；超過
       _MAX_PENDING_OBSERVATIONS 時 drop + log warn。
    """
    import asyncio

    from src.lib.background import background_task_count, fire_and_forget
    from src.services.valuation_observations import record_observation

    result = await asyncio.to_thread(
        xgb_predict_with_fallback, request, request.n_samples
    )
    if request.image_url is not None:
        if background_task_count() < _MAX_PENDING_OBSERVATIONS:
            fire_and_forget(
                record_observation(
                    inp=request,
                    result=result,
                    image_url=request.image_url,
                    item_id=request.item_id,
                ),
                name="record_valuation_observation",
            )
        else:
            _logger.warning(
                "valuation observation backlog at cap (%d); dropping",
                _MAX_PENDING_OBSERVATIONS,
            )
    return result


@app.get("/api/v1/ml/valuation/status")
async def valuation_status() -> dict[str, bool | str]:
    """透過 ModelRegistry 檢查 XGBoost 模型是否已載入。"""
    from src.models.xgboost_valuation import _get_valuation_model

    _model, _encoder, available = await _get_valuation_model()
    return {
        "xgboost_available": available,
        "method": "xgboost" if available else "rule_based",
    }


class TextEmbeddingRequest(BaseModel):
    text: str


@app.post("/api/v1/ml/embed/text", response_model=EmbeddingResponse)
async def embed_text(request: TextEmbeddingRequest) -> EmbeddingResponse:
    """Generate a FashionSigLIP text embedding for semantic search.

    透過 ModelRegistry 取得 embed_text capability 對應的 FashionSigLIP 實例。
    """
    from src.registry import get_registry
    import torch

    reg = get_registry()
    loaded = await reg.get("embed_text")
    model = loaded.model
    tokenizer = loaded.tokenizer
    device = loaded.config.get("device", "cpu")

    def _encode_text() -> list[float]:
        tokens = tokenizer([request.text]).to(device)
        with torch.no_grad():
            features = model.encode_text(tokens)
            features = features / features.norm(dim=-1, keepdim=True)
        return features.squeeze(0).cpu().tolist()

    import asyncio

    vector = await asyncio.to_thread(_encode_text)
    return EmbeddingResponse(embedding=vector, dimension=len(vector))


# --- Local AI Verification (Gemma 4 Vision) ---


class LocalVerifyRequest(BaseModel):
    # v17 P2 — extra="forbid": URL fields go through SSRF validation; a typo'd
    # field name (e.g. `imageUrl` camelCase vs snake_case) would silently skip
    # validation rather than rejecting the request.
    model_config = ConfigDict(extra="forbid")

    brand: str | None = None
    image_url: str | None = None
    label_image_url: str | None = None


def _to_confidence_band(confidence: float) -> str:
    """Bucket Gemma's raw confidence float into a 3-level band.

    Codebase review v19 §ML-02. The raw float is a scalar oracle even
    though the endpoint is owner-gated: the owner can submit forgeries
    against their own item and use the float as a gradient signal to push
    inputs toward the "authentic" decision boundary. 3-level enum (matches
    anchor_band) is the minimum semantic the UI / writer needs.

    Thresholds:
      high   ≥ 0.8  → Gemma is genuinely confident.
      medium ≥ 0.5  → leaning but not strongly so.
      low           → coin-flip or worse.
    """
    if confidence >= 0.8:
        return "high"
    if confidence >= 0.5:
        return "medium"
    return "low"


class LocalVerifyResponse(BaseModel):
    result: str  # "authentic" | "counterfeit" | "inconclusive"
    # Codebase review v19 §ML-02: raw confidence float was a scalar oracle
    # even on this owner-gated endpoint (memory feedback_llm_response_scalar_oracle
    # "Applies even to owner-only endpoints"). Now a 3-level band that
    # matches anchor_band's shape — log2(3) ≈ 1.5 bit/query is too thin
    # for gradient attacks.
    confidence_band: str  # "high" | "medium" | "low"
    reasoning: str
    # F5 — 防偽錨點粗分級；None = 未跑錨點（空品牌 / 該品牌無錨點 / 查詢失敗）。
    # 只暴露 band，不暴露精確 similarity + gallery size：scalar float 會讓
    # authenticated attacker 把 response 當黑箱 oracle（梯度優化把偽品推到
    # looks_authentic threshold），count 洩露 gallery size 讓他挑最弱保護品牌。
    # band 三級 log2(3) ≈ 1.5 bit/query 攻擊效益太低，同時 UI 仍有可顯示的 verdict。
    # F5 post-review SR2 conf P1（2026-04-23 commit 6d27790 之後的 re-review）。
    anchor_band: str | None = None  # "looks_authentic" | "inconclusive" | "deviates"


from src.ssrf import download_image as _download_image  # noqa: E402
from src.lib.prompt_safety import sanitize_for_prompt as _sanitize_for_prompt  # noqa: E402
from src.lib.prompt_safety import wrap_boundary as _wrap_boundary  # noqa: E402


def _sanitize_brand_for_prompt(raw: str | None, max_len: int = 80) -> str | None:
    """把 request.brand 做 prompt-injection hardening 再拿去塞 Gemma prompt（flatten 單行 token）。

    委派給單一來源 `src.lib.prompt_safety.sanitize_for_prompt(preserve_newlines=False)`：
    控制空白（含 form-feed/vtab，flatten_form_feeds 預設 True）先轉空白 →
    isprintable() 濾零寬/雙向 override → whitespace 壓合 → clamp 80。

    F5 post-review SR3 conf P1（`request.brand` 之前直接 f-string 進 prompt）。
    """
    return _sanitize_for_prompt(raw, max_len=max_len)


@app.post("/api/v1/ml/verify/local", response_model=LocalVerifyResponse)
async def verify_local(request: LocalVerifyRequest) -> LocalVerifyResponse:
    """Run local AI verification using Gemma 4 vision analysis.

    Accepts product image URL and optional label image URL.
    Analyzes visual authenticity cues and label consistency.
    Protected by SSRF validation and image size limits.
    """
    from src.services.gemma_inference import gemma_vision

    images: list[Image.Image] = []
    image_desc_parts: list[str] = []

    # 下載商品照（含 SSRF 防護 + 大小限制）
    if request.image_url:
        try:
            images.append(await _download_image(request.image_url))
            image_desc_parts.append("product photo")
        except Exception as e:
            _logger.warning("Failed to fetch product image: %s", e)

    # 下載標籤照
    if request.label_image_url:
        try:
            images.append(await _download_image(request.label_image_url))
            image_desc_parts.append("care label/tag photo")
        except Exception as e:
            _logger.warning("Failed to fetch label image: %s", e)

    # Section 3 I: 防偽驗證 inconclusive+0.5 fallback 長得跟真實 verdict 一模一樣，
    # 生產環境吐出假驗證會誤導使用者 + Provenance Chain。strict default —
    # 只有 DEV_MOCK_ML=true 才走 inconclusive，否則 503。
    if not images:
        if os.environ.get("DEV_MOCK_ML") != "true":
            raise HTTPException(
                status_code=503,
                detail={
                    "error": "ML_UNAVAILABLE",
                    "message": "AI 驗證暫時無法使用，請稍後再試",
                    "capability": "verify_local",
                },
            )
        return LocalVerifyResponse(
            result="inconclusive",
            confidence_band=_to_confidence_band(0.5),
            reasoning="No images available for analysis",
        )

    # SR3: caller-supplied brand 不可直接 f-string 進 prompt——caller 是已認證的
    # hub-service 但 brand 字串來源是 hub_items.brand（user-supplied，只有 varchar
    # 255 cap，沒有 prompt-safety validation）。sanitize 過之後用 `<<<...>>>` 三
    # 重角括號包起來，讓 Gemma 清楚邊界，攻擊者就算 strip 後還能塞 80 chars
    # 也無法 override 指令（Gemma 解析時角括號內內容視為實體資料）。
    brand_safe = _sanitize_brand_for_prompt(request.brand)
    # wrap_boundary 統一 fence + 中和字面 <<< / >>>（fence-escape，PR #197 P1）
    brand_context = f" for brand {_wrap_boundary(brand_safe)}" if brand_safe else ""
    image_context = " and ".join(image_desc_parts)

    # F5 — query brand anchor gallery when brand + product image both known.
    # We embed the product image (images[0]) and compare against stored
    # references; the result feeds the Gemma prompt as grounding so the VLM
    # doesn't have to invent its own internal reference for each brand.
    # anchor_result is None when brand is blank OR no anchors exist for it —
    # fall back to pure zero-shot in that case (backwards-compatible).
    # 注意：query_anchor_similarity 用 sanitized brand 查 DB 也 OK（brand_anchors
    # 表儲存的就是 canonical form，不含換行/控制字元）——用 raw brand 會多
    # 繞 strip/whitespace 一次才 match，用 sanitized 反而少一層意外。
    anchor_result = None
    if brand_safe and images:
        try:
            from src.registry import get_registry
            from src.services.counterfeit_anchor import query_anchor_similarity

            cap_cfg = get_registry().get_capability_config("counterfeit_anchor")
            anchor_result = await query_anchor_similarity(
                images[0],
                brand_safe,
                authentic_threshold=float(cap_cfg.get("authentic_threshold", 0.80)),
                deviates_threshold=float(cap_cfg.get("deviates_threshold", 0.60)),
            )
        except Exception as e:
            # Anchor is advisory; failure never blocks the Gemma narrative.
            _logger.warning("Anchor similarity query failed: %s", e)

    anchor_block = ""
    if anchor_result is not None:
        # Band 是粗分級，描述要嚴謹但不洩露 scalar similarity / gallery size。
        # 對 Gemma 來說「strongly matches」「borderline」「deviates」的語意就夠
        # 做 prior 加權；float 只對 debug 有用，對 VLM 決策反而是噪音。
        band_phrase = {
            "looks_authentic": "strongly matches",
            "inconclusive": "partially matches",
            "deviates": "deviates from",
        }.get(anchor_result.verdict_band, "is compared against")
        anchor_block = (
            f"\nReference gallery check: the product image {band_phrase} "
            f"known-authentic reference photos for this brand. Treat this as "
            "supporting evidence, not a verdict — weigh it against your own "
            "visual inspection.\n"
        )

    prompt = (
        f"You are an expert authenticator. Analyze this {image_context}{brand_context}.\n"
        f"{anchor_block}"
        "Check for:\n"
        "1. Stitching quality and consistency\n"
        "2. Logo/branding accuracy (font, spacing, alignment)\n"
        "3. Hardware and zipper quality\n"
        "4. Material texture and grain patterns\n"
        "5. Label printing quality and text accuracy\n"
        "6. Overall construction and finishing\n\n"
        'Return ONLY valid JSON: {"result": "authentic" or "counterfeit" or "inconclusive", '
        '"confidence": 0.0-1.0, "reasoning": "brief explanation"}'
    )

    try:
        main_image = images[0]
        extra_images = images[1:] if len(images) > 1 else None

        raw = await gemma_vision(
            "counterfeit_narrative",
            main_image,
            prompt,
            additional_images=extra_images,
            max_tokens=512,
        )

        # 解析 JSON
        import json

        text = raw.strip()
        if text.startswith("```"):
            lines = text.split("\n")
            text = "\n".join(lines[1:-1] if lines[-1].strip() == "```" else lines[1:])

        start = text.index("{")
        end = text.rindex("}") + 1
        data = json.loads(text[start:end])

        result = data.get("result", "inconclusive")
        if result not in ("authentic", "counterfeit", "inconclusive"):
            result = "inconclusive"

        confidence = float(data.get("confidence", 0.5))
        confidence = max(0.0, min(1.0, confidence))

        return LocalVerifyResponse(
            result=result,
            confidence_band=_to_confidence_band(confidence),
            reasoning=data.get("reasoning", ""),
            anchor_band=(anchor_result.verdict_band if anchor_result else None),
        )

    except Exception as e:
        _logger.warning("Local verification failed: %s", e)
        if os.environ.get("DEV_MOCK_ML") != "true":
            raise HTTPException(
                status_code=503,
                detail={
                    "error": "ML_UNAVAILABLE",
                    "message": "AI 驗證暫時無法使用，請稍後再試",
                    "capability": "verify_local",
                },
            ) from e
        # anchor_band 在 Gemma 爆炸之前可能已經成功算出來——保留回傳，讓
        # DEV_MOCK_ML mock path 至少帶得動 anchor 的分級資訊。
        return LocalVerifyResponse(
            result="inconclusive",
            confidence_band=_to_confidence_band(0.5),
            reasoning="Verification could not be completed. Please try again.",
            anchor_band=(anchor_result.verdict_band if anchor_result else None),
        )


# --- #71: Trend Prediction ---


class PricePoint(BaseModel):
    date: str
    price: float


class TrendRequest(BaseModel):
    brand: str
    price_history: list[PricePoint] | None = None


@app.post("/api/v1/ml/trends/predict", response_model=TrendPrediction)
async def trends_predict(request: TrendRequest) -> TrendPrediction:
    """品牌價格趨勢預測（Chronos-Bolt-Tiny + pytrends，透過 ModelRegistry）。

    【流程】
    Dev mode: 返回 mock 線性預測（TREND_DEV_MODE=true）。
    Production: 從 registry 取得 Chronos 模型 → 零射時間序列預測 + Google Trends。
    """
    import asyncio

    history = None
    if request.price_history:
        history = [{"date": p.date, "price": p.price} for p in request.price_history]

    # predict_trend 是同步函式，放在 thread 中執行
    # 內部的 _chronos_predict 會呼叫 _load_chronos()（registry fallback）
    return await asyncio.to_thread(predict_trend, request.brand, history)


# --- #72: Size Recommendation ---


class SizeRequest(BaseModel):
    # v17 P2 — extra="forbid": caller typo (e.g. garment_measurments) must fail
    # loud, not silently drop the fusion input and degrade to brand-chart only.
    model_config = ConfigDict(extra="forbid")
    brand: str
    category: str = ""  # reserved (future category-specific charts); unused in SP1
    user_profile: UserProfile
    garment_measurements: dict[str, dict[str, float]] | None = None
    crowd: CrowdInput | None = None
    sentiment_band: str | None = None


@app.post("/api/v1/ml/size/recommend", response_model=FitVerdict)
async def size_recommend(request: SizeRequest) -> FitVerdict:
    """Fuse brand chart + body + garment measurements + crowd + sentiment."""
    return build_fit_verdict(
        request.brand,
        request.user_profile,
        garment_measurements=request.garment_measurements,
        crowd=request.crowd,
        sentiment_band=request.sentiment_band,
    )


# --- F2: Q&A Chatbot ---


@app.post("/api/v1/ml/qa/answer", response_model=QaOutput)
async def qa_answer(request: QaInput) -> QaOutput:
    """Answer a buyer's question about an item using local model."""
    return await answer_question(request)


# --- Pivot 賣家側 #3：跨語言議價 / Q&A 翻譯 ---


@app.post("/api/v1/ml/translate", response_model=TranslateOutput)
async def translate_endpoint(request: TranslateInput) -> TranslateOutput:
    """把任意語言文字翻成 target_lang（source 由模型自動偵測）。

    用途：賣家讀外語買家議價訊息 / Q&A 提問（入站翻譯）+ copilot 草稿的買家
    語言偵測（出站）。重用已 resident 的 translate capability（預設 Qwen3-VL-8B，
    與 chat / qa 同一 model 鍵 → 零增量記憶體）。

    request.text 是完全 user-controlled 的自由文字（經典 prompt-injection 向量）—
    translator.build_translate_prompt 用 <<< >>> boundary token 包住 + 嚴格
    translate-only 指令 + sanitize（保留 \\n、strip 零寬/雙向/控制字元）。
    模型不可用 / 失敗 → available=False（不捏造翻譯）；HTTP 仍 200，caller 見
    available 旗標優雅降級（fail-soft toast，不擋手動操作）。
    """
    return await translate_text(request)


# --- Newsletter Promo Parser ---


class NewsletterParseRequest(BaseModel):
    email_body: str
    platform: str  # "uniqlo" or "ssense"


@app.post("/api/v1/ml/parse/newsletter", response_model=NewsletterParseResult)
async def parse_newsletter_endpoint(
    request: NewsletterParseRequest,
) -> NewsletterParseResult:
    """解析品牌行銷電子報，提取折扣活動和優惠碼。parse_newsletter 已改為 async，直接 await。"""
    # parse_newsletter 已改為 async（透過 ModelRegistry 取得 Qwen2.5-0.5B），直接 await
    return await parse_newsletter(request.email_body, request.platform)


# --- 多輪對話（Multi-turn Chat + Tool Calling） ---


@app.post("/api/v1/ml/chat/respond", response_model=ChatResponse)
async def chat_respond_endpoint(request: ChatRequest) -> ChatResponse:
    """多輪對話端點（含 tool calling 支援）"""
    return await chat_respond(request)


# --- S14-1: AI 商品描述生成 ---


@app.post("/api/v1/ml/generate-description")
async def generate_description_endpoint(request: DescriptionInput) -> dict:
    """根據商品資訊生成 AI 商品描述。

    v17 P2 — 之前簽名是 `request: dict`，FastAPI 對任意 JSON 不做驗證。
    上層 hub-service 若送錯 shape（list 而非 object）會在 `request.get(...)`
    噴 `AttributeError`，而非乾淨的 422；且任意大小欄位直送 LLM prompt。
    改用既有 DescriptionInput Pydantic model（model_config 設 extra="forbid"
    防止 caller 多送拼錯欄位被靜默 ignore）。
    """
    from src.services.description_generator import generate_description

    result = await generate_description(request)
    return {"description": result.description}


# --- S14-8: 時尚趨勢分析 ---


class TrendAnalysisRequest(BaseModel):
    # v25 wire-contract fix — hub trends-report.ts + ml-client trends.analyze 送
    # `{content_items: [...]}`，但原本 required `items` → 每次週報趨勢分析 422 →
    # hub catch → 使用者見 502。model_validator(before) 接受 `content_items`
    # alias 映射到 `items`（extra="forbid" 鎖住其餘拼錯欄位）。
    model_config = ConfigDict(extra="forbid")

    items: list[dict]

    @model_validator(mode="before")
    @classmethod
    def _accept_content_items_alias(cls, data: object) -> object:
        if isinstance(data, dict) and "items" not in data and "content_items" in data:
            aliased = {k: v for k, v in data.items() if k != "content_items"}
            aliased["items"] = data["content_items"]
            return aliased
        return data


@app.post("/api/v1/ml/trends/analyze")
async def trends_analyze_endpoint(request: TrendAnalysisRequest) -> dict:
    """分析近期時尚內容，提取趨勢關鍵詞、風格方向與摘要。"""
    from src.services.trend_analyzer import (
        ContentItem,
        analyze_trends,
    )

    content_items = [
        ContentItem(
            title=item.get("title", ""),
            summary=item.get("summary"),
            tags=item.get("tags"),
        )
        for item in request.items
    ]
    result = await analyze_trends(content_items)
    return result.model_dump()


# --- S14-6: 穿搭照自動匹配商品 ---


class OutfitMatchRequest(BaseModel):
    image_url: str
    item_embeddings: list[dict]  # [{id, embedding}]
    threshold: float = 0.5
    top_k: int = 3


@app.post("/api/v1/ml/outfit/match-items")
async def outfit_match_endpoint(request: OutfitMatchRequest) -> dict:
    """穿搭照 → FashionSigLIP → cosine similarity → 回傳匹配的商品 ID。

    match_outfit_items 已改為非同步函式，直接 await 即可。
    """
    from src.services.outfit_matcher import (
        MatchRequest,
        match_outfit_items,
    )

    req = MatchRequest(
        image_url=request.image_url,
        item_embeddings=request.item_embeddings,
        threshold=request.threshold,
        top_k=request.top_k,
    )
    result = await match_outfit_items(req)
    return result.model_dump()


# --- A2: 零樣本視覺屬性抽取（colorway / category / features） ---


class VisionAttributesRequest(BaseModel):
    # extra="forbid"：URL 欄位走 SSRF 下載；caller typo（imageUrl camelCase vs
    # snake_case）要 loud fail，不靜默跳過驗證。與 verify/local 同策略。
    model_config = ConfigDict(extra="forbid")

    image_url: str


class VisionAttributesResponse(BaseModel):
    # available=False = 模型不可用 / 拿不到圖（不捏造）；caller 見此跳過 colorway。
    available: bool
    colorway: str | None = None
    category: str | None = None
    features: list[str] | None = None


@app.post("/api/v1/ml/vision/attributes", response_model=VisionAttributesResponse)
async def vision_attributes_endpoint(
    request: VisionAttributesRequest,
) -> VisionAttributesResponse:
    """A2 — 從商品照 zero-shot 抽 colorway / category / features 給 wiki 富集。

    重用已 resident 的 gemma-4-e4b vision 實例（vision_attributes capability）。
    SSRF + PIL bomb cap 全在 extract_vision_attributes → download_image 內。
    模型不可用 / 拿不到圖 → available=False（不捏造），HTTP 仍 200 讓 caller
    優雅降級（caller 是 INTERNAL_SECRET hub-service，503 會被 retry 浪費；明確
    available 旗標比 status code 語意更清楚）。輸出已 server 端 sanitize（S-5）。
    """
    from src.services.vision_attributes import extract_vision_attributes

    result = await extract_vision_attributes(request.image_url)
    return VisionAttributesResponse(
        available=result.available,
        colorway=result.colorway,
        category=result.category,
        features=result.features,
    )


# --- #23: LLM-assisted admin tariff suggestion (advisory) ---


class TariffSuggestRequest(BaseModel):
    # extra="forbid"：caller typo 要 loud fail，不靜默用 default（同 verify/local）。
    model_config = ConfigDict(extra="forbid")

    country_code: str
    country_name: str | None = None

    @field_validator("country_code")
    @classmethod
    def _validate_country_code(cls, v: str) -> str:
        # Defense-in-depth：country_code 直接拼進 prompt（country_name 已 sanitize，
        # 但 code 沒有對等防線）。即使 caller 帶有效 INTERNAL_SECRET 直打 ml-service，
        # 也只接受 ISO 3166-1 alpha-2，杜絕換行/注入 payload 從 code 欄繞過。
        import re

        if not re.fullmatch(r"[A-Za-z]{2}", v):
            raise ValueError("country_code must be ISO 3166-1 alpha-2")
        return v.upper()


class TariffSuggestResponse(BaseModel):
    # available=False = 模型不可用 / 推論失敗（不捏造）；caller 見此回 503 讓 admin 手填。
    available: bool
    duty_rate: float | None = None
    vat_rate: float | None = None
    de_minimis_usd: float | None = None
    rationale: str | None = None
    confidence: str | None = None


@app.post("/api/v1/ml/tariff/suggest", response_model=TariffSuggestResponse)
async def tariff_suggest_endpoint(
    request: TariffSuggestRequest,
) -> TariffSuggestResponse:
    """#23 — 為一個國家「建議」import duty / VAT / de-minimis 三個費率（advisory）。

    **永不寫 DB**：純 zero-shot Gemma 推論，admin 在前端核實 / 修改後才透過既有
    write 端點落地。country_name 進 prompt 前 sanitize；費率 server 端 clamp；
    模型不可用 / 拿不到合理輸出 → available=False（不捏造），api-gateway 端轉 503。
    """
    from src.services.tariff_suggester import suggest_tariff

    result = await suggest_tariff(request.country_code, request.country_name)
    return TariffSuggestResponse(
        available=result.available,
        duty_rate=result.duty_rate,
        vat_rate=result.vat_rate,
        de_minimis_usd=result.de_minimis_usd,
        rationale=result.rationale,
        confidence=result.confidence,
    )