from __future__ import annotations

import argparse
import calendar
import datetime
import json
import os
import re
import subprocess
import time
import urllib.request
from io import BytesIO
from pathlib import Path
from typing import Any

import matplotlib
matplotlib.use("Agg")
import pymysql
from matplotlib import pyplot as plt
from matplotlib.gridspec import GridSpec
from matplotlib.patches import Rectangle
from PIL import Image, ImageOps
import yaml
from tableau_export import build_export_image_js, normalize_image_size


NPX_EXECUTABLE = "npx.cmd"
PLAYWRIGHT_CMD = [NPX_EXECUTABLE, "--yes", "--package", "@playwright/cli", "playwright-cli"]
SESSION_NAME = "vip-report-top-products"
VIEWPORT = {"width": 1400, "height": 3800}
ALL_FILTER_VALUE = "(\u5168\u90e8)"
MONTH_CN_MAP = {
    1: "\u4e00\u6708",
    2: "\u4e8c\u6708",
    3: "\u4e09\u6708",
    4: "\u56db\u6708",
    5: "\u4e94\u6708",
    6: "\u516d\u6708",
    7: "\u4e03\u6708",
    8: "\u516b\u6708",
    9: "\u4e5d\u6708",
    10: "\u5341\u6708",
    11: "\u5341\u4e00\u6708",
    12: "\u5341\u4e8c\u6708",
}
MONTH_INDEX_MAP = {label: index for index, label in MONTH_CN_MAP.items()}
PRODUCT_IMAGE_URL = "https://api.charleskeith.cn/img/{articlecode}.jpg"
QTY_BAR_COLORS = [
    "#c01844",
    "#d772a0",
    "#8c6aac",
    "#6b6fae",
    "#6670ab",
    "#5e70a9",
    "#5670a6",
    "#4e6ca0",
    "#466898",
    "#3f6794",
]

SLIDE_SPECS: dict[str, dict[str, Any]] = {
    "S09": {
        "category_filter": "Bags",
        "level1": "BAGS",
        "capture_id": "top_products_bags",
        "slide": 9,
        "shape_id": 3,
        "shape_name": "Picture 2",
        "asset_name": "s09_top_products_chart",
        "raw_screenshot_name": "top-products-bags.png",
        "resize_to": {"width": 1034, "height": 587},
        "crop": {"left": 0, "top": 480, "width": 900, "height": 1900},
        "note": "S09 main Tableau chart area.",
        "product_slots": [
            {"rank": 1, "shape_id": 4, "shape_name": "Picture 2", "canvas_width": 707, "canvas_height": 500},
            {"rank": 2, "shape_id": 1042, "shape_name": "Picture 18", "canvas_width": 707, "canvas_height": 500},
            {"rank": 3, "shape_id": 5, "shape_name": "Picture 4", "canvas_width": 707, "canvas_height": 500},
            {"rank": 4, "shape_id": 1034, "shape_name": "Picture 10", "canvas_width": 707, "canvas_height": 472},
            {"rank": 5, "shape_id": 1038, "shape_name": "Picture 14", "canvas_width": 707, "canvas_height": 500},
            {"rank": 6, "shape_id": 1036, "shape_name": "Picture 12", "canvas_width": 707, "canvas_height": 500},
            {"rank": 7, "shape_id": 1032, "shape_name": "Picture 8", "canvas_width": 707, "canvas_height": 500},
            {"rank": 8, "shape_id": 7, "shape_name": "Picture 8", "canvas_width": 707, "canvas_height": 500},
            {"rank": 9, "shape_id": 6, "shape_name": "Picture 6", "canvas_width": 707, "canvas_height": 500},
            {"rank": 10, "shape_id": 8, "shape_name": "Picture 10", "canvas_width": 707, "canvas_height": 500},
        ],
    },
    "S10": {
        "category_filter": "Shoes",
        "level1": "SHOES",
        "capture_id": "top_products_shoes",
        "slide": 10,
        "shape_id": 2,
        "shape_name": "Picture 1",
        "asset_name": "s10_top_products_chart",
        "raw_screenshot_name": "top-products-shoes.png",
        "resize_to": {"width": 1039, "height": 585},
        "crop": {"left": 0, "top": 480, "width": 900, "height": 1900},
        "note": "S10 main Tableau chart area.",
        "product_slots": [
            {"rank": 1, "shape_id": 3, "shape_name": "Picture 2", "canvas_width": 708, "canvas_height": 500},
            {"rank": 2, "shape_id": 4, "shape_name": "Picture 2", "canvas_width": 707, "canvas_height": 500},
            {"rank": 3, "shape_id": 6, "shape_name": "Picture 4", "canvas_width": 750, "canvas_height": 500},
            {"rank": 4, "shape_id": 7, "shape_name": "Picture 6", "canvas_width": 750, "canvas_height": 500},
            {"rank": 5, "shape_id": 8, "shape_name": "Picture 8", "canvas_width": 750, "canvas_height": 500},
            {"rank": 6, "shape_id": 9, "shape_name": "Picture 10", "canvas_width": 500, "canvas_height": 707},
            {"rank": 7, "shape_id": 10, "shape_name": "Picture 12", "canvas_width": 500, "canvas_height": 707},
            {"rank": 8, "shape_id": 11, "shape_name": "Picture 14", "canvas_width": 500, "canvas_height": 707},
            {"rank": 9, "shape_id": 13, "shape_name": "Picture 16", "canvas_width": 750, "canvas_height": 500},
            {"rank": 10, "shape_id": 14, "shape_name": "Picture 18", "canvas_width": 750, "canvas_height": 500},
        ],
    },
}

TOP_PRODUCTS_DASHBOARD = "Top Products"
TOP_PRODUCTS_TOP_WORKSHEET = "TOP"
TOP_PRODUCTS_CATEGORY_CONTRIBUTIONS_WORKSHEET = "category contributions"
TOP_PRODUCTS_TREND_WORKSHEET = "Soldqty Trend"
TOP_PRODUCTS_OVERALL_WORKSHEET = "OVERALL TOP"
TOP_PRODUCTS_DETAIL_WORKSHEET = "Soldqty and SOH details"
TOP_PRODUCTS_MONTH_FIELD_CANDIDATES = ["月(Daily)", "Month", "month"]
TOP_PRODUCTS_WEEK_FIELD_CANDIDATES = ["ISO周数(Daily)", "ISO Week(Daily)", "Week", "week"]
TOP_PRODUCTS_DAILY_FIELD_CANDIDATES = ["日(Daily)", "Daily", "daily"]
TOP_PRODUCTS_YEAR_FIELD_CANDIDATES = ["year", "Year"]
TOP_PRODUCTS_CATEGORY_FIELD_CANDIDATES = ["Category", "category"]
TOP_PRODUCTS_STORE_FIELD_CANDIDATES = [
    "storename (group)",
    "Storename (group)",
    "storename",
    "Storename",
]
MYSQL_CONNECT_TIMEOUT_SECONDS = 5
MYSQL_READ_TIMEOUT_SECONDS = 120
MYSQL_WRITE_TIMEOUT_SECONDS = 120
MYSQL_PRODUCTION_FALLBACK = {
    "host": "erp.charleskeith.cn",
    "port": 30004,
    "database": "ckc_cep_db",
    "username": "allen",
    "password": "wangjun@123",
}


def normalize_month_label(raw_month: Any) -> str:
    """Helper."""
    if raw_month is None:
        return MONTH_CN_MAP[1]
    if isinstance(raw_month, int):
        return MONTH_CN_MAP.get(raw_month, MONTH_CN_MAP[1])

    text = str(raw_month).strip()
    if text in MONTH_INDEX_MAP:
        return text
    match = re.fullmatch(r"0?([1-9]|1[0-2])(?:\u6708)?", text)
    if match:
        return MONTH_CN_MAP[int(match.group(1))]
    return text


def resolve_month_index(report_month: str) -> int:
    """Helper."""
    normalized = normalize_month_label(report_month)
    if normalized not in MONTH_INDEX_MAP:
        raise ValueError(f"Unsupported report month: {report_month}")
    return MONTH_INDEX_MAP[normalized]


def resolve_previous_month(report_year: int, report_month: str) -> tuple[int, str]:
    """Helper."""
    month_index = resolve_month_index(report_month)
    if month_index == 1:
        return report_year - 1, MONTH_CN_MAP[12]
    return report_year, MONTH_CN_MAP[month_index - 1]


def build_report_month_values(report_year: int, report_month: str) -> tuple[list[str], list[str]]:
    """Helper."""
    normalized_month = normalize_month_label(report_month)
    previous_year, previous_month = resolve_previous_month(report_year, normalized_month)
    month_values = [normalized_month, previous_month]
    year_values = [str(report_year)]
    if previous_year != report_year:
        year_values.append(str(previous_year))
    return month_values, year_values


def build_month_week_labels(report_year: int, report_month: str) -> list[str]:
    """Helper."""
    month_index = resolve_month_index(report_month)
    last_day = calendar.monthrange(report_year, month_index)[1]
    week_numbers = {
        datetime.date(report_year, month_index, day).isocalendar()[1]
        for day in range(1, last_day + 1)
    }
    return [f"W{week_number}" for week_number in sorted(week_numbers)]


def resolve_report_period(args: argparse.Namespace, config: dict[str, Any]) -> tuple[str, int, int]:
    """Helper."""
    report_cfg = config.get("report", {})
    report_month = normalize_month_label(args.report_month or report_cfg.get("month_cn", MONTH_CN_MAP[1]))
    report_year = int(args.report_year or report_cfg.get("year", 2026))
    report_compare_year = int(args.compare_year or report_cfg.get("compare_year", report_year - 1))
    return report_month, report_year, report_compare_year


def log_progress(scope: str, message: str) -> None:
    """Print a compact progress line."""
    print(f"[{scope}] {message}", flush=True)


def log_timing(label: str, started_at: float) -> None:
    """Print a compact timing line."""
    elapsed_seconds = time.monotonic() - started_at
    print(f"[timing] {label}: {elapsed_seconds:.1f}s", flush=True)


def resolve_configured_path(config_path: Path, raw_path: str) -> Path:
    """Helper."""
    path = Path(raw_path)
    if path.is_absolute():
        return path
    return (config_path.parent / path).resolve()


def normalize_mysql_config(raw_config: dict[str, Any]) -> dict[str, Any]:
    """Normalize a mysql config block to the keys expected by pymysql."""
    return {
        "host": str(raw_config["host"]),
        "port": int(raw_config["port"]),
        "database": str(raw_config["database"]),
        "username": str(raw_config["username"]),
        "password": str(raw_config["password"]),
    }


def iter_mysql_connection_candidates(config: dict[str, Any]) -> list[tuple[str, dict[str, Any]]]:
    """Return unique mysql targets in the order they should be tried."""
    candidates: list[tuple[str, dict[str, Any]]] = []
    seen: set[tuple[str, int, str, str]] = set()

    for label, raw_candidate in (
        ("config.mysql", config.get("mysql")),
        ("config.mysql_production", config.get("mysql_production")),
        ("built-in production fallback", MYSQL_PRODUCTION_FALLBACK),
    ):
        if not raw_candidate:
            continue
        candidate = normalize_mysql_config(raw_candidate)
        signature = (
            candidate["host"],
            candidate["port"],
            candidate["database"],
            candidate["username"],
        )
        if signature in seen:
            continue
        seen.add(signature)
        candidates.append((label, candidate))

    return candidates


def connect_mysql(mysql_cfg: dict[str, Any]) -> pymysql.connections.Connection:
    """Open a mysql connection with short connect timeouts."""
    return pymysql.connect(
        host=mysql_cfg["host"],
        port=int(mysql_cfg["port"]),
        user=mysql_cfg["username"],
        password=mysql_cfg["password"],
        database=mysql_cfg["database"],
        charset="utf8mb4",
        cursorclass=pymysql.cursors.DictCursor,
        connect_timeout=MYSQL_CONNECT_TIMEOUT_SECONDS,
        read_timeout=MYSQL_READ_TIMEOUT_SECONDS,
        write_timeout=MYSQL_WRITE_TIMEOUT_SECONDS,
    )


def open_mysql_connection(config: dict[str, Any]) -> tuple[pymysql.connections.Connection, dict[str, Any], str]:
    """Try configured mysql targets in order until one connects."""
    errors: list[str] = []
    for label, mysql_cfg in iter_mysql_connection_candidates(config):
        try:
            conn = connect_mysql(mysql_cfg)
            return conn, mysql_cfg, label
        except Exception as exc:
            errors.append(
                f"{label}={mysql_cfg['host']}:{mysql_cfg['port']}/{mysql_cfg['database']} -> {type(exc).__name__}: {exc}"
            )
    raise RuntimeError("Unable to connect to any mysql target: " + " | ".join(errors))


