ewoooc/services/market_intel/service.py

"""市場情報 Phase 1 骨架服務。

本階段只回報 feature flag 與 rollout 狀態，不啟動爬蟲、不寫資料庫。
"""

from dataclasses import asdict, dataclass
from datetime import datetime, timedelta, timezone
from uuid import uuid4

from config import (
    MARKET_INTEL_CRAWLER_ENABLED,
    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.deployment_readiness import (
    build_deployment_readiness_preview,
)
from services.market_intel.discovery_runner import ManualDiscoveryRunner
from services.market_intel.legacy_source_bridge import build_legacy_source_bridge_plan
from services.market_intel.live_db_inventory import build_live_db_inventory_preview
from services.market_intel.match_review_plan import build_match_review_plan_preview
from services.market_intel.mcp_activation_runbook import build_mcp_activation_runbook_preview
from services.market_intel.mcp_completion_audit import build_mcp_completion_audit_for_runtime
from services.market_intel.mcp_contract import build_mcp_tool_contract_preview
from services.market_intel.mcp_deploy_preflight import build_mcp_deploy_preflight_plan
from services.market_intel.mcp_fetch_gate import build_mcp_fetch_gate_preview
from services.market_intel.mcp_readiness import build_mcp_readiness_plan
from services.market_intel.manual_sample_plan import (
    build_manual_sample_fetch_plan_preview,
)
from services.market_intel.manual_sample_acceptance import (
    build_manual_sample_acceptance_preview,
)
from services.market_intel.manual_sample_candidate_queue import (
    build_manual_sample_candidate_queue_approval_preview,
    build_manual_sample_candidate_queue_draft_preview,
    build_manual_sample_candidate_queue_transaction_preview,
)
from services.market_intel.manual_sample_review import (
    build_manual_sample_candidate_handoff_preview,
    build_manual_sample_review_evaluation_preview,
    build_manual_sample_review_preview,
)
from services.market_intel.migration_blueprint import build_migration_blueprint
from services.market_intel.migration_catalog_review import (
    build_migration_catalog_review_preview,
)
from services.market_intel.migration_drill import build_migration_apply_drill_preview
from services.market_intel.migration_live_smoke import (
    build_migration_live_smoke_preview,
)
from services.market_intel.opportunity_alerts import (
    build_opportunity_alert_plan_preview,
)
from services.market_intel.opportunity_evidence import (
    build_opportunity_evidence_plan_preview,
)
from services.market_intel.opportunity_plan import build_opportunity_plan_preview
from services.market_intel.opportunity_scoring import (
    build_opportunity_scoring_plan_preview,
)
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.phase import MARKET_INTEL_PHASE
from services.market_intel.scheduler_plan import build_scheduler_attach_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))
MARKET_INTEL_TABLES = (
    "market_platforms",
    "market_campaigns",
    "market_campaign_snapshots",
    "market_campaign_products",
    "market_product_price_history",
    "market_product_matches",
    "market_crawler_runs",
    "market_alert_review_queue",
)


@dataclass(frozen=True)
class MarketIntelRuntimeStatus:
    """市場情報模組目前的啟用狀態。"""

    phase: str
    enabled: bool
    crawler_enabled: bool
    write_enabled: bool
    dry_run_only: bool
    scheduler_attached: bool
    database_write_allowed: bool

    def to_dict(self):
        return asdict(self)


