#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""PChome 業績成長自動化作戰系統的只讀作戰清單。"""

from __future__ import annotations

import hashlib
import json
import logging
from datetime import date, datetime, time
from typing import Any

from sqlalchemy import bindparam, inspect, text

from services.external_market_offer_service import build_external_source_readiness
from services.pchome_sales_freshness_policy import (
    TAIPEI_TZ,
    evaluate_pchome_sales_freshness,
)

logger = logging.getLogger(__name__)

SYSTEM_DISPLAY_NAME = "PChome 業績成長自動化作戰系統"
PRIMARY_SALES_SOURCE = "PChome 後台業績"
ACTIVE_EXTERNAL_SOURCES = ("MOMO 外部價格參考",)
PAUSED_EXTERNAL_SOURCES = ("蝦皮", "酷澎")


def _to_float(value: Any, default: float = 0.0) -> float:
    if value is None:
        return default
    try:
        return float(value)
    except (TypeError, ValueError):
        return default


def _quote_identifier(identifier: str) -> str:
    return '"' + identifier.replace('"', '""') + '"'


def _first_available(columns: set[str], candidates: list[str]) -> str | None:
    return next((col for col in candidates if col in columns), None)


def _load_json_tags(value: Any) -> list[str]:
    if not value:
        return []
    if isinstance(value, list):
        return value
    try:
        parsed = json.loads(value)
        return parsed if isinstance(parsed, list) else []
    except Exception:
        return []


def _source_names_by_status(source_readiness: dict[str, Any], status_code: str) -> list[str]:
    return [
        str(source.get("display_name") or source.get("code") or "").strip()
        for source in source_readiness.get("sources") or []
        if source.get("status_code") == status_code
        and (status_code != "active" or source.get("enabled", True))
        and str(source.get("display_name") or source.get("code") or "").strip()
    ]


def _sales_freshness(
    latest_sales_date: Any,
    *,
    today: date | None = None,
    now: datetime | None = None,
) -> dict[str, Any]:
    if now is None and today is not None:
        now = TAIPEI_TZ.localize(datetime.combine(today, time(hour=23, minute=59)))
    return evaluate_pchome_sales_freshness(latest_sales_date, now=now)


def _table_exists(engine, table_name: str) -> bool:
    try:
        return inspect(engine).has_table(table_name)
    except Exception:
        logger.warning("[PChomeGrowth] table probe failed: %s", table_name, exc_info=True)
        return False


def _daily_sales_columns(conn) -> dict[str, str | None]:
    if conn.dialect.name == "postgresql":
        rows = conn.execute(text("""
            SELECT column_name
            FROM information_schema.columns
            WHERE table_name = 'daily_sales_snapshot'
        """)).fetchall()
        columns = {row[0] for row in rows}
    elif conn.dialect.name == "sqlite":
        rows = conn.execute(text("PRAGMA table_info(daily_sales_snapshot)")).fetchall()
        columns = {row[1] for row in rows}
    else:
        result = conn.execute(text("SELECT * FROM daily_sales_snapshot LIMIT 0"))
        columns = set(result.keys())

    return {
        "sku": _first_available(columns, ["商品ID", "Product ID", "ID", "Item Code"]),
        "name": _first_available(columns, ["商品名稱", "商品名", "品名", "Product Name", "Name"]),
        "date": _first_available(columns, ["snapshot_date", "日期", "訂單日期", "交易日期", "Date"]),
        "revenue": _first_available(columns, ["總業績", "銷售金額", "業績", "金額", "Amount", "Sales", "Total"]),
        "qty": _first_available(columns, ["數量", "銷售數量", "銷量", "Qty", "Quantity"]),
        "price": _first_available(columns, ["商品單位售價", "單價", "售價", "Price", "Unit Price"]),
        "category": _first_available(columns, ["商品館", "館別", "分類", "Category"]),
        "vendor": _first_available(columns, ["廠商名稱", "供應商", "Vendor"]),
    }


def _as_text_expr(identifier_or_expr: str, dialect_name: str, *, raw: bool = False) -> str:
    expr = identifier_or_expr if raw else _quote_identifier(identifier_or_expr)
    if dialect_name == "postgresql":
        return f"{expr}::text"
    return f"CAST({expr} AS TEXT)"


def _numeric_expr(identifier: str, dialect_name: str) -> str:
    quoted = _quote_identifier(identifier)
    if dialect_name == "postgresql":
        return f"COALESCE(NULLIF(regexp_replace({quoted}::text, '[^0-9.-]', '', 'g'), '')::numeric, 0)"
    return f"COALESCE(CAST(REPLACE(CAST({quoted} AS TEXT), ',', '') AS REAL), 0)"


def _fetch_sales_rows(conn, limit: int) -> tuple[list[dict[str, Any]], str | None]:
    cols = _daily_sales_columns(conn)
    missing = [key for key in ["sku", "name", "date", "revenue"] if not cols.get(key)]
    if missing:
        raise RuntimeError("PChome 業績檔缺少必要欄位：" + "、".join(missing))

    dialect = conn.dialect.name
    sku_col = _quote_identifier(cols["sku"])
    date_col = _quote_identifier(cols["date"])
    revenue_expr = _numeric_expr(cols["revenue"], dialect)
    qty_expr = _numeric_expr(cols["qty"], dialect) if cols.get("qty") else "0"
    price_expr = _numeric_expr(cols["price"], dialect) if cols.get("price") else "0"
    category_text = _as_text_expr(cols["category"], dialect) if cols.get("category") else "NULL"
    vendor_text = _as_text_expr(cols["vendor"], dialect) if cols.get("vendor") else "NULL"
    sku_text = _as_text_expr(cols["sku"], dialect)
    name_text = _as_text_expr(cols["name"], dialect)
    candidate_limit = max(limit * 4, 80)

    if dialect == "postgresql":
        sale_date_expr = f"NULLIF({date_col}::text, '')::date"
        curr_window = "lw.latest_date - INTERVAL '6 days'"
        prev_window_start = "lw.latest_date - INTERVAL '13 days'"
        prev_window_end = "lw.latest_date - INTERVAL '6 days'"
    else:
        sale_date_expr = f"date({date_col})"
        curr_window = "date(lw.latest_date, '-6 days')"
        prev_window_start = "date(lw.latest_date, '-13 days')"
        prev_window_end = "date(lw.latest_date, '-6 days')"

    order_metric = "CASE WHEN sales_7d >= sales_prev_7d THEN sales_7d ELSE sales_prev_7d END"
    rows = conn.execute(text(f"""
        WITH sales_rows AS (
            SELECT
                NULLIF(TRIM({sku_text}), '') AS pchome_product_id,
                NULLIF(TRIM({name_text}), '') AS product_name,
                NULLIF(TRIM({_as_text_expr(category_text, dialect, raw=True)}), '') AS category,
                NULLIF(TRIM({_as_text_expr(vendor_text, dialect, raw=True)}), '') AS vendor,
                {sale_date_expr} AS sale_date,
                {revenue_expr} AS revenue,
                {qty_expr} AS qty,
                {price_expr} AS unit_price
            FROM daily_sales_snapshot
            WHERE {sku_col} IS NOT NULL
        ),
        latest_window AS (
            SELECT MAX(sale_date) AS latest_date
            FROM sales_rows
            WHERE sale_date IS NOT NULL
        ),
        sales AS (
            SELECT
                sr.pchome_product_id,
                MAX(sr.product_name) AS product_name,
                MAX(sr.category) AS category,
                MAX(sr.vendor) AS vendor,
                SUM(CASE WHEN sr.sale_date >= {curr_window}
                         THEN sr.revenue ELSE 0 END) AS sales_7d,
                SUM(CASE WHEN sr.sale_date >= {prev_window_start}
                          AND sr.sale_date < {prev_window_end}
                         THEN sr.revenue ELSE 0 END) AS sales_prev_7d,
                SUM(CASE WHEN sr.sale_date >= {curr_window}
                         THEN sr.qty ELSE 0 END) AS qty_7d,
                CASE
                    WHEN SUM(CASE WHEN sr.sale_date >= {curr_window} THEN sr.qty ELSE 0 END) > 0
                    THEN SUM(CASE WHEN sr.sale_date >= {curr_window} THEN sr.revenue ELSE 0 END)
                         / NULLIF(SUM(CASE WHEN sr.sale_date >= {curr_window} THEN sr.qty ELSE 0 END), 0)
                    ELSE NULLIF(MAX(CASE WHEN sr.sale_date >= {curr_window} THEN sr.unit_price ELSE 0 END), 0)
                END AS pchome_price,
                MAX(sr.sale_date) AS last_sale_date,
                MAX(lw.latest_date) AS latest_sales_date
            FROM sales_rows sr
            CROSS JOIN latest_window lw
            WHERE sr.pchome_product_id IS NOT NULL
            GROUP BY sr.pchome_product_id
        )
        SELECT *
        FROM sales
        WHERE sales_7d > 0 OR sales_prev_7d > 0
        ORDER BY {order_metric} DESC, qty_7d DESC, pchome_product_id ASC
        LIMIT :limit
    """), {"limit": candidate_limit}).mappings().all()

    mapped_rows = [dict(row) for row in rows]
    latest_date = None
    for row in mapped_rows:
        value = row.get("latest_sales_date")
        if value:
            latest_date = value.isoformat() if hasattr(value, "isoformat") else str(value)
            break
    return mapped_rows, latest_date


