[V10.359] 導入 browse.sh 診斷與色號防錯配

services/browse_sh_tool.py

@@ -0,0 +1,172 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""browse.sh CLI 的可選執行 wrapper。
+
+正式爬蟲仍以既有 Python/API client 為準；browse.sh 只用於動態頁面診斷、
+selector 探勘與 network trace，避免把外部 CLI 變成 scheduler 的硬依賴。
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from dataclasses import dataclass
+from typing import Mapping, Sequence
+
+
+BROWSE_SH_CLI_ENV = "BROWSE_SH_CLI"
+DEFAULT_TIMEOUT_SECONDS = 90
+
+
+@dataclass(frozen=True)
+class BrowseShAvailability:
+    available: bool
+    command: tuple[str, ...]
+    reason: str = ""
+    version: str = ""
+
+    def as_dict(self) -> dict:
+        return {
+            "available": self.available,
+            "command": list(self.command),
+            "reason": self.reason,
+            "version": self.version,
+        }
+
+
+@dataclass(frozen=True)
+class BrowseShResult:
+    ok: bool
+    command: tuple[str, ...]
+    stdout: str = ""
+    stderr: str = ""
+    returncode: int | None = None
+    timed_out: bool = False
+    unavailable_reason: str = ""
+
+    def as_dict(self) -> dict:
+        return {
+            "ok": self.ok,
+            "command": list(self.command),
+            "stdout": self.stdout,
+            "stderr": self.stderr,
+            "returncode": self.returncode,
+            "timed_out": self.timed_out,
+            "unavailable_reason": self.unavailable_reason,
+        }
+
+
+class BrowseShTool:
+    """browse CLI 的最小安全包裝。"""
+
+    def __init__(
+        self,
+        cli_path: str | None = None,
+        timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
+        env: Mapping[str, str] | None = None,
+    ) -> None:
+        self.cli_path = cli_path
+        self.timeout_seconds = timeout_seconds
+        self.env = dict(env or {})
+
+    def resolve_cli_path(self) -> str | None:
+        override = self.cli_path or os.getenv(BROWSE_SH_CLI_ENV)
+        if override:
+            return override
+        return shutil.which("browse")
+
+    def build_command(self, args: Sequence[str]) -> tuple[str, ...]:
+        cli_path = self.resolve_cli_path()
+        if not cli_path:
+            return tuple()
+        return (cli_path, *[str(arg) for arg in args])
+
+    def availability(self) -> BrowseShAvailability:
+        command = self.build_command(("--version",))
+        if not command:
+            return BrowseShAvailability(
+                available=False,
+                command=tuple(),
+                reason="browse CLI 未安裝；請先安裝並確認 PATH 可找到 browse。",
+            )
+        try:
+            completed = subprocess.run(
+                command,
+                capture_output=True,
+                check=False,
+                env={**os.environ, **self.env},
+                text=True,
+                timeout=8,
+            )
+        except FileNotFoundError:
+            return BrowseShAvailability(False, command, "browse CLI 路徑不存在。")
+        except subprocess.TimeoutExpired:
+            return BrowseShAvailability(False, command, "browse --version 執行逾時。")
+        except OSError as exc:
+            return BrowseShAvailability(False, command, f"browse CLI 無法啟動：{exc}")
+
+        stdout = (completed.stdout or "").strip()
+        stderr = (completed.stderr or "").strip()
+        if completed.returncode != 0:
+            reason = stderr or stdout or f"browse --version 回傳 {completed.returncode}"
+            return BrowseShAvailability(False, command, reason)
+        return BrowseShAvailability(True, command, version=stdout or stderr)
+
+    def run(
+        self,
+        args: Sequence[str],
+        timeout_seconds: int | None = None,
+        require_available: bool = True,
+    ) -> BrowseShResult:
+        command = self.build_command(args)
+        if not command:
+            return BrowseShResult(
+                ok=False,
+                command=tuple(),
+                unavailable_reason="browse CLI 未安裝；此工具只會略過，不影響正式爬蟲。",
+            )
+        if require_available:
+            availability = self.availability()
+            if not availability.available:
+                return BrowseShResult(
+                    ok=False,
+                    command=command,
+                    unavailable_reason=availability.reason,
+                )
+
+        try:
+            completed = subprocess.run(
+                command,
+                capture_output=True,
+                check=False,
+                env={**os.environ, **self.env},
+                text=True,
+                timeout=timeout_seconds or self.timeout_seconds,
+            )
+        except subprocess.TimeoutExpired as exc:
+            return BrowseShResult(
+                ok=False,
+                command=command,
+                stdout=exc.stdout or "",
+                stderr=exc.stderr or "",
+                timed_out=True,
+            )
+        except OSError as exc:
+            return BrowseShResult(
+                ok=False,
+                command=command,
+                stderr=str(exc),
+                unavailable_reason=str(exc),
+            )
+
+        return BrowseShResult(
+            ok=completed.returncode == 0,
+            command=command,
+            stdout=completed.stdout or "",
+            stderr=completed.stderr or "",
+            returncode=completed.returncode,
+        )
+
+    def run_skill(self, skill_name: str, *skill_args: str, timeout_seconds: int | None = None) -> BrowseShResult:
+        return self.run((skill_name, *skill_args), timeout_seconds=timeout_seconds)