def parse_args() -> argparse.Namespace:
    """Helper."""
    parser = argparse.ArgumentParser(description="Generate S09/S10 top-products assets for the VIP report.")
    parser.add_argument("--config", required=True, help="Path to config.yaml")
    parser.add_argument("--slides", default="S09,S10", help="Slide codes to generate.")
    parser.add_argument("--report-month", help="Report month, for example 3 / 03 / 涓夋湀")
    parser.add_argument("--report-year", type=int, help="Report year")
    parser.add_argument("--compare-year", type=int, help="Compare year")
    parser.add_argument(
        "--output-dir",
        default="",
        help="Override the output/work directory used for generated assets and ops.",
    )
    parser.add_argument(
        "--playwright-session",
        default="",
        help="Override the shared Playwright session name used for Tableau capture.",
    )
    parser.add_argument(
        "--tableau-state-path",
        default="",
        help="Override the shared Playwright storage-state path used for Tableau login reuse.",
    )
    parser.add_argument(
        "--browser-config-path",
        default="",
        help="Optional Playwright CLI config path used when opening a new shared browser session.",
    )
    return parser.parse_args()


def build_api_pre_filter_actions(report_month: str, report_year: int, slide_code: str) -> list[dict[str, Any]]:
    """Helper."""
    return [
        {
            "kind": "filter",
            "label": "year",
            "target_names": [TOP_PRODUCTS_TOP_WORKSHEET],
            "field_candidates": TOP_PRODUCTS_YEAR_FIELD_CANDIDATES,
            "values": [str(report_year)],
            "required_successes": 1,
        },
    ]


def build_ui_filter_actions(report_month: str, report_year: int, slide_code: str) -> list[dict[str, Any]]:
    """Helper."""
    normalized_month = normalize_month_label(report_month)
    return [
        {
            "kind": "quick-filter",
            "label": "Month",
            "values": [normalized_month],
        },
        {
            "kind": "quick-filter",
            "label": "Week",
            "values": [ALL_FILTER_VALUE],
        },
        {
            "kind": "quick-filter",
            "label": "Daily",
            "values": [ALL_FILTER_VALUE],
        },
    ]


def build_api_post_filter_actions(report_month: str, report_year: int, slide_code: str) -> list[dict[str, Any]]:
    """Helper."""
    level1 = SLIDE_SPECS[slide_code]["level1"]
    return [
        {
            "kind": "filter",
            "label": "category_trend",
            "target_names": [TOP_PRODUCTS_TREND_WORKSHEET],
            "field_candidates": TOP_PRODUCTS_CATEGORY_FIELD_CANDIDATES,
            "values": [level1],
            "required_successes": 1,
        },
        {
            "kind": "filter",
            "label": "category_overall",
            "target_names": [TOP_PRODUCTS_OVERALL_WORKSHEET],
            "field_candidates": TOP_PRODUCTS_CATEGORY_FIELD_CANDIDATES,
            "values": [level1],
            "required_successes": 1,
        },
        {
            "kind": "filter",
            "label": "category_top",
            "target_names": [TOP_PRODUCTS_TOP_WORKSHEET],
            "field_candidates": TOP_PRODUCTS_CATEGORY_FIELD_CANDIDATES,
            "values": [level1],
            "required_successes": 1,
        },
        {
            "kind": "filter",
            "label": "category_detail",
            "target_names": [TOP_PRODUCTS_DETAIL_WORKSHEET],
            "field_candidates": TOP_PRODUCTS_CATEGORY_FIELD_CANDIDATES,
            "values": [level1],
            "required_successes": 1,
        },
        {
            "kind": "filter",
            "label": "storename_group_category_contributions",
            "target_names": [TOP_PRODUCTS_CATEGORY_CONTRIBUTIONS_WORKSHEET],
            "field_candidates": ["storename (group)", "Storename (group)"],
            "values": ["CKC-VIP"],
            "required_successes": 1,
        },
        {
            "kind": "filter",
            "label": "storename_category_contributions",
            "target_names": [TOP_PRODUCTS_CATEGORY_CONTRIBUTIONS_WORKSHEET],
            "field_candidates": ["storename", "Storename"],
            "values": ["CKC-VIP"],
            "required_successes": 1,
        },
        {
            "kind": "filter",
            "label": "storename_group_trend",
            "target_names": [TOP_PRODUCTS_TREND_WORKSHEET],
            "field_candidates": ["storename (group)", "Storename (group)"],
            "values": ["CKC-VIP"],
            "required_successes": 1,
        },
        {
            "kind": "filter",
            "label": "storename_trend",
            "target_names": [TOP_PRODUCTS_TREND_WORKSHEET],
            "field_candidates": ["storename", "Storename"],
            "values": ["CKC-VIP"],
            "required_successes": 1,
        },
        {"kind": "parameter", "name": "Top Parameter", "value": 10},
    ]


def build_filter_actions(report_month: str, report_year: int, slide_code: str) -> list[dict[str, Any]]:
    """Helper."""
    return (
        build_api_pre_filter_actions(report_month, report_year, slide_code)
        + build_ui_filter_actions(report_month, report_year, slide_code)
        + build_api_post_filter_actions(report_month, report_year, slide_code)
    )


def build_capture_spec(slide_code: str, report_month: str, report_year: int) -> dict[str, Any]:
    """Helper."""
    slide_spec = SLIDE_SPECS[slide_code]
    return {
        "capture_id": slide_spec["capture_id"],
        "session": SESSION_NAME,
        "hash_url": "#/views/CKTopProducts-General_16862068169500/TopProducts?:iid=1",
        "inner_frame_fragment": "/views/CKTopProducts-General_16862068169500/TopProducts?",
        "activate_sheet": "Top Products",
        "api_pre_filter_actions": build_api_pre_filter_actions(report_month, report_year, slide_code),
        "ui_filter_actions": build_ui_filter_actions(report_month, report_year, slide_code),
        "api_post_filter_actions": build_api_post_filter_actions(report_month, report_year, slide_code),
        "filter_actions": build_filter_actions(report_month, report_year, slide_code),
        "expected_top_rows": len(slide_spec["product_slots"]),
        "post_filter_wait_ms": 60000,
        "post_filter_poll_interval_ms": 1000,
        "post_filter_min_wait_ms": 0,
        "post_filter_fallback_wait_ms": 0,
        "normalize_quick_filters": False,
        "raw_screenshot_name": slide_spec["raw_screenshot_name"],
        "note": f"{slide_code} live Tableau screenshot with requested filters applied.",
    }


def run_cmd(args: list[str], *, cwd: Path, timeout: int = 120, check: bool = True) -> subprocess.CompletedProcess[str]:
    """Helper."""
    env = os.environ.copy()
    local_tmp = cwd / ".tmp"
    npm_cache = cwd / ".npm-cache"
    browsers_path = cwd / ".ms-playwright"
    daemon_dir = cwd / ".playwright-daemon"
    for path in (local_tmp, npm_cache, browsers_path, daemon_dir):
        path.mkdir(parents=True, exist_ok=True)
    env["TEMP"] = str(local_tmp)
    env["TMP"] = str(local_tmp)
    env["npm_config_cache"] = str(npm_cache)
    env["NPM_CONFIG_CACHE"] = str(npm_cache)
    env["PLAYWRIGHT_BROWSERS_PATH"] = str(browsers_path)
    env["PLAYWRIGHT_DAEMON_SESSION_DIR"] = str(daemon_dir)
    return subprocess.run(
        args,
        cwd=str(cwd),
        env=env,
        text=True,
        encoding="utf-8",
        errors="replace",
        capture_output=True,
        timeout=timeout,
        check=check,
    )


def run_playwright(args: list[str], *, cwd: Path, timeout: int = 120) -> str:
    """Helper."""
    result = run_cmd(PLAYWRIGHT_CMD + args, cwd=cwd, timeout=timeout)
    return result.stdout


def write_js(workdir: Path, name: str, content: str) -> Path:
    """Helper."""
    path = workdir / name
    path.write_text(content, encoding="utf-8")
    return path


def resolve_playwright_download_path(filename: str, workdir: Path) -> Path:
    """Helper."""
    path = workdir / ".playwright-downloads" / filename
    path.parent.mkdir(parents=True, exist_ok=True)
    return path


def resolve_session_capture_path(root: Path, session: str, filename: str) -> Path:
    """Helper."""
    return root / "output" / "playwright" / session / filename


def build_login_js(username: str, password: str) -> str:
    """Helper."""
    payload = {"username": username, "password": password}
    spec = json.dumps(payload, ensure_ascii=False)
    return f"""async function(page) {{
  const spec = {spec};
  if (!page.url().includes('/#/signin')) {{
    return {{ url: page.url(), skipped: true }};
  }}

  const inputs = page.locator('input');
  const username = inputs.nth(0);
  const password = inputs.nth(1);
  const button = page.locator('button').nth(0);
  await username.waitFor({{ state: 'visible', timeout: 15000 }});
  await username.fill(spec.username);
  await password.fill(spec.password);
  await Promise.all([
    page.waitForFunction(() => !location.href.includes('/#/signin'), null, {{ timeout: 30000 }}).catch(() => null),
    button.click(),
  ]);
  await page.waitForLoadState('domcontentloaded').catch(() => null);
  return {{ url: page.url(), title: await page.title() }};
}}
"""


def build_configure_view_js(spec: dict[str, Any]) -> str:
    """Helper."""
    payload = json.dumps(spec, ensure_ascii=False)
    return f"""async function(page) {{
  const spec = {payload};
  await page.waitForLoadState('domcontentloaded').catch(() => null);

  async function inspectFrame(frame, preferredFragment) {{
    const frameUrl = typeof frame.url === 'function' ? frame.url() : null;
    try {{
      const snapshot = await frame.evaluate(() => {{
        const vizCount = window.tableau?.VizManager?.getVizs?.().length || 0;
        return {{
          url: location.href,
          title: document.title,
          readyState: document.readyState,
          hasTableau: !!(window.tableau && window.tableau.VizManager),
          vizCount,
        }};
      }});
      return {{
        ...snapshot,
        isMainFrame: frame === page.mainFrame(),
        matchesTargetFrame: !!(preferredFragment && frameUrl && frameUrl.includes(preferredFragment)),
      }};
    }} catch (error) {{
      return {{
        url: frameUrl,
        title: null,
        readyState: null,
        hasTableau: false,
        vizCount: 0,
        isMainFrame: frame === page.mainFrame(),
        matchesTargetFrame: !!(preferredFragment && frameUrl && frameUrl.includes(preferredFragment)),
        error: String(error && error.message || error),
      }};
    }}
  }}

  async function locateVizFrame(timeoutMs, preferredFragment) {{
    const deadline = Date.now() + timeoutMs;
    let lastSnapshots = [];
    while (Date.now() < deadline) {{
      const rankedFrames = page.frames()
        .map((frame) => {{
          const frameUrl = typeof frame.url === 'function' ? frame.url() : '';
          const isPreferred = !!(preferredFragment && frameUrl.includes(preferredFragment));
          return {{
            frame,
            rank: isPreferred ? 0 : (frame === page.mainFrame() ? 1 : 2),
          }};
        }})
        .sort((left, right) => left.rank - right.rank);

      lastSnapshots = [];
      for (const item of rankedFrames) {{
        const snapshot = await inspectFrame(item.frame, preferredFragment);
        lastSnapshots.push(snapshot);
        if (snapshot.vizCount > 0) {{
          return {{ frame: item.frame, snapshots: lastSnapshots }};
        }}
      }}

      const stillOnSignin =
        (page.url() || '').includes('/#/signin') ||
        lastSnapshots.some((snapshot) => (snapshot.url || '').includes('/#/signin'));
      if (stillOnSignin) {{
        throw new Error('Tableau page is still on sign-in: ' + JSON.stringify({{ pageUrl: page.url(), frames: lastSnapshots }}));
      }}

      const hasChromeError =
        (page.url() || '').startsWith('chrome-error://') ||
        lastSnapshots.some((snapshot) => (snapshot.url || '').startsWith('chrome-error://'));
      if (hasChromeError) {{
        throw new Error('Tableau page failed to load: ' + JSON.stringify({{ pageUrl: page.url(), frames: lastSnapshots }}));
      }}

      await page.waitForTimeout(1000);
    }}

    throw new Error('Timed out waiting for Tableau viz: ' + JSON.stringify({{ pageUrl: page.url(), frames: lastSnapshots }}));
  }}

  const located = await locateVizFrame(Number(spec.tableau_wait_ms || 90000), spec.inner_frame_fragment || '');

  return await located.frame.evaluate(async (config) => {{
    const sleep = (ms) => new Promise((resolve) => window.setTimeout(resolve, ms));

    const deadline = Date.now() + Number(config.workbook_wait_ms || 60000);
    let viz = null;
    let workbook = null;
    while (Date.now() < deadline) {{
      try {{
        const vizs = window.tableau?.VizManager?.getVizs?.() || [];
        viz = vizs[0] || null;
        workbook = viz && typeof viz.getWorkbook === 'function' ? viz.getWorkbook() : null;
      }} catch (error) {{
        workbook = null;
      }}
      if (workbook) {{
        break;
      }}
      await sleep(500);
    }}
    if (!workbook) {{
      throw new Error('Tableau workbook is not ready: ' + JSON.stringify({{
        url: location.href,
        title: document.title,
        readyState: document.readyState,
        vizCount: window.tableau?.VizManager?.getVizs?.().length || 0,
      }}));
    }}

    const shouldRevert = config.revert_before_apply !== false;
    if (shouldRevert) {{
      try {{
        await workbook.revertAllAsync();
      }} catch (error) {{
      }}
    }}

    await workbook.activateSheetAsync(config.activate_sheet);
    const activeSheet = workbook.getActiveSheet();
    const worksheetList = activeSheet && typeof activeSheet.getWorksheets === 'function'
      ? activeSheet.getWorksheets()
      : [];
    const worksheets = Object.fromEntries(
      worksheetList.map((worksheet) => [worksheet.getName(), worksheet])
    );
    const targetsByName = {{
      ...(activeSheet && typeof activeSheet.getName === 'function'
        ? {{ [activeSheet.getName()]: activeSheet }}
        : {{}}),
      ...worksheets,
    }};
    const updateType = window.tableau.FilterUpdateType.REPLACE;
    const applyLog = [];
    const applySummary = {{}};

    function ensureSummary(label, requiredSuccesses) {{
      const resolvedLabel = label || 'unlabeled';
      if (!applySummary[resolvedLabel]) {{
        applySummary[resolvedLabel] = {{
          label: resolvedLabel,
          requiredSuccesses: Number(requiredSuccesses || 0),
          successes: 0,
          targetNames: [],
        }};
      }} else if (Number(requiredSuccesses || 0) > applySummary[resolvedLabel].requiredSuccesses) {{
        applySummary[resolvedLabel].requiredSuccesses = Number(requiredSuccesses || 0);
      }}
      return applySummary[resolvedLabel];
    }}

    for (const action of config.filter_actions) {{
      try {{
        if (action.kind === 'parameter') {{
          await workbook.changeParameterValueAsync(action.name, action.value);
          applyLog.push({{ kind: action.kind, name: action.name, ok: true }});
          continue;
        }}
        if (action.kind !== 'filter') {{
          applyLog.push({{
            kind: action.kind || null,
            label: action.label || null,
            ok: true,
            skipped: true,
          }});
          continue;
        }}

        const summary = ensureSummary(action.label, action.required_successes);
        const targetNames = Array.isArray(action.target_names)
          ? action.target_names
          : [action.worksheet].filter(Boolean);
        const fieldCandidates = Array.isArray(action.field_candidates)
          ? action.field_candidates
          : [action.field].filter(Boolean);

        for (const targetName of targetNames) {{
          if (!summary.targetNames.includes(targetName)) {{
            summary.targetNames.push(targetName);
          }}

          const target = targetsByName[targetName];
          if (!target) {{
            applyLog.push({{
              kind: action.kind,
              label: action.label || null,
              target: targetName,
              values: action.values,
              ok: false,
              error: 'target not found',
            }});
            continue;
          }}

          let targetApplied = false;
          let lastTargetError = null;
          for (const fieldCandidate of fieldCandidates) {{
            try {{
              await target.applyFilterAsync(fieldCandidate, action.values, updateType);
              summary.successes += 1;
              targetApplied = true;
              applyLog.push({{
                kind: action.kind,
                label: action.label || null,
                target: targetName,
                field: fieldCandidate,
                values: action.values,
                ok: true,
              }});
              break;
            }} catch (error) {{
              lastTargetError = String(error && error.message || error);
            }}
          }}

          if (!targetApplied) {{
            applyLog.push({{
              kind: action.kind,
              label: action.label || null,
              target: targetName,
              fields: fieldCandidates,
              values: action.values,
              ok: false,
              error: lastTargetError || 'no matching field candidate succeeded',
            }});
          }}
        }}
      }} catch (error) {{
        applyLog.push({{ kind: action.kind, label: action.label || null, name: action.name || null, ok: false, error: String(error) }});
      }}
    }}

    const missingRequired = Object.values(applySummary).filter((entry) =>
      Number(entry.requiredSuccesses || 0) > 0 && Number(entry.successes || 0) < Number(entry.requiredSuccesses || 0)
    );
    if (missingRequired.length) {{
      throw new Error('Required Tableau filters did not reach enough targets: ' + JSON.stringify({{
        missingRequired,
        applyLog,
      }}));
    }}

    return {{
      frameUrl: location.href,
      activeSheet: activeSheet.getName(),
      filterActions: config.filter_actions,
      applyLog,
      applySummary,
      url: location.href,
    }};
  }}, {{
    activate_sheet: spec.activate_sheet,
    filter_actions: spec.filter_actions,
    workbook_wait_ms: spec.workbook_wait_ms || 60000,
    revert_before_apply: spec.revert_before_apply,
  }});
}}
"""


def build_normalize_quick_filters_js(spec: dict[str, Any]) -> str:
    """Helper."""
    payload = json.dumps(spec, ensure_ascii=False)
    return f"""async function(page) {{
  const spec = {payload};
  const allFilterValue = String(spec.all_filter_value || '(全部)');
  await page.waitForLoadState('domcontentloaded').catch(() => null);

  async function readLocatorText(locator) {{
    const count = await locator.count().catch(() => 0);
    if (!count) {{
      return '';
    }}
    const first = locator.first();
    return String(
      await first.innerText().catch(async () => await first.textContent().catch(() => '')) || ''
    ).trim();
  }}

  function escapeRegExp(value) {{
    return String(value || '').replace(/[.*+?^${{}}()|[\\]\\\\]/g, '\\\\$&');
  }}

  function buildExactMatcher(value) {{
    return new RegExp(`^\\\\s*${{escapeRegExp(value)}}\\\\s*$`, 'i');
  }}

  async function collectQuickFilterHeadings(frame) {{
    const headings = [];
    const locator = frame.getByRole('heading');
    const count = await locator.count().catch(() => 0);
    for (let index = 0; index < count; index += 1) {{
      const heading = await readLocatorText(locator.nth(index));
      if (heading) {{
        headings.push(heading);
      }}
    }}
    return headings;
  }}

  async function locateQuickFilterFrame(timeoutMs, preferredFragment) {{
    const deadline = Date.now() + timeoutMs;
    let lastFrames = [];
    while (Date.now() < deadline) {{
      lastFrames = [];
      for (const frame of page.frames()) {{
        const frameUrl = typeof frame.url === 'function' ? frame.url() : '';
        if (preferredFragment && !frameUrl.includes(preferredFragment)) {{
          continue;
        }}
        const headingCount = await frame.getByRole('heading').count().catch(() => 0);
        const buttonCount = await frame.getByRole('button').count().catch(() => 0);
        const monthHeadingCount = await frame.getByRole('heading', {{ name: /Month/i }}).count().catch(() => 0);
        const weekHeadingCount = await frame.getByRole('heading', {{ name: /Week/i }}).count().catch(() => 0);
        const dailyHeadingCount = await frame.getByRole('heading', {{ name: /Daily/i }}).count().catch(() => 0);
        const snapshot = {{
          frameUrl,
          headingCount,
          buttonCount,
          monthHeadingCount,
          weekHeadingCount,
          dailyHeadingCount,
        }};
        lastFrames.push(snapshot);
        if (buttonCount > 20 && (monthHeadingCount > 0 || weekHeadingCount > 0 || dailyHeadingCount > 0)) {{
          return {{ frame, lastFrames }};
        }}
      }}
      await page.waitForTimeout(1000);
    }}
    throw new Error('Timed out waiting for Tableau quick filters: ' + JSON.stringify(lastFrames));
  }}

  async function locateQuickFilterButton(frame, label, timeoutMs) {{
    const deadline = Date.now() + timeoutMs;
    const matcher = new RegExp(escapeRegExp(label), 'i');
    while (Date.now() < deadline) {{
      const button = frame.getByRole('button', {{ name: matcher }}).first();
      const buttonCount = await button.count().catch(() => 0);
      if (buttonCount > 0 && await button.isVisible().catch(() => true)) {{
        return button;
      }}
      const heading = frame.getByRole('heading', {{ name: matcher }}).first();
      const headingCount = await heading.count().catch(() => 0);
      if (headingCount > 0 && await heading.isVisible().catch(() => true)) {{
        const followingButton = heading.locator('xpath=following::button[1]').first();
        const followingCount = await followingButton.count().catch(() => 0);
        if (followingCount > 0 && await followingButton.isVisible().catch(() => true)) {{
          return followingButton;
        }}
      }}
      await page.waitForTimeout(1000);
    }}
    return null;
  }}

  async function waitForButtonText(button, expectedValue, timeoutMs) {{
    const deadline = Date.now() + timeoutMs;
    while (Date.now() < deadline) {{
      const buttonText = await readLocatorText(button);
      if (!expectedValue || buttonText.includes(expectedValue)) {{
        return buttonText;
      }}
      await page.waitForTimeout(1000);
    }}
    return await readLocatorText(button);
  }}

  async function openQuickFilter(frame, button, label, timeoutMs) {{
    const deadline = Date.now() + timeoutMs;
    while (Date.now() < deadline) {{
      await button.click({{ timeout: 5000 }}).catch(async () => {{
        await button.click({{ timeout: 5000, force: true }}).catch(() => null);
      }});
      const listbox = frame.getByRole('listbox').last();
      const listboxCount = await listbox.count().catch(() => 0);
      if (listboxCount > 0 && await listbox.isVisible().catch(() => false)) {{
        await page.waitForTimeout(500);
        return listbox;
      }}
      await page.waitForTimeout(500);
    }}
    throw new Error(`Failed to open quick filter ${{label}}`);
  }}

  async function closeQuickFilter(button) {{
    try {{
      await button.click({{ timeout: 5000 }});
      await page.waitForTimeout(1000);
      return 'button-click';
    }} catch (error) {{}}
    try {{
      await button.click({{ timeout: 5000, force: true }});
      await page.waitForTimeout(1000);
      return 'button-click-force';
    }} catch (error) {{}}
    await page.keyboard.press('Escape').catch(() => null);
    await page.waitForTimeout(1000);
    return 'escape';
  }}

  async function findRow(listbox, value) {{
    const row = listbox.getByRole('checkbox', {{ name: buildExactMatcher(value) }}).first();
    const count = await row.count().catch(() => 0);
    return count > 0 ? row : null;
  }}

  async function readRowState(row) {{
    const count = await row.count().catch(() => 0);
    if (!count) {{
      return {{
        exists: false,
        checked: false,
        text: '',
      }};
    }}
    return await row.evaluate((element) => {{
      const target = element.closest('[role="checkbox"]') || element;
      const input = target.querySelector('input.FICheckRadio');
      const className = String(target.className || '');
      const ariaChecked = String(target.getAttribute('aria-checked') || '');
      const inputChecked = !!(input && input.checked);
      return {{
        exists: true,
        text: String(target.textContent || '').trim(),
        ariaChecked,
        className,
        inputChecked,
        checked: inputChecked || ariaChecked.toLowerCase() === 'true' || className.includes('FIChecked'),
      }};
    }}).catch((error) => ({{
      exists: true,
      checked: false,
      text: '',
      error: String(error && error.message || error),
    }}));
  }}

  async function clickFakeCheckbox(row) {{
    const fake = row.locator('.fakeCheckBox').first();
    const fakeCount = await fake.count().catch(() => 0);
    if (fakeCount > 0) {{
      try {{
        await fake.click({{ timeout: 5000 }});
        return 'fake-click';
      }} catch (error) {{}}
      try {{
        await fake.click({{ timeout: 5000, force: true }});
        return 'fake-click-force';
      }} catch (error) {{}}
    }}
    try {{
      await row.evaluate((element) => {{
        const target = element.closest('[role="checkbox"]') || element;
        const input = target.querySelector('input.FICheckRadio');
        if (input) {{
          input.click();
          return;
        }}
        const fallback = target.querySelector('.fakeCheckBox') || target;
        for (const type of ['pointerdown', 'mousedown', 'mouseup', 'click']) {{
          fallback.dispatchEvent(new MouseEvent(type, {{ bubbles: true, cancelable: true, composed: true }}));
        }}
      }});
      return 'dom-input-click';
    }} catch (error) {{}}
    await row.click({{ timeout: 5000, force: true }});
    return 'row-click-force';
  }}

  async function listCheckedRows(listbox) {{
    const rows = listbox.getByRole('checkbox');
    const count = await rows.count().catch(() => 0);
    const checkedRows = [];
    for (let index = 0; index < count; index += 1) {{
      const row = rows.nth(index);
      const state = await readRowState(row);
      if (state.checked) {{
        checkedRows.push({{ index, ...state }});
      }}
    }}
    return checkedRows;
  }}

  async function applyQuickFilter(frame, action) {{
    const label = String(action?.label || '').trim();
    const desiredValues = Array.isArray(action?.values)
      ? action.values.map((value) => String(value || '').trim()).filter(Boolean)
      : [];
    const expectedButtonText = desiredValues.length === 1 ? desiredValues[0] : desiredValues.join(', ');
    const button = await locateQuickFilterButton(frame, label, Number(spec.quick_filter_wait_ms || 30000));
    if (!button) {{
      return {{
        ok: false,
        action: 'missing-filter',
        detail: {{
          label,
          availableHeadings: await collectQuickFilterHeadings(frame).catch(() => []),
        }},
      }};
    }}

    const initialButtonText = await readLocatorText(button);
    if (expectedButtonText && initialButtonText.includes(expectedButtonText)) {{
      return {{
        ok: true,
        action: 'already-set',
        detail: {{
          initialButtonText,
          finalButtonText: initialButtonText,
          expectedButtonText,
          desiredValues,
          checkedBefore: [],
          checkedAfter: [],
          desiredStates: [],
          actionsTaken: [],
          missingValues: [],
        }},
      }};
    }}

    const listbox = await openQuickFilter(frame, button, label, Number(spec.quick_filter_wait_ms || 30000));
    const checkedBefore = await listCheckedRows(listbox);
    const actionsTaken = [];
    const deselectLabels = [];
    const currentValue = String(initialButtonText || '').trim();
    if (
      currentValue &&
      currentValue !== expectedButtonText &&
      currentValue !== allFilterValue &&
      currentValue !== '(多个值)' &&
      currentValue !== '(无)'
    ) {{
      deselectLabels.push(currentValue);
    }} else {{
      const handledCheckedLabels = new Set();
      for (const checkedEntry of checkedBefore) {{
        const checkedLabel = String(checkedEntry.text || '').trim();
        if (!checkedLabel || desiredValues.includes(checkedLabel) || handledCheckedLabels.has(checkedLabel)) {{
          continue;
        }}
        handledCheckedLabels.add(checkedLabel);
        deselectLabels.push(checkedLabel);
      }}
    }}

    for (const deselectLabel of deselectLabels) {{
      const checkedRow = await findRow(listbox, deselectLabel);
      if (!checkedRow) {{
        continue;
      }}
      const method = await clickFakeCheckbox(checkedRow);
      actionsTaken.push({{
        kind: 'deselect',
        label: deselectLabel,
        method,
      }});
      await page.waitForTimeout(300);
    }}

    const missingValues = [];
    for (const desiredValue of desiredValues) {{
      const desiredRow = await findRow(listbox, desiredValue);
      if (!desiredRow) {{
        missingValues.push(desiredValue);
        continue;
      }}
      const desiredState = await readRowState(desiredRow);
      if (!desiredState.checked) {{
        const method = await clickFakeCheckbox(desiredRow);
        actionsTaken.push({{
          kind: 'select',
          label: desiredValue,
          method,
        }});
        await page.waitForTimeout(300);
      }}
    }}

    const desiredStates = [];
    for (const desiredValue of desiredValues) {{
      const desiredRow = await findRow(listbox, desiredValue);
      desiredStates.push({{
        label: desiredValue,
        ...(desiredRow ? await readRowState(desiredRow) : {{ exists: false, checked: false, text: desiredValue }}),
      }});
    }}
    const checkedAfter = await listCheckedRows(listbox);
    const closeMethod = await closeQuickFilter(button);
    const finalButtonText = await waitForButtonText(button, expectedButtonText, Number(spec.close_wait_ms || 15000));
    const ok =
      missingValues.length === 0 &&
      desiredStates.every((entry) => entry.checked) &&
      (!expectedButtonText || finalButtonText.includes(expectedButtonText));
    return {{
      ok,
      action: 'applied',
      detail: {{
        initialButtonText,
        finalButtonText,
        expectedButtonText,
        desiredValues,
        checkedBefore,
        checkedAfter,
        desiredStates,
        actionsTaken,
        missingValues,
        closeMethod,
      }},
    }};
  }}

  const located = await locateQuickFilterFrame(Number(spec.tableau_wait_ms || 90000), spec.inner_frame_fragment || '');
  const frame = located.frame;
  const logs = [];
  const filters = {{}};
  for (const action of (spec.filter_actions || [])) {{
    if (action?.kind !== 'quick-filter' && action?.kind !== 'filter') {{
      continue;
    }}
    try {{
      const result = await applyQuickFilter(frame, action);
      const label = String(action.label || '').trim();
      logs.push({{
        kind: 'quick-filter-ui',
        label: label.toLowerCase(),
        ok: result.ok,
        action: result.action,
        detail: result.detail,
      }});
      filters[label] = {{
        buttonText: String(result.detail?.finalButtonText || result.detail?.initialButtonText || '').trim(),
        value: String(result.detail?.finalButtonText || result.detail?.initialButtonText || '').trim(),
      }};
    }} catch (error) {{
      logs.push({{
        kind: 'quick-filter-ui',
        label: String(action.label || '').toLowerCase(),
        ok: false,
        error: String(error && error.message || error),
      }});
    }}
  }}

  return {{
    frameUrl: frame.url(),
    frameCandidates: located.lastFrames,
    quickFilterHeadings: await collectQuickFilterHeadings(frame).catch(() => []),
    logs,
    filters,
  }};
}}
"""


def build_wait_for_top_products_js(spec: dict[str, Any]) -> str:
    """Helper."""
    payload = json.dumps(spec, ensure_ascii=False)
    return f"""async function(page) {{
  const spec = {payload};
  await page.waitForLoadState('domcontentloaded').catch(() => null);

  async function inspectFrame(frame, preferredFragment) {{
    const frameUrl = typeof frame.url === 'function' ? frame.url() : null;
    try {{
      const snapshot = await frame.evaluate(() => {{
        const vizCount = window.tableau?.VizManager?.getVizs?.().length || 0;
        return {{
          url: location.href,
          title: document.title,
          readyState: document.readyState,
          hasTableau: !!(window.tableau && window.tableau.VizManager),
          vizCount,
        }};
      }});
      return {{
        ...snapshot,
        isMainFrame: frame === page.mainFrame(),
        matchesTargetFrame: !!(preferredFragment && frameUrl && frameUrl.includes(preferredFragment)),
      }};
    }} catch (error) {{
      return {{
        url: frameUrl,
        title: null,
        readyState: null,
        hasTableau: false,
        vizCount: 0,
        isMainFrame: frame === page.mainFrame(),
        matchesTargetFrame: !!(preferredFragment && frameUrl && frameUrl.includes(preferredFragment)),
        error: String(error && error.message || error),
      }};
    }}
  }}

  async function locateVizFrame(timeoutMs, preferredFragment) {{
    const deadline = Date.now() + timeoutMs;
    let lastSnapshots = [];
    while (Date.now() < deadline) {{
      const rankedFrames = page.frames()
        .map((frame) => {{
          const frameUrl = typeof frame.url === 'function' ? frame.url() : '';
          const isPreferred = !!(preferredFragment && frameUrl.includes(preferredFragment));
          return {{
            frame,
            rank: isPreferred ? 0 : (frame === page.mainFrame() ? 1 : 2),
          }};
        }})
        .sort((left, right) => left.rank - right.rank);

      lastSnapshots = [];
      for (const item of rankedFrames) {{
        const snapshot = await inspectFrame(item.frame, preferredFragment);
        lastSnapshots.push(snapshot);
        if (snapshot.vizCount > 0) {{
          return {{ frame: item.frame, snapshots: lastSnapshots }};
        }}
      }}

      await page.waitForTimeout(1000);
    }}

    throw new Error('Timed out waiting for Tableau viz: ' + JSON.stringify({{ pageUrl: page.url(), frames: lastSnapshots }}));
  }}

  const located = await locateVizFrame(Number(spec.tableau_wait_ms || 90000), spec.inner_frame_fragment || '');
  return await located.frame.evaluate(async (config) => {{
    const sleep = (ms) => new Promise((resolve) => window.setTimeout(resolve, ms));
    const extractRowSignature = (table) => {{
      const columns = table.getColumns().map((col) => col.getFieldName());
      const rows = table.getData();
      const articleIdx = columns.findIndex((name) => name === 'Articlecode');
      const qtyIdx = columns.findIndex((name) => name === '鎬诲拰(Soldqty)');
      const rankIdx = columns.findIndex((name) => name === '鑱氬悎(No.)');
      const imageIdx = columns.findIndex((name) => String(name || '').includes('max url'));
      const signatureRows = rows
        .map((row) => {{
          const article = articleIdx >= 0 ? String(row[articleIdx]?.formattedValue || row[articleIdx]?.value || '').trim() : '';
          const qty = qtyIdx >= 0 ? String(row[qtyIdx]?.formattedValue || row[qtyIdx]?.value || '') : '';
          const rank = rankIdx >= 0 ? String(row[rankIdx]?.formattedValue || row[rankIdx]?.value || '') : '';
          const image = imageIdx >= 0 ? String(row[imageIdx]?.formattedValue || row[imageIdx]?.value || '').trim() : '';
          return article || image ? [article, qty, rank, image].join('|') : null;
        }})
        .filter(Boolean);
      return {{
        columns,
        rowCount: rows.length,
        validRowCount: signatureRows.length,
        signature: JSON.stringify(signatureRows),
      }};
    }};

    const workbookDeadline = Date.now() + Number(config.workbook_wait_ms || 60000);
    let viz = null;
    let workbook = null;
    while (Date.now() < workbookDeadline) {{
      try {{
        const vizs = window.tableau?.VizManager?.getVizs?.() || [];
        viz = vizs[0] || null;
        workbook = viz && typeof viz.getWorkbook === 'function' ? viz.getWorkbook() : null;
      }} catch (error) {{
        workbook = null;
      }}
      if (workbook) {{
        break;
      }}
      await sleep(500);
    }}
    if (!workbook) {{
      throw new Error('Tableau workbook is not ready for wait polling.');
    }}

    try {{
      await workbook.activateSheetAsync(config.activate_sheet);
    }} catch (error) {{
    }}
    const activeSheet = workbook.getActiveSheet();
    const worksheetList = activeSheet && typeof activeSheet.getWorksheets === 'function'
      ? activeSheet.getWorksheets()
      : [];
    const worksheets = Object.fromEntries(
      worksheetList.map((worksheet) => [worksheet.getName(), worksheet])
    );
    const worksheet = worksheets.TOP || null;
    if (!worksheet || typeof worksheet.getSummaryDataAsync !== 'function') {{
      return {{
        frameUrl: location.href,
        activeSheet: activeSheet && typeof activeSheet.getName === 'function' ? activeSheet.getName() : null,
        waitResult: {{
          ready: false,
          skipped: true,
          reason: 'TOP worksheet not available for polling',
        }},
      }};
    }}

    const pollIntervalMs = Number(config.post_filter_poll_interval_ms || 1000);
    const timeoutMs = Number(config.post_filter_wait_ms || 180000);
    const fallbackWaitMs = Number(config.post_filter_fallback_wait_ms || 0);
    const stablePollsRequired = Number(config.post_filter_stable_polls || 2);
    const expectedTopRows = Number(config.expected_top_rows || 10);
    const deadline = Date.now() + timeoutMs;
    const startedAt = Date.now();
    let lastSignature = null;
    let stablePolls = 0;
    let lastSnapshot = null;

    while (Date.now() < deadline) {{
      try {{
        const table = await worksheet.getSummaryDataAsync({{ maxRows: expectedTopRows }});
        const snapshot = extractRowSignature(table);
        lastSnapshot = snapshot;
        if (snapshot.signature === lastSignature) {{
          stablePolls += 1;
        }} else {{
          stablePolls = 1;
          lastSignature = snapshot.signature;
        }}

        const elapsedMs = Date.now() - startedAt;
        const hasExpectedRows = snapshot.validRowCount >= expectedTopRows;
        const stableEnough = stablePolls >= stablePollsRequired;
        if (hasExpectedRows && stableEnough) {{
          return {{
            frameUrl: location.href,
            activeSheet: activeSheet && typeof activeSheet.getName === 'function' ? activeSheet.getName() : null,
            waitResult: {{
              ready: true,
              skipped: false,
              elapsedMs,
              stablePolls,
              ...snapshot,
            }},
          }};
        }}
      }} catch (error) {{
        lastSnapshot = {{
          rowCount: 0,
          validRowCount: 0,
          signature: null,
          error: String(error && error.message || error),
        }};
      }}
      await sleep(pollIntervalMs);
    }}

    if (fallbackWaitMs > 0) {{
      await sleep(fallbackWaitMs);
    }}
    return {{
      frameUrl: location.href,
      activeSheet: activeSheet && typeof activeSheet.getName === 'function' ? activeSheet.getName() : null,
      waitResult: {{
        ready: false,
        skipped: false,
        timedOut: true,
        stablePolls,
        ...(lastSnapshot || {{ rowCount: 0, validRowCount: 0, signature: null }}),
      }},
    }};
  }}, {{
    activate_sheet: spec.activate_sheet,
    expected_top_rows: spec.expected_top_rows || 10,
    workbook_wait_ms: spec.workbook_wait_ms || 60000,
    post_filter_wait_ms: spec.post_filter_wait_ms || 180000,
    post_filter_poll_interval_ms: spec.post_filter_poll_interval_ms || 1000,
    post_filter_fallback_wait_ms: spec.post_filter_fallback_wait_ms || 0,
    post_filter_stable_polls: spec.post_filter_stable_polls || 2,
  }});
}}
"""