def _fetch_sales_summary(conn) -> dict[str, Any]:
    cols = _daily_sales_columns(conn)
    missing = [key for key in ["sku", "date", "revenue"] if not cols.get(key)]
    if missing:
        return {}

    dialect = conn.dialect.name
    sku_col = _quote_identifier(cols["sku"])
    date_col = _quote_identifier(cols["date"])
    revenue_expr = _numeric_expr(cols["revenue"], dialect)
    qty_expr = _numeric_expr(cols["qty"], dialect) if cols.get("qty") else "0"
    category_text = _as_text_expr(cols["category"], dialect) if cols.get("category") else "NULL"
    sku_text = _as_text_expr(cols["sku"], dialect)

    if dialect == "postgresql":
        sale_date_expr = f"NULLIF({date_col}::text, '')::date"
        curr_window = "lw.latest_date - INTERVAL '6 days'"
        prev_window_start = "lw.latest_date - INTERVAL '13 days'"
        prev_window_end = "lw.latest_date - INTERVAL '6 days'"
    else:
        sale_date_expr = f"date({date_col})"
        curr_window = "date(lw.latest_date, '-6 days')"
        prev_window_start = "date(lw.latest_date, '-13 days')"
        prev_window_end = "date(lw.latest_date, '-6 days')"

    row = conn.execute(text(f"""
        WITH sales_rows AS (
            SELECT
                NULLIF(TRIM({sku_text}), '') AS pchome_product_id,
                NULLIF(TRIM({_as_text_expr(category_text, dialect, raw=True)}), '') AS category,
                {sale_date_expr} AS sale_date,
                {revenue_expr} AS revenue,
                {qty_expr} AS qty
            FROM daily_sales_snapshot
            WHERE {sku_col} IS NOT NULL
        ),
        latest_window AS (
            SELECT MAX(sale_date) AS latest_date
            FROM sales_rows
            WHERE sale_date IS NOT NULL
        ),
        per_product AS (
            SELECT
                sr.pchome_product_id,
                MAX(sr.category) AS category,
                SUM(CASE WHEN sr.sale_date >= {curr_window}
                         THEN sr.revenue ELSE 0 END) AS sales_7d,
                SUM(CASE WHEN sr.sale_date >= {prev_window_start}
                          AND sr.sale_date < {prev_window_end}
                         THEN sr.revenue ELSE 0 END) AS sales_prev_7d,
                SUM(CASE WHEN sr.sale_date >= {curr_window}
                         THEN sr.qty ELSE 0 END) AS qty_7d,
                MAX(lw.latest_date) AS latest_sales_date
            FROM sales_rows sr
            CROSS JOIN latest_window lw
            WHERE sr.pchome_product_id IS NOT NULL
            GROUP BY sr.pchome_product_id
        ),
        category_sales AS (
            SELECT
                COALESCE(NULLIF(category, ''), '未分類') AS category,
                SUM(sales_7d) AS sales_7d
            FROM per_product
            GROUP BY COALESCE(NULLIF(category, ''), '未分類')
        )
        SELECT
            MAX(latest_sales_date) AS latest_sales_date,
            COALESCE(SUM(sales_7d), 0) AS overall_sales_7d,
            COALESCE(SUM(sales_prev_7d), 0) AS overall_sales_prev_7d,
            COALESCE(SUM(qty_7d), 0) AS overall_qty_7d,
            SUM(CASE WHEN sales_7d > 0 THEN 1 ELSE 0 END) AS active_product_count,
            SUM(CASE WHEN sales_prev_7d > 0 AND sales_7d < sales_prev_7d THEN 1 ELSE 0 END) AS declining_product_count,
            (
                SELECT category
                FROM category_sales
                WHERE sales_7d > 0
                ORDER BY sales_7d DESC
                LIMIT 1
            ) AS top_category,
            (
                SELECT COALESCE(sales_7d, 0)
                FROM category_sales
                WHERE sales_7d > 0
                ORDER BY sales_7d DESC
                LIMIT 1
            ) AS top_category_sales_7d
        FROM per_product
    """)).mappings().first()

    if not row:
        return {}
    current = _to_float(row.get("overall_sales_7d"))
    previous = _to_float(row.get("overall_sales_prev_7d"))
    delta_pct = ((current - previous) / previous * 100) if previous else None
    latest_date = row.get("latest_sales_date")
    return {
        "overall_latest_sales_date": latest_date.isoformat() if hasattr(latest_date, "isoformat") else str(latest_date or ""),
        "overall_sales_7d": round(current, 2),
        "overall_sales_prev_7d": round(previous, 2),
        "overall_sales_delta_pct": round(delta_pct, 1) if delta_pct is not None else None,
        "overall_qty_7d": round(_to_float(row.get("overall_qty_7d")), 2),
        "active_product_count": int(row.get("active_product_count") or 0),
        "declining_product_count": int(row.get("declining_product_count") or 0),
        "top_category": row.get("top_category") or "",
        "top_category_sales_7d": round(_to_float(row.get("top_category_sales_7d")), 2),
    }


def _json_dict(value: Any) -> dict[str, Any]:
    if not value:
        return {}
    if isinstance(value, dict):
        return value
    try:
        parsed = json.loads(value)
        return parsed if isinstance(parsed, dict) else {}
    except Exception:
        return {}


def _match_score_from_quality(value: Any) -> float:
    score = _to_float(value)
    if score > 1:
        score = score / 100
    return max(0, min(1, score))


def _external_price_basis(raw_payload: dict[str, Any]) -> str:
    price_basis = str(raw_payload.get("price_basis") or "").strip()
    return price_basis if price_basis in {"total_price", "unit_price"} else "total_price"


def _external_price_basis_label(price_basis: str) -> str:
    return "單位價" if price_basis == "unit_price" else "商品總價"