services/competitor_price_feeder.py

@@ -45,6 +45,7 @@ BATCH_SIZE       = 30    # 每批 DB 寫入筆數
 RATE_DELAY       = float(os.getenv("PCHOME_FEEDER_RATE_DELAY", "1.0"))  # 每次 PChome 請求間隔（秒）
 TTL_HOURS        = 6     # competitor_prices 快取有效期
 REQUEST_TIMEOUT  = float(os.getenv("PCHOME_FEEDER_TIMEOUT", "12"))  # 避免外部搜尋 API 長時間卡住排程
+VARIANT_RECALL_SORTS = ("sale/dc", "new/dc")
 RECOVERABLE_LOW_SCORE_FLOOR = max(MIN_MATCH_SCORE - 0.03, 0.72)
 RECOVERABLE_DIAGNOSTIC_REASONS = {
     "strong_product_line_match",
@@ -188,6 +189,25 @@ def _build_search_keywords(momo_name: str) -> list:
     return _dedupe_terms(primary_terms)
 
 
+def _build_variant_recall_search_plan(momo_name: str, keywords: list[str]) -> list[tuple[str, str | None]]:
+    plan = [(keyword, None) for keyword in (keywords or [])]
+    try:
+        from services.marketplace_product_matcher import parse_product_identity
+
+        identity = parse_product_identity(momo_name)
+    except Exception:
+        return plan
+
+    brand_tokens = {token.lower() for token in getattr(identity, "brand_tokens", set())}
+    if not ({"dashing", "diva"} <= brand_tokens and "美甲片" in getattr(identity, "searchable_name", "")):
+        return plan
+
+    broad_keyword = "dashing diva 時尚潮流美甲片"
+    for sort in VARIANT_RECALL_SORTS:
+        plan.append((broad_keyword, sort))
+    return plan
+
+
 def _format_match_diagnostics(diagnostics) -> str:
     if not diagnostics:
         return ""
@@ -294,8 +314,18 @@ def _search_pchome_candidates(crawler, momo_name: str, keywords: list = None, mo
     candidates = []
     seen_ids = set()
     search_limit = SEARCH_LIMIT * max(1, SEARCH_MAX_PAGES)
-    for keyword in keywords or _build_search_keywords(momo_name):
-        ok, _, products = crawler.search_products(keyword, limit=search_limit, max_pages=SEARCH_MAX_PAGES)
+    active_keywords = keywords or _build_search_keywords(momo_name)
+    search_plan = _build_variant_recall_search_plan(momo_name, active_keywords)
+    for keyword, sort in search_plan:
+        if sort:
+            ok, _, products = crawler.search_products(
+                keyword,
+                limit=search_limit,
+                max_pages=SEARCH_MAX_PAGES,
+                sort=sort,
+            )
+        else:
+            ok, _, products = crawler.search_products(keyword, limit=search_limit, max_pages=SEARCH_MAX_PAGES)
         if not ok or not products:
             continue
         for product in products:

services/marketplace_product_matcher.py

@@ -379,6 +379,10 @@ SEARCH_BROAD_ANCHORS = {
 }
 
 VARIANT_SENSITIVE_KEYWORDS = {
+    "妝前防護乳",
+    "妝前乳",
+    "素顏霜",
+    "粉底",
     "美甲片",
     "眼影盤",
     "唇釉",
@@ -394,6 +398,37 @@ VARIANT_SENSITIVE_KEYWORDS = {
     "遮瑕棒",
 }
 
+VARIANT_OPTION_COLOR_WORDS = {
+    "黑色",
+    "棕色",
+    "咖啡色",
+    "灰色",
+    "白色",
+    "紅色",
+    "粉色",
+    "粉紅",
+    "桃紅",
+    "玫瑰",
+    "玫瑰色",
+    "珊瑚",
+    "珊瑚色",
+    "橘色",
+    "橙色",
+    "裸色",
+    "奶茶色",
+    "豆沙色",
+    "紫色",
+    "薰衣草",
+    "藍色",
+    "綠色",
+    "膚色",
+    "自然色",
+    "明亮色",
+    "透明色",
+    "極光之藍",
+    "月光銀影",
+}
+
 SEARCH_AMBIGUOUS_PRODUCT_TERMS = {
     "保護膜",
     "保護貼",
@@ -1314,6 +1349,11 @@ def score_marketplace_match(
         reasons.append("component_count_conflict")
     if chinese_name_score < 0.16:
         reasons.append("product_line_conflict")
+    shared_anchor = _shared_identity_anchor(left, right)
+    variant_descriptor_conflict = _has_variant_descriptor_conflict(left, right, shared_anchor)
+    variant_option_conflict = _has_explicit_variant_option_conflict(left, right, shared_anchor)
+    if variant_option_conflict:
+        reasons.append("variant_option_conflict")
 
     hard_veto = brand_conflict or spec_conflict
     if bundle_offer_conflict:
@@ -1330,6 +1370,8 @@ def score_marketplace_match(
         hard_veto = True
     if left.product_type and right.product_type and left.product_type != right.product_type:
         hard_veto = True
+    if variant_option_conflict:
+        hard_veto = True
 
     comparison_mode = "exact_identity"
     if _is_unit_comparable_candidate(
@@ -1370,8 +1412,6 @@ def score_marketplace_match(
 
     if token_score >= 0.72 and spec_score >= 0.82 and not brand_conflict:
         score += 0.08
-    shared_anchor = _shared_identity_anchor(left, right)
-    variant_descriptor_conflict = _has_variant_descriptor_conflict(left, right, shared_anchor)
 
     if (
         brand_score >= 0.95
@@ -1528,7 +1568,9 @@ def _clean_search_phrase(value: str) -> str:
     text = normalize_product_text(value)
     for phrase in sorted(SEARCH_NOISE_PHRASES, key=len, reverse=True):
         text = text.replace(phrase.lower(), " ")
+    text = re.sub(r"(?<=\d)\.(?=\d)", "DECIMALPOINT", text)
     text = re.sub(r"[^\w\u4e00-\u9fff]+", " ", text)
+    text = text.replace("DECIMALPOINT", ".").replace("decimalpoint", ".")
     text = " ".join(
         token for token in text.split()
         if token not in SEARCH_NOISE_TOKENS and token not in GENERIC_TOKENS
@@ -1674,6 +1716,39 @@ def _has_variant_descriptor_conflict(left: ProductIdentity, right: ProductIdenti
     return True
 
 
+def _explicit_variant_option_tokens(identity: ProductIdentity) -> set[str]:
+    text = identity.searchable_name
+    options: set[str] = set()
+    for match in re.finditer(r"(?:#|no\.?|色號|號色)\s*([a-z]?\d{1,3}[a-z]?)(?![a-z0-9])", text, re.I):
+        value = re.sub(r"[^a-z0-9]", "", match.group(1).lower())
+        if value:
+            options.add(value)
+    for color_word in VARIANT_OPTION_COLOR_WORDS:
+        if color_word in text:
+            options.add(color_word)
+    return options
+
+
+def _has_explicit_variant_option_conflict(
+    left: ProductIdentity,
+    right: ProductIdentity,
+    shared_anchor: str,
+) -> bool:
+    if not _is_variant_sensitive_identity(left, right, shared_anchor):
+        return False
+    left_options = _explicit_variant_option_tokens(left)
+    right_options = _explicit_variant_option_tokens(right)
+    if not left_options or not right_options:
+        return False
+    if left_options & right_options:
+        return False
+    for left_option in left_options:
+        for right_option in right_options:
+            if left_option in right_option or right_option in left_option:
+                return False
+    return True
+
+
 def _search_core_score(token: str, all_tokens: set[str]) -> tuple[int, int, str]:
     cleaned = _clean_search_phrase(token)
     if not cleaned:
@@ -1790,6 +1865,8 @@ def build_search_terms(name: str, max_terms: int = 3) -> list[str]:
     )
     variant_descriptors = sorted(_variant_descriptors(identity), key=lambda token: (len(token), token))
     variant_primary = variant_descriptors[0] if variant_descriptors else ""
+    variant_options = sorted(_explicit_variant_option_tokens(identity), key=lambda token: (len(token), token))
+    variant_option_part = " ".join(variant_options[:2])
     model_phrases = [
         phrase
         for phrase in core_phrases[1:]
@@ -1801,8 +1878,11 @@ def build_search_terms(name: str, max_terms: int = 3) -> list[str]:
     )
     variant_sensitive = any(keyword in identity.searchable_name for keyword in VARIANT_SENSITIVE_KEYWORDS)
     for value in (
+        " ".join(part for part in (brand_part, core_primary, variant_option_part, spec_part) if part)
+        if variant_sensitive and variant_option_part
+        else "",
         " ".join(part for part in (brand_part, core_primary, variant_primary, spec_part) if part)
-        if variant_sensitive and variant_primary
+        if variant_sensitive and variant_primary and variant_options
         else "",
         " ".join(part for part in (brand_part, modifier_with_primary, spec_part) if part)
         if modifier_with_primary and identity.product_type and identity.product_type in core_primary

services/pchome_crawler.py

@@ -382,6 +382,7 @@ class PChomeCrawler:
         keyword: str,
         limit: int = 50,
         max_pages: Optional[int] = None,
+        sort: str = "rnk/dc",
     ) -> Tuple[bool, str, List[PChomeProduct]]:
         """
         搜尋商品 (使用搜尋 API)
@@ -390,6 +391,7 @@ class PChomeCrawler:
             keyword: 搜尋關鍵字
             limit: 最多回傳數量
             max_pages: 搜尋結果最多掃描頁數；預設依 limit 最多掃到 3 頁
+            sort: 搜尋排序；預設 relevance ranking (`rnk/dc`)
 
         Returns:
             (成功與否, 訊息, 商品資料列表)
@@ -407,7 +409,7 @@ class PChomeCrawler:
                 params = {
                     'q': keyword,
                     'page': page,
-                    'sort': 'rnk/dc',
+                    'sort': sort,
                     'cateid': '24h',
                 }
                 response = self._get_with_retry(search_url, params=params, timeout=self.timeout)
@@ -435,7 +437,7 @@ class PChomeCrawler:
             # 取得詳細資料
             success, message, products = self.fetch_product_details(product_ids[:limit])
             if success:
-                message = f"{message}；搜尋頁數 {pages_scanned}"
+                message = f"{message}；搜尋頁數 {pages_scanned}；排序 {sort}"
             return success, message, products
 
         except requests.RequestException as e: