src/services/paste_parser.py

"""剪貼簿批次匯入解析器 — Gemma 4 E4B（text / html / 螢幕截圖三模式）

【這個模組在做什麼？】
  使用者從 Grailed / Depop / Poshmark / eBay / Vinted 自己的 profile 頁複製內容，
  丟給 Gemma 4 後產出多筆結構化商品 JSON，作為 bulk import 的預覽資料。

【為什麼不寫 per-site 爬蟲？】
  - SPA rendering 不能直接 HTTP 抓
  - 自家 profile 需要 auth（使用者自己複製就有 auth）
  - Cloudflare / anti-bot 免了
  - 新平台上線等於 0 code
  任何商品 URL 的單品匯入已由 `apps/scraper-service/src/universal/` 的 JSON-LD / OG /
  microdata extractor 處理；paste_parser 補的是「多筆一起拉進來」這條路。

【三種輸入模式】
  text           — 原始文字 copy-paste（最簡單，但可能掉 image URL）
  html           — F12 → Copy outerHTML（最優，保留 product 和 image URL）
  image_base64   — 螢幕截圖（Gemma 4 vision OCR，~10-15s on M-series）

【輸出策略】
  Gemma 會幻覺 —— 只要 item 缺 `title` 一律丟掉。image_url / source_url 上游 handler
  會再跑一次 SSRF + universal extractor 補欄位，故此處不嚴格驗 URL。
"""

from __future__ import annotations

import asyncio
import base64
import io
import json
import logging
from typing import Literal

from pydantic import BaseModel, Field

from .gemma_inference import gemma_text, gemma_vision

# email_parser._strip_code_fence 是純工具函式，不帶狀態，直接複用
from .email_parser import _strip_code_fence

logger = logging.getLogger(__name__)


# ── 資料模型 ──────────────────────────────────────────────────────────────


class PastedItem(BaseModel):
    """從剪貼簿解析出的單筆商品。與 hub_items schema 對齊但留彈性。"""

    title: str
    brand: str | None = None
    price: float | None = None
    currency: str | None = None
    size: str | None = None
    # "new" | "like_new" | "good" | "fair" | "used" — 不強制 enum，Gemma 常回自由字串
    condition: str | None = None
    image_url: str | None = None
    # 商品頁 URL（若 paste 裡有），給 hub-service 餵給 universal extractor 做 enrichment
    source_url: str | None = None
    description: str | None = None


class PasteParseResult(BaseModel):
    """Gemma 4 回傳的多筆商品 + 解析 metadata。"""

    items: list[PastedItem] = Field(default_factory=list)
    # 若 Gemma 完全無法解析會是 0；handler 端據此決定是否要 fallback
    dropped: int = 0


# ── Prompt ────────────────────────────────────────────────────────────────


PASTE_INSTRUCTION = (
    "You extract second-hand fashion marketplace listings from pasted content. "
    "The paste comes from the user's OWN profile page on Grailed / Depop / "
    "Poshmark / eBay / Vinted / other similar sites. "
    "Return ONLY valid JSON — no prose, no markdown code fences — matching:\n"
    '{"items": [{"title": string, "brand": string|null, '
    '"price": number|null, "currency": string|null, "size": string|null, '
    '"condition": string|null, "image_url": string|null, '
    '"source_url": string|null, "description": string|null}]}\n'
    "Rules:\n"
    "- Extract ONLY real listings you can clearly identify. Drop anything ambiguous.\n"
    '- Do NOT invent items. If nothing parseable, return {"items": []}.\n'
    '- price: number only, strip currency symbols (e.g. "$45" → 45).\n'
    "- Skip any item missing a title.\n"
    "- image_url / source_url: include verbatim only if present in the paste.\n"
    "- condition: one of new / like_new / good / fair / used if inferable.\n"
)


# ── 主函式 ────────────────────────────────────────────────────────────────


# text / html 不必再跑 OCR，用文字路徑；image_base64 走 vision 路徑。
Mode = Literal["text", "html", "image_base64"]

# Paste 的文字上限：避免吃爆 Gemma 4 的 context（128k 名目，但本機 CPU 不會想跑滿）
_TEXT_CAP = 12_000