def _fetch_normalized_external_offer_map(conn, pchome_product_ids: list[str]) -> dict[str, list[dict[str, Any]]]:
    inspector = inspect(conn)
    if not inspector.has_table("external_offers"):
        return {}
    ids = [str(item).strip() for item in pchome_product_ids if str(item or "").strip()]
    if not ids:
        return {}

    if conn.dialect.name == "postgresql":
        sql = """
            WITH latest_offer AS (
                SELECT DISTINCT ON (eo.pchome_product_id, eo.source_code)
                    eo.pchome_product_id,
                    eo.source_code,
                    eo.platform_code,
                    eo.source_product_id AS momo_sku,
                    eo.title AS momo_name,
                    eo.product_url,
                    eo.price AS momo_price,
                    eo.quality_score,
                    eo.quality_notes_json,
                    eo.raw_payload_json,
                    eo.observed_at,
                    eo.ingestion_method
                FROM external_offers eo
                WHERE eo.pchome_product_id IS NOT NULL
                  AND eo.pchome_product_id IN :ids
                  AND eo.price IS NOT NULL
                  AND eo.price > 0
                  AND COALESCE(eo.quality_score, 0) >= 76
                  AND eo.match_status IN ('verified', 'usable', 'reviewed', 'exact', 'confirmed')
                  AND eo.data_quality_status IN ('verified', 'usable', 'reviewed')
                  AND (eo.expires_at IS NULL OR eo.expires_at > CURRENT_TIMESTAMP)
                ORDER BY eo.pchome_product_id, eo.source_code, eo.observed_at DESC NULLS LAST, eo.id DESC
            )
            SELECT * FROM latest_offer
        """
    else:
        sql = """
            WITH latest_offer AS (
                SELECT
                    eo.pchome_product_id,
                    eo.source_code,
                    eo.platform_code,
                    eo.source_product_id AS momo_sku,
                    eo.title AS momo_name,
                    eo.product_url,
                    eo.price AS momo_price,
                    eo.quality_score,
                    eo.quality_notes_json,
                    eo.raw_payload_json,
                    eo.observed_at,
                    eo.ingestion_method,
                    ROW_NUMBER() OVER (
                        PARTITION BY eo.pchome_product_id, eo.source_code
                        ORDER BY eo.observed_at DESC, eo.id DESC
                    ) AS rn
                FROM external_offers eo
                WHERE eo.pchome_product_id IS NOT NULL
                  AND eo.pchome_product_id IN :ids
                  AND eo.price IS NOT NULL
                  AND eo.price > 0
                  AND COALESCE(eo.quality_score, 0) >= 76
                  AND eo.match_status IN ('verified', 'usable', 'reviewed', 'exact', 'confirmed')
                  AND eo.data_quality_status IN ('verified', 'usable', 'reviewed')
            )
            SELECT *
            FROM latest_offer
            WHERE rn = 1
        """

    stmt = text(sql).bindparams(bindparam("ids", expanding=True))
    rows = conn.execute(stmt, {"ids": ids}).mappings().all()
    result: dict[str, list[dict[str, Any]]] = {}
    for row in rows:
        raw_payload = _json_dict(row.get("raw_payload_json"))
        price_basis = _external_price_basis(raw_payload)
        unit_price_comparison = _json_dict(raw_payload.get("unit_price_comparison"))
        key = str(row.get("pchome_product_id") or "").strip()
        if not key:
            continue
        offer = {
            "pchome_product_id": key,
            "source_code": str(row.get("source_code") or "").strip(),
            "platform_code": str(row.get("platform_code") or row.get("source_code") or "").strip(),
            "pchome_public_name": raw_payload.get("pchome_public_name"),
            "momo_sku": row.get("momo_sku"),
            "momo_name": row.get("momo_name"),
            "product_url": row.get("product_url"),
            "momo_price": row.get("momo_price"),
            "pchome_price": raw_payload.get("pchome_public_price"),
            "price_basis": price_basis,
            "price_basis_label": _external_price_basis_label(price_basis),
            "unit_label": unit_price_comparison.get("unit_label"),
            "momo_unit_price": unit_price_comparison.get("momo_unit_price"),
            "pchome_unit_price": unit_price_comparison.get("competitor_unit_price"),
            "unit_gap_pct": unit_price_comparison.get("unit_gap_pct"),
            "momo_total_quantity": unit_price_comparison.get("momo_total_quantity"),
            "pchome_total_quantity": unit_price_comparison.get("competitor_total_quantity"),
            "match_score": _match_score_from_quality(row.get("quality_score")),
            "tags": raw_payload.get("tags") or ["external_offers", "verified"],
            "crawled_at": row.get("observed_at"),
            "momo_price_at": row.get("observed_at"),
            "data_source": "external_offers",
            "ingestion_method": row.get("ingestion_method"),
        }
        result.setdefault(key, []).append(offer)
    return result


def _fetch_legacy_external_price_map(conn, pchome_product_ids: list[str]) -> dict[str, dict[str, Any]]:
    inspector = inspect(conn)
    if not all(inspector.has_table(table) for table in {"competitor_prices", "products", "price_records"}):
        return {}
    ids = [str(item).strip() for item in pchome_product_ids if str(item or "").strip()]
    if not ids:
        return {}

    if conn.dialect.name == "postgresql":
        sql = """
            WITH valid_cp AS (
                SELECT DISTINCT ON (cp.competitor_product_id)
                    cp.competitor_product_id AS pchome_product_id,
                    cp.competitor_product_name AS pchome_public_name,
                    cp.sku AS momo_sku,
                    cp.price AS pchome_price,
                    cp.match_score,
                    cp.tags,
                    cp.crawled_at
                FROM competitor_prices cp
                WHERE cp.source = 'pchome'
                  AND cp.competitor_product_id IS NOT NULL
                  AND cp.price IS NOT NULL
                  AND cp.price > 0
                  AND COALESCE(cp.match_score, 0) >= 0.76
                  AND (cp.expires_at IS NULL OR cp.expires_at > CURRENT_TIMESTAMP)
                  AND COALESCE(cp.tags, '[]'::jsonb) ? 'identity_v2'
                  AND cp.competitor_product_id IN :ids
                ORDER BY cp.competitor_product_id, cp.crawled_at DESC NULLS LAST
            )
            SELECT vc.*, lm.momo_name, lm.momo_price, lm.momo_price_at
            FROM valid_cp vc
            LEFT JOIN LATERAL (
                SELECT
                    p.name AS momo_name,
                    pr.price AS momo_price,
                    pr.timestamp AS momo_price_at
                FROM products p
                JOIN price_records pr ON pr.product_id = p.id
                WHERE p.i_code = vc.momo_sku
                  AND p.status = 'ACTIVE'
                ORDER BY pr.timestamp DESC, pr.id DESC
                LIMIT 1
            ) lm ON TRUE
        """
    else:
        sql = """
            WITH latest_cp AS (
                SELECT
                    cp.competitor_product_id AS pchome_product_id,
                    cp.competitor_product_name AS pchome_public_name,
                    cp.sku AS momo_sku,
                    cp.price AS pchome_price,
                    cp.match_score,
                    cp.tags,
                    cp.crawled_at,
                    ROW_NUMBER() OVER (
                        PARTITION BY cp.competitor_product_id
                        ORDER BY cp.crawled_at DESC
                    ) AS rn
                FROM competitor_prices cp
                WHERE cp.source = 'pchome'
                  AND cp.competitor_product_id IS NOT NULL
                  AND cp.price IS NOT NULL
                  AND cp.price > 0
                  AND COALESCE(cp.match_score, 0) >= 0.76
                  AND COALESCE(cp.tags, '') LIKE '%identity_v2%'
                  AND cp.competitor_product_id IN :ids
            ),
            latest_momo AS (
                SELECT
                    p.i_code AS momo_sku,
                    p.name AS momo_name,
                    pr.price AS momo_price,
                    pr.timestamp AS momo_price_at,
                    ROW_NUMBER() OVER (
                        PARTITION BY p.i_code
                        ORDER BY pr.timestamp DESC, pr.id DESC
                    ) AS rn
                FROM products p
                JOIN price_records pr ON pr.product_id = p.id
                WHERE p.status = 'ACTIVE'
                  AND p.i_code IN (SELECT momo_sku FROM latest_cp)
            )
            SELECT cp.pchome_product_id, cp.pchome_public_name, cp.momo_sku,
                   cp.pchome_price, cp.match_score, cp.tags, cp.crawled_at,
                   lm.momo_name, lm.momo_price, lm.momo_price_at
            FROM latest_cp cp
            LEFT JOIN latest_momo lm ON lm.momo_sku = cp.momo_sku AND lm.rn = 1
            WHERE cp.rn = 1
        """

    stmt = text(sql).bindparams(bindparam("ids", expanding=True))
    rows = conn.execute(stmt, {"ids": ids}).mappings().all()
    result: dict[str, dict[str, Any]] = {}
    for row in rows:
        key = str(row.get("pchome_product_id") or "").strip()
        if key:
            item = dict(row)
            item["data_source"] = "competitor_prices"
            result[key] = item
    return result