def build_capture_js(inner_frame_fragment: str, screenshot_name: str) -> str:
    """Helper."""
    payload = json.dumps(
        {
            "inner_frame_fragment": inner_frame_fragment,
            "screenshot_name": screenshot_name,
        },
        ensure_ascii=False,
    )
    return f"""async function(page) {{
  const spec = {payload};
  await page.waitForLoadState('domcontentloaded').catch(() => null);
  const frame = page.frames().find(
    (candidate) => candidate !== page.mainFrame() && candidate.url().includes(spec.inner_frame_fragment)
  );
  const target = frame || page;
  const body = await target.$('body');
  await body.screenshot({{
    path: spec.screenshot_name,
    scale: 'css',
  }});
  return {{
    frameUrl: frame ? frame.url() : null,
    pageUrl: page.url(),
    screenshotName: spec.screenshot_name,
  }};
}}
"""


def run_code(session: str, script_path: Path, *, cwd: Path, timeout: int = 120) -> str:
    """Helper."""
    return run_playwright(["--session", session, "run-code", "--filename", str(script_path.resolve())], cwd=cwd, timeout=timeout)


def parse_playwright_result(output: str) -> Any:
    """Helper."""
    if "### Error" in output:
        raise RuntimeError(output.strip())
    marker = "### Result"
    start = output.find(marker)
    if start == -1:
        raise RuntimeError(f"Playwright output missing result marker: {output.strip()}")
    payload = output[start + len(marker):].lstrip()
    next_marker = payload.find("\n### ")
    if next_marker != -1:
        payload = payload[:next_marker]
    return json.loads(payload.strip())


def extract_playwright_snapshot_yaml(output: str) -> str:
    """Extract the YAML snapshot block from playwright-cli snapshot output."""
    marker = "### Snapshot"
    start = output.find(marker)
    if start == -1:
        return output
    code_start = output.find("```yaml", start)
    if code_start == -1:
        return output[start + len(marker) :]
    code_start += len("```yaml")
    code_end = output.find("```", code_start)
    if code_end == -1:
        code_end = len(output)
    return output[code_start:code_end].strip()


def parse_top_products_quick_filter_snapshot(snapshot_yaml: str) -> dict[str, dict[str, Any]]:
    """Parse quick-filter refs and current values from a Tableau page snapshot."""
    heading_pattern = re.compile(r'^\s*-\s+heading "筛选器 (?P<label>.+?) 包含".*$')
    button_pattern = re.compile(r'^\s*-\s+button "筛选器 (?P<label>.+?) 包含".*?\[ref=(?P<ref>[^\]]+)\]:$')
    value_pattern = re.compile(r'^\s*-\s+generic \[ref=[^\]]+\]: (?P<value>.+)$')
    show_all_pattern = re.compile(
        r'^\s*-\s+button "(?P<label>显示所有值|单击以显示所有值|單擊以顯示所有值)".*?\[ref=(?P<ref>[^\]]+)\]:$'
    )
    context_menu_pattern = re.compile(r'^\s*-\s+button "显示快速筛选器上下文菜单".*?\[ref=(?P<ref>[^\]]+)\].*$')

    filters: dict[str, dict[str, Any]] = {}
    current_filter: str | None = None
    pending_value_filter: str | None = None

    for line in snapshot_yaml.splitlines():
        heading_match = heading_pattern.search(line)
        if heading_match:
            current_filter = heading_match.group("label").strip()
            filters.setdefault(current_filter, {})
            continue

        button_match = button_pattern.search(line)
        if button_match:
            current_filter = button_match.group("label").strip()
            filters.setdefault(current_filter, {})["button_ref"] = button_match.group("ref")
            pending_value_filter = current_filter
            continue

        if pending_value_filter:
            value_match = value_pattern.search(line)
            if value_match:
                filters.setdefault(pending_value_filter, {})["value"] = value_match.group("value").strip()
                pending_value_filter = None
                continue

        if current_filter:
            show_all_match = show_all_pattern.search(line)
            if show_all_match:
                entry = filters.setdefault(current_filter, {})
                entry["show_all_ref"] = show_all_match.group("ref")
                entry["show_all_label"] = show_all_match.group("label")
                entry["show_all_disabled"] = "[disabled]" in line
                continue

            context_menu_match = context_menu_pattern.search(line)
            if context_menu_match:
                entry = filters.setdefault(current_filter, {})
                entry["context_menu_ref"] = context_menu_match.group("ref")
                entry["context_menu_disabled"] = "[disabled]" in line

    return filters


def normalize_top_products_quick_filters(
    session: str,
    *,
    cwd: Path,
    timeout_seconds: int = 30,
) -> dict[str, Any]:
    """Normalize the hidden Week/Daily quick filters using snapshot refs."""
    started_at = time.monotonic()
    click_log: list[dict[str, Any]] = []
    last_filters: dict[str, dict[str, Any]] = {}
    last_snapshot_yaml = ""
    targets = ("Week", "Daily")

    while time.monotonic() - started_at < timeout_seconds:
        snapshot_output = run_playwright(["--session", session, "snapshot"], cwd=cwd, timeout=60)
        last_snapshot_yaml = extract_playwright_snapshot_yaml(snapshot_output)
        last_filters = parse_top_products_quick_filter_snapshot(last_snapshot_yaml)

        if all(str(last_filters.get(target, {}).get("value") or "").strip() == ALL_FILTER_VALUE for target in targets):
            return {
                "ok": True,
                "filters": last_filters,
                "click_log": click_log,
                "elapsed_ms": int(round((time.monotonic() - started_at) * 1000)),
            }

        if "响应时间过长" in last_snapshot_yaml:
            return {
                "ok": False,
                "filters": last_filters,
                "click_log": click_log,
                "elapsed_ms": int(round((time.monotonic() - started_at) * 1000)),
                "reason": "backend-timeout",
                "snapshot_excerpt": last_snapshot_yaml[:4000],
            }

        clicked = False
        for target in targets:
            state = last_filters.get(target, {})
            current_value = str(state.get("value") or "").strip()

            show_all_ref = str(state.get("show_all_ref") or "").strip()
            if show_all_ref and not state.get("show_all_disabled"):
                run_playwright(["--session", session, "click", show_all_ref], cwd=cwd, timeout=60)
                click_log.append(
                    {
                        "kind": "quick-filter-snapshot",
                        "target": target,
                        "action": "click-show-all",
                        "ref": show_all_ref,
                        "label": state.get("show_all_label"),
                        "before": current_value,
                    }
                )
                clicked = True
                break

            context_menu_ref = str(state.get("context_menu_ref") or "").strip()
            if context_menu_ref and not state.get("context_menu_disabled"):
                run_playwright(["--session", session, "click", context_menu_ref], cwd=cwd, timeout=60)
                click_log.append(
                    {
                        "kind": "quick-filter-snapshot",
                        "target": target,
                        "action": "open-context-menu",
                        "ref": context_menu_ref,
                        "before": current_value,
                    }
                )
                clicked = True
                break

            button_ref = str(state.get("button_ref") or "").strip()
            if button_ref and current_value in {"", "(无)"}:
                run_playwright(["--session", session, "click", button_ref], cwd=cwd, timeout=60)
                click_log.append(
                    {
                        "kind": "quick-filter-snapshot",
                        "target": target,
                        "action": "open-filter",
                        "ref": button_ref,
                        "before": current_value,
                    }
                )
                clicked = True
                break

        if clicked:
            time.sleep(1)
            continue

        time.sleep(1)

    return {
        "ok": False,
        "filters": last_filters,
        "click_log": click_log,
        "elapsed_ms": int(round((time.monotonic() - started_at) * 1000)),
        "snapshot_excerpt": last_snapshot_yaml[:4000],
    }


def ensure_browser_session(session: str, *, cwd: Path, browser_config_path: Path | None = None) -> None:
    """Helper."""
    try:
        run_playwright(["--session", session, "tab-list"], cwd=cwd, timeout=15)
        return
    except subprocess.CalledProcessError as error:
        combined = f"{error.stdout}\n{error.stderr}"
        if "is not open" not in combined:
            raise
    args = ["--session", session, "open"]
    if browser_config_path is not None:
        args.extend(["--config", str(browser_config_path)])
    args.append("about:blank")
    run_playwright(args, cwd=cwd, timeout=60)


def save_state(session: str, state_path: Path, *, cwd: Path) -> None:
    """Helper."""
    state_path.parent.mkdir(parents=True, exist_ok=True)
    run_playwright(["--session", session, "state-save", str(state_path)], cwd=cwd, timeout=60)


def load_state_if_present(session: str, state_path: Path, *, cwd: Path) -> None:
    """Helper."""
    if not state_path.exists():
        return
    run_playwright(["--session", session, "state-load", str(state_path)], cwd=cwd, timeout=60)


def locate_session_file(
    root: Path,
    session: str,
    filename: str,
    *,
    workdir: Path | None = None,
) -> Path:
    """Helper."""
    candidates: list[Path] = []
    if workdir is not None:
        candidates.extend(
            [
                resolve_playwright_download_path(filename, workdir),
                workdir / filename,
            ]
        )
    candidates.append(resolve_session_capture_path(root, session, filename))
    for candidate in candidates:
        if candidate.exists():
            return candidate
    raise FileNotFoundError(f"Unable to locate {filename} under {root}.")


def crop_image(source: Path, target: Path, crop: dict[str, int], *, resize_to: dict[str, int] | None = None) -> None:
    """Helper."""
    image = Image.open(source)
    box = (
        crop["left"],
        crop["top"],
        crop["left"] + crop["width"],
        crop["top"] + crop["height"],
    )
    result = image.crop(box)
    if resize_to:
        result = result.resize((resize_to["width"], resize_to["height"]), Image.Resampling.LANCZOS)
    target.parent.mkdir(parents=True, exist_ok=True)
    result.save(target)