async def parse_paste(mode: Mode, content: str) -> PasteParseResult:
    """解析 pasted profile，回傳結構化 items 清單。

    【流程】
    - text / html：`gemma_text("paste_parse", ...)` 直接跑
    - image_base64：解 base64 → PIL Image → `gemma_vision("paste_parse", img, ...)`
    """
    if mode == "image_base64":
        # 使用者貼進來的 base64 可能是壞掉的 padding/字元，或解出來根本不是圖片。
        # base64.b64decode 會丟 binascii.Error(ValueError 子類)，PIL 解非圖片
        # bytes 會丟 UnidentifiedImageError/OSError。兩者都是「可復原的壞輸入」，
        # 比照 _coerce_and_validate 的 fail-soft：回空結果而非讓未捕捉例外冒到
        # FastAPI 變成 HTTP 500。
        try:
            img = await asyncio.to_thread(_decode_image_base64, content)
        except (ValueError, OSError) as exc:
            logger.warning(
                "paste image_base64 decode failed: %s", type(exc).__name__
            )
            return PasteParseResult(items=[], dropped=0)
        # 螢幕截圖 → vision 模式，prompt 給 context + instruction
        raw = await gemma_vision(
            "paste_parse",
            img,
            PASTE_INSTRUCTION
            + "\nThe image is a screenshot of the user's profile page.",
            max_tokens=4096,
        )
    else:
        snippet = content[:_TEXT_CAP]
        label = "HTML source" if mode == "html" else "Pasted text"
        prompt = f"{PASTE_INSTRUCTION}\n\n{label}:\n{snippet}"
        raw = await gemma_text("paste_parse", prompt, max_tokens=4096)

    return _coerce_and_validate(raw)


# PIL decompression bomb 上限 — Pillow 預設 178 MP（89 MB uncompressed），攻擊者
# 可用 4KB 壓縮好的 PNG 宣告 10000×10000 把我們的 ml-service 吃到 OOM。
# 40 MP（~6300×6300 = 120 MB RGB）對螢幕截圖足夠，且不會把 Gemma 4 vision
# tower 的 input tensor 撐爆。一旦超過 Pillow 直接 raise DecompressionBombError。
_MAX_PASTE_IMAGE_PIXELS = 40_000_000


def _decode_image_base64(content: str):  # -> PIL.Image.Image
    """解 base64 → PIL.Image。在 to_thread 內執行（PIL open 是 blocking I/O）。

    支援：
      - 純 base64 字串 "iVBORw0KGgo..."
      - data URI "data:image/png;base64,iVBORw0KGgo..."

    【decompression bomb 防線】
      `Image.MAX_IMAGE_PIXELS` 是 class attribute，設在函式內部每次 call 會覆寫
      一次（幾乎 no-op 開銷），避免外部程式碼把它 reset 成 None 的順序依賴問題。
    """
    # 延遲 import PIL 避免提高 ml-service 啟動延遲
    from PIL import Image

    Image.MAX_IMAGE_PIXELS = _MAX_PASTE_IMAGE_PIXELS

    payload = content.split(",", 1)[-1] if content.startswith("data:") else content
    img_bytes = base64.b64decode(payload)
    # Image.open() 是 lazy——真正 decode 在 .convert()；兩段合起來才會觸發
    # DecompressionBombError（若聲明尺寸 × 3 channels 超過 MAX_IMAGE_PIXELS）。
    img = Image.open(io.BytesIO(img_bytes)).convert("RGB")
    return img


def _coerce_and_validate(raw: str) -> PasteParseResult:
    """剝 code fence + parse JSON + Pydantic 驗證 + 丟掉無 title 的 item。"""
    cleaned = _strip_code_fence(raw.strip())
    try:
        data = json.loads(cleaned)
    except json.JSONDecodeError as e:
        logger.warning(
            "paste_parse: JSON decode failed (%s); raw head=%r", e, raw[:300]
        )
        return PasteParseResult(items=[], dropped=0)

    # Gemma 有時會包一個 `{"output": {"items": [...]}}` 的額外層 — 攤平
    if (
        isinstance(data, dict)
        and "items" not in data
        and isinstance(data.get("output"), dict)
    ):
        data = data["output"]

    try:
        parsed = PasteParseResult.model_validate(data)
    except Exception as e:
        logger.warning(
            "paste_parse: Pydantic validation failed (%s); data head=%r",
            e,
            str(data)[:300],
        )
        return PasteParseResult(items=[], dropped=0)

    total = len(parsed.items)
    parsed.items = [i for i in parsed.items if i.title and i.title.strip()]
    parsed.dropped = total - len(parsed.items)
    return parsed