def _external_offer_gap(offer: dict[str, Any]) -> float:
    price_basis = str(offer.get("price_basis") or "total_price")
    if price_basis == "unit_price":
        explicit_gap = offer.get("unit_gap_pct")
        if explicit_gap is not None:
            return _to_float(explicit_gap)
        pchome_price = _to_float(offer.get("pchome_unit_price"))
        external_price = _to_float(offer.get("momo_unit_price"))
    else:
        pchome_price = _to_float(offer.get("pchome_price"))
        external_price = _to_float(offer.get("momo_price"))
    if pchome_price > 0 and external_price > 0:
        return (external_price - pchome_price) / pchome_price * 100
    return 9999.0


def _fetch_external_offer_maps(
    conn,
    pchome_product_ids: list[str],
    source_readiness: dict[str, Any],
) -> tuple[dict[str, dict[str, Any]], dict[str, list[dict[str, Any]]]]:
    normalized_map = _fetch_normalized_external_offer_map(conn, pchome_product_ids)
    source_policy = {
        str(source.get("code") or ""): source
        for source in source_readiness.get("sources") or []
    }
    primary_map: dict[str, dict[str, Any]] = {}
    market_map: dict[str, list[dict[str, Any]]] = {}

    for product_id, offers in normalized_map.items():
        normalized_offers = []
        for offer in offers:
            source_code = str(offer.get("source_code") or "").strip()
            policy = source_policy.get(source_code, {})
            enriched = {
                **offer,
                "source_display_name": policy.get("display_name") or source_code or "外部平台",
                "source_status": policy.get("status_code") or "unregistered",
                "source_enabled": bool(policy.get("enabled", False)),
                "actionable": bool(policy.get("can_alert")),
            }
            enriched["gap_pct"] = round(_external_offer_gap(enriched), 1)
            normalized_offers.append(enriched)
        normalized_offers.sort(key=lambda item: (_external_offer_gap(item), item.get("source_code") or ""))
        market_map[product_id] = normalized_offers
        actionable = [offer for offer in normalized_offers if offer.get("actionable")]
        if actionable:
            primary_map[product_id] = actionable[0]

    missing_ids = [
        str(item).strip()
        for item in pchome_product_ids
        if str(item or "").strip() and str(item).strip() not in primary_map
    ]
    legacy_map = _fetch_legacy_external_price_map(conn, missing_ids)
    momo_policy = source_policy.get("momo_reference", {})
    for product_id, legacy in legacy_map.items():
        legacy_offer = {
            **legacy,
            "source_code": "momo_reference",
            "platform_code": "momo",
            "source_display_name": momo_policy.get("display_name") or "MOMO 外部價格參考",
            "source_status": momo_policy.get("status_code") or "active",
            "source_enabled": bool(momo_policy.get("enabled", True)),
            "actionable": bool(momo_policy.get("can_alert", True)),
        }
        legacy_offer["gap_pct"] = round(_external_offer_gap(legacy_offer), 1)
        existing_sources = {
            str(item.get("source_code") or "") for item in market_map.get(product_id, [])
        }
        if "momo_reference" not in existing_sources:
            market_map.setdefault(product_id, []).append(legacy_offer)
        if legacy_offer["actionable"]:
            primary_map[product_id] = legacy_offer

    for offers in market_map.values():
        offers.sort(key=lambda item: (_external_offer_gap(item), item.get("source_code") or ""))

    return primary_map, market_map


def _fetch_catalog_mapping_summary(conn) -> dict[str, Any]:
    cols = _daily_sales_columns(conn)
    sku_column = cols.get("sku")
    if not sku_column:
        return {
            "catalog_product_count": 0,
            "mapped_product_count": 0,
            "catalog_mapping_rate": 0.0,
            "active_catalog_product_count": 0,
            "active_mapped_product_count": 0,
            "active_catalog_mapping_rate": 0.0,
        }

    sales_rows = conn.execute(text(
        f"SELECT DISTINCT TRIM({_as_text_expr(sku_column, conn.dialect.name)}) AS product_id "
        "FROM daily_sales_snapshot "
        f"WHERE {_quote_identifier(sku_column)} IS NOT NULL"
    )).fetchall()
    sales_ids = {str(row[0]).strip() for row in sales_rows if str(row[0] or "").strip()}
    active_sales_ids: set[str] = set()
    date_column = cols.get("date")
    revenue_column = cols.get("revenue")
    if date_column and revenue_column:
        dialect = conn.dialect.name
        sku_text = _as_text_expr(sku_column, dialect)
        date_col = _quote_identifier(date_column)
        sale_date_expr = (
            f"NULLIF({date_col}::text, '')::date"
            if dialect == "postgresql"
            else f"date({date_col})"
        )
        active_window = (
            "lw.latest_date - INTERVAL '6 days'"
            if dialect == "postgresql"
            else "date(lw.latest_date, '-6 days')"
        )
        active_rows = conn.execute(text(f"""
            WITH sales_rows AS (
                SELECT
                    NULLIF(TRIM({sku_text}), '') AS product_id,
                    {sale_date_expr} AS sale_date,
                    {_numeric_expr(revenue_column, dialect)} AS revenue
                FROM daily_sales_snapshot
                WHERE {_quote_identifier(sku_column)} IS NOT NULL
            ),
            latest_window AS (
                SELECT MAX(sale_date) AS latest_date
                FROM sales_rows
                WHERE sale_date IS NOT NULL
            )
            SELECT sr.product_id
            FROM sales_rows sr
            CROSS JOIN latest_window lw
            WHERE sr.product_id IS NOT NULL
              AND sr.sale_date >= {active_window}
            GROUP BY sr.product_id
            HAVING SUM(sr.revenue) > 0
        """)).fetchall()
        active_sales_ids = {
            str(row[0]).strip() for row in active_rows if str(row[0] or "").strip()
        }

    mapped_ids: set[str] = set()
    inspector = inspect(conn)
    if inspector.has_table("external_offers"):
        rows = conn.execute(text("""
            SELECT DISTINCT pchome_product_id
            FROM external_offers
            WHERE pchome_product_id IS NOT NULL
              AND price IS NOT NULL
              AND price > 0
              AND COALESCE(quality_score, 0) >= 76
              AND match_status IN ('verified', 'usable', 'reviewed', 'exact', 'confirmed')
              AND data_quality_status IN ('verified', 'usable', 'reviewed')
              AND (expires_at IS NULL OR expires_at > CURRENT_TIMESTAMP)
        """)).fetchall()
        mapped_ids.update(str(row[0]).strip() for row in rows if str(row[0] or "").strip())
    if inspector.has_table("competitor_prices"):
        identity_predicate = (
            "COALESCE(tags, '[]'::jsonb) ? 'identity_v2'"
            if conn.dialect.name == "postgresql"
            else "COALESCE(tags, '') LIKE '%identity_v2%'"
        )
        rows = conn.execute(text(f"""
            SELECT DISTINCT competitor_product_id
            FROM competitor_prices
            WHERE source = 'pchome'
              AND competitor_product_id IS NOT NULL
              AND price IS NOT NULL
              AND price > 0
              AND COALESCE(match_score, 0) >= 0.76
              AND (expires_at IS NULL OR expires_at > CURRENT_TIMESTAMP)
              AND {identity_predicate}
        """)).fetchall()
        mapped_ids.update(str(row[0]).strip() for row in rows if str(row[0] or "").strip())

    mapped_catalog_ids = sales_ids & mapped_ids
    active_mapped_ids = active_sales_ids & mapped_ids
    return {
        "catalog_product_count": len(sales_ids),
        "mapped_product_count": len(mapped_catalog_ids),
        "catalog_mapping_rate": round(len(mapped_catalog_ids) / max(len(sales_ids), 1) * 100, 2),
        "unmapped_product_count": max(0, len(sales_ids) - len(mapped_catalog_ids)),
        "active_catalog_product_count": len(active_sales_ids),
        "active_mapped_product_count": len(active_mapped_ids),
        "active_catalog_mapping_rate": round(
            len(active_mapped_ids) / max(len(active_sales_ids), 1) * 100,
            2,
        ),
        "active_unmapped_product_count": max(
            0,
            len(active_sales_ids) - len(active_mapped_ids),
        ),
    }