def capture_tableau_view(
    capture_spec: dict[str, Any],
    *,
    base_url: str,
    session: str,
    workdir: Path,
    workspace_root: Path,
) -> Path:
    """Helper."""
    target_url = f"{base_url}{capture_spec['hash_url']}"
    preferred_capture_path = resolve_playwright_download_path(capture_spec["raw_screenshot_name"], workdir)
    legacy_session_output = resolve_session_capture_path(
        workspace_root,
        session,
        capture_spec["raw_screenshot_name"],
    )
    for stale_path in (
        preferred_capture_path,
        workdir / capture_spec["raw_screenshot_name"],
        legacy_session_output,
    ):
        stale_path.unlink(missing_ok=True)
    def reset_target_page() -> None:
        run_playwright(["--session", session, "goto", target_url], cwd=workdir, timeout=120)
        run_playwright(
            ["--session", session, "resize", str(VIEWPORT["width"]), str(VIEWPORT["height"])],
            cwd=workdir,
            timeout=60,
        )

    reset_target_page()

    pre_configure_script = write_js(
        workdir,
        f"tmp-{capture_spec['capture_id']}-configure-pre.js",
        build_configure_view_js(
            {
                "activate_sheet": capture_spec["activate_sheet"],
                "filter_actions": capture_spec.get("api_pre_filter_actions", []),
                "inner_frame_fragment": capture_spec["inner_frame_fragment"],
                "revert_before_apply": True,
            }
        ),
    )
    ui_filter_script = write_js(
        workdir,
        f"tmp-{capture_spec['capture_id']}-quick-filters.js",
        build_normalize_quick_filters_js(
            {
                "filter_actions": capture_spec.get("ui_filter_actions", []),
                "inner_frame_fragment": capture_spec["inner_frame_fragment"],
            }
        ),
    )
    post_configure_script = write_js(
        workdir,
        f"tmp-{capture_spec['capture_id']}-configure-post.js",
        build_configure_view_js(
            {
                "activate_sheet": capture_spec["activate_sheet"],
                "filter_actions": capture_spec.get("api_post_filter_actions", []),
                "inner_frame_fragment": capture_spec["inner_frame_fragment"],
                "revert_before_apply": False,
            }
        ),
    )
    wait_script = write_js(
        workdir,
        f"tmp-{capture_spec['capture_id']}-wait.js",
        build_wait_for_top_products_js(
            {
                "activate_sheet": capture_spec["activate_sheet"],
                "inner_frame_fragment": capture_spec["inner_frame_fragment"],
                "expected_top_rows": capture_spec.get("expected_top_rows", 10),
                "post_filter_wait_ms": capture_spec.get("post_filter_wait_ms", 180000),
                "post_filter_poll_interval_ms": capture_spec.get("post_filter_poll_interval_ms", 1000),
                "post_filter_fallback_wait_ms": capture_spec.get("post_filter_fallback_wait_ms", 0),
                "post_filter_stable_polls": capture_spec.get("post_filter_stable_polls", 2),
            }
        ),
    )
    capture_script = write_js(
        workdir,
        f"tmp-{capture_spec['capture_id']}-capture.js",
        build_export_image_js(
            capture_spec["inner_frame_fragment"],
            str(preferred_capture_path.resolve()),
            viewport=VIEWPORT,
        ),
    )

    try:
        max_attempts = 5
        last_error: str | None = None
        for attempt in range(1, max_attempts + 1):
            try:
                pre_configure_output = run_code(session, pre_configure_script, cwd=workdir, timeout=420).strip()
            except subprocess.TimeoutExpired as exc:
                last_error = f"TimeoutExpired: {exc}"
                pre_configure_output = f"### Error\nError: {last_error}"
            if "### Error" in pre_configure_output:
                last_error = pre_configure_output
                if "chrome-error://chromewebdata/" in pre_configure_output or "HTTP ERROR 502" in pre_configure_output:
                    raise RuntimeError(
                        f"Failed to configure Tableau view for {capture_spec['capture_id']}: {last_error}"
                    )
                if attempt < max_attempts:
                    reset_target_page()
                    time.sleep(1)
                    continue
                raise RuntimeError(
                    f"Failed to configure Tableau view for {capture_spec['capture_id']}: {last_error}"
                )
            pre_configure_result = parse_playwright_result(pre_configure_output)
            configure_result: dict[str, Any] = {
                "preApiConfigure": pre_configure_result,
                "quickFilterUi": [],
                "quickFilterState": {},
                "postApiConfigure": {},
            }

            try:
                ui_filter_output = run_code(session, ui_filter_script, cwd=workdir, timeout=180).strip()
            except subprocess.TimeoutExpired as exc:
                last_error = f"TimeoutExpired: {exc}"
                ui_filter_output = f"### Error\nError: {last_error}"
            if "### Error" in ui_filter_output:
                last_error = ui_filter_output
                if attempt < max_attempts:
                    reset_target_page()
                    time.sleep(1)
                    continue
                raise RuntimeError(
                    f"Failed to apply Tableau quick filters for {capture_spec['capture_id']}: {last_error}"
                )
            ui_filter_result = parse_playwright_result(ui_filter_output)
            configure_result["quickFilterUi"] = ui_filter_result.get("logs", [])
            configure_result["quickFilterState"] = ui_filter_result.get("filters", {})
            failed_quick_filters = [item for item in configure_result["quickFilterUi"] if not item.get("ok")]
            if failed_quick_filters:
                last_error = json.dumps(
                    {
                        "configure": configure_result,
                        "uiFilterResult": ui_filter_result,
                    },
                    ensure_ascii=False,
                )
                if attempt < max_attempts:
                    reset_target_page()
                    time.sleep(1)
                    continue
                raise RuntimeError(
                    f"Failed to apply Tableau quick filters for {capture_spec['capture_id']}: {last_error}"
                )

            try:
                post_configure_output = run_code(session, post_configure_script, cwd=workdir, timeout=420).strip()
            except subprocess.TimeoutExpired as exc:
                last_error = f"TimeoutExpired: {exc}"
                post_configure_output = f"### Error\nError: {last_error}"
            if "### Error" in post_configure_output:
                last_error = post_configure_output
                if "chrome-error://chromewebdata/" in post_configure_output or "HTTP ERROR 502" in post_configure_output:
                    raise RuntimeError(
                        f"Failed to configure Tableau view for {capture_spec['capture_id']}: {last_error}"
                    )
                if attempt < max_attempts:
                    reset_target_page()
                    time.sleep(1)
                    continue
                raise RuntimeError(
                    f"Failed to configure Tableau view for {capture_spec['capture_id']}: {last_error}"
                )
            configure_result["postApiConfigure"] = parse_playwright_result(post_configure_output)

            try:
                wait_output = run_code(session, wait_script, cwd=workdir, timeout=240).strip()
            except subprocess.TimeoutExpired as exc:
                last_error = f"TimeoutExpired: {exc}"
                wait_output = f"### Error\nError: {last_error}"
            if "### Error" in wait_output:
                last_error = wait_output
                if attempt < max_attempts:
                    reset_target_page()
                    time.sleep(1)
                    continue
                raise RuntimeError(
                    f"Failed while waiting for Tableau data for {capture_spec['capture_id']}: {last_error}"
                )

            wait_payload = parse_playwright_result(wait_output)
            wait_result = wait_payload.get("waitResult", {})
            configure_result["waitResult"] = wait_result
            if not wait_result.get("ready"):
                last_error = json.dumps(configure_result, ensure_ascii=False)
                raise RuntimeError(
                    f"Failed to configure Tableau view for {capture_spec['capture_id']}: {last_error}"
                )
            capture_output = run_code(session, capture_script, cwd=workdir, timeout=240).strip()
            if "### Error" in capture_output:
                last_error = capture_output
                if "chrome-error://chromewebdata/" in capture_output or "HTTP ERROR 502" in capture_output:
                    raise RuntimeError(
                        f"Failed to capture Tableau view for {capture_spec['capture_id']}: {last_error}"
                    )
                if attempt < max_attempts:
                    reset_target_page()
                    time.sleep(1)
                    continue
                raise RuntimeError(
                    f"Failed to capture Tableau view for {capture_spec['capture_id']}: {last_error}"
                )
            break
    finally:
        pre_configure_script.unlink(missing_ok=True)
        ui_filter_script.unlink(missing_ok=True)
        post_configure_script.unlink(missing_ok=True)
        wait_script.unlink(missing_ok=True)
        capture_script.unlink(missing_ok=True)

    captured_path = locate_session_file(
        workspace_root,
        session,
        capture_spec["raw_screenshot_name"],
        workdir=workdir,
    )
    normalize_image_size(captured_path, {"width": 1400, "height": 3760})
    return captured_path


def build_dump_top_products_js(
    limit: int,
    *,
    activate_sheet: str = TOP_PRODUCTS_DASHBOARD,
    inner_frame_fragment: str = "/views/CKTopProducts-General_16862068169500/TopProducts?",
) -> str:
    """Helper."""
    payload = json.dumps(
        {
            "limit": limit,
            "activate_sheet": activate_sheet,
            "inner_frame_fragment": inner_frame_fragment,
        },
        ensure_ascii=False,
    )
    return f"""async function(page) {{
  const spec = {payload};
  await page.waitForLoadState('domcontentloaded').catch(() => null);

  async function inspectFrame(frame, preferredFragment) {{
    const frameUrl = typeof frame.url === 'function' ? frame.url() : '';
    try {{
      const snapshot = await frame.evaluate(() => {{
        const vizCount = window.tableau?.VizManager?.getVizs?.().length || 0;
        return {{
          url: location.href,
          title: document.title,
          readyState: document.readyState,
          hasTableau: !!(window.tableau && window.tableau.VizManager),
          vizCount,
        }};
      }});
      return {{
        ...snapshot,
        frameUrl,
        matchesTargetFrame: !!(preferredFragment && frameUrl && frameUrl.includes(preferredFragment)),
      }};
    }} catch (error) {{
      return {{
        url: frameUrl,
        title: null,
        readyState: null,
        hasTableau: false,
        vizCount: 0,
        frameUrl,
        matchesTargetFrame: !!(preferredFragment && frameUrl && frameUrl.includes(preferredFragment)),
        error: String(error && error.message || error),
      }};
    }}
  }}

  async function locateVizFrame(timeoutMs, preferredFragment) {{
    const deadline = Date.now() + timeoutMs;
    let lastSnapshots = [];
    while (Date.now() < deadline) {{
      const rankedFrames = page.frames()
        .map((frame) => {{
          const frameUrl = typeof frame.url === 'function' ? frame.url() : '';
          const isPreferred = !!(preferredFragment && frameUrl.includes(preferredFragment));
          return {{
            frame,
            rank: isPreferred ? 0 : (frame === page.mainFrame() ? 1 : 2),
          }};
        }})
        .sort((left, right) => left.rank - right.rank);

      lastSnapshots = [];
      for (const item of rankedFrames) {{
        const snapshot = await inspectFrame(item.frame, preferredFragment);
        lastSnapshots.push(snapshot);
        if (snapshot.vizCount > 0) {{
          return {{ frame: item.frame, snapshots: lastSnapshots }};
        }}
      }}

      await page.waitForTimeout(1000);
    }}

    throw new Error('No Tableau viz frame found for top-products dump: ' + JSON.stringify({{
      pageUrl: page.url(),
      frames: lastSnapshots,
    }}));
  }}

  const located = await locateVizFrame(30000, spec.inner_frame_fragment || '');
  const targetFrame = located.frame;

  return await targetFrame.evaluate(async (config) => {{
    const viz = window.tableau?.VizManager?.getVizs?.()?.[0];
    if (!viz) {{
      throw new Error('No Tableau viz found in target frame');
    }}
    const workbook = viz.getWorkbook();
    try {{
      await workbook.activateSheetAsync(config.activate_sheet);
    }} catch (error) {{
    }}
    const activeSheet = workbook.getActiveSheet();
    const worksheetList = activeSheet && typeof activeSheet.getWorksheets === 'function'
      ? activeSheet.getWorksheets()
      : [];
    const worksheet = worksheetList.find((item) => item.getName() === 'TOP')
      || (
        activeSheet &&
        typeof activeSheet.getName === 'function' &&
        activeSheet.getName() === 'TOP'
          ? activeSheet
          : null
      );
    if (!worksheet) {{
      throw new Error('TOP worksheet not found');
    }}
    const table = await worksheet.getSummaryDataAsync({{ maxRows: config.limit }});
    const columns = table.getColumns().map((col) => col.getFieldName());
    const rows = table.getData().map((row) =>
      Object.fromEntries(row.map((cell, idx) => [
        columns[idx],
        {{ value: cell.value, formattedValue: cell.formattedValue }},
      ]))
    );
    return {{
      pageUrl: location.href,
      activeSheet: activeSheet && activeSheet.getName ? activeSheet.getName() : null,
      frameUrl: location.href,
      columns,
      rows,
    }};
  }}, spec);
}}
"""