class MarketIntelService:
    """市場情報入口服務，先集中 feature gate 與安全狀態。"""

    phase = MARKET_INTEL_PHASE

    def get_runtime_status(self) -> MarketIntelRuntimeStatus:
        return MarketIntelRuntimeStatus(
            phase=self.phase,
            enabled=MARKET_INTEL_ENABLED,
            crawler_enabled=MARKET_INTEL_CRAWLER_ENABLED,
            write_enabled=MARKET_INTEL_WRITE_ENABLED,
            dry_run_only=not MARKET_INTEL_WRITE_ENABLED,
            scheduler_attached=False,
            database_write_allowed=(
                MARKET_INTEL_ENABLED
                and MARKET_INTEL_CRAWLER_ENABLED
                and MARKET_INTEL_WRITE_ENABLED
            ),
        )

    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,
            "created_at": datetime.now(TAIPEI_TZ).replace(tzinfo=None).isoformat(),
            "phase": self.phase,
            "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 = []
        status = self.get_runtime_status()
        mcp_fetch_gate = self.build_mcp_fetch_gate(
            fetch_requested=fetch,
            execute_readiness=bool(fetch and status.enabled and status.crawler_enabled),
        )

        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=status,
            http_get=http_get,
            network_allowed_override=mcp_fetch_gate["network_request_allowed"],
            network_gate=mcp_fetch_gate,
        )
        return {
            "platform_code": platform_code or "all",
            "found": True,
            "fetch_requested": bool(fetch),
            "manual_fetch_allowed": self.manual_fetch_allowed(),
            "mcp_fetch_gate": mcp_fetch_gate,
            "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,
        engine=None,
        database_url=None,
        database_type=None,
    ):
        """回報正式 DB schema 只讀探針；預設不連 DB。"""
        probe = build_schema_db_probe_plan(
            MARKET_INTEL_TABLES,
            execute_requested=execute_requested,
            engine=engine,
            database_url=database_url,
            database_type=database_type,
        )
        probe["phase"] = self.phase
        return probe

    def build_platform_seed_db_diff(
        self,
        platform_code="all",
        *,
        execute_requested=False,
        engine=None,
        database_url=None,
        database_type=None,
    ):
        """回報 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,
            engine=engine,
            database_url=database_url,
            database_type=database_type,
        )
        diff["phase"] = self.phase
        diff["platform_code"] = platform_code or "all"
        diff["seed_plan_found"] = bool(seed_plan["found"])
        return diff

    def build_legacy_source_bridge(
        self,
        *,
        execute_requested=False,
        sample_limit=5,
        engine=None,
        database_url=None,
        database_type=None,
    ):
        """回報既有 EDM/PChome 資料源橋接 preview；預設不連 DB。"""
        bridge = build_legacy_source_bridge_plan(
            execute_requested=execute_requested,
            sample_limit=sample_limit,
            engine=engine,
            database_url=database_url,
            database_type=database_type,
        )
        bridge["phase"] = self.phase
        return bridge

    def build_mcp_readiness(
        self,
        *,
        execute_requested=False,
        timeout_sec=3,
        engine=None,
        database_url=None,
        database_type=None,
    ):
        readiness = build_mcp_readiness_plan(
            execute_requested=execute_requested,
            timeout_sec=timeout_sec,
            engine=engine,
            database_url=database_url,
            database_type=database_type,
        )
        readiness["phase"] = self.phase
        return readiness

    def build_mcp_tool_contract(self):
        contract = build_mcp_tool_contract_preview()
        contract["phase"] = self.phase
        return contract

    def build_mcp_deploy_preflight(self):
        preflight = build_mcp_deploy_preflight_plan()
        preflight["phase"] = self.phase
        return preflight

    def build_mcp_activation_runbook(self):
        preflight = self.build_mcp_deploy_preflight()
        readiness = self.build_mcp_readiness()
        runbook = build_mcp_activation_runbook_preview(
            preflight=preflight,
            readiness=readiness,
        )
        runbook["phase"] = self.phase
        return runbook

    def build_mcp_fetch_gate(self, *, fetch_requested=False, execute_readiness=False):
        gate = build_mcp_fetch_gate_preview(
            self.get_runtime_status(),
            fetch_requested=fetch_requested,
            execute_readiness=execute_readiness,
        )
        gate["phase"] = self.phase
        return gate

    def build_mcp_completion_audit(self):
        return build_mcp_completion_audit_for_runtime(
            runtime_status=self.get_runtime_status(),
            phase=self.phase,
        )

    def build_scheduler_plan(self):
        """回報市場情報排程掛載計畫；不註冊 job、不啟動 crawler。"""
        plan = build_scheduler_attach_plan(
            runtime_status=self.get_runtime_status(),
            mcp_fetch_gate=self.build_mcp_fetch_gate(),
            schema_smoke=build_schema_smoke(MARKET_INTEL_TABLES),
        )
        plan["phase"] = self.phase
        return plan

    def build_manual_sample_plan(self):
        """回報第一次人工 sample fetch 計畫；不抓外部頁、不寫 DB。"""
        plan = build_manual_sample_fetch_plan_preview(
            runtime_status=self.get_runtime_status(),
            adapters=get_adapter_registry().values(),
            mcp_fetch_gate=self.build_mcp_fetch_gate(),
            live_db_inventory=self.build_live_db_inventory(),
        )
        plan["phase"] = self.phase
        return plan

    def build_manual_sample_acceptance(self):
        """回報人工 sample result 驗收契約；不載入外部結果、不寫 DB。"""
        acceptance = build_manual_sample_acceptance_preview(
            runtime_status=self.get_runtime_status(),
            manual_sample_plan=self.build_manual_sample_plan(),
        )
        acceptance["phase"] = self.phase
        return acceptance

    def build_manual_sample_review(self, sample_result=None):
        """回報人工 sample result 審核預覽；預設不載入結果、不寫 DB。"""
        review = build_manual_sample_review_preview(
            runtime_status=self.get_runtime_status(),
            acceptance_contract=self.build_manual_sample_acceptance(),
            sample_result=sample_result,
        )
        review["phase"] = self.phase
        return review

    def build_manual_sample_review_evaluation(
        self,
        sample_result=None,
        payload_error=None,
    ):
        """回報 POST sample result 即時審核；不保存 payload、不寫 DB。"""
        review = build_manual_sample_review_evaluation_preview(
            runtime_status=self.get_runtime_status(),
            acceptance_contract=self.build_manual_sample_acceptance(),
            sample_result=sample_result,
            payload_error=payload_error,
        )
        review["phase"] = self.phase
        return review

    def build_manual_sample_candidate_handoff(
        self,
        sample_result=None,
        payload_error=None,
        limit=20,
    ):
        """回報人工樣本候選活動 handoff；只產生 preview，不寫 DB。"""
        handoff = build_manual_sample_candidate_handoff_preview(
            runtime_status=self.get_runtime_status(),
            acceptance_contract=self.build_manual_sample_acceptance(),
            sample_result=sample_result,
            payload_error=payload_error,
            limit=limit,
        )
        handoff["phase"] = self.phase
        return handoff

    def build_manual_sample_candidate_queue_draft(
        self,
        sample_result=None,
        payload_error=None,
        limit=20,
    ):
        """回報人工候選活動審核 queue 草案；只產生 preview，不寫 DB。"""
        queue_draft = build_manual_sample_candidate_queue_draft_preview(
            runtime_status=self.get_runtime_status(),
            acceptance_contract=self.build_manual_sample_acceptance(),
            sample_result=sample_result,
            payload_error=payload_error,
            limit=limit,
        )
        queue_draft["phase"] = self.phase
        return queue_draft

    def build_manual_sample_candidate_queue_approval(
        self,
        sample_result=None,
        payload_error=None,
        limit=20,
    ):
        """回報候選審核 queue 寫入前 gate；只產生 preview，不寫 DB。"""
        approval = build_manual_sample_candidate_queue_approval_preview(
            runtime_status=self.get_runtime_status(),
            acceptance_contract=self.build_manual_sample_acceptance(),
            sample_result=sample_result,
            payload_error=payload_error,
            limit=limit,
        )
        approval["phase"] = self.phase
        return approval

    def build_manual_sample_candidate_queue_transaction(self, sample_result=None, payload_error=None, limit=20):
        """回報候選審核 queue transaction preview；不開 transaction、不寫 DB。"""
        transaction = build_manual_sample_candidate_queue_transaction_preview(
            runtime_status=self.get_runtime_status(),
            acceptance_contract=self.build_manual_sample_acceptance(),
            sample_result=sample_result,
            payload_error=payload_error,
            limit=limit,
        )
        transaction["phase"] = self.phase
        return transaction

    def build_match_review_plan(self):
        """回報商品比對審核計畫；不建立 queue、不自動確認。"""
        schema_smoke = self.build_schema_smoke()
        plan = build_match_review_plan_preview(
            runtime_status=self.get_runtime_status(),
            mcp_tool_contract=self.build_mcp_tool_contract(),
            legacy_source_bridge=self.build_legacy_source_bridge(),
            schema_smoke={
                **schema_smoke["schema_smoke"],
                "expected_tables": schema_smoke["expected_tables"],
            },
        )
        plan["phase"] = self.phase
        return plan

    def build_opportunity_plan(self):
        """回報市場機會與威脅判讀計畫；不派送、不產生 AI 報告。"""
        plan = build_opportunity_plan_preview(
            runtime_status=self.get_runtime_status(),
            match_review_plan=self.build_match_review_plan(),
            scheduler_plan=self.build_scheduler_plan(),
            legacy_source_bridge=self.build_legacy_source_bridge(),
        )
        plan["phase"] = self.phase
        return plan

    def build_opportunity_scoring_plan(self):
        """回報機會與威脅分數模型計畫；不計分、不寫入。"""
        schema_smoke = self.build_schema_smoke()
        plan = build_opportunity_scoring_plan_preview(
            runtime_status=self.get_runtime_status(),
            opportunity_plan=self.build_opportunity_plan(),
            match_review_plan=self.build_match_review_plan(),
            scheduler_plan=self.build_scheduler_plan(),
            schema_smoke=schema_smoke,
            mcp_fetch_gate=self.build_mcp_fetch_gate(),
        )
        plan["phase"] = self.phase
        return plan

    def build_opportunity_evidence_plan(self):
        """回報機會與威脅 evidence bundle 計畫；不查詢、不造樣本。"""
        schema_smoke = self.build_schema_smoke()
        plan = build_opportunity_evidence_plan_preview(
            runtime_status=self.get_runtime_status(),
            opportunity_plan=self.build_opportunity_plan(),
            scoring_plan=self.build_opportunity_scoring_plan(),
            match_review_plan=self.build_match_review_plan(),
            legacy_source_bridge=self.build_legacy_source_bridge(),
            schema_smoke=schema_smoke,
        )
        plan["phase"] = self.phase
        return plan

    def build_opportunity_alert_plan(self):
        """回報機會與威脅告警候選計畫；不派送、不呼叫 LLM。"""
        plan = build_opportunity_alert_plan_preview(
            runtime_status=self.get_runtime_status(),
            opportunity_plan=self.build_opportunity_plan(),
            scoring_plan=self.build_opportunity_scoring_plan(),
            evidence_plan=self.build_opportunity_evidence_plan(),
            scheduler_plan=self.build_scheduler_plan(),
        )
        plan["phase"] = self.phase
        return plan

    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_migration_apply_drill(
        self,
        *,
        execute_requested=False,
        schema_engine=None,
        seed_diff_engine=None,
        database_url=None,
        database_type=None,
    ):
        """建立正式 migration 前的只讀演練；不執行 migration 或 rollback。"""
        schema_db_probe = self.build_schema_db_probe(
            execute_requested=execute_requested,
            engine=schema_engine,
            database_url=database_url,
            database_type=database_type,
        )
        platform_seed_db_diff = self.build_platform_seed_db_diff(
            execute_requested=execute_requested,
            engine=seed_diff_engine,
            database_url=database_url,
            database_type=database_type,
        )
        drill = build_migration_apply_drill_preview(
            runtime_status=self.get_runtime_status(),
            migration_blueprint=self.build_migration_blueprint(),
            schema_db_probe=schema_db_probe,
            platform_seed_db_diff=platform_seed_db_diff,
        )
        drill["phase"] = self.phase
        return drill

    def build_migration_catalog_review(
        self,
        *,
        execute_requested=False,
        schema_engine=None,
        seed_diff_engine=None,
        database_url=None,
        database_type=None,
    ):
        """建立正式 DB catalog 判讀；不執行 migration、不寫 DB。"""
        schema_db_probe = self.build_schema_db_probe(
            execute_requested=execute_requested,
            engine=schema_engine,
            database_url=database_url,
            database_type=database_type,
        )
        platform_seed_db_diff = self.build_platform_seed_db_diff(
            execute_requested=execute_requested,
            engine=seed_diff_engine,
            database_url=database_url,
            database_type=database_type,
        )
        review = build_migration_catalog_review_preview(
            runtime_status=self.get_runtime_status(),
            migration_blueprint=self.build_migration_blueprint(),
            schema_db_probe=schema_db_probe,
            platform_seed_db_diff=platform_seed_db_diff,
        )
        review["phase"] = self.phase
        return review

    def build_migration_live_smoke(
        self,
        *,
        execute_requested=False,
        schema_engine=None,
        seed_diff_engine=None,
        database_url=None,
        database_type=None,
    ):
        """建立正式 DB 只讀 smoke；不執行 migration、不寫 DB。"""
        catalog_review = self.build_migration_catalog_review(
            execute_requested=execute_requested,
            schema_engine=schema_engine,
            seed_diff_engine=seed_diff_engine,
            database_url=database_url,
            database_type=database_type,
        )
        smoke = build_migration_live_smoke_preview(
            runtime_status=self.get_runtime_status(),
            catalog_review=catalog_review,
        )
        smoke["phase"] = self.phase
        return smoke

    def build_live_db_inventory(
        self,
        *,
        execute_requested=False,
        engine=None,
        database_url=None,
        database_type=None,
    ):
        """建立 market_* 正式 DB 只讀庫存總覽；預設不連 DB。"""
        inventory = build_live_db_inventory_preview(
            self.get_schema_tables(),
            execute_requested=execute_requested,
            engine=engine,
            database_url=database_url,
            database_type=database_type,
        )
        inventory["phase"] = self.phase
        return inventory

    def build_seed_writer_cli_status(
        self,
        platform_code="all",
        *,
        execute_requested=False,
        approval_token=None,
        approval_token_secret=None,
        apply_real_write=False,
        engine=None,
        database_url=None,
        database_type=None,
    ):
        """建立 seed writer CLI status；只有 CLI 明確批准時才寫入。"""
        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,
            approval_token_secret=approval_token_secret,
            apply_real_write=apply_real_write,
            seed_plan=seed_plan,
            write_guard=write_guard,
            writer_plan=writer_plan,
            migration_blueprint=migration_blueprint,
            engine=engine,
            database_url=database_url,
            database_type=database_type,
        )
        status["phase"] = self.phase
        return status

    def build_deployment_readiness(self):
        """建立市場情報推版準備狀態；不執行 git、部署或遠端操作。"""
        return build_deployment_readiness_preview(
            service=self,
            market_intel_tables=MARKET_INTEL_TABLES,
            schema_smoke_builder=build_schema_smoke,
        )