def _fetch_review_candidate_map(conn, pchome_product_ids: list[str]) -> dict[str, dict[str, Any]]:
    inspector = inspect(conn)
    if not inspector.has_table("external_offers"):
        return {}
    ids = [str(item).strip() for item in pchome_product_ids if str(item or "").strip()]
    if not ids:
        return {}

    if conn.dialect.name == "postgresql":
        sql = """
            WITH latest_review AS (
                SELECT DISTINCT ON (eo.pchome_product_id)
                    eo.id,
                    eo.pchome_product_id,
                    eo.source_product_id AS momo_sku,
                    eo.title AS momo_name,
                    eo.product_url,
                    eo.price AS momo_price,
                    eo.quality_score,
                    eo.raw_payload_json,
                    eo.observed_at
                FROM external_offers eo
                WHERE eo.source_code = 'momo_reference'
                  AND eo.ingestion_method = 'targeted_momo_review'
                  AND eo.pchome_product_id IS NOT NULL
                  AND eo.pchome_product_id IN :ids
                  AND (
                      eo.match_status = 'needs_review'
                      OR eo.data_quality_status = 'needs_review'
                  )
                ORDER BY eo.pchome_product_id, eo.observed_at DESC NULLS LAST, eo.id DESC
            )
            SELECT * FROM latest_review
        """
    else:
        sql = """
            WITH latest_review AS (
                SELECT
                    eo.id,
                    eo.pchome_product_id,
                    eo.source_product_id AS momo_sku,
                    eo.title AS momo_name,
                    eo.product_url,
                    eo.price AS momo_price,
                    eo.quality_score,
                    eo.raw_payload_json,
                    eo.observed_at,
                    ROW_NUMBER() OVER (
                        PARTITION BY eo.pchome_product_id
                        ORDER BY eo.observed_at DESC, eo.id DESC
                    ) AS rn
                FROM external_offers eo
                WHERE eo.source_code = 'momo_reference'
                  AND eo.ingestion_method = 'targeted_momo_review'
                  AND eo.pchome_product_id IS NOT NULL
                  AND eo.pchome_product_id IN :ids
                  AND (
                      eo.match_status = 'needs_review'
                      OR eo.data_quality_status = 'needs_review'
                  )
            )
            SELECT *
            FROM latest_review
            WHERE rn = 1
        """

    stmt = text(sql).bindparams(bindparam("ids", expanding=True))
    rows = conn.execute(stmt, {"ids": ids}).mappings().all()
    result: dict[str, dict[str, Any]] = {}
    for row in rows:
        key = str(row.get("pchome_product_id") or "").strip()
        if not key:
            continue
        raw_payload = _json_dict(row.get("raw_payload_json"))
        reasons = raw_payload.get("match_reasons") if isinstance(raw_payload.get("match_reasons"), list) else []
        result[key] = {
            "id": row.get("id"),
            "pchome_product_id": key,
            "pchome_product_name": raw_payload.get("pchome_public_name"),
            "pchome_price": raw_payload.get("pchome_public_price"),
            "momo_sku": row.get("momo_sku"),
            "momo_name": row.get("momo_name"),
            "momo_price": row.get("momo_price"),
            "product_url": row.get("product_url"),
            "quality_score": round(_to_float(row.get("quality_score")), 2),
            "match_score": _match_score_from_quality(row.get("quality_score")),
            "match_reasons": [str(reason) for reason in reasons[:5]],
            "gap_pct": raw_payload.get("target_gap_pct"),
            "observed_at": str(row.get("observed_at") or ""),
        }
    return result


def _serialize_market_offer(offer: dict[str, Any]) -> dict[str, Any]:
    gap = _external_offer_gap(offer)
    return {
        "source_code": offer.get("source_code") or "momo_reference",
        "source": offer.get("source_display_name") or "MOMO 外部價格參考",
        "platform_code": offer.get("platform_code") or "momo",
        "source_product_id": offer.get("momo_sku"),
        "title": offer.get("momo_name"),
        "product_url": offer.get("product_url"),
        "price": round(_to_float(offer.get("momo_price")), 2) if _to_float(offer.get("momo_price")) else None,
        "pchome_price": round(_to_float(offer.get("pchome_price")), 2) if _to_float(offer.get("pchome_price")) else None,
        "price_basis": offer.get("price_basis") or "total_price",
        "gap_pct": round(gap, 1) if gap < 9999 else None,
        "match_score": round(_to_float(offer.get("match_score")), 3),
        "source_status": offer.get("source_status") or "active",
        "actionable": bool(offer.get("actionable")),
        "evidence_state": "decision_ready" if offer.get("actionable") else "observation_only",
        "updated_at": str(offer.get("crawled_at") or ""),
    }


