wooo/ewoooc
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修復 Wave 0 阻塞與 market intel 入庫

Browse Source
This commit is contained in:
OoO
2026-05-12 22:49:56 +08:00
parent 26669a2c6a
commit fccc80858d
29 changed files with 4888 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
services/cache_service.py
View File

@@ -11,6 +11,7 @@ from services.cache_manager import (
_SALES_PROCESSED_CACHE,
_SALES_OPTIONS_CACHE,
_SALES_ANALYSIS_RESULT_CACHE,
_SALES_CACHE_TTL,
_DASHBOARD_DATA_CACHE,
_DASHBOARD_CACHE_TTL,
clear_sales_cache,
@@ -20,10 +21,9 @@ from services.cache_manager import (
# 台北時區
TAIPEI_TZ = timezone(timedelta(hours=8))
# 快取 TTL 設定
_SALES_CACHE_TTL = 3600 # 業績分析快取: 60 分鐘
# 快取 TTL 設定：sales TTL 以 cache_manager 為單一來源
_SALES_OPTIONS_TTL = 21600 # 選項快取: 6 小時
_SALES_RESULT_TTL = 3600 # 結果快取: 60 分鐘
_SALES_RESULT_TTL = _SALES_CACHE_TTL
# ==========================================
# 成長分析快取

9
services/market_intel/adapters/__init__.py Normal file
View File

@@ -0,0 +1,9 @@
"""市場情報平台 adapter registry。"""
from services.market_intel.adapters.registry import (
get_adapter,
get_adapter_registry,
get_adapter_summaries,
)
__all__ = ["get_adapter", "get_adapter_registry", "get_adapter_summaries"]

93
services/market_intel/adapters/base.py Normal file
View File

@@ -0,0 +1,93 @@
"""市場情報 adapter 基礎類別。
Phase 3 只提供 read-only discovery plan，不發 HTTP request。
"""
from dataclasses import asdict, dataclass
from typing import Iterable, Sequence
@dataclass(frozen=True)
class CampaignSource:
"""公開活動入口的描述，不代表已實際爬取。"""
source_key: str
name: str
url: str
campaign_type: str
notes: str = ""
def to_dict(self):
return asdict(self)
@dataclass(frozen=True)
class AdapterSafetyPolicy:
"""平台爬取安全策略。"""
request_interval_sec: float
timeout_sec: int
max_pages_per_run: int
allow_login: bool = False
allow_database_write: bool = False
allow_scheduler_attach: bool = False
def to_dict(self):
return asdict(self)
class MarketIntelAdapter:
"""市場情報平台 adapter base class。"""
platform_code = ""
platform_name = ""
base_url = ""
campaign_url_keywords = ()
campaign_text_keywords = ()
safety_policy = AdapterSafetyPolicy(
request_interval_sec=2.0,
timeout_sec=20,
max_pages_per_run=5,
)
def campaign_sources(self) -> Sequence[CampaignSource]:
return ()
def summary(self):
return {
"platform_code": self.platform_code,
"platform_name": self.platform_name,
"base_url": self.base_url,
"source_count": len(self.campaign_sources()),
"safety_policy": self.safety_policy.to_dict(),
"phase": "read_only_adapter_skeleton",
}
def build_discovery_plan(self):
"""建立 discovery plan，不發 request、不寫 DB。"""
return {
**self.summary(),
"network_request_allowed": False,
"database_write_allowed": False,
"scheduler_attach_allowed": False,
"sources": [source.to_dict() for source in self.campaign_sources()],
}
def discover_campaigns(self, *, dry_run=True) -> Iterable[CampaignSource]:
"""Phase 3 僅允許 dry-run 回傳入口描述。"""
if not dry_run:
raise RuntimeError("市場情報 adapter 尚未允許正式 discovery")
return self.campaign_sources()
def score_campaign_link(self, href, text):
"""平台別活動連結加權，只用於診斷排序。"""
url_text = (href or "").lower()
link_text = (text or "").lower()
score = 0
for keyword in self.campaign_url_keywords:
if str(keyword).lower() in url_text:
score += 4
for keyword in self.campaign_text_keywords:
if str(keyword).lower() in link_text:
score += 3
return score

54
services/market_intel/adapters/coupang_adapter.py Normal file
View File

@@ -0,0 +1,54 @@
"""Coupang 市場活動 read-only adapter skeleton。"""
from services.market_intel.adapters.base import (
AdapterSafetyPolicy,
CampaignSource,
MarketIntelAdapter,
)
class CoupangMarketAdapter(MarketIntelAdapter):
platform_code = "coupang"
platform_name = "Coupang 酷澎"
base_url = "https://www.tw.coupang.com"
campaign_url_keywords = (
"tw.coupang.com",
"/np/",
"coupangglobal",
"promotion",
"event",
"sale",
)
campaign_text_keywords = (
"coupang",
"酷澎",
"活動",
"優惠",
"折扣",
"特價",
"今日精選",
"火箭",
)
safety_policy = AdapterSafetyPolicy(
request_interval_sec=2.5,
timeout_sec=25,
max_pages_per_run=5,
)
def campaign_sources(self):
return (
CampaignSource(
source_key="coupang_home",
name="Coupang 酷澎首頁入口",
url="https://www.tw.coupang.com/",
campaign_type="homepage_campaign",
notes="只保存官方公開首頁入口；正式抓取需另開 feature flag。",
),
CampaignSource(
source_key="coupang_global",
name="Coupang 火箭跨境入口",
url="https://www.tw.coupang.com/np/coupangglobal",
campaign_type="cross_border_sale",
notes="先作為跨境/火箭相關活動 discovery 起點，不代表已建立活動。",
),
)

54
services/market_intel/adapters/momo_adapter.py Normal file
View File

@@ -0,0 +1,54 @@
"""MOMO 市場活動 read-only adapter skeleton。"""
from services.market_intel.adapters.base import (
AdapterSafetyPolicy,
CampaignSource,
MarketIntelAdapter,
)
class MomoMarketAdapter(MarketIntelAdapter):
platform_code = "momo"
platform_name = "MOMO 購物網"
base_url = "https://www.momoshop.com.tw"
campaign_url_keywords = (
"edm",
"cmmedm",
"lgrpcategory",
"category",
"promo",
"event",
)
campaign_text_keywords = (
"momo",
"活動",
"優惠",
"限時",
"限時搶購",
"品牌日",
"滿額",
"折價券",
)
safety_policy = AdapterSafetyPolicy(
request_interval_sec=2.0,
timeout_sec=25,
max_pages_per_run=8,
)
def campaign_sources(self):
return (
CampaignSource(
source_key="momo_edm",
name="MOMO EDM 活動入口",
url="https://www.momoshop.com.tw/edm/cmmedm.jsp",
campaign_type="festival",
notes="先接既有 EDM / festival 爬蟲語意，正式抓取需另開 feature flag。",
),
CampaignSource(
source_key="momo_flash_sale",
name="MOMO 限時搶購入口",
url="https://www.momoshop.com.tw/category/LgrpCategory.jsp?l_code=2140000000",
campaign_type="flash_sale",
notes="只保存公開入口描述；Phase 3 不發 request。",
),
)

54
services/market_intel/adapters/pchome_adapter.py Normal file
View File

@@ -0,0 +1,54 @@
"""PChome 市場活動 read-only adapter skeleton。"""
from services.market_intel.adapters.base import (
AdapterSafetyPolicy,
CampaignSource,
MarketIntelAdapter,
)
class PChomeMarketAdapter(MarketIntelAdapter):
platform_code = "pchome"
platform_name = "PChome 24h"
base_url = "https://24h.pchome.com.tw"
campaign_url_keywords = (
"24h.pchome.com.tw",
"/activity",
"/campaign",
"/region/",
"promo",
"event",
)
campaign_text_keywords = (
"pchome",
"24h",
"活動",
"優惠",
"限時",
"品牌日",
"特價",
"折扣",
)
safety_policy = AdapterSafetyPolicy(
request_interval_sec=1.5,
timeout_sec=20,
max_pages_per_run=8,
)
def campaign_sources(self):
return (
CampaignSource(
source_key="pchome_home",
name="PChome 24h 首頁活動入口",
url="https://24h.pchome.com.tw/",
campaign_type="homepage_campaign",
notes="可作為活動 banner / 分類入口 discovery 起點。",
),
CampaignSource(
source_key="pchome_region_beauty",
name="PChome 美妝保養館別入口",
url="https://24h.pchome.com.tw/region/DDAB",
campaign_type="category_sale",
notes="延續既有 PChome crawler 的 region page 低成本入口。",
),
)

29
services/market_intel/adapters/registry.py Normal file
View File

@@ -0,0 +1,29 @@
"""市場情報 adapter 註冊表。"""
from services.market_intel.adapters.coupang_adapter import CoupangMarketAdapter
from services.market_intel.adapters.momo_adapter import MomoMarketAdapter
from services.market_intel.adapters.pchome_adapter import PChomeMarketAdapter
from services.market_intel.adapters.shopee_adapter import ShopeeMarketAdapter
ADAPTER_CLASSES = {
"momo": MomoMarketAdapter,
"pchome": PChomeMarketAdapter,
"coupang": CoupangMarketAdapter,
"shopee": ShopeeMarketAdapter,
}
def get_adapter(platform_code):
adapter_class = ADAPTER_CLASSES.get((platform_code or "").lower())
if not adapter_class:
return None
return adapter_class()
def get_adapter_registry():
return {code: adapter_class() for code, adapter_class in ADAPTER_CLASSES.items()}
def get_adapter_summaries():
return [adapter.summary() for adapter in get_adapter_registry().values()]

57
services/market_intel/adapters/shopee_adapter.py Normal file
View File

@@ -0,0 +1,57 @@
"""Shopee 市場活動 read-only adapter skeleton。"""
from services.market_intel.adapters.base import (
AdapterSafetyPolicy,
CampaignSource,
MarketIntelAdapter,
)
class ShopeeMarketAdapter(MarketIntelAdapter):
platform_code = "shopee"
platform_name = "Shopee 蝦皮購物"
base_url = "https://shopee.tw"
campaign_url_keywords = (
"shopee.tw",
"/mall",
"campaign",
"event",
"flash_sale",
"promotion",
"sale",
)
campaign_text_keywords = (
"shopee",
"蝦皮",
"商城",
"活動",
"優惠",
"折扣",
"特價",
"限時",
"品牌",
"免運",
)
safety_policy = AdapterSafetyPolicy(
request_interval_sec=4.0,
timeout_sec=25,
max_pages_per_run=3,
)
def campaign_sources(self):
return (
CampaignSource(
source_key="shopee_home",
name="Shopee 蝦皮首頁入口",
url="https://shopee.tw/",
campaign_type="homepage_campaign",
notes="只保存公開首頁入口；不得登入、碰會員券、購物車或反爬繞過。",
),
CampaignSource(
source_key="shopee_mall",
name="Shopee Mall 商城入口",
url="https://shopee.tw/mall",
campaign_type="brand_mall",
notes="只作為商城/品牌活動 discovery 起點；正式抓取需另開 feature flag。",
),
)

81
services/market_intel/candidate_preview.py Normal file
View File

@@ -0,0 +1,81 @@
"""市場情報候選連結 preview 聚合。
只整理本次 diagnostics 結果供人工審核，不建立 campaign/product，不寫 DB。
"""
BAND_RANK = {
"high": 3,
"medium": 2,
"low": 1,
}
def _band_allowed(candidate, min_band):
if not min_band or min_band == "all":
return True
candidate_rank = BAND_RANK.get(candidate.get("confidence_band"), 0)
min_rank = BAND_RANK.get(min_band, 0)
return candidate_rank >= min_rank
def build_candidate_preview_from_discovery(discovery_result, *, min_band="all", limit=50):
"""把 manual discovery diagnostics 整理成人工審核 preview。"""
candidates = []
run_statuses = []
for run in discovery_result.get("runs", []):
platform_code = run.get("platform_code")
run_statuses.append({
"platform_code": platform_code,
"status": run.get("status"),
"sources_planned": run.get("sources_planned", 0),
"sources_fetched": run.get("sources_fetched", 0),
"errors": run.get("errors", 0),
})
for source_result in run.get("results", []):
diagnostics = source_result.get("diagnostics") or {}
for candidate in diagnostics.get("campaign_link_candidates", []):
if not _band_allowed(candidate, min_band):
continue
candidates.append({
"platform_code": platform_code,
"source_key": source_result.get("source_key"),
"source_name": source_result.get("name"),
"source_url": source_result.get("url"),
"source_status": source_result.get("status"),
"page_title": diagnostics.get("title"),
"page_hash": diagnostics.get("page_hash"),
"href": candidate.get("href"),
"text": candidate.get("text"),
"is_same_host": candidate.get("is_same_host"),
"score": candidate.get("score", 0),
"generic_score": candidate.get("generic_score", 0),
"platform_score": candidate.get("platform_score", 0),
"confidence_band": candidate.get("confidence_band"),
"confidence_reason": candidate.get("confidence_reason"),
})
candidates = sorted(
candidates,
key=lambda item: (
BAND_RANK.get(item.get("confidence_band"), 0),
item.get("score", 0),
item.get("platform_score", 0),
),
reverse=True,
)
return {
"platform_code": discovery_result.get("platform_code", "all"),
"fetch_requested": bool(discovery_result.get("fetch_requested")),
"manual_fetch_allowed": bool(discovery_result.get("manual_fetch_allowed")),
"min_band": min_band or "all",
"limit": limit,
"candidate_count": len(candidates),
"candidates": candidates[:limit],
"run_statuses": run_statuses,
"database_write_allowed": False,
"scheduler_attached": False,
}

215
services/market_intel/discovery_runner.py Normal file
View File

@@ -0,0 +1,215 @@
"""市場情報手動 discovery dry-run runner。
預設不發 HTTP request；即使手動 fetch，也只做公開頁面探測與摘要，不寫 DB。
"""
import hashlib
import time
from dataclasses import asdict, dataclass
from datetime import datetime, timedelta, timezone
from typing import Callable, Optional
from uuid import uuid4
import requests
from services.market_intel.html_diagnostics import parse_html_diagnostics
TAIPEI_TZ = timezone(timedelta(hours=8))
DEFAULT_HEADERS = {
"User-Agent": "EwoooC-MarketIntel-DryRun/1.0 (+https://mo.wooo.work)",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "zh-TW,zh;q=0.9,en;q=0.8",
}
@dataclass(frozen=True)
class DiscoverySourceResult:
"""單一活動入口的 dry-run 結果。"""
source_key: str
name: str
url: str
campaign_type: str
status: str
network_requested: bool
network_executed: bool
status_code: Optional[int] = None
content_length: int = 0
page_hash: Optional[str] = None
title: Optional[str] = None
diagnostics: Optional[dict] = None
error_message: Optional[str] = None
def to_dict(self):
return asdict(self)
@dataclass(frozen=True)
class ManualDiscoveryRunResult:
"""手動 discovery dry-run 的整體結果。"""
batch_id: str
platform_code: str
started_at: str
finished_at: str
status: str
fetch_requested: bool
network_allowed: bool
database_write_allowed: bool
scheduler_attached: bool
sources_planned: int
sources_fetched: int
errors: int
results: list
error_message: Optional[str] = None
def to_dict(self):
return asdict(self)
def _now_iso():
return datetime.now(TAIPEI_TZ).replace(tzinfo=None).isoformat()
class ManualDiscoveryRunner:
"""手動 discovery runner，透過 feature gate 控制是否允許網路探測。"""
def __init__(self, *, runtime_status, http_get: Optional[Callable] = None):
self.runtime_status = runtime_status
self.http_get = http_get or requests.get
def _network_allowed(self):
return bool(self.runtime_status.enabled and self.runtime_status.crawler_enabled)
def run(self, adapter, *, fetch=False):
started_at = _now_iso()
sources = list(adapter.campaign_sources())
network_allowed = self._network_allowed()
if fetch and not network_allowed:
return ManualDiscoveryRunResult(
batch_id=f"market-manual-{uuid4().hex[:12]}",
platform_code=adapter.platform_code,
started_at=started_at,
finished_at=_now_iso(),
status="blocked",
fetch_requested=True,
network_allowed=False,
database_write_allowed=False,
scheduler_attached=False,
sources_planned=len(sources),
sources_fetched=0,
errors=0,
results=[
self._source_result(source, "blocked", True, False).to_dict()
for source in sources
],
error_message="MARKET_INTEL_ENABLED 與 MARKET_INTEL_CRAWLER_ENABLED 必須同時開啟才允許手動 fetch",
)
capped_sources = sources[:adapter.safety_policy.max_pages_per_run]
results = []
errors = 0
fetched = 0
last_request_at = 0.0
for source in capped_sources:
if not fetch:
results.append(self._source_result(source, "planned", False, False).to_dict())
continue
elapsed = time.time() - last_request_at
if last_request_at and elapsed < adapter.safety_policy.request_interval_sec:
time.sleep(adapter.safety_policy.request_interval_sec - elapsed)
try:
response = self.http_get(
source.url,
headers=DEFAULT_HEADERS,
timeout=adapter.safety_policy.timeout_sec,
)
last_request_at = time.time()
text = response.text or ""
fetched += 1
results.append(
self._source_result(
source,
"fetched",
True,
True,
status_code=getattr(response, "status_code", None),
content=text,
score_link=adapter.score_campaign_link,
).to_dict()
)
except Exception as exc:
last_request_at = time.time()
errors += 1
results.append(
self._source_result(
source,
"failed",
True,
True,
error_message=str(exc),
).to_dict()
)
status = "planned"
if fetch:
status = "success" if errors == 0 else "partial_failed"
return ManualDiscoveryRunResult(
batch_id=f"market-manual-{uuid4().hex[:12]}",
platform_code=adapter.platform_code,
started_at=started_at,
finished_at=_now_iso(),
status=status,
fetch_requested=bool(fetch),
network_allowed=network_allowed,
database_write_allowed=False,
scheduler_attached=False,
sources_planned=len(sources),
sources_fetched=fetched,
errors=errors,
results=results,
)
def _source_result(
self,
source,
status,
network_requested,
network_executed,
*,
status_code=None,
content=None,
score_link=None,
error_message=None,
):
content = content or ""
diagnostics = (
parse_html_diagnostics(
content,
base_url=source.url,
score_link=score_link,
).to_dict()
if content
else None
)
return DiscoverySourceResult(
source_key=source.source_key,
name=source.name,
url=source.url,
campaign_type=source.campaign_type,
status=status,
network_requested=network_requested,
network_executed=network_executed,
status_code=status_code,
content_length=len(content),
page_hash=hashlib.sha256(content.encode("utf-8", errors="ignore")).hexdigest() if content else None,
title=diagnostics.get("title") if diagnostics else None,
diagnostics=diagnostics,
error_message=error_message,
)

172
services/market_intel/html_diagnostics.py Normal file
View File

@@ -0,0 +1,172 @@
"""市場情報 HTML 診斷解析工具。
只萃取頁面標題、連結候選與內容指紋，不建立正式 campaign/product。
"""
import hashlib
import re
from dataclasses import asdict, dataclass
from html.parser import HTMLParser
from urllib.parse import urljoin, urlparse
@dataclass(frozen=True)
class LinkCandidate:
"""活動頁中可疑連結候選。"""
href: str
text: str
is_same_host: bool
score: int
generic_score: int
platform_score: int
confidence_band: str
confidence_reason: str
def to_dict(self):
return asdict(self)
@dataclass(frozen=True)
class HtmlDiagnostics:
"""HTML 診斷摘要。"""
content_length: int
page_hash: str
title: str
link_count: int
same_host_link_count: int
campaign_link_candidates: list
def to_dict(self):
return asdict(self)
class _DiagnosticHtmlParser(HTMLParser):
def __init__(self):
super().__init__(convert_charrefs=True)
self.title_parts = []
self.links = []
self._in_title = False
self._active_href = None
self._active_text = []
def handle_starttag(self, tag, attrs):
attrs_map = dict(attrs or [])
if tag.lower() == "title":
self._in_title = True
if tag.lower() == "a":
self._active_href = attrs_map.get("href")
self._active_text = []
def handle_data(self, data):
if self._in_title:
self.title_parts.append(data)
if self._active_href is not None:
self._active_text.append(data)
def handle_endtag(self, tag):
tag = tag.lower()
if tag == "title":
self._in_title = False
if tag == "a" and self._active_href is not None:
self.links.append((self._active_href, " ".join(self._active_text)))
self._active_href = None
self._active_text = []
def _clean_text(value, limit=160):
text = re.sub(r"\s+", " ", value or "").strip()
return text[:limit]
def _score_link(href, text):
haystack = f"{href} {text}".lower()
score = 0
for keyword in ("edm", "event", "promo", "campaign", "sale", "activity"):
if keyword in haystack:
score += 2
for keyword in ("活動", "優惠", "折扣", "檔期", "品牌日", "限時", "促銷"):
if keyword in haystack:
score += 3
return score
def _confidence_band(score, *, is_same_host, platform_score, generic_score):
"""把診斷分數轉成人工審核用信心帶。"""
reasons = []
reasons.append("same_host" if is_same_host else "external_host")
if platform_score > 0:
reasons.append(f"platform_score={platform_score}")
if generic_score > 0:
reasons.append(f"generic_score={generic_score}")
if score >= 12 and is_same_host:
return "high", ", ".join(reasons)
if score >= 6:
return "medium", ", ".join(reasons)
return "low", ", ".join(reasons)
def parse_html_diagnostics(html, base_url="", candidate_limit=12, score_link=None):
"""解析 HTML 診斷資訊，不做商業資料入庫。"""
html = html or ""
parser = _DiagnosticHtmlParser()
parser.feed(html)
base_host = urlparse(base_url or "").netloc
candidates = []
same_host_count = 0
for raw_href, raw_text in parser.links:
if not raw_href:
continue
href = urljoin(base_url, raw_href)
parsed = urlparse(href)
if parsed.scheme not in {"http", "https"}:
continue
is_same_host = bool(base_host and parsed.netloc == base_host)
if is_same_host:
same_host_count += 1
text = _clean_text(raw_text)
generic_score = _score_link(href, text)
platform_score = int(score_link(href, text)) if score_link else 0
score = generic_score + platform_score
if score <= 0:
continue
confidence_band, confidence_reason = _confidence_band(
score,
is_same_host=is_same_host,
platform_score=platform_score,
generic_score=generic_score,
)
candidates.append(
LinkCandidate(
href=href,
text=text,
is_same_host=is_same_host,
score=score,
generic_score=generic_score,
platform_score=platform_score,
confidence_band=confidence_band,
confidence_reason=confidence_reason,
)
)
band_rank = {"high": 3, "medium": 2, "low": 1}
candidates = sorted(
candidates,
key=lambda item: (band_rank.get(item.confidence_band, 0), item.score),
reverse=True,
)[:candidate_limit]
page_hash = hashlib.sha256(html.encode("utf-8", errors="ignore")).hexdigest() if html else ""
title = _clean_text(" ".join(parser.title_parts), limit=200)
return HtmlDiagnostics(
content_length=len(html),
page_hash=page_hash,
title=title,
link_count=len(parser.links),
same_host_link_count=same_host_count,
campaign_link_candidates=[candidate.to_dict() for candidate in candidates],
)

381
services/market_intel/migration_blueprint.py Normal file
View File

@@ -0,0 +1,381 @@
"""市場情報 migration 與正式 seed writer 命令草案。
本模組只產生可審核的 SQL / command preview，不連線 DB、不執行命令。
"""
from pathlib import Path
MIGRATION_NUMBER = "032"
MIGRATION_FILENAME = "migrations/032_market_intel_core_schema.sql"
SEED_WRITER_SCRIPT = "scripts/market_intel_seed_writer.py"
FORWARD_SQL = """
-- =============================================================================
-- Migration 032: market_intel core schema
-- MOMO PRO - Cross-platform market campaign intelligence
-- 2026-05-07 Taipei
-- =============================================================================
-- Notes:
-- Creates the ADR-035 market_* schema. This migration is additive only:
-- it creates tables, indexes, and grants. It does not drop or alter existing
-- sales tables, and it does not touch the momo-db container lifecycle.
-- =============================================================================
CREATE TABLE IF NOT EXISTS market_platforms (
id BIGSERIAL PRIMARY KEY,
code VARCHAR(50) NOT NULL UNIQUE,
name VARCHAR(120) NOT NULL,
base_url VARCHAR(500),
enabled BOOLEAN NOT NULL DEFAULT FALSE,
crawl_policy_json TEXT,
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
updated_at TIMESTAMP NOT NULL DEFAULT NOW()
);
CREATE INDEX IF NOT EXISTS idx_market_platforms_code
ON market_platforms (code);
CREATE TABLE IF NOT EXISTS market_campaigns (
id BIGSERIAL PRIMARY KEY,
platform_code VARCHAR(50) NOT NULL REFERENCES market_platforms(code),
campaign_key VARCHAR(200) NOT NULL,
campaign_name VARCHAR(500) NOT NULL,
campaign_type VARCHAR(80),
campaign_url TEXT,
start_at TIMESTAMP,
end_at TIMESTAMP,
status VARCHAR(30) NOT NULL DEFAULT 'unknown',
discovered_at TIMESTAMP NOT NULL DEFAULT NOW(),
last_seen_at TIMESTAMP NOT NULL DEFAULT NOW(),
metadata_json TEXT,
CONSTRAINT uq_market_campaign_platform_key
UNIQUE (platform_code, campaign_key)
);
CREATE INDEX IF NOT EXISTS idx_market_campaigns_platform_code
ON market_campaigns (platform_code);
CREATE INDEX IF NOT EXISTS idx_market_campaigns_campaign_type
ON market_campaigns (campaign_type);
CREATE INDEX IF NOT EXISTS idx_market_campaigns_status
ON market_campaigns (status);
CREATE INDEX IF NOT EXISTS idx_market_campaign_status_time
ON market_campaigns (status, start_at, end_at);
CREATE TABLE IF NOT EXISTS market_campaign_snapshots (
id BIGSERIAL PRIMARY KEY,
campaign_id BIGINT NOT NULL REFERENCES market_campaigns(id),
batch_id VARCHAR(80) NOT NULL,
crawled_at TIMESTAMP NOT NULL DEFAULT NOW(),
title VARCHAR(500),
hero_text TEXT,
coupon_text TEXT,
raw_discount_text TEXT,
page_hash VARCHAR(128),
raw_snapshot_path TEXT,
status VARCHAR(30) NOT NULL DEFAULT 'success',
error_message TEXT,
metadata_json TEXT
);
CREATE INDEX IF NOT EXISTS idx_market_campaign_snapshots_campaign_id
ON market_campaign_snapshots (campaign_id);
CREATE INDEX IF NOT EXISTS idx_market_campaign_snapshots_batch_id
ON market_campaign_snapshots (batch_id);
CREATE INDEX IF NOT EXISTS idx_market_campaign_snapshots_crawled_at
ON market_campaign_snapshots (crawled_at);
CREATE INDEX IF NOT EXISTS idx_market_campaign_snapshots_page_hash
ON market_campaign_snapshots (page_hash);
CREATE INDEX IF NOT EXISTS idx_market_campaign_snapshots_status
ON market_campaign_snapshots (status);
CREATE INDEX IF NOT EXISTS idx_market_campaign_snapshot_campaign_time
ON market_campaign_snapshots (campaign_id, crawled_at);
CREATE TABLE IF NOT EXISTS market_campaign_products (
id BIGSERIAL PRIMARY KEY,
campaign_id BIGINT NOT NULL REFERENCES market_campaigns(id),
platform_code VARCHAR(50) NOT NULL,
platform_product_id VARCHAR(200) NOT NULL,
product_url TEXT,
name VARCHAR(500) NOT NULL,
brand VARCHAR(200),
image_url TEXT,
category_text VARCHAR(300),
price DOUBLE PRECISION,
original_price DOUBLE PRECISION,
discount_text VARCHAR(200),
discount_rate DOUBLE PRECISION,
coupon_text TEXT,
stock_text VARCHAR(200),
sold_count INTEGER,
rating DOUBLE PRECISION,
review_count INTEGER,
rank_position INTEGER,
is_active BOOLEAN NOT NULL DEFAULT TRUE,
first_seen_at TIMESTAMP NOT NULL DEFAULT NOW(),
last_seen_at TIMESTAMP NOT NULL DEFAULT NOW(),
metadata_json TEXT,
CONSTRAINT uq_market_campaign_product
UNIQUE (campaign_id, platform_code, platform_product_id)
);
CREATE INDEX IF NOT EXISTS idx_market_campaign_products_campaign_id
ON market_campaign_products (campaign_id);
CREATE INDEX IF NOT EXISTS idx_market_campaign_products_platform_code
ON market_campaign_products (platform_code);
CREATE INDEX IF NOT EXISTS idx_market_campaign_products_platform_product_id
ON market_campaign_products (platform_product_id);
CREATE INDEX IF NOT EXISTS idx_market_campaign_products_brand
ON market_campaign_products (brand);
CREATE INDEX IF NOT EXISTS idx_market_campaign_products_category_text
ON market_campaign_products (category_text);
CREATE INDEX IF NOT EXISTS idx_market_campaign_products_is_active
ON market_campaign_products (is_active);
CREATE INDEX IF NOT EXISTS idx_market_campaign_products_last_seen_at
ON market_campaign_products (last_seen_at);
CREATE INDEX IF NOT EXISTS idx_market_product_platform_seen
ON market_campaign_products (platform_code, last_seen_at);
CREATE INDEX IF NOT EXISTS idx_market_product_discount
ON market_campaign_products (discount_rate, price);
CREATE TABLE IF NOT EXISTS market_product_price_history (
id BIGSERIAL PRIMARY KEY,
market_product_id BIGINT NOT NULL REFERENCES market_campaign_products(id),
campaign_id BIGINT NOT NULL REFERENCES market_campaigns(id),
platform_code VARCHAR(50) NOT NULL,
platform_product_id VARCHAR(200) NOT NULL,
price DOUBLE PRECISION,
original_price DOUBLE PRECISION,
discount_rate DOUBLE PRECISION,
stock_text VARCHAR(200),
sold_count INTEGER,
rank_position INTEGER,
crawled_at TIMESTAMP NOT NULL DEFAULT NOW(),
batch_id VARCHAR(80) NOT NULL,
metadata_json TEXT
);
CREATE INDEX IF NOT EXISTS idx_market_product_price_history_market_product_id
ON market_product_price_history (market_product_id);
CREATE INDEX IF NOT EXISTS idx_market_product_price_history_campaign_id
ON market_product_price_history (campaign_id);
CREATE INDEX IF NOT EXISTS idx_market_product_price_history_platform_code
ON market_product_price_history (platform_code);
CREATE INDEX IF NOT EXISTS idx_market_product_price_history_platform_product_id
ON market_product_price_history (platform_product_id);
CREATE INDEX IF NOT EXISTS idx_market_product_price_history_crawled_at
ON market_product_price_history (crawled_at);
CREATE INDEX IF NOT EXISTS idx_market_product_price_history_batch_id
ON market_product_price_history (batch_id);
CREATE INDEX IF NOT EXISTS idx_market_price_platform_time
ON market_product_price_history (platform_code, platform_product_id, crawled_at);
CREATE INDEX IF NOT EXISTS idx_market_price_campaign_time
ON market_product_price_history (campaign_id, crawled_at);
CREATE TABLE IF NOT EXISTS market_product_matches (
id BIGSERIAL PRIMARY KEY,
market_product_id BIGINT NOT NULL REFERENCES market_campaign_products(id),
momo_product_id INTEGER REFERENCES products(id),
momo_i_code VARCHAR(50),
match_score DOUBLE PRECISION NOT NULL DEFAULT 0.0,
match_status VARCHAR(30) NOT NULL DEFAULT 'needs_review',
match_reason_json TEXT,
created_at TIMESTAMP NOT NULL DEFAULT NOW(),
reviewed_at TIMESTAMP,
reviewed_by VARCHAR(120),
CONSTRAINT uq_market_product_momo_match
UNIQUE (market_product_id, momo_i_code)
);
CREATE INDEX IF NOT EXISTS idx_market_product_matches_market_product_id
ON market_product_matches (market_product_id);
CREATE INDEX IF NOT EXISTS idx_market_product_matches_momo_product_id
ON market_product_matches (momo_product_id);
CREATE INDEX IF NOT EXISTS idx_market_product_matches_momo_i_code
ON market_product_matches (momo_i_code);
CREATE INDEX IF NOT EXISTS idx_market_product_matches_match_status
ON market_product_matches (match_status);
CREATE INDEX IF NOT EXISTS idx_market_match_status_score
ON market_product_matches (match_status, match_score);
CREATE TABLE IF NOT EXISTS market_crawler_runs (
id BIGSERIAL PRIMARY KEY,
platform_code VARCHAR(50),
crawler_name VARCHAR(120) NOT NULL,
campaign_id BIGINT REFERENCES market_campaigns(id),
batch_id VARCHAR(80) NOT NULL,
started_at TIMESTAMP NOT NULL DEFAULT NOW(),
finished_at TIMESTAMP,
status VARCHAR(30) NOT NULL DEFAULT 'started',
dry_run BOOLEAN NOT NULL DEFAULT TRUE,
pages_found INTEGER NOT NULL DEFAULT 0,
products_found INTEGER NOT NULL DEFAULT 0,
products_changed INTEGER NOT NULL DEFAULT 0,
error_count INTEGER NOT NULL DEFAULT 0,
error_message TEXT,
metadata_json TEXT
);
CREATE INDEX IF NOT EXISTS idx_market_crawler_runs_platform_code
ON market_crawler_runs (platform_code);
CREATE INDEX IF NOT EXISTS idx_market_crawler_runs_crawler_name
ON market_crawler_runs (crawler_name);
CREATE INDEX IF NOT EXISTS idx_market_crawler_runs_campaign_id
ON market_crawler_runs (campaign_id);
CREATE INDEX IF NOT EXISTS idx_market_crawler_runs_batch_id
ON market_crawler_runs (batch_id);
CREATE INDEX IF NOT EXISTS idx_market_crawler_runs_started_at
ON market_crawler_runs (started_at);
CREATE INDEX IF NOT EXISTS idx_market_crawler_runs_status
ON market_crawler_runs (status);
CREATE INDEX IF NOT EXISTS idx_market_crawler_run_platform_time
ON market_crawler_runs (platform_code, started_at);
CREATE INDEX IF NOT EXISTS idx_market_crawler_run_status_time
ON market_crawler_runs (status, started_at);
GRANT ALL PRIVILEGES ON market_platforms TO momo;
GRANT ALL PRIVILEGES ON market_campaigns TO momo;
GRANT ALL PRIVILEGES ON market_campaign_snapshots TO momo;
GRANT ALL PRIVILEGES ON market_campaign_products TO momo;
GRANT ALL PRIVILEGES ON market_product_price_history TO momo;
GRANT ALL PRIVILEGES ON market_product_matches TO momo;
GRANT ALL PRIVILEGES ON market_crawler_runs TO momo;
GRANT USAGE, SELECT ON SEQUENCE market_platforms_id_seq TO momo;
GRANT USAGE, SELECT ON SEQUENCE market_campaigns_id_seq TO momo;
GRANT USAGE, SELECT ON SEQUENCE market_campaign_snapshots_id_seq TO momo;
GRANT USAGE, SELECT ON SEQUENCE market_campaign_products_id_seq TO momo;
GRANT USAGE, SELECT ON SEQUENCE market_product_price_history_id_seq TO momo;
GRANT USAGE, SELECT ON SEQUENCE market_product_matches_id_seq TO momo;
GRANT USAGE, SELECT ON SEQUENCE market_crawler_runs_id_seq TO momo;
""".strip()
ROLLBACK_SQL = """
-- Manual rollback draft only. Do not run without operator approval and backup verification.
DROP TABLE IF EXISTS market_crawler_runs;
DROP TABLE IF EXISTS market_product_matches;
DROP TABLE IF EXISTS market_product_price_history;
DROP TABLE IF EXISTS market_campaign_products;
DROP TABLE IF EXISTS market_campaign_snapshots;
DROP TABLE IF EXISTS market_campaigns;
DROP TABLE IF EXISTS market_platforms;
""".strip()
def _statement_count(sql_text):
return len([part for part in sql_text.split(";") if part.strip()])
def _contains_destructive_forward_sql(sql_text):
lowered = sql_text.lower()
destructive_markers = (
"drop table",
"truncate ",
"delete from",
"alter table products",
"alter table daily_sales",
"alter table monthly_sales",
)
return any(marker in lowered for marker in destructive_markers)
def build_migration_blueprint(expected_tables):
"""建立 market_intel migration 與真寫入命令草案；不執行任何 SQL。"""
expected_tables = list(expected_tables)
migration_path = Path(MIGRATION_FILENAME)
seed_writer_path = Path(SEED_WRITER_SCRIPT)
migration_file_exists = migration_path.exists()
seed_writer_script_exists = seed_writer_path.exists()
migration_file_text = (
migration_path.read_text(encoding="utf-8").strip()
if migration_file_exists
else ""
)
migration_file_matches_blueprint = (
migration_file_exists and migration_file_text == FORWARD_SQL
)
forward_has_destructive_sql = _contains_destructive_forward_sql(FORWARD_SQL)
blocked_reasons = [
"migration_not_executed",
"backup_not_verified",
"operator_approval_missing",
"production_maintenance_window_required",
"seed_writer_real_write_not_implemented",
]
if not migration_file_exists:
blocked_reasons.insert(0, "migration_file_not_created")
elif not migration_file_matches_blueprint:
blocked_reasons.insert(0, "migration_file_differs_from_blueprint")
return {
"mode": "migration_file_draft_read_only",
"migration_number": MIGRATION_NUMBER,
"suggested_filename": MIGRATION_FILENAME,
"file_created": migration_file_exists,
"file_matches_blueprint": migration_file_matches_blueprint,
"file_status": (
"local_draft_matches_blueprint"
if migration_file_matches_blueprint
else "local_draft_differs_from_blueprint"
if migration_file_exists
else "not_created"
),
"migration_executed": False,
"database_session_created": False,
"database_commit_executed": False,
"external_network_executed": False,
"scheduler_attached": False,
"expected_tables": expected_tables,
"table_count": len(expected_tables),
"forward_sql": FORWARD_SQL,
"forward_statement_count": _statement_count(FORWARD_SQL),
"forward_has_destructive_sql": forward_has_destructive_sql,
"rollback_sql": ROLLBACK_SQL,
"rollback_requires_manual_approval": True,
"rollback_statement_count": _statement_count(ROLLBACK_SQL),
"blocked_reasons": blocked_reasons,
"table_operations": [
{
"table": table_name,
"operation": "CREATE TABLE IF NOT EXISTS",
"write_status": "preview_only_not_executed",
}
for table_name in expected_tables
],
"command_plan": {
"migration_apply_command": {
"command": (
"psql \"$DATABASE_URL\" -v ON_ERROR_STOP=1 "
f"-f {MIGRATION_FILENAME}"
),
"executed": False,
"requires_backup": True,
"requires_adr011_deploy_boundary_review": True,
},
"seed_writer_command": {
"command": (
"MARKET_INTEL_ENABLED=true MARKET_INTEL_WRITE_ENABLED=true "
"MARKET_INTEL_CRAWLER_ENABLED=false "
"MARKET_INTEL_SEED_WRITE_APPROVAL=<one_time_token> "
f"python {SEED_WRITER_SCRIPT} --execute --platform all"
),
"executed": False,
"script_created": seed_writer_script_exists,
"script_path": SEED_WRITER_SCRIPT,
"requires_new_approval_token": True,
"notes": (
"Seed writer skeleton 已存在，但真寫入仍未實作；不要為了 "
"seed upsert 而打開 crawler/manual fetch 權限。"
),
},
},
"safety_checks": {
"forward_sql_additive_only": not forward_has_destructive_sql,
"does_not_touch_momo_db_container": True,
"does_not_attach_scheduler": True,
"does_not_enable_external_crawling": True,
"does_not_write_seed_rows": True,
},
}

43
services/market_intel/platform_seed.py Normal file
View File

@@ -0,0 +1,43 @@
"""市場情報平台種子資料規劃。
本模組只產生可審核的 seed plan，不執行資料庫寫入。
"""
from services.market_intel.adapters import get_adapter, get_adapter_registry
def _adapter_to_seed(adapter):
policy = adapter.safety_policy.to_dict()
sources = [source.to_dict() for source in adapter.campaign_sources()]
return {
"code": adapter.platform_code,
"name": adapter.platform_name,
"base_url": adapter.base_url,
"enabled": False,
"crawl_policy_json": {
"request_interval_sec": policy["request_interval_sec"],
"timeout_sec": policy["timeout_sec"],
"max_pages_per_run": policy["max_pages_per_run"],
"allow_login": policy["allow_login"],
"allow_database_write": policy["allow_database_write"],
"allow_scheduler_attach": policy["allow_scheduler_attach"],
"seed_source_keys": [source["source_key"] for source in sources],
},
"source_count": len(sources),
"sources": sources,
"write_action": "upsert_market_platform_after_gate_approval",
}
def build_platform_seed_rows(platform_code="all"):
"""根據已註冊 adapter 建立平台 seed rows，不碰 DB。"""
if platform_code and platform_code != "all":
adapter = get_adapter(platform_code)
if not adapter:
return []
return [_adapter_to_seed(adapter)]
return [
_adapter_to_seed(adapter)
for adapter in get_adapter_registry().values()
]

238
services/market_intel/platform_seed_db_diff.py Normal file
View File

@@ -0,0 +1,238 @@
"""市場情報平台 seed 與正式 DB 的只讀差異探針。
本模組只查詢 market_platforms 既有 rows，不使用 DatabaseManager、不建立 ORM session、
不開 explicit transaction、不寫入、不 commit。
"""
import json
from sqlalchemy import bindparam, create_engine, text
def _normalize_policy(value):
"""把 seed policy 與 DB text policy 正規化成可比較字串。"""
if value in (None, ""):
return ""
if isinstance(value, str):
try:
value = json.loads(value)
except json.JSONDecodeError:
return value.strip()
return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
def _build_planned_seed_diffs(seed_rows):
return [
{
"code": seed["code"],
"exists": False,
"expected_enabled": bool(seed.get("enabled", False)),
"actual_enabled": None,
"name_matches": False,
"base_url_matches": False,
"policy_matches": False,
"diff_status": "planned_no_db_connection",
}
for seed in seed_rows
]
def _build_seed_diff(seed, actual_row):
if not actual_row:
return {
"code": seed["code"],
"exists": False,
"expected_enabled": bool(seed.get("enabled", False)),
"actual_enabled": None,
"name_matches": False,
"base_url_matches": False,
"policy_matches": False,
"diff_status": "missing",
}
expected_policy = _normalize_policy(seed.get("crawl_policy_json"))
actual_policy = _normalize_policy(actual_row.get("crawl_policy_json"))
name_matches = (actual_row.get("name") or "") == (seed.get("name") or "")
base_url_matches = (actual_row.get("base_url") or "") == (seed.get("base_url") or "")
expected_enabled = bool(seed.get("enabled", False))
actual_enabled = bool(actual_row.get("enabled", False))
policy_matches = expected_policy == actual_policy
diff_status = (
"matches_expected"
if name_matches and base_url_matches and policy_matches and expected_enabled == actual_enabled
else "differs_from_expected"
)
return {
"code": seed["code"],
"exists": True,
"expected_enabled": expected_enabled,
"actual_enabled": actual_enabled,
"name_matches": name_matches,
"base_url_matches": base_url_matches,
"policy_matches": policy_matches,
"diff_status": diff_status,
}
def _query_platform_rows(conn, expected_codes):
if not expected_codes:
return {}
rows = conn.execute(
text(
"""
SELECT code, name, base_url, enabled, crawl_policy_json
FROM market_platforms
WHERE code IN :platform_codes
ORDER BY code
"""
).bindparams(bindparam("platform_codes", expanding=True)),
{"platform_codes": tuple(expected_codes)},
).fetchall()
return {
row._mapping["code"]: dict(row._mapping)
for row in rows
}
def build_platform_seed_db_diff_plan(
seed_plan,
*,
execute_requested=False,
database_url=None,
database_type=None,
engine=None,
):
"""建立平台 seed DB 差異探針結果；預設只回 planned，不連 DB。"""
seed_rows = list(seed_plan.get("seeds", []))
expected_codes = [seed["code"] for seed in seed_rows]
if not execute_requested:
return {
"mode": "platform_seed_db_diff_planned",
"execute_requested": False,
"read_only_query_executed": False,
"database_connection_opened": False,
"database_session_created": False,
"explicit_transaction_opened": False,
"database_write_executed": False,
"database_commit_executed": False,
"migration_executed": False,
"seed_write_executed": False,
"expected_seed_count": len(seed_rows),
"existing_seed_count": 0,
"expected_codes": expected_codes,
"existing_codes": [],
"missing_codes": expected_codes,
"changed_codes": [],
"matching_codes": [],
"seed_rows_ready": False,
"seed_diffs": _build_planned_seed_diffs(seed_rows),
"blocked_reasons": [
"execute_false_planned_only",
"seed_db_diff_not_loaded",
"seed_write_still_blocked",
],
}
from config import DATABASE_PATH, DATABASE_TYPE
effective_database_type = (database_type or DATABASE_TYPE or "").lower()
effective_database_url = database_url or DATABASE_PATH
created_engine = False
connection_opened = False
try:
if engine is None:
connect_args = {}
if effective_database_type == "postgresql":
connect_args = {
"connect_timeout": 8,
"options": "-c statement_timeout=15000",
}
engine = create_engine(
effective_database_url,
isolation_level="AUTOCOMMIT",
pool_pre_ping=True,
connect_args=connect_args,
)
created_engine = True
with engine.connect() as conn:
connection_opened = True
existing_by_code = _query_platform_rows(conn, expected_codes)
seed_diffs = [
_build_seed_diff(seed, existing_by_code.get(seed["code"]))
for seed in seed_rows
]
missing_codes = [
item["code"] for item in seed_diffs
if item["diff_status"] == "missing"
]
changed_codes = [
item["code"] for item in seed_diffs
if item["diff_status"] == "differs_from_expected"
]
matching_codes = [
item["code"] for item in seed_diffs
if item["diff_status"] == "matches_expected"
]
blocked_reasons = ["seed_write_still_blocked"]
if missing_codes:
blocked_reasons.insert(0, "seed_rows_missing")
if changed_codes:
blocked_reasons.insert(0, "seed_rows_differ")
return {
"mode": "platform_seed_db_diff_read_only",
"execute_requested": True,
"read_only_query_executed": True,
"database_connection_opened": connection_opened,
"database_session_created": False,
"explicit_transaction_opened": False,
"database_write_executed": False,
"database_commit_executed": False,
"migration_executed": False,
"seed_write_executed": False,
"expected_seed_count": len(seed_rows),
"existing_seed_count": len(existing_by_code),
"expected_codes": expected_codes,
"existing_codes": sorted(existing_by_code),
"missing_codes": missing_codes,
"changed_codes": changed_codes,
"matching_codes": matching_codes,
"seed_rows_ready": not missing_codes and not changed_codes,
"seed_diffs": seed_diffs,
"blocked_reasons": blocked_reasons,
}
except Exception as exc:
return {
"mode": "platform_seed_db_diff_error",
"execute_requested": True,
"read_only_query_executed": False,
"database_connection_opened": connection_opened,
"database_session_created": False,
"explicit_transaction_opened": False,
"database_write_executed": False,
"database_commit_executed": False,
"migration_executed": False,
"seed_write_executed": False,
"expected_seed_count": len(seed_rows),
"existing_seed_count": 0,
"expected_codes": expected_codes,
"existing_codes": [],
"missing_codes": expected_codes,
"changed_codes": [],
"matching_codes": [],
"seed_rows_ready": False,
"seed_diffs": _build_planned_seed_diffs(seed_rows),
"blocked_reasons": [
"platform_seed_db_diff_error",
"seed_write_still_blocked",
],
"error_message": str(exc),
}
finally:
if created_engine:
engine.dispose()

89
services/market_intel/platform_seed_writer.py Normal file
View File

@@ -0,0 +1,89 @@
"""市場情報平台種子資料寫入 dry-run 規劃。
本模組只建立 upsert preview 與 schema smoke 結果，不執行資料庫寫入。
"""
import json
from database.manager import Base
MARKET_PLATFORM_REQUIRED_COLUMNS = (
"code",
"name",
"base_url",
"enabled",
"crawl_policy_json",
)
def build_schema_smoke(expected_tables, metadata=None):
"""檢查 ORM metadata 是否具備 market_* 表與 platform upsert 必要欄位。"""
metadata = metadata or Base.metadata
metadata_tables = metadata.tables
table_names = set(metadata_tables)
missing_tables = [
table_name for table_name in expected_tables
if table_name not in table_names
]
market_platform = metadata_tables.get("market_platforms")
platform_columns = (
set(market_platform.columns.keys())
if market_platform is not None
else set()
)
missing_platform_columns = [
column_name for column_name in MARKET_PLATFORM_REQUIRED_COLUMNS
if column_name not in platform_columns
]
return {
"passed": not missing_tables and not missing_platform_columns,
"expected_table_count": len(expected_tables),
"metadata_table_count": len(table_names),
"missing_tables": missing_tables,
"market_platform_required_columns": list(MARKET_PLATFORM_REQUIRED_COLUMNS),
"missing_market_platform_columns": missing_platform_columns,
}
def build_platform_seed_writer_plan(seed_plan, write_guard, schema_smoke):
"""建立 market_platforms upsert dry-run plan，不建立 session、不 commit。"""
operations = []
for seed in seed_plan.get("seeds", []):
operations.append(
{
"operation": "upsert",
"table": "market_platforms",
"lookup": {"code": seed["code"]},
"values": {
"code": seed["code"],
"name": seed["name"],
"base_url": seed["base_url"],
"enabled": False,
"crawl_policy_json": json.dumps(
seed["crawl_policy_json"],
ensure_ascii=False,
sort_keys=True,
),
},
"sql_shape": (
"INSERT INTO market_platforms (...) VALUES (...) "
"ON CONFLICT(code) DO UPDATE SET ..."
),
"write_status": "blocked_dry_run_only",
}
)
return {
"mode": "dry_run",
"operation_count": len(operations),
"operations": operations,
"schema_smoke": schema_smoke,
"ready_to_write": False,
"writes_executed": False,
"would_write_database": False,
"database_write_allowed": bool(write_guard.get("database_write_allowed")),
"blocked_reasons": write_guard.get("blocked_reasons", []),
"write_action": "preview_only_no_session_no_commit",
}

172
services/market_intel/schema_db_probe.py Normal file
View File

@@ -0,0 +1,172 @@
"""市場情報正式 DB schema 只讀探針。
本模組只查詢系統 catalog，不使用 DatabaseManager、不呼叫 create_all、不寫入。
"""
from sqlalchemy import bindparam, create_engine, text
def _build_table_status(expected_tables, existing_tables):
existing_set = set(existing_tables)
return [
{
"table": table_name,
"exists": table_name in existing_set,
}
for table_name in expected_tables
]
def _probe_postgresql(conn, expected_tables):
existing_tables = []
for table_name in expected_tables:
exists = conn.execute(
text(
"""
SELECT EXISTS (
SELECT 1
FROM information_schema.tables
WHERE table_schema = ANY (current_schemas(false))
AND table_name = :table_name
)
"""
),
{"table_name": table_name},
).scalar()
if exists:
existing_tables.append(table_name)
return existing_tables
def _probe_sqlite(conn, expected_tables):
rows = conn.execute(
text(
"""
SELECT name
FROM sqlite_master
WHERE type = 'table'
AND name IN :table_names
"""
).bindparams(bindparam("table_names", expanding=True)),
{"table_names": tuple(expected_tables)},
).fetchall()
return [row[0] for row in rows]
def build_schema_db_probe_plan(
expected_tables,
*,
execute_requested=False,
database_url=None,
database_type=None,
engine=None,
):
"""建立 DB schema 探針結果；預設只回 planned，不連 DB。"""
expected_tables = list(expected_tables)
if not execute_requested:
return {
"mode": "schema_db_probe_planned",
"execute_requested": False,
"read_only_query_executed": False,
"database_connection_opened": False,
"database_session_created": False,
"explicit_transaction_opened": False,
"database_write_executed": False,
"database_commit_executed": False,
"migration_executed": False,
"schema_tables_exist": False,
"expected_tables": expected_tables,
"existing_tables": [],
"missing_tables": expected_tables,
"table_statuses": _build_table_status(expected_tables, []),
"blocked_reasons": [
"execute_false_planned_only",
"migration_not_executed_by_this_probe",
"seed_write_still_blocked",
],
}
from config import DATABASE_PATH, DATABASE_TYPE
effective_database_type = (database_type or DATABASE_TYPE or "").lower()
effective_database_url = database_url or DATABASE_PATH
created_engine = False
connection_opened = False
try:
if engine is None:
connect_args = {}
if effective_database_type == "postgresql":
connect_args = {
"connect_timeout": 8,
"options": "-c statement_timeout=15000",
}
engine = create_engine(
effective_database_url,
isolation_level="AUTOCOMMIT",
pool_pre_ping=True,
connect_args=connect_args,
)
created_engine = True
with engine.connect() as conn:
connection_opened = True
if effective_database_type == "postgresql":
existing_tables = _probe_postgresql(conn, expected_tables)
else:
existing_tables = _probe_sqlite(conn, expected_tables)
missing_tables = [
table_name for table_name in expected_tables
if table_name not in set(existing_tables)
]
schema_tables_exist = not missing_tables
blocked_reasons = [
"migration_not_executed_by_this_probe",
"seed_write_still_blocked",
]
if not schema_tables_exist:
blocked_reasons.insert(0, "market_tables_missing")
return {
"mode": "schema_db_probe_read_only",
"execute_requested": True,
"read_only_query_executed": True,
"database_connection_opened": connection_opened,
"database_session_created": False,
"explicit_transaction_opened": False,
"database_write_executed": False,
"database_commit_executed": False,
"migration_executed": False,
"schema_tables_exist": schema_tables_exist,
"expected_tables": expected_tables,
"existing_tables": existing_tables,
"missing_tables": missing_tables,
"table_statuses": _build_table_status(expected_tables, existing_tables),
"blocked_reasons": blocked_reasons,
}
except Exception as exc:
return {
"mode": "schema_db_probe_error",
"execute_requested": True,
"read_only_query_executed": False,
"database_connection_opened": connection_opened,
"database_session_created": False,
"explicit_transaction_opened": False,
"database_write_executed": False,
"database_commit_executed": False,
"migration_executed": False,
"schema_tables_exist": False,
"expected_tables": expected_tables,
"existing_tables": [],
"missing_tables": expected_tables,
"table_statuses": _build_table_status(expected_tables, []),
"blocked_reasons": [
"schema_db_probe_error",
"seed_write_still_blocked",
],
"error_message": str(exc),
}
finally:
if created_engine:
engine.dispose()

197
services/market_intel/seed_writer_cli.py Normal file
View File

@@ -0,0 +1,197 @@
"""市場情報 seed writer CLI skeleton。
本階段只回報 CLI 執行計畫，不建立 DB session、不寫入、不 commit。
"""
import hashlib
import json
APPROVAL_ENV_VAR = "MARKET_INTEL_SEED_WRITE_APPROVAL"
PLATFORM_UPSERT_SQL = """
INSERT INTO market_platforms (
code,
name,
base_url,
enabled,
crawl_policy_json
) VALUES (
:code,
:name,
:base_url,
:enabled,
:crawl_policy_json
)
ON CONFLICT (code) DO UPDATE SET
name = EXCLUDED.name,
base_url = EXCLUDED.base_url,
enabled = EXCLUDED.enabled,
crawl_policy_json = EXCLUDED.crawl_policy_json,
updated_at = NOW()
""".strip()
def _payload_hash(payload):
encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True).encode("utf-8")
return hashlib.sha256(encoded).hexdigest()[:16]
def build_seed_transaction_preview(writer_plan, migration_blueprint):
"""建立 seed writer transaction preview；不建立 DB session。"""
statements = []
for index, operation in enumerate(writer_plan.get("operations", []), start=1):
values = operation.get("values", {})
lookup_code = operation.get("lookup", {}).get("code") or values.get("code")
statements.append(
{
"index": index,
"operation": operation.get("operation", "upsert"),
"table": operation.get("table", "market_platforms"),
"lookup": {"code": lookup_code},
"sql_template": PLATFORM_UPSERT_SQL,
"parameter_keys": sorted(values),
"parameter_payload_hash": _payload_hash(values),
"idempotency_key": f"market_platforms:{lookup_code}",
"diff_status": "not_loaded_no_db_session",
"write_status": "blocked_transaction_preview_only",
}
)
migration_ready = bool(
migration_blueprint.get("file_created")
and migration_blueprint.get("file_matches_blueprint")
and not migration_blueprint.get("migration_executed")
)
return {
"mode": "seed_transaction_preview_no_session",
"target_table": "market_platforms",
"statement_count": len(statements),
"statements": statements,
"migration_draft_ready": migration_ready,
"database_snapshot_loaded": False,
"existing_rows_seen": 0,
"database_session_created": False,
"transaction_opened": False,
"writes_executed": False,
"would_write_database": False,
"database_commit_executed": False,
"database_rollback_executed": False,
"external_network_executed": False,
"scheduler_attached": False,
"required_runtime_order": [
"backup_verified",
"migration_applied_by_operator",
"schema_smoke_passed",
"feature_flags_reviewed",
"one_time_approval_token_verified",
"real_write_implementation_enabled",
],
"safety_contract": {
"idempotent_upsert_preview_only": True,
"does_not_load_existing_rows": True,
"does_not_open_transaction": True,
"does_not_commit": True,
},
}
def build_seed_writer_cli_plan(
*,
platform_code,
execute_requested,
approval_token,
seed_plan,
write_guard,
writer_plan,
migration_blueprint,
):
"""建立 seed writer CLI blocked plan。"""
approval_token_present = bool(approval_token)
migration_ready = bool(
migration_blueprint.get("file_created")
and migration_blueprint.get("file_matches_blueprint")
and not migration_blueprint.get("migration_executed")
)
gates = [
{
"key": "script_created",
"label": "scripts/market_intel_seed_writer.py exists",
"passed": True,
},
{
"key": "migration_file_matches_blueprint",
"label": "migration draft exists and matches the reviewed blueprint",
"passed": migration_ready,
},
{
"key": "execute_requested",
"label": "--execute flag was explicitly provided",
"passed": bool(execute_requested),
},
{
"key": "approval_token_present",
"label": f"{APPROVAL_ENV_VAR} or --approval-token was provided",
"passed": approval_token_present,
},
{
"key": "database_write_allowed",
"label": "runtime database_write_allowed gate is true",
"passed": bool(write_guard.get("database_write_allowed")),
},
{
"key": "manual_operator_approval",
"label": "operator approval has been verified out-of-band",
"passed": False,
},
{
"key": "real_write_implementation_enabled",
"label": "CLI real write implementation has been enabled",
"passed": False,
},
]
blocked_reasons = [gate["key"] for gate in gates if not gate["passed"]]
if execute_requested:
blocked_reasons.insert(0, "execute_request_blocked_by_skeleton")
transaction_preview = build_seed_transaction_preview(
writer_plan=writer_plan,
migration_blueprint=migration_blueprint,
)
return {
"mode": "seed_writer_cli_blocked_skeleton",
"platform_code": platform_code or "all",
"execute_requested": bool(execute_requested),
"approval_token_present": approval_token_present,
"approval_env_var": APPROVAL_ENV_VAR,
"ready_for_real_write": False,
"writes_executed": False,
"would_write_database": False,
"database_session_created": False,
"database_commit_executed": False,
"external_network_executed": False,
"scheduler_attached": False,
"exit_code": 2 if execute_requested else 0,
"blocked_reasons": blocked_reasons,
"approval_gates": gates,
"seed_count": int(seed_plan.get("seed_count") or 0),
"writer_operation_count": int(writer_plan.get("operation_count") or 0),
"transaction_preview": transaction_preview,
"write_guard_summary": {
"ready_to_write": bool(write_guard.get("ready_to_write")),
"would_write_database": bool(write_guard.get("would_write_database")),
"database_write_allowed": bool(write_guard.get("database_write_allowed")),
"blocked_reasons": write_guard.get("blocked_reasons", []),
},
"migration_file_summary": {
"suggested_filename": migration_blueprint.get("suggested_filename"),
"file_created": bool(migration_blueprint.get("file_created")),
"file_matches_blueprint": bool(migration_blueprint.get("file_matches_blueprint")),
"migration_executed": bool(migration_blueprint.get("migration_executed")),
},
"safety_contract": {
"refuses_execute_in_this_phase": True,
"requires_independent_approval_token": True,
"keeps_crawler_disabled_for_seed_write": True,
"no_db_session_in_skeleton": True,
},
}

425
services/market_intel/service.py
View File

@@ -12,6 +12,23 @@ from config import (
MARKET_INTEL_ENABLED,
MARKET_INTEL_WRITE_ENABLED,
)
from services.market_intel.adapters import (
get_adapter,
get_adapter_registry,
get_adapter_summaries,
)
from services.market_intel.candidate_preview import build_candidate_preview_from_discovery
from services.market_intel.discovery_runner import ManualDiscoveryRunner
from services.market_intel.migration_blueprint import build_migration_blueprint
from services.market_intel.platform_seed import build_platform_seed_rows
from services.market_intel.platform_seed_db_diff import build_platform_seed_db_diff_plan
from services.market_intel.platform_seed_writer import (
build_platform_seed_writer_plan,
build_schema_smoke,
)
from services.market_intel.seed_writer_cli import build_seed_writer_cli_plan
from services.market_intel.schema_db_probe import build_schema_db_probe_plan
from services.market_intel.write_approval_runbook import build_write_approval_runbook
TAIPEI_TZ = timezone(timedelta(hours=8))
@@ -45,7 +62,7 @@ class MarketIntelRuntimeStatus:
class MarketIntelService:
"""市場情報入口服務，先集中 feature gate 與安全狀態。"""
phase = "phase_2_schema_ready_disabled"
phase = "phase_25_platform_seed_db_diff"
def get_runtime_status(self) -> MarketIntelRuntimeStatus:
return MarketIntelRuntimeStatus(
@@ -62,13 +79,22 @@ class MarketIntelService:
),
)
def manual_fetch_allowed(self):
status = self.get_runtime_status()
return bool(status.enabled and status.crawler_enabled)
def get_schema_tables(self):
"""回傳 ADR-035 定義的 market_* schema 名稱。"""
return list(MARKET_INTEL_TABLES)
def get_adapter_summaries(self):
"""回傳目前已註冊 adapter，不觸發網路。"""
return get_adapter_summaries()
def build_dry_run_plan(self, platform_code="all"):
"""建立 dry-run 計畫，不執行爬蟲、不寫 DB。"""
status = self.get_runtime_status()
adapter_registry = get_adapter_registry()
return {
"batch_id": f"market-dry-run-{uuid4().hex[:12]}",
"platform_code": platform_code,
@@ -77,6 +103,403 @@ class MarketIntelService:
"would_discover_campaigns": bool(status.enabled and status.crawler_enabled),
"would_write_database": bool(status.database_write_allowed),
"scheduler_attached": status.scheduler_attached,
"manual_fetch_allowed": self.manual_fetch_allowed(),
"schema_tables": self.get_schema_tables(),
"adapter_count": len(adapter_registry),
"adapters": [adapter.summary() for adapter in adapter_registry.values()],
"status": status.to_dict(),
}
def build_discovery_plan(self, platform_code="all"):
"""建立平台 discovery dry-run plan，不發 request、不寫 DB。"""
if platform_code and platform_code != "all":
adapter = get_adapter(platform_code)
if not adapter:
return {
"platform_code": platform_code,
"found": False,
"plans": [],
"error": "未知平台 adapter",
}
return {
"platform_code": platform_code,
"found": True,
"plans": [adapter.build_discovery_plan()],
}
return {
"platform_code": "all",
"found": True,
"plans": [
adapter.build_discovery_plan()
for adapter in get_adapter_registry().values()
],
}
def run_manual_discovery(self, platform_code="all", *, fetch=False, http_get=None):
"""手動執行 discovery dry-run；預設不發 request，永遠不寫 DB。"""
registry = get_adapter_registry()
adapters = []
if platform_code and platform_code != "all":
adapter = get_adapter(platform_code)
if not adapter:
return {
"platform_code": platform_code,
"found": False,
"runs": [],
"error": "未知平台 adapter",
}
adapters = [adapter]
else:
adapters = list(registry.values())
runner = ManualDiscoveryRunner(
runtime_status=self.get_runtime_status(),
http_get=http_get,
)
return {
"platform_code": platform_code or "all",
"found": True,
"fetch_requested": bool(fetch),
"manual_fetch_allowed": self.manual_fetch_allowed(),
"runs": [
runner.run(adapter, fetch=fetch).to_dict()
for adapter in adapters
],
}
def build_candidate_preview(
self,
platform_code="all",
*,
fetch=False,
min_band="all",
limit=50,
http_get=None,
):
"""聚合候選連結 preview，只供人工審核，不寫 DB。"""
discovery_result = self.run_manual_discovery(
platform_code=platform_code,
fetch=fetch,
http_get=http_get,
)
preview = build_candidate_preview_from_discovery(
discovery_result,
min_band=min_band,
limit=limit,
)
preview["discovery_found"] = bool(discovery_result.get("found"))
preview["error"] = discovery_result.get("error")
return preview
def build_platform_seed_plan(self, platform_code="all"):
"""建立 market_platforms 初始資料計畫，不寫入 DB。"""
status = self.get_runtime_status()
seed_rows = build_platform_seed_rows(platform_code=platform_code)
found = bool(seed_rows) or platform_code in (None, "", "all")
return {
"platform_code": platform_code or "all",
"found": found,
"phase": self.phase,
"seed_count": len(seed_rows),
"seeds": seed_rows,
"would_write_database": False,
"database_write_allowed": bool(status.database_write_allowed),
"required_gates": {
"market_intel_enabled": bool(status.enabled),
"market_intel_write_enabled": bool(status.write_enabled),
"schema_smoke_required": True,
"migration_required": True,
"manual_operator_approval_required": True,
},
"status": status.to_dict(),
"error": None if found else "未知平台 adapter",
}
def build_platform_seed_write_guard(self, platform_code="all"):
"""回報 platform seed 寫入前置 gate；本方法不執行寫入。"""
status = self.get_runtime_status()
seed_plan = self.build_platform_seed_plan(platform_code=platform_code)
schema_smoke = build_schema_smoke(MARKET_INTEL_TABLES)
guard_checks = {
"seed_plan_found": bool(seed_plan["found"]),
"has_seed_rows": bool(seed_plan["seed_count"]),
"market_intel_enabled": bool(status.enabled),
"market_intel_write_enabled": bool(status.write_enabled),
"database_write_allowed": bool(status.database_write_allowed),
"migration_confirmed": False,
"schema_smoke_confirmed": bool(schema_smoke["passed"]),
"manual_operator_approval": False,
}
blocked_reasons = [
name for name, passed in guard_checks.items()
if not passed
]
return {
"platform_code": platform_code or "all",
"phase": self.phase,
"seed_count": seed_plan["seed_count"],
"ready_to_write": False,
"would_write_database": False,
"database_write_allowed": bool(status.database_write_allowed),
"guard_checks": guard_checks,
"blocked_reasons": blocked_reasons,
"write_action": "blocked_dry_run_only",
"schema_smoke": schema_smoke,
"seed_plan": seed_plan,
}
def build_schema_smoke(self):
"""回報 market_intel ORM metadata smoke 結果，不查詢 DB。"""
return {
"phase": self.phase,
"schema_smoke": build_schema_smoke(MARKET_INTEL_TABLES),
"expected_tables": self.get_schema_tables(),
}
def build_schema_db_probe(self, *, execute_requested=False):
"""回報正式 DB schema 只讀探針；預設不連 DB。"""
probe = build_schema_db_probe_plan(
MARKET_INTEL_TABLES,
execute_requested=execute_requested,
)
probe["phase"] = self.phase
return probe
def build_platform_seed_db_diff(self, platform_code="all", *, execute_requested=False):
"""回報 platform seed 與 DB 的只讀差異；預設不連 DB。"""
seed_plan = self.build_platform_seed_plan(platform_code=platform_code)
diff = build_platform_seed_db_diff_plan(
seed_plan,
execute_requested=execute_requested,
)
diff["phase"] = self.phase
diff["platform_code"] = platform_code or "all"
diff["seed_plan_found"] = bool(seed_plan["found"])
return diff
def build_platform_seed_writer_plan(self, platform_code="all"):
"""建立 platform seed writer dry-run plan，不建立 DB session。"""
seed_plan = self.build_platform_seed_plan(platform_code=platform_code)
write_guard = self.build_platform_seed_write_guard(platform_code=platform_code)
schema_smoke = write_guard["schema_smoke"]
writer_plan = build_platform_seed_writer_plan(
seed_plan=seed_plan,
write_guard=write_guard,
schema_smoke=schema_smoke,
)
writer_plan["phase"] = self.phase
writer_plan["platform_code"] = platform_code or "all"
writer_plan["seed_plan_found"] = bool(seed_plan["found"])
writer_plan["seed_count"] = seed_plan["seed_count"]
return writer_plan
def build_write_approval_runbook(self, platform_code="all"):
"""建立正式 seed writer 前的人工批准 runbook；本方法不執行寫入。"""
status = self.get_runtime_status()
seed_plan = self.build_platform_seed_plan(platform_code=platform_code)
write_guard = self.build_platform_seed_write_guard(platform_code=platform_code)
writer_plan = self.build_platform_seed_writer_plan(platform_code=platform_code)
return build_write_approval_runbook(
phase=self.phase,
status=status,
schema_smoke=write_guard["schema_smoke"],
seed_plan=seed_plan,
write_guard=write_guard,
writer_plan=writer_plan,
)
def build_migration_blueprint(self):
"""建立 market_intel migration 與 seed writer 命令草案；不執行 SQL。"""
blueprint = build_migration_blueprint(self.get_schema_tables())
blueprint["phase"] = self.phase
return blueprint
def build_seed_writer_cli_status(
self,
platform_code="all",
*,
execute_requested=False,
approval_token=None,
):
"""建立 seed writer CLI blocked status；不建立 DB session、不寫入。"""
seed_plan = self.build_platform_seed_plan(platform_code=platform_code)
write_guard = self.build_platform_seed_write_guard(platform_code=platform_code)
writer_plan = self.build_platform_seed_writer_plan(platform_code=platform_code)
migration_blueprint = self.build_migration_blueprint()
status = build_seed_writer_cli_plan(
platform_code=platform_code or "all",
execute_requested=execute_requested,
approval_token=approval_token,
seed_plan=seed_plan,
write_guard=write_guard,
writer_plan=writer_plan,
migration_blueprint=migration_blueprint,
)
status["phase"] = self.phase
return status
def build_deployment_readiness(self):
"""建立市場情報推版準備狀態；不執行 git、部署或遠端操作。"""
status = self.get_runtime_status()
schema_smoke = build_schema_smoke(MARKET_INTEL_TABLES)
writer_plan = self.build_platform_seed_writer_plan()
checks = {
"schema_smoke_passed": bool(schema_smoke["passed"]),
"feature_flags_default_safe": bool(
not status.enabled
and not status.crawler_enabled
and not status.write_enabled
),
"database_write_blocked": bool(not status.database_write_allowed),
"scheduler_detached": bool(not status.scheduler_attached),
"manual_fetch_disabled": bool(not self.manual_fetch_allowed()),
"writer_plan_dry_run_only": bool(
writer_plan["mode"] == "dry_run"
and not writer_plan["writes_executed"]
and not writer_plan["would_write_database"]
),
"registered_adapters_present": bool(len(self.get_adapter_summaries()) >= 4),
"schema_db_probe_planned_safe": bool(
not self.build_schema_db_probe()["read_only_query_executed"]
),
"platform_seed_db_diff_planned_safe": bool(
not self.build_platform_seed_db_diff()["read_only_query_executed"]
),
}
ready_for_production_deploy = all(checks.values())
blocked_reasons = [
reason for reason, blocked in (
("readiness_checks_not_all_passed", not ready_for_production_deploy),
("production_deploy_not_executed_by_api", True),
("git_commit_not_created_by_api", True),
("git_push_not_executed_by_api", True),
("backup_must_be_verified_by_operator", True),
("production_smoke_must_be_verified_by_operator", True),
)
if blocked
]
required_manual_steps = [
{
"key": "review_worktree_scope",
"label": "審核 worktree，只納入市場情報相關變更，排除 unrelated dirty files",
"status": "required",
},
{
"key": "run_backup_system",
"label": "重大更新前執行 python backup_system.py",
"status": "required",
},
{
"key": "commit_market_intel_changes_only",
"label": "只 commit 市場情報模組、ADR/TODO 與必要測試",
"status": "operator_optional",
},
{
"key": "push_reviewed_branch_or_main",
"label": "推送已審核分支或 main，再進入部署 SOP",
"status": "operator_optional",
},
{
"key": "run_deployment_sop",
"label": "依 deployment SOP app-only 部署，不碰 momo-db",
"status": "required",
},
{
"key": "verify_health_endpoint",
"label": "部署後先驗證 /health，不使用首頁作為探測",
"status": "required",
},
{
"key": "verify_market_intel_page_after_deploy",
"label": "驗證 /market_intel 與市場情報 API 仍維持 blocked dry-run",
"status": "required",
},
]
fallback_plan = [
{
"key": "feature_flag_kill_switch",
"label": "MARKET_INTEL_ENABLED、MARKET_INTEL_CRAWLER_ENABLED、MARKET_INTEL_WRITE_ENABLED 保持全關，可立即停用新功能面",
"trigger": "任何 UI/API 異常或非預期連外行為",
},
{
"key": "app_only_rollback",
"label": "回退到上一個已知正常版本後，只 recreate momo-app，避免影響 momo-db 資料生命週期",
"trigger": "部署後 /health 或 /market_intel smoke 失敗",
},
{
"key": "scheduler_detached",
"label": "市場情報 scheduler 尚未掛載；異常時不需停爬蟲排程，因為本階段沒有排程入口",
"trigger": "排程或外部流量疑慮",
},
{
"key": "database_write_blocked",
"label": "writer 仍是 preview_only_no_session_no_commit；異常時不需要 DB rollback",
"trigger": "seed writer 或 schema smoke 異常",
},
]
safe_deploy_boundaries = [
{
"key": "no_remove_orphans",
"label": "禁止使用 docker compose --remove-orphans",
},
{
"key": "no_momo_db_lifecycle_change",
"label": "禁止 stop/rm/recreate momo-db 或變更資料生命週期",
},
{
"key": "health_probe_only",
"label": "HTTP health / blackbox / CD 探測只打 /health",
},
{
"key": "flags_default_off",
"label": "市場情報三個 feature flags 預設維持 OFF",
},
]
return {
"phase": self.phase,
"mode": "app_only_release_gate",
"production_deployed": False,
"git_committed": False,
"git_pushed": False,
"ready_for_production_deploy": ready_for_production_deploy,
"deployment_actions_executed": False,
"execution_boundary": {
"api_executes_git": False,
"api_executes_backup": False,
"api_executes_scp": False,
"api_executes_ssh": False,
"api_recreates_container": False,
"api_runs_migration": False,
"api_writes_database": False,
},
"checks": checks,
"blocked_reasons": blocked_reasons,
"requires_backup_before_major_update": True,
"backup_command": "python backup_system.py",
"required_manual_steps": required_manual_steps,
"fallback_plan": fallback_plan,
"safe_deploy_boundaries": safe_deploy_boundaries,
"production_smoke_targets": [
"/health",
"/market_intel",
"/api/market_intel/status",
"/api/market_intel/deployment_readiness",
"/api/market_intel/schema_smoke",
"/api/market_intel/schema_db_probe",
"/api/market_intel/platform_seed_db_diff",
],
"status": status.to_dict(),
"schema_smoke": schema_smoke,
"writer_plan_summary": {
"operation_count": writer_plan["operation_count"],
"writes_executed": writer_plan["writes_executed"],
"would_write_database": writer_plan["would_write_database"],
},
"write_approval_runbook": self.build_write_approval_runbook(),
"migration_blueprint": self.build_migration_blueprint(),
"seed_writer_cli_status": self.build_seed_writer_cli_status(),
"schema_db_probe": self.build_schema_db_probe(),
"platform_seed_db_diff": self.build_platform_seed_db_diff(),
}