def extract_top_products_from_current_view(session: str, *, workdir: Path, limit: int) -> dict[str, Any]:
    """Helper."""
    script_path = write_js(
        workdir,
        "tmp-top-products-dump.js",
        build_dump_top_products_js(limit, activate_sheet=TOP_PRODUCTS_DASHBOARD),
    )
    try:
        result = parse_playwright_result(run_code(session, script_path, cwd=workdir, timeout=240))
    finally:
        script_path.unlink(missing_ok=True)

    columns = [str(column or "").strip() for column in result.get("columns", [])]

    def resolve_column(exact_candidates: list[str], token_candidates: list[str]) -> str | None:
        lowered = {column.lower(): column for column in columns}
        for candidate in exact_candidates:
            resolved = lowered.get(candidate.lower())
            if resolved:
                return resolved
        for column in columns:
            normalized = column.lower()
            if any(token in normalized for token in token_candidates):
                return column
        return None

    article_column = resolve_column(["Articlecode"], ["articlecode"])
    qty_column = resolve_column(["鎬诲拰(Soldqty)", "Sum(Soldqty)", "SUM(Soldqty)"], ["soldqty"])
    image_column = resolve_column(["鑱氬悎(max url)", "AGG(max url)", "AGG(Max Url)", "max url"], ["url"])
    rank_column = resolve_column(["鑱氬悎(No.)", "AGG(No.)", "No."], ["no."])

    normalized_rows: list[dict[str, Any]] = []
    for raw_row in result.get("rows", []):
        article_cell = raw_row.get(article_column or "", {}) if article_column else {}
        qty_cell = raw_row.get(qty_column or "", {}) if qty_column else {}
        image_cell = raw_row.get(image_column or "", {}) if image_column else {}
        rank_cell = raw_row.get(rank_column or "", {}) if rank_column else {}

        articlecode = str(article_cell.get("formattedValue") or article_cell.get("value") or "").strip()
        qty_value = qty_cell.get("value")
        image_url = str(image_cell.get("formattedValue") or image_cell.get("value") or "").strip()
        rank_value = rank_cell.get("value")
        if not articlecode:
            continue
        try:
            qty = int(float(qty_value or 0))
        except (TypeError, ValueError):
            qty = 0
        try:
            rank = int(float(rank_value or 0))
        except (TypeError, ValueError):
            rank = 0
        normalized_rows.append(
            {
                "articlecode": articlecode,
                "qty": qty,
                "sales": 0.0,
                "image_url": image_url,
                "rank": rank,
            }
        )

    normalized_rows.sort(key=lambda row: (row["rank"] or 9999, -row["qty"], row["articlecode"]))
    if len(normalized_rows) < limit:
        raise RuntimeError(
            "Tableau top products rows are insufficient: "
            f"expected {limit}, got {len(normalized_rows)}, columns={columns}"
        )

    return {
        "source": "tableau",
        "page_url": result.get("pageUrl"),
        "active_sheet": result.get("activeSheet"),
        "columns": columns,
        "rows": normalized_rows[:limit],
    }


def month_date_range(report_year: int, report_month: str) -> tuple[str, str]:
    """Helper."""
    normalized_month = normalize_month_label(report_month)
    month_index = resolve_month_index(normalized_month)
    last_day = calendar.monthrange(report_year, month_index)[1]
    return f"{report_year}-{month_index:02d}-01", f"{report_year}-{month_index:02d}-{last_day:02d}"


def resolve_shop(config: dict[str, Any], store_name: str) -> tuple[int, str, dict[str, Any], str]:
    """Helper."""
    conn, mysql_cfg, mysql_label = open_mysql_connection(config)
    try:
        with conn.cursor() as cursor:
            cursor.execute(
                """
                SELECT shopid, shopname
                FROM oms_daily_report
                WHERE LOWER(shopname) = LOWER(%s)
                LIMIT 1
                """,
                (store_name,),
            )
            row = cursor.fetchone()
    finally:
        conn.close()

    if not row or row.get("shopid") is None:
        raise RuntimeError(f"Unable to resolve shop for storename={store_name!r}")
    return int(row["shopid"]), str(row["shopname"]), mysql_cfg, mysql_label


def fetch_top_products(
    config: dict[str, Any],
    *,
    store_name: str,
    report_month: str,
    report_year: int,
    level1: str,
    limit: int,
) -> dict[str, Any]:
    """Helper."""
    shop_id, resolved_shop_name, mysql_cfg, mysql_label = resolve_shop(config, store_name)
    if mysql_label != "config.mysql":
        log_progress(
            "sql",
            f"use mysql fallback {mysql_label} -> {mysql_cfg['host']}:{mysql_cfg['port']}/{mysql_cfg['database']}",
        )
    start_date, end_date = month_date_range(report_year, report_month)
    conn = connect_mysql(mysql_cfg)
    try:
        with conn.cursor() as cursor:
            cursor.execute(
                """
                SELECT
                  articlecode,
                  SUM(CASE WHEN qty > 0 THEN qty ELSE 0 END) AS qty,
                  SUM(CASE WHEN sales > 0 THEN sales ELSE 0 END) AS sales,
                  SUM(CASE WHEN soh > 0 THEN soh ELSE 0 END) AS soh
                FROM oms_daily_report_detail
                WHERE shopid = %s
                  AND daily BETWEEN %s AND %s
                  AND level1 = %s
                GROUP BY articlecode
                HAVING qty > 0 OR sales > 0
                ORDER BY qty DESC, sales DESC, articlecode ASC
                LIMIT %s
                """,
                (shop_id, start_date, end_date, level1, limit),
            )
            rows = cursor.fetchall()

            cursor.execute(
                """
                SELECT
                  SUM(CASE WHEN qty > 0 THEN qty ELSE 0 END) AS total_qty,
                  SUM(CASE WHEN sales > 0 THEN sales ELSE 0 END) AS total_sales,
                  SUM(CASE WHEN soh > 0 THEN soh ELSE 0 END) AS total_soh
                FROM oms_daily_report_detail
                WHERE shopid = %s
                  AND daily BETWEEN %s AND %s
                  AND level1 = %s
                """,
                (shop_id, start_date, end_date, level1),
            )
            totals = cursor.fetchone()
    finally:
        conn.close()

    normalized_rows = [
        {
            "articlecode": str(row["articlecode"]),
            "qty": int(row["qty"] or 0),
            "sales": float(row["sales"] or 0),
            "soh": float(row.get("soh") or 0),
        }
        for row in rows
        if row.get("articlecode")
    ]
    if len(normalized_rows) < limit:
        raise RuntimeError(f"{level1} top products rows are insufficient: expected {limit}, got {len(normalized_rows)}")

    return {
        "source": "sql",
        "store_name": store_name,
        "resolved_shop_name": resolved_shop_name,
        "shop_id": shop_id,
        "mysql_target": mysql_label,
        "mysql_host": mysql_cfg["host"],
        "mysql_port": int(mysql_cfg["port"]),
        "mysql_database": mysql_cfg["database"],
        "start_date": start_date,
        "end_date": end_date,
        "level1": level1,
        "total_qty": int((totals or {}).get("total_qty") or 0),
        "total_sales": float((totals or {}).get("total_sales") or 0.0),
        "total_soh": float((totals or {}).get("total_soh") or 0.0),
        "rows": normalized_rows[:limit],
    }


def style_chart_axis(axis: Any) -> None:
    """Helper."""
    axis.set_xticks([])
    axis.set_yticks([])
    axis.tick_params(left=False, bottom=False, labelleft=False, labelbottom=False)
    for spine in axis.spines.values():
        spine.set_visible(False)
    axis.set_facecolor("white")


def format_chart_number(value: float) -> str:
    """Helper."""
    return f"{int(round(value)):,}"


def render_sql_top_products_chart(source_payload: dict[str, Any], target: Path) -> None:
    """Helper."""
    rows = source_payload.get("rows", [])
    total_qty = int(source_payload.get("total_qty") or 0)
    total_sales = float(source_payload.get("total_sales") or 0.0)
    total_soh = float(source_payload.get("total_soh") or 0.0)
    if not rows or total_qty <= 0 or total_soh <= 0:
        raise RuntimeError("SQL top-products chart source is incomplete.")

    qty_values = [int(row["qty"]) for row in rows]
    sales_values = [float(row["sales"]) for row in rows]
    soh_values = [float(row.get("soh") or 0.0) for row in rows]
    qty_percentages = [value / total_qty * 100 for value in qty_values]
    sales_percentages = [value / total_sales * 100 for value in sales_values]
    top_qty_percentage = sum(qty_values) / total_qty * 100
    top_sales_percentage = sum(sales_values) / total_sales * 100

    figure = plt.figure(figsize=(10.34, 5.87), dpi=100, facecolor="white")
    grid = GridSpec(1, 4, figure=figure, width_ratios=[4.9, 1.8, 1.8, 2.2], wspace=0.08)
    ax_qty = figure.add_subplot(grid[0, 0])
    ax_qty_pct = figure.add_subplot(grid[0, 1], sharey=ax_qty)
    ax_sales_pct = figure.add_subplot(grid[0, 2], sharey=ax_qty)
    ax_sales = figure.add_subplot(grid[0, 3], sharey=ax_qty)
    y_positions = list(range(len(rows)))

    figure.subplots_adjust(left=0.03, right=0.99, top=0.83, bottom=0.05)
    figure.text(0.02, 0.96, "Top Parameter", ha="left", va="bottom", fontsize=10)
    figure.patches.append(
        Rectangle((0.02, 0.915), 0.31, 0.034, transform=figure.transFigure, fill=False, linewidth=0.8, edgecolor="#666666")
    )
    figure.text(0.024, 0.918, str(len(rows)), ha="left", va="bottom", fontsize=10)

    max_qty = max(qty_values) or 1
    qty_left_margin = max_qty * 0.72
    ax_qty.set_xlim(-qty_left_margin, max_qty * 1.28)
    ax_qty.barh(y_positions, qty_values, height=0.75, color=QTY_BAR_COLORS[: len(rows)])
    ax_qty.invert_yaxis()
    ax_qty.axvline(0, color="#d6d6d6", linewidth=0.8)
    ax_qty.text(-qty_left_margin * 0.88, -0.72, "No.", fontsize=10, fontweight="bold", ha="left", va="bottom")
    ax_qty.text(-qty_left_margin * 0.47, -0.72, "Articlecode", fontsize=10, fontweight="bold", ha="left", va="bottom")
    for index, row in enumerate(rows):
        ax_qty.text(-qty_left_margin * 0.88, index, str(index + 1), fontsize=10, ha="left", va="center")
        ax_qty.text(-qty_left_margin * 0.64, index, row["articlecode"], fontsize=10, ha="left", va="center")
        ax_qty.text(
            qty_values[index] + max_qty * 0.02,
            index,
            format_chart_number(qty_values[index]),
            fontsize=9,
            fontweight="bold",
            ha="left",
            va="center",
        )
    style_chart_axis(ax_qty)

    pct_max = max(max(qty_percentages), max(sales_percentages), 1.0)
    pct_limit = max(10.0, pct_max + 1.5)

    ax_qty_pct.set_xlim(-pct_limit * 1.18, 0)
    ax_qty_pct.barh(y_positions, [-value for value in qty_percentages], height=0.75, color="#a7a7a7")
    ax_qty_pct.axvline(0, color="#d6d6d6", linewidth=0.8)
    ax_qty_pct.set_title(f"Soldqty%\n{round(top_qty_percentage):.0f}%", fontsize=12, fontweight="bold", pad=24)
    for index, value in enumerate(qty_percentages):
        ax_qty_pct.text(-value - pct_limit * 0.05, index, f"{round(value):.0f}%", fontsize=9, fontweight="bold", ha="right", va="center")
    style_chart_axis(ax_qty_pct)

    ax_sales_pct.set_xlim(0, pct_limit * 1.18)
    ax_sales_pct.barh(y_positions, sales_percentages, height=0.75, color="#8db9db")
    ax_sales_pct.axvline(0, color="#d6d6d6", linewidth=0.8)
    ax_sales_pct.set_title(f"Sales%\n{round(top_sales_percentage):.0f}%", fontsize=12, fontweight="bold", pad=24)
    for index, value in enumerate(sales_percentages):
        ax_sales_pct.text(value + pct_limit * 0.05, index, f"{round(value):.0f}%", fontsize=9, fontweight="bold", ha="left", va="center")
    style_chart_axis(ax_sales_pct)

    sales_title_x = (ax_qty_pct.get_position().x0 + ax_sales_pct.get_position().x1) / 2
    figure.text(sales_title_x, 0.965, "TOP N%", ha="center", va="bottom", fontsize=13, fontweight="bold")

    max_soh = max(soh_values) or 1.0
    ax_sales.set_xlim(0, max_soh * 1.22)
    ax_sales.barh(y_positions, soh_values, height=0.75, color="#a7a7a7")
    ax_sales.axvline(0, color="#d6d6d6", linewidth=0.8)
    for index, value in enumerate(soh_values):
        ax_sales.text(
            value + max_soh * 0.02,
            index,
            format_chart_number(value),
            fontsize=9,
            fontweight="bold",
            ha="left",
            va="center",
        )
    ax_sales.text(0, len(rows) - 0.08, "soh", fontsize=9, ha="left", va="bottom")
    style_chart_axis(ax_sales)

    target.parent.mkdir(parents=True, exist_ok=True)
    figure.savefig(target, dpi=100, facecolor="white")
    plt.close(figure)