def _score_opportunity(
    sales_row: dict[str, Any],
    external_row: dict[str, Any] | None,
    review_candidate: dict[str, Any] | None = None,
    market_offers: list[dict[str, Any]] | None = None,
) -> dict[str, Any]:
    sales_7d = _to_float(sales_row.get("sales_7d"))
    sales_prev_7d = _to_float(sales_row.get("sales_prev_7d"))
    qty_7d = _to_float(sales_row.get("qty_7d"))
    sales_delta_pct = None
    if sales_prev_7d > 0:
        sales_delta_pct = (sales_7d - sales_prev_7d) / sales_prev_7d * 100

    volume_score = min(34, max(sales_7d, sales_prev_7d) / 80000 * 34)
    qty_score = min(10, qty_7d / 25 * 10)
    decline_score = min(24, abs(sales_delta_pct) / 45 * 24) if sales_delta_pct is not None and sales_delta_pct < 0 else 0
    data_quality_score = 54
    external_payload = None
    market_offer_payloads = [_serialize_market_offer(offer) for offer in (market_offers or [])]
    review_candidate_payload = None
    action_code = "map_external_product"
    action_label = "先補商品對應"
    action_message = "這項商品已有業績訊號，但還沒有可確認的外部平台同款。先補對應，後續才能判斷價格與促銷壓力。"
    reason_lines = []

    if external_row:
        source_name = str(external_row.get("source_display_name") or "MOMO 外部價格參考")
        price_basis = str(external_row.get("price_basis") or "total_price")
        price_basis_label = external_row.get("price_basis_label") or _external_price_basis_label(price_basis)
        pchome_price = _to_float(external_row.get("pchome_price"))
        momo_price = _to_float(external_row.get("momo_price"))
        pchome_compare_price = pchome_price
        momo_compare_price = momo_price
        if price_basis == "unit_price":
            pchome_unit_price = _to_float(external_row.get("pchome_unit_price"))
            momo_unit_price = _to_float(external_row.get("momo_unit_price"))
            pchome_compare_price = pchome_unit_price
            momo_compare_price = momo_unit_price
            gap_pct = _to_float(external_row.get("unit_gap_pct"), default=None)
            if gap_pct is None and pchome_compare_price and momo_compare_price:
                gap_pct = (momo_compare_price - pchome_compare_price) / pchome_compare_price * 100
        else:
            pchome_unit_price = None
            momo_unit_price = None
            gap_pct = ((momo_price - pchome_price) / pchome_price * 100) if pchome_price else None
        tags = _load_json_tags(external_row.get("tags"))
        data_quality_score = 78 + min(12, _to_float(external_row.get("match_score")) * 12)
        external_payload = {
            "source": source_name,
            "source_code": external_row.get("source_code") or "momo_reference",
            "platform_code": external_row.get("platform_code") or "momo",
            "data_source": external_row.get("data_source") or "competitor_prices",
            "data_source_label": "自動同步資料層"
            if external_row.get("data_source") == "external_offers"
            else "舊比價快取",
            "momo_sku": external_row.get("momo_sku"),
            "momo_name": external_row.get("momo_name"),
            "momo_price": round(momo_price, 2) if momo_price else None,
            "pchome_price": round(pchome_price, 2) if pchome_price else None,
            "price_basis": price_basis,
            "price_basis_label": price_basis_label,
            "unit_label": external_row.get("unit_label") or "",
            "momo_unit_price": round(momo_unit_price, 4) if momo_unit_price else None,
            "pchome_unit_price": round(pchome_unit_price, 4) if pchome_unit_price else None,
            "momo_total_quantity": external_row.get("momo_total_quantity"),
            "pchome_total_quantity": external_row.get("pchome_total_quantity"),
            "gap_pct": round(gap_pct, 1) if gap_pct is not None else None,
            "match_score": round(_to_float(external_row.get("match_score")), 3),
            "tags": tags,
            "updated_at": str(external_row.get("crawled_at") or ""),
        }

        if gap_pct is not None and gap_pct < -5:
            action_code = "review_price_or_promo"
            action_label = "檢查售價與活動"
            action_message = f"{source_name}的同款價格較低，建議檢查 PChome 售價、活動組合或曝光策略。"
        elif gap_pct is not None and gap_pct > 5:
            action_code = "amplify_price_advantage"
            action_label = "放大價格優勢"
            action_message = "PChome 目前有價格優勢，適合檢查曝光、文案與活動位置。"
        elif sales_delta_pct is not None and sales_delta_pct < -10:
            action_code = "recover_sales_momentum"
            action_label = "找回銷售動能"
            action_message = "價格差距不大，但近 7 天業績轉弱，建議檢查曝光、庫存與商品頁內容。"
        else:
            action_code = "monitor"
            action_label = "持續觀察"
            action_message = "業績與外部價格暫無明顯異常，先保留在觀察清單。"

        if gap_pct is not None:
            basis_prefix = "單位價" if price_basis == "unit_price" else "價格"
            if gap_pct > 0:
                reason_lines.append(f"PChome {basis_prefix}目前比{source_name}低約 {abs(gap_pct):.1f}%。")
            elif gap_pct < 0:
                reason_lines.append(f"{source_name}{basis_prefix}目前比 PChome 低約 {abs(gap_pct):.1f}%。")
            else:
                reason_lines.append(f"PChome 與{source_name}{basis_prefix}幾乎相同。")
    else:
        if review_candidate:
            review_candidate_payload = {
                "id": review_candidate.get("id"),
                "momo_sku": review_candidate.get("momo_sku"),
                "momo_name": review_candidate.get("momo_name"),
                "momo_price": review_candidate.get("momo_price"),
                "pchome_price": review_candidate.get("pchome_price"),
                "quality_score": review_candidate.get("quality_score"),
                "gap_pct": review_candidate.get("gap_pct"),
                "match_reasons": review_candidate.get("match_reasons") or [],
                "product_url": review_candidate.get("product_url"),
                "observed_at": review_candidate.get("observed_at"),
            }
            data_quality_score = max(data_quality_score, 62)
            action_code = "review_external_candidate"
            action_label = "確認候選"
            action_message = "已找到 MOMO 候選，但還要確認同款、色號或組合後才能進價格判斷。"
            reason_lines.append("已找到 MOMO 候選，先確認同款、色號或組合。")
        else:
            if market_offer_payloads:
                data_quality_score = max(data_quality_score, 60)
                action_code = "validate_external_source"
                action_label = "驗證外部來源"
                action_message = "已有外部平台同款證據，但來源尚未通過啟用政策；系統會先完成來源與商品身份驗證。"
                observed_sources = sorted({str(item.get("source") or "") for item in market_offer_payloads})
                reason_lines.append(f"已有 {len(observed_sources)} 個外部平台證據，尚未進入正式價格決策。")
            else:
                data_quality_score -= 12
                reason_lines.append("尚未找到可確認的外部平台同款商品。")

    if sales_delta_pct is None:
        reason_lines.append("前 7 天沒有可比基準，先看近 7 天表現。")
    elif sales_delta_pct < 0:
        reason_lines.append(f"近 7 天業績比前 7 天少約 {abs(sales_delta_pct):.1f}%。")
    elif sales_delta_pct > 0:
        reason_lines.append(f"近 7 天業績比前 7 天多約 {sales_delta_pct:.1f}%。")
    else:
        reason_lines.append("近 7 天業績與前 7 天大致持平。")

    if sales_7d > 0:
        reason_lines.append(f"近 7 天業績約 NT$ {sales_7d:,.0f}，銷量 {qty_7d:,.0f}。")

    mapping_gap_score = 18 if not external_row and not market_offer_payloads and max(sales_7d, sales_prev_7d) > 0 else 0
    priority_score = min(100, volume_score + qty_score + decline_score + mapping_gap_score + data_quality_score * 0.18)
    if external_payload and external_payload.get("gap_pct") is not None:
        gap = float(external_payload["gap_pct"])
        if gap < -5:
            priority_score = min(100, priority_score + min(18, abs(gap) / 20 * 18))
        elif gap > 5:
            priority_score = min(100, priority_score + min(10, gap / 25 * 10))

    issues = []
    if not external_row:
        issues.append(
            "外部來源尚未啟用"
            if market_offer_payloads
            else "需要確認 MOMO 候選"
            if review_candidate_payload
            else "需要補商品對應"
        )
    if sales_delta_pct is None:
        issues.append("前期業績不足")

    return {
        "pchome_product_id": str(sales_row.get("pchome_product_id") or ""),
        "product_name": sales_row.get("product_name") or "未命名商品",
        "category": sales_row.get("category") or "",
        "vendor": sales_row.get("vendor") or "",
        "sales_7d": round(sales_7d, 2),
        "sales_prev_7d": round(sales_prev_7d, 2),
        "sales_delta_pct": round(sales_delta_pct, 1) if sales_delta_pct is not None else None,
        "qty_7d": round(qty_7d, 2),
        "pchome_price": round(_to_float(sales_row.get("pchome_price")), 2)
        if _to_float(sales_row.get("pchome_price")) > 0
        else None,
        "last_sale_date": str(sales_row.get("last_sale_date") or ""),
        "external_price": external_payload,
        "market_offers": market_offer_payloads,
        "cross_platform": {
            "observed_platform_count": len({item.get("source_code") for item in market_offer_payloads}),
            "decision_ready_platform_count": len({
                item.get("source_code") for item in market_offer_payloads if item.get("actionable")
            }),
            "lowest_external_offer": market_offer_payloads[0] if market_offer_payloads else None,
        },
        "review_candidate": review_candidate_payload,
        "priority_score": round(priority_score, 1),
        "recommended_action": {
            "code": action_code,
            "label": action_label,
            "message": action_message,
        },
        "reason_lines": reason_lines[:4],
        "data_quality": {
            "label": (
                "資料可用單位價判斷"
                if external_payload and external_payload.get("price_basis") == "unit_price"
                else "資料可直接判斷"
                if external_row
                else "候選待確認"
                if review_candidate_payload
                else "需要補資料"
            ),
            "score": round(max(0, min(100, data_quality_score)), 1),
            "issues": issues,
        },
    }