134
services/market_intel/write_approval_runbook.py Normal file
View File

@@ -0,0 +1,134 @@
"""市場情報正式寫入前的人工批准 runbook。
本模組只產生 gate 與操作順序，不建立 DB session、不執行 migration、不寫入資料。
"""
def _status_value(status, key, default=False):
return bool(getattr(status, key, default))
def build_write_approval_runbook(
*,
phase,
status,
schema_smoke,
seed_plan,
write_guard,
writer_plan,
):
"""建立正式 seed write 前的 read-only runbook。"""
gates = [
{
"key": "schema_smoke_passed",
"label": "ORM metadata smoke 已通過，七張 market_* table 與 market_platforms 欄位完整",
"passed": bool(schema_smoke.get("passed")),
},
{
"key": "backup_completed",
"label": "已在正式推版前執行 python backup_system.py",
"passed": False,
},
{
"key": "migration_file_reviewed",
"label": "market_* schema migration 已人工審核，且不 drop/alter 既有業績資料表",
"passed": False,
},
{
"key": "feature_flags_enabled_for_write_window",
"label": "寫入窗口才可同時啟用 MARKET_INTEL_ENABLED 與 MARKET_INTEL_WRITE_ENABLED",
"passed": bool(_status_value(status, "enabled") and _status_value(status, "write_enabled")),
},
{
"key": "database_write_allowed",
"label": "runtime database_write_allowed 為 true",
"passed": bool(_status_value(status, "database_write_allowed")),
},
{
"key": "manual_operator_approval",
"label": "操作者已明確批准一次性 market_platforms seed upsert",
"passed": False,
},
{
"key": "rollback_plan_reviewed",
"label": "已確認回復策略：關閉 flags、app-only rollback、必要時清理 seed rows",
"passed": False,
},
{
"key": "production_smoke_targets_defined",
"label": "已定義 /health 與市場情報 API smoke targets",
"passed": True,
},
]
blocked_reasons = [gate["key"] for gate in gates if not gate["passed"]]
return {
"phase": phase,
"mode": "approval_runbook_read_only",
"ready_for_real_write": False,
"writes_executed": False,
"would_write_database": False,
"database_session_created": False,
"database_commit_executed": False,
"external_network_executed": False,
"scheduler_attached": False,
"approval_required": True,
"approval_token_present": False,
"blocked_reasons": blocked_reasons,
"approval_gates": gates,
"seed_count": int(seed_plan.get("seed_count") or 0),
"writer_operation_count": int(writer_plan.get("operation_count") or 0),
"schema_smoke": schema_smoke,
"write_guard_summary": {
"ready_to_write": bool(write_guard.get("ready_to_write")),
"would_write_database": bool(write_guard.get("would_write_database")),
"blocked_reasons": write_guard.get("blocked_reasons", []),
},
"operator_sequence": [
{
"key": "scope_review",
"label": "確認 git diff 只包含 market_intel、ADR/TODO、版本與測試",
},
{
"key": "backup",
"label": "執行 python backup_system.py，保存正式環境回復點",
},
{
"key": "migration_apply_window",
"label": "在維護窗口套用 market_* migration，不觸碰 momo-db 容器生命週期",
},
{
"key": "seed_preview_compare",
"label": "比對 platform_seed_writer_plan 的 4 筆 upsert preview",
},
{
"key": "one_time_seed_write",
"label": "另開明確批准後才允許一次性 seed writer 真寫入",
},
{
"key": "post_write_smoke",
"label": "驗證 /health、/market_intel、schema_smoke 與 deployment_readiness",
},
],
"rollback_plan": [
{
"key": "disable_flags",
"label": "立刻關閉 MARKET_INTEL_ENABLED、MARKET_INTEL_CRAWLER_ENABLED、MARKET_INTEL_WRITE_ENABLED",
},
{
"key": "app_only_rollback",
"label": "回退 momo-app 程式版本，只 recreate momo-app，不碰 momo-db",
},
{
"key": "seed_rows_cleanup",
"label": "若唯一異常來自 platform seed，可在人工審核後刪除或停用 market_platforms seed rows",
},
],
"hard_safety_boundaries": [
"no_external_crawling_during_seed_write",
"no_scheduler_attach_during_seed_write",
"no_momo_db_container_lifecycle_change",
"no_remove_orphans",
"health_probe_uses_health_endpoint_only",
],
}