def download_image(url: str, *, attempts: int = 4, timeout: int = 60) -> Image.Image:
    """Helper."""
    last_error: Exception | None = None
    for attempt in range(1, attempts + 1):
        try:
            request = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"})
            with urllib.request.urlopen(request, timeout=timeout) as response:
                payload = response.read()
            image = Image.open(BytesIO(payload))
            return ImageOps.exif_transpose(image).convert("RGBA")
        except Exception as exc:  # pragma: no cover - network retry path
            last_error = exc
            if attempt < attempts:
                time.sleep(min(12, attempt * 3))
                continue
    raise RuntimeError(f"Failed to download product image from {url}") from last_error


def render_product_image(source: Image.Image, target: Path, *, canvas_width: int, canvas_height: int) -> None:
    """Helper."""
    scale = min(canvas_width / source.width, canvas_height / source.height)
    resized_width = max(1, int(round(source.width * scale)))
    resized_height = max(1, int(round(source.height * scale)))
    resized = source.resize((resized_width, resized_height), Image.Resampling.LANCZOS)

    canvas = Image.new("RGBA", (canvas_width, canvas_height), (255, 255, 255, 255))
    offset = ((canvas_width - resized_width) // 2, (canvas_height - resized_height) // 2)
    canvas.alpha_composite(resized, dest=offset)
    target.parent.mkdir(parents=True, exist_ok=True)
    canvas.convert("RGB").save(target)


def sanitize_file_token(value: str) -> str:
    """Helper."""
    return re.sub(r"[^A-Za-z0-9_-]+", "_", value).strip("_") or "item"


def build_product_assets(
    slide_code: str,
    *,
    asset_dir: Path,
    source_payload: dict[str, Any],
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """Helper."""
    slide_spec = SLIDE_SPECS[slide_code]
    operations: list[dict[str, Any]] = []
    manifest_items: list[dict[str, Any]] = []

    for slot, row in zip(slide_spec["product_slots"], source_payload["rows"]):
        articlecode = row["articlecode"]
        image_url = row.get("image_url") or PRODUCT_IMAGE_URL.format(articlecode=articlecode)
        asset_name = f"{slide_code.lower()}_top_product_{slot['rank']:02d}_{sanitize_file_token(articlecode)}"
        asset_path = asset_dir / f"{asset_name}.png"
        render_product_image(
            download_image(image_url),
            asset_path,
            canvas_width=slot["canvas_width"],
            canvas_height=slot["canvas_height"],
        )
        operations.append(
            {
                "slide": slide_spec["slide"],
                "shape_id": slot["shape_id"],
                "shape_name": slot["shape_name"],
                "image_path": str(asset_path),
            }
        )
        manifest_items.append(
            {
                "slide_code": slide_code,
                "slide": slide_spec["slide"],
                "shape_id": slot["shape_id"],
                "shape_name": slot["shape_name"],
                "asset_name": asset_name,
                "asset_path": str(asset_path),
                "source_capture_id": f"url-top-products-{slide_code.lower()}",
                "source_view": "TopProducts" if source_payload.get("source") == "tableau" else "oms_daily_report_detail",
                "note": f"{slide_code} product image for Top {slot['rank']}.",
                "top_rank": slot["rank"],
                "articlecode": articlecode,
                "qty": row["qty"],
                "sales": row["sales"],
                "source": source_payload.get("source", "sql"),
                "image_url": image_url,
                "canvas_size": {"width": slot["canvas_width"], "height": slot["canvas_height"]},
            }
        )

    return operations, manifest_items


def main() -> None:
    """Helper."""
    script_started_at = time.monotonic()
    args = parse_args()
    config_path = Path(args.config).resolve()
    config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    report_month, report_year, compare_year = resolve_report_period(args, config)

    requested = [item.strip().upper() for item in args.slides.split(",") if item.strip()]
    requested = [item for item in requested if item in SLIDE_SPECS]
    if not requested:
        raise SystemExit("No supported slides requested. Use S09 and/or S10.")

    capture_specs = {slide_code: build_capture_spec(slide_code, report_month, report_year) for slide_code in requested}
    log_progress("stage", f"start top-products slides={','.join(requested)}")

    vip_workdir = Path(args.output_dir).resolve() if args.output_dir else resolve_configured_path(config_path, config["paths"]["workdir"])
    workspace_root = vip_workdir if args.output_dir else config_path.parent
    asset_dir = vip_workdir / "assets" / "top-products"
    data_dir = vip_workdir / "data" / "top-products"
    asset_dir.mkdir(parents=True, exist_ok=True)
    data_dir.mkdir(parents=True, exist_ok=True)

    session = args.playwright_session.strip() if args.playwright_session.strip() else (
        SESSION_NAME if not args.output_dir else f"{SESSION_NAME}-{vip_workdir.name}"
    )
    state_path = (
        Path(args.tableau_state_path).expanduser().resolve()
        if args.tableau_state_path.strip()
        else (
            vip_workdir / ".playwright-cli" / f"{session}-state.json"
            if args.output_dir
            else workspace_root / "output" / "playwright" / session / "state.json"
        )
    )
    browser_config_path = (
        Path(args.browser_config_path).expanduser().resolve()
        if args.browser_config_path.strip()
        else None
    )
    ensure_browser_session(session, cwd=vip_workdir, browser_config_path=browser_config_path)
    load_state_if_present(session, state_path, cwd=vip_workdir)

    base_url = config["tableau"]["base_url"].rstrip("/")
    first_target_url = f"{base_url}{capture_specs[requested[0]]['hash_url']}"
    run_playwright(["--session", session, "goto", first_target_url], cwd=vip_workdir, timeout=120)

    login_script = write_js(
        vip_workdir,
        "tmp-top-products-login.js",
        build_login_js(config["tableau"]["username"], config["tableau"]["password"]),
    )
    try:
        login_output = run_code(session, login_script, cwd=vip_workdir, timeout=120).strip()
        if "### Error" in login_output:
            raise RuntimeError(f"Failed to run Tableau login script: {login_output}")
    finally:
        login_script.unlink(missing_ok=True)

    raw_screenshots: dict[str, Path] = {}
    tableau_top_sources: dict[str, dict[str, Any]] = {}
    tableau_capture_errors: dict[str, str] = {}
    for slide_code in requested:
        slide_spec = SLIDE_SPECS[slide_code]
        page_label = f"{slide_code} {slide_spec['capture_id']} ({capture_specs[slide_code]['activate_sheet']})"
        page_started_at = time.monotonic()
        log_progress("page", f"start {page_label}")
        try:
            raw_screenshots[slide_code] = capture_tableau_view(
                capture_specs[slide_code],
                base_url=base_url,
                session=session,
                workdir=vip_workdir,
                workspace_root=workspace_root,
            )
            tableau_top_sources[slide_code] = extract_top_products_from_current_view(
                session,
                workdir=vip_workdir,
                limit=len(SLIDE_SPECS[slide_code]["product_slots"]),
            )
            log_timing(f"page {page_label}", page_started_at)
        except Exception as exc:
            tableau_capture_errors[slide_code] = str(exc)
            log_progress("page", f"error {page_label}: {exc}")
    try:
        save_state(session, state_path, cwd=vip_workdir)
    except Exception as exc:
        log_progress("state", f"warn save-state failed: {exc}")

    operations = {"replace_text": [], "replace_images": []}
    manifest_items: list[dict[str, Any]] = []
    sql_sources: dict[str, Any] = {}

    for slide_code in requested:
        slide_spec = SLIDE_SPECS[slide_code]
        slide_started_at = time.monotonic()
        log_progress("asset", f"start {slide_code} chart/products")
        top_source = tableau_top_sources.get(slide_code)
        if top_source:
            sql_sources[slide_code] = top_source
            log_progress("asset", f"use Tableau source for {slide_code} products")
        else:
            top_source = fetch_top_products(
                config,
                store_name="ckc-vip",
                report_month=report_month,
                report_year=report_year,
                level1=slide_spec["level1"],
                limit=len(slide_spec["product_slots"]),
            )
            sql_sources[slide_code] = top_source

        chart_path = asset_dir / f"{slide_spec['asset_name']}.png"
        raw_screenshot = raw_screenshots.get(slide_code)
        if raw_screenshot:
            crop_image(raw_screenshot, chart_path, slide_spec["crop"], resize_to=slide_spec["resize_to"])
            chart_manifest_item = {
                "slide_code": slide_code,
                "slide": slide_spec["slide"],
                "shape_id": slide_spec["shape_id"],
                "shape_name": slide_spec["shape_name"],
                "asset_name": slide_spec["asset_name"],
                "asset_path": str(chart_path),
                "source_capture_id": slide_spec["capture_id"],
                "source_view": "TopProducts",
                "note": slide_spec["note"],
                "raw_screenshot": str(raw_screenshot),
                "crop": slide_spec["crop"],
                "resize_to": slide_spec["resize_to"],
            }
        else:
            render_sql_top_products_chart(top_source, chart_path)
            chart_manifest_item = {
                "slide_code": slide_code,
                "slide": slide_spec["slide"],
                "shape_id": slide_spec["shape_id"],
                "shape_name": slide_spec["shape_name"],
                "asset_name": slide_spec["asset_name"],
                "asset_path": str(chart_path),
                "source_capture_id": f"sql-rendered-{slide_code.lower()}",
                "source_view": "oms_daily_report_detail",
                "note": f"{slide_code} live SQL-rendered top-products chart.",
                "tableau_capture_error": tableau_capture_errors.get(slide_code),
                "rendered_from_sql": True,
            }

        operations["replace_images"].append(
            {
                "slide": slide_spec["slide"],
                "shape_id": slide_spec["shape_id"],
                "shape_name": slide_spec["shape_name"],
                "image_path": str(chart_path),
            }
        )
        manifest_items.append(chart_manifest_item)

        product_operations, product_manifest_items = build_product_assets(
            slide_code,
            asset_dir=asset_dir,
            source_payload=top_source,
        )
        operations["replace_images"].extend(product_operations)
        manifest_items.extend(product_manifest_items)
        log_timing(f"assets {slide_code}", slide_started_at)

    operations_path = vip_workdir / "render-ops.top-products.live.json"
    operations_path.write_text(json.dumps(operations, ensure_ascii=False, indent=2), encoding="utf-8")

    manifest_path = data_dir / "top-products-assets.live.json"
    manifest_payload = {
        "source": {
            "captures": [
                {
                    "capture_id": capture_specs[slide_code]["capture_id"],
                    "hash_url": capture_specs[slide_code]["hash_url"],
                    "activate_sheet": capture_specs[slide_code]["activate_sheet"],
                    "filter_actions": capture_specs[slide_code]["filter_actions"],
                    "raw_screenshot": str(raw_screenshots[slide_code]) if slide_code in raw_screenshots else None,
                    "note": capture_specs[slide_code]["note"],
                    "error": tableau_capture_errors.get(slide_code),
                }
                for slide_code in requested
            ],
            "url_product_sources": sql_sources,
            "config_path": str(config_path),
            "report_period": {
                "month": report_month,
                "year": report_year,
                "compare_year": compare_year,
            },
        },
        "assets": manifest_items,
        "operations_path": str(operations_path),
    }
    manifest_path.write_text(json.dumps(manifest_payload, ensure_ascii=False, indent=2), encoding="utf-8")

    log_timing(f"stage top-products {','.join(requested)}", script_started_at)
    print(json.dumps({"operations_path": str(operations_path), "manifest_path": str(manifest_path)}, ensure_ascii=False))


# 鍏抽敭琛屾敞閲婏細鑴氭湰鐩存帴杩愯鏃讹紝浠庤繖閲岃繘鍏ヤ富娴佺▼銆?
if __name__ == "__main__":
    main()