def build_pchome_growth_opportunities(engine, limit: int = 20) -> dict[str, Any]:
    """讀取 PChome 業績與已驗證外部價格，產生營運用作戰清單。"""
    limit = max(5, min(int(limit or 20), 50))
    generated_at = datetime.now().isoformat(timespec="seconds")
    source_readiness = build_external_source_readiness(engine)
    review_candidate_count = int(source_readiness.get("review_offer_count") or 0)
    registered_external_source_count = len(source_readiness.get("sources") or [])
    active_external_source_count = int(source_readiness.get("active_count") or 0)
    runtime_external_source_count = int(source_readiness.get("sources_with_data_count") or 0)
    active_external_sources = _source_names_by_status(source_readiness, "active") or list(ACTIVE_EXTERNAL_SOURCES)
    paused_external_sources = _source_names_by_status(source_readiness, "paused") or list(PAUSED_EXTERNAL_SOURCES)
    observed_external_sources = [
        str(source.get("display_name") or source.get("code") or "")
        for source in source_readiness.get("sources") or []
        if source.get("has_runtime_data")
    ]
    source_scope = {
        "primary_goal": "提升 PChome 業績",
        "primary_sales_source": PRIMARY_SALES_SOURCE,
        "active_external_sources": active_external_sources,
        "paused_external_sources": paused_external_sources,
        "observed_external_sources": observed_external_sources,
        "plain_note": "多平台證據會完整保留；只有 runtime 已啟用且通過同款與品質驗證的報價能驅動正式決策。",
        "source_readiness": source_readiness,
        "offer_evidence_contract": source_readiness.get("offer_evidence_contract") or {},
    }

    if not _table_exists(engine, "daily_sales_snapshot"):
        return {
            "success": True,
            "system_name": SYSTEM_DISPLAY_NAME,
            "generated_at": generated_at,
            "source_scope": source_scope,
            "stats": {
                "latest_sales_date": None,
                "candidate_count": 0,
                "mapped_count": 0,
                "mapping_rate": 0,
                "needs_mapping_count": 0,
                "review_candidate_count": review_candidate_count,
                "current_review_candidate_count": 0,
                "candidate_validation_count": 0,
                "unmatched_count": 0,
                "comparison_ready_count": 0,
                "decision_ready_count": 0,
                "registered_external_source_count": registered_external_source_count,
                "active_external_source_count": active_external_source_count,
                "platforms_with_runtime_data": runtime_external_source_count,
                "sales_freshness": _sales_freshness(None),
            },
            "opportunities": [],
            "message": "目前還沒有 PChome 業績資料，請先完成業績匯入。",
        }

    with engine.connect() as conn:
        sales_summary = _fetch_sales_summary(conn)
        catalog_mapping_summary = _fetch_catalog_mapping_summary(conn)
        summary_freshness = _sales_freshness(sales_summary.get("overall_latest_sales_date"))
        try:
            sales_rows, latest_sales_date = _fetch_sales_rows(conn, limit=limit)
        except RuntimeError as exc:
            return {
                "success": True,
                "system_name": SYSTEM_DISPLAY_NAME,
                "generated_at": generated_at,
                "source_scope": source_scope,
                "stats": {
                    "latest_sales_date": sales_summary.get("overall_latest_sales_date"),
                    "candidate_count": 0,
                    "mapped_count": 0,
                    "mapping_rate": 0,
                    "needs_mapping_count": 0,
                    "review_candidate_count": review_candidate_count,
                    "current_review_candidate_count": 0,
                    "candidate_validation_count": 0,
                    "unmatched_count": 0,
                    "comparison_ready_count": 0,
                    "decision_ready_count": 0,
                    "registered_external_source_count": registered_external_source_count,
                    "active_external_source_count": active_external_source_count,
                    "platforms_with_runtime_data": runtime_external_source_count,
                    "total_sales_7d": 0,
                    "opportunity_sales_7d": 0,
                    "action_counts": {},
                    "action_code_counts": {},
                    "external_data_source_counts": {},
                    "external_platform_counts": {},
                    "sales_freshness": summary_freshness,
                    **catalog_mapping_summary,
                    **sales_summary,
                },
                "opportunities": [],
                "message": str(exc),
            }
        # Coverage must use a stable revenue-ranked cohort. Mapping status may
        # reorder work inside the cohort, but must never remove a success from it.
        sales_rows = sales_rows[:limit]
        sales_ids = [str(row.get("pchome_product_id") or "") for row in sales_rows]
        external_map, market_offer_map = _fetch_external_offer_maps(
            conn,
            sales_ids,
            source_readiness,
        )
        review_candidate_map = _fetch_review_candidate_map(conn, sales_ids)

    opportunities = []
    for row in sales_rows:
        key = str(row.get("pchome_product_id") or "").strip()
        opportunities.append(_score_opportunity(
            row,
            external_map.get(key),
            review_candidate_map.get(key),
            market_offer_map.get(key),
        ))

    opportunities.sort(key=lambda item: item["priority_score"], reverse=True)
    opportunities = opportunities[:limit]
    needs_mapping_count = sum(1 for item in opportunities if not item.get("market_offers"))
    observed_mapping_count = len(opportunities) - needs_mapping_count
    mapped_count = sum(1 for item in opportunities if item.get("external_price"))
    current_review_candidate_count = sum(
        1
        for item in opportunities
        if not item.get("external_price") and item.get("review_candidate")
    )
    candidate_validation_count = sum(
        1
        for item in opportunities
        if not item.get("external_price")
        and (item.get("review_candidate") or item.get("market_offers"))
    )
    unmatched_count = sum(
        1
        for item in opportunities
        if not item.get("external_price")
        and not item.get("review_candidate")
        and not item.get("market_offers")
    )
    mapping_rate = round(mapped_count / max(len(opportunities), 1) * 100, 1)
    observed_mapping_rate = round(observed_mapping_count / max(len(opportunities), 1) * 100, 1)
    total_opportunity_sales_7d = sum(
        _to_float(item.get("sales_7d")) for item in opportunities
    )
    comparison_ready_sales_7d = sum(
        _to_float(item.get("sales_7d"))
        for item in opportunities
        if item.get("external_price")
    )
    candidate_validation_sales_7d = sum(
        _to_float(item.get("sales_7d"))
        for item in opportunities
        if not item.get("external_price")
        and (item.get("review_candidate") or item.get("market_offers"))
    )
    unmatched_sales_7d = max(
        0.0,
        total_opportunity_sales_7d
        - comparison_ready_sales_7d
        - candidate_validation_sales_7d,
    )
    comparison_revenue_coverage_rate = round(
        comparison_ready_sales_7d / max(total_opportunity_sales_7d, 1.0) * 100,
        1,
    )
    action_counts: dict[str, int] = {}
    action_code_counts: dict[str, int] = {}
    external_data_source_counts: dict[str, int] = {}
    external_platform_counts: dict[str, int] = {}
    for item in opportunities:
        label = item["recommended_action"]["label"]
        code = item["recommended_action"]["code"]
        action_counts[label] = action_counts.get(label, 0) + 1
        action_code_counts[code] = action_code_counts.get(code, 0) + 1
        external_price = item.get("external_price") or {}
        data_source_label = external_price.get("data_source_label")
        if data_source_label:
            external_data_source_counts[data_source_label] = (
                external_data_source_counts.get(data_source_label, 0) + 1
            )
        for offer in item.get("market_offers") or []:
            source_name = str(offer.get("source") or offer.get("source_code") or "外部平台")
            external_platform_counts[source_name] = external_platform_counts.get(source_name, 0) + 1

    latest_sales = latest_sales_date or sales_summary.get("overall_latest_sales_date")
    sales_freshness = _sales_freshness(latest_sales)
    decision_ready_count = mapped_count if sales_freshness["decision_ready"] else 0
    decision_ready_rate = round(
        decision_ready_count / max(len(opportunities), 1) * 100,
        1,
    )
    blocked_reason_codes = []
    if not sales_freshness["decision_ready"]:
        blocked_reason_codes.append("sales_data_stale")
    if unmatched_count:
        blocked_reason_codes.append("verified_same_item_missing")
    if candidate_validation_count:
        blocked_reason_codes.append("candidate_validation_pending")
    if runtime_external_source_count < registered_external_source_count:
        blocked_reason_codes.append("platform_runtime_coverage_gap")

    scope_ids = [str(row.get("pchome_product_id") or "") for row in sales_rows]
    scope_fingerprint = hashlib.sha256(
        json.dumps(scope_ids, ensure_ascii=True, separators=(",", ":")).encode("utf-8")
    ).hexdigest()[:24]
    comparison_metric_contract = {
        "version": "pchome_comparison_coverage_v3",
        "scope": f"top_{len(opportunities)}_revenue_products",
        "scope_membership": {
            "selection_stage": "before_action_priority_scoring",
            "ranking": "max_current_or_previous_7d_revenue_desc_then_qty_desc_then_product_id_asc",
            "member_count": len(scope_ids),
            "fingerprint_sha256": scope_fingerprint,
            "mapping_status_can_change_membership": False,
        },
        "numerator": mapped_count,
        "denominator": len(opportunities),
        "rate": mapping_rate,
        "meaning": "營收排名母體中，通過同款與價格品質門檻且具正式外部價格證據的商品",
        "state_partition": {
            "comparison_ready": mapped_count,
            "candidate_validation": candidate_validation_count,
            "unmatched": unmatched_count,
            "total": len(opportunities),
            "mutually_exclusive": (
                mapped_count + candidate_validation_count + unmatched_count
                == len(opportunities)
            ),
        },
        "operational_decision": {
            "ready": bool(sales_freshness["decision_ready"] and mapped_count),
            "ready_count": decision_ready_count,
            "ready_rate": decision_ready_rate,
            "blocked_reason_codes": blocked_reason_codes,
        },
        "active_catalog": {
            "numerator": int(catalog_mapping_summary.get("active_mapped_product_count") or 0),
            "denominator": int(catalog_mapping_summary.get("active_catalog_product_count") or 0),
            "rate": _to_float(catalog_mapping_summary.get("active_catalog_mapping_rate")),
            "scope": "products_with_positive_sales_in_latest_7_day_window",
        },
        "revenue_coverage": {
            "numerator": round(comparison_ready_sales_7d, 2),
            "denominator": round(total_opportunity_sales_7d, 2),
            "rate": comparison_revenue_coverage_rate,
            "meaning": "營收排名母體近 7 天業績中，已有正式同商品外部價格證據的占比",
            "state_partition": {
                "comparison_ready": round(comparison_ready_sales_7d, 2),
                "candidate_validation": round(candidate_validation_sales_7d, 2),
                "unmatched": round(unmatched_sales_7d, 2),
                "total": round(total_opportunity_sales_7d, 2),
            },
        },
        "platform_runtime": {
            "numerator": runtime_external_source_count,
            "denominator": registered_external_source_count,
            "rate": _to_float(source_readiness.get("source_coverage_rate")),
            "scope": "registered_external_market_sources_with_runtime_data",
        },
    }

    return {
        "success": True,
        "system_name": SYSTEM_DISPLAY_NAME,
        "generated_at": generated_at,
        "source_scope": source_scope,
        "stats": {
            "latest_sales_date": latest_sales,
            "sales_freshness": sales_freshness,
            "candidate_count": len(opportunities),
            "mapped_count": mapped_count,
            "mapping_rate": mapping_rate,
            "comparison_ready_count": mapped_count,
            "decision_ready_count": decision_ready_count,
            "decision_ready_rate": decision_ready_rate,
            "observed_mapping_count": observed_mapping_count,
            "observed_mapping_rate": observed_mapping_rate,
            "needs_mapping_count": needs_mapping_count,
            "review_candidate_count": review_candidate_count,
            "current_review_candidate_count": current_review_candidate_count,
            "candidate_validation_count": candidate_validation_count,
            "unmatched_count": unmatched_count,
            "unresolved_count": max(0, len(opportunities) - mapped_count),
            "total_sales_7d": round(total_opportunity_sales_7d, 2),
            "opportunity_sales_7d": round(total_opportunity_sales_7d, 2),
            "comparison_ready_sales_7d": round(comparison_ready_sales_7d, 2),
            "candidate_validation_sales_7d": round(candidate_validation_sales_7d, 2),
            "unmatched_sales_7d": round(unmatched_sales_7d, 2),
            "comparison_revenue_coverage_rate": comparison_revenue_coverage_rate,
            "action_counts": action_counts,
            "action_code_counts": action_code_counts,
            "external_data_source_counts": external_data_source_counts,
            "external_platform_counts": external_platform_counts,
            "registered_external_source_count": registered_external_source_count,
            "active_external_source_count": active_external_source_count,
            "platforms_with_runtime_data": runtime_external_source_count,
            "platform_source_coverage_rate": _to_float(source_readiness.get("source_coverage_rate")),
            "comparison_metric_contract": comparison_metric_contract,
            "blocked_reason_codes": blocked_reason_codes,
            **catalog_mapping_summary,
            **sales_summary,
        },
        "opportunities": opportunities,
        "message": (
            "業績資料已過期，作戰清單只供觀察；系統應先自動取得最新業績。"
            if not sales_freshness["decision_ready"]
            else "已依最新業績與外部平台證據整理今日作戰清單。"
        ),
    }
