ewoooc/scripts/ops/report_pixelrag_crawler_integration.py

#!/usr/bin/env python3
"""Report PixelRAG-style crawler integration readiness."""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path


ROOT = Path(__file__).resolve().parents[2]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from services.pixelrag_crawler_integration_service import (  # noqa: E402
    build_pixelrag_crawler_integration_assessment,
    build_pixelrag_ecommerce_expansion_plan,
    build_pixelrag_marketplace_search_manifest,
    build_pixelrag_visual_evidence_manifest,
)


CAPABILITY_FLAGS = {
    "no_playwright": ("playwright_artifact_pipeline", False),
    "ollama_multimodal_ready": ("ollama_multimodal_embedding_ready", True),
    "pgvector_visual_ready": ("pgvector_visual_index_ready", True),
    "faiss_allowed": ("faiss_allowed_in_production", True),
}


def _size_arg(value: str) -> dict[str, int | str]:
    try:
        name, size = value.split("=", 1)
    except ValueError:
        name, size = "", value
    try:
        width_text, height_text = size.lower().split("x", 1)
        width = int(width_text)
        height = int(height_text)
    except (ValueError, TypeError):
        raise argparse.ArgumentTypeError("size must be name=WIDTHxHEIGHT or WIDTHxHEIGHT")
    payload: dict[str, int | str] = {"width": width, "height": height}
    if name:
        payload["name"] = name
    return payload


def main() -> int:
    parser = argparse.ArgumentParser(
        description="輸出 PixelRAG 導入現有爬蟲解決方案的機器可讀評估。"
    )
    parser.add_argument(
        "--platform",
        action="append",
        dest="platforms",
        help="限制目標平台，可重複指定；預設 momo + pchome。",
    )
    parser.add_argument(
        "--capability",
        action="append",
        choices=sorted(CAPABILITY_FLAGS),
        default=[],
        help="覆寫 readiness 假設，用於模擬下一階段條件。",
    )
    parser.add_argument("--manifest-url", help="輸出單一 URL 的視覺證據 manifest。")
    parser.add_argument("--marketplace-keyword", help="依平台與關鍵字輸出多電商視覺搜尋 manifest。")
    parser.add_argument(
        "--ecommerce-expansion-plan",
        action="store_true",
        help="輸出 MOMO/PChome 以外的 PixelRAG 多電商擴充計畫。",
    )
    parser.add_argument("--crawler", default="crawler_diagnostics", help="manifest 來源 crawler 名稱。")
    parser.add_argument(
        "--trigger-reason",
        default="parser_empty_or_low_confidence",
        help="啟動視覺 fallback 的原因。",
    )
    parser.add_argument(
        "--evidence-intent",
        default="recover_visual_offer_evidence",
        help="視覺證據用途。",
    )
    parser.add_argument(
        "--viewport",
        type=_size_arg,
        help="viewport，例如 desktop-1440=1440x950。",
    )
    parser.add_argument(
        "--page-size",
        type=_size_arg,
        help="預估頁面尺寸，例如 page=1440x1900。",
    )
    parser.add_argument(
        "--tile-size",
        type=_size_arg,
        help="tile 尺寸，例如 tile=512x512。",
    )
    args = parser.parse_args()

    if args.ecommerce_expansion_plan:
        payload = build_pixelrag_ecommerce_expansion_plan(
            target_platforms=tuple(args.platforms) if args.platforms else None,
        )
        print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
        return 0 if payload.get("success") else 1

    if args.marketplace_keyword:
        platform = (args.platforms or ["shopee_tw"])[0]
        crawler = (
            "PixelRAGMarketplaceSearch.visual_fallback"
            if args.crawler == "crawler_diagnostics"
            else args.crawler
        )
        trigger_reason = (
            "marketplace_search_visual_scan"
            if args.trigger_reason == "parser_empty_or_low_confidence"
            else args.trigger_reason
        )
        evidence_intent = (
            "collect_public_marketplace_offer_cards"
            if args.evidence_intent == "recover_visual_offer_evidence"
            else args.evidence_intent
        )
        payload = build_pixelrag_marketplace_search_manifest(
            platform=platform,
            keyword=args.marketplace_keyword,
            crawler=crawler,
            trigger_reason=trigger_reason,
            evidence_intent=evidence_intent,
            viewport=args.viewport,
            page_size=args.page_size,
            tile_size=args.tile_size,
        )
        print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
        return 0 if payload.get("success") else 1

    if args.manifest_url:
        platform = (args.platforms or ["momo"])[0]
        payload = build_pixelrag_visual_evidence_manifest(
            url=args.manifest_url,
            platform=platform,
            crawler=args.crawler,
            trigger_reason=args.trigger_reason,
            evidence_intent=args.evidence_intent,
            viewport=args.viewport,
            page_size=args.page_size,
            tile_size=args.tile_size,
        )
        print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
        return 0 if payload.get("success") else 1

    capabilities = {
        key: value
        for flag in args.capability
        for key, value in [CAPABILITY_FLAGS[flag]]
    }
    payload = build_pixelrag_crawler_integration_assessment(
        capabilities=capabilities or None,
        target_platforms=tuple(args.platforms) if args.platforms else None,
    )
    print(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True))
    return 0 if payload.get("success") else 1


if __name__ == "__main__":
    raise SystemExit(main())