from __future__ import annotations

import argparse
import calendar
import json
import os
import shutil
import subprocess
import time
from datetime import date, timedelta
from pathlib import Path
from typing import Any

from PIL import Image
import yaml
from tableau_export import build_export_image_js, normalize_image_size


NPX_EXECUTABLE = shutil.which("npx.cmd") or shutil.which("npx") or "npx"
PLAYWRIGHT_CMD = [NPX_EXECUTABLE, "--yes", "--package", "@playwright/cli", "playwright-cli"]
SESSION_NAME = "vip-report-warehouse-100060"
VIEWPORT = {"width": 1400, "height": 3400}
TEMPLATE_LOCK_DIR = Path(r"C:\workspace\cursor\output\vip-report\assets\template-lock-s04-s08-s13")
S13_HASH_URL = "#/views/WH100060SalesPerformance/60SalesSohbyDiscount?:iid=3"
S13_INNER_FRAME_FRAGMENT = "/views/WH100060SalesPerformance/60SalesSohbyDiscount?"

MONTH_LABELS = {
    1: "一月",
    2: "二月",
    3: "三月",
    4: "四月",
    5: "五月",
    6: "六月",
    7: "七月",
    8: "八月",
    9: "九月",
    10: "十月",
    11: "十一月",
    12: "十二月",
}

MOJIBAKE_MONTH_LABELS = {
    "涓€鏈?": 1,
    "涓€鏈": 1,
    "涓€鏈�": 1,
    "浜屾湀": 2,
    "涓夋湀": 3,
    "鍥涙湀": 4,
    "浜旀湀": 5,
    "鍏湀": 6,
    "涓冩湀": 7,
    "鍏湀": 8,
    "涔濇湀": 9,
    "鍗佹湀": 10,
    "鍗佷竴鏈?": 11,
    "鍗佷竴鏈": 11,
    "鍗佷簩鏈?": 12,
    "鍗佷簩鏈": 12,
}
MONTH_LABEL_TO_NUMBER = {label: number for number, label in MONTH_LABELS.items()}
MONTH_LABEL_TO_NUMBER.update(MOJIBAKE_MONTH_LABELS)


def should_use_template_lock(report_month_label: str, report_year: int, compare_year: int) -> bool:
    """Helper."""
    return False


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

    text = str(raw_month).strip()
    if text in MONTH_LABEL_TO_NUMBER:
        return MONTH_LABELS[MONTH_LABEL_TO_NUMBER[text]]
    if text.isdigit():
        value = int(text)
        if 1 <= value <= 12:
            return MONTH_LABELS[value]

    aliases = {
        "jan": 1,
        "january": 1,
        "feb": 2,
        "february": 2,
        "mar": 3,
        "march": 3,
        "apr": 4,
        "april": 4,
        "may": 5,
        "jun": 6,
        "june": 6,
        "jul": 7,
        "july": 7,
        "aug": 8,
        "august": 8,
        "sep": 9,
        "sept": 9,
        "september": 9,
        "oct": 10,
        "october": 10,
        "nov": 11,
        "november": 11,
        "dec": 12,
        "december": 12,
    }
    alias_value = aliases.get(text.lower())
    if alias_value:
        return MONTH_LABELS[alias_value]
    return MONTH_LABELS[1]


def month_label_to_number(label: str) -> int:
    """Helper."""
    return MONTH_LABEL_TO_NUMBER.get(label, 1)


def resolve_report_period(args: argparse.Namespace, config: dict[str, Any]) -> tuple[str, int, int, int]:
    """Helper."""
    report_cfg = config.get("report", {})
    raw_month = args.report_month or report_cfg.get("month_cn") or 1
    month_label = normalize_month_label(raw_month)
    month_number = month_label_to_number(month_label)
    report_year = int(args.report_year or report_cfg.get("year", 2026))
    compare_year = int(args.compare_year or report_cfg.get("compare_year", report_year - 1))
    return month_label, month_number, report_year, 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 build_capture_slide_codes(filtered_assets: list[dict[str, Any]]) -> dict[str, list[str]]:
    """Map each capture_id to the slide codes that depend on it."""
    capture_slide_codes: dict[str, set[str]] = {}
    for asset in filtered_assets:
        capture_slide_codes.setdefault(asset["capture_id"], set()).add(asset["slide_code"])
    return {capture_id: sorted(slide_codes) for capture_id, slide_codes in capture_slide_codes.items()}


def describe_capture(
    capture_id: str,
    capture_spec: dict[str, Any],
    capture_slide_codes: dict[str, list[str]],
) -> str:
    """Build a readable page label for logging."""
    slide_codes = "/".join(capture_slide_codes.get(capture_id, [])) or "-"
    return f"{slide_codes} {capture_id} ({capture_spec['activate_sheet']})"


def next_month_label(month_number: int) -> str:
    """Helper."""
    return MONTH_LABELS[1 if month_number == 12 else month_number + 1]


def sales_day_labels(report_year: int, report_month_number: int) -> list[str]:
    """Helper."""
    last_day = calendar.monthrange(report_year, report_month_number)[1]
    start = date(report_year, report_month_number, 1)
    return [
        format_cn_date(start + timedelta(days=offset))
        for offset in range(last_day)
    ]


def format_cn_date(value: date) -> str:
    """Helper."""
    return f"{value.year}年{value.month}月{value.day}日"


def build_filter_config(
    report_month_label: str,
    report_month_number: int,
    report_year: int,
    *,
    top_category: str | None = None,
    top_store: str | None = None,
) -> list[dict[str, Any]]:
    """Helper."""
    filters: list[dict[str, Any]] = [
        {"caption": "Year", "action": "replace", "values": [str(report_year)]},
        {
            "caption": "Month",
            "action": "replace",
            "values": [report_month_label, next_month_label(report_month_number)],
        },
        {"caption": "Week", "action": "all"},
        {
            "caption": "sales day",
            "action": "replace",
            "values": sales_day_labels(report_year, report_month_number),
        },
        {"caption": "Outbound Tag", "action": "all"},
        {"caption": "store", "action": "replace", "values": ["CKC-VIP"]},
    ]
    if top_category:
        filters.append({"caption": "Top Category", "action": "replace", "values": [top_category]})
    if top_store:
        filters.append({"caption": "Top Store", "action": "replace", "values": [top_store]})
    return filters


def build_specs(
    report_month_label: str,
    report_month_number: int,
    report_year: int,
    compare_year: int,
) -> dict[str, Any]:
    """Helper."""
    top_filters = build_filter_config(report_month_label, report_month_number, report_year)
    bags_filters = build_filter_config(
        report_month_label,
        report_month_number,
        report_year,
        top_category="BAGS",
        top_store="CKC-VIP",
    )
    shoes_filters = build_filter_config(
        report_month_label,
        report_month_number,
        report_year,
        top_category="SHOES",
        top_store="CKC-VIP",
    )
    return {
        "captures": [
            {
                "capture_id": "warehouse_discount_top",
                "session": SESSION_NAME,
                "hash_url": S13_HASH_URL,
                "inner_frame_fragment": S13_INNER_FRAME_FRAGMENT,
                "activate_sheet": "60 Sales& Soh by Discount",
                "filter_config": top_filters,
                "raw_screenshot_name": "warehouse-100060-top.png",
                "note": "S13 第一行 Performance by Discount。",
            },
            {
                "capture_id": "warehouse_top10_bags",
                "session": SESSION_NAME,
                "hash_url": S13_HASH_URL,
                "inner_frame_fragment": S13_INNER_FRAME_FRAGMENT,
                "activate_sheet": "60 Sales& Soh by Discount",
                "filter_config": bags_filters,
                "raw_screenshot_name": "warehouse-100060-top10-bags.png",
                "note": "S13 第二行左侧，Top Category=BAGS，Top Store=CKC-VIP。",
            },
            {
                "capture_id": "warehouse_top10_shoes",
                "session": SESSION_NAME,
                "hash_url": S13_HASH_URL,
                "inner_frame_fragment": S13_INNER_FRAME_FRAGMENT,
                "activate_sheet": "60 Sales& Soh by Discount",
                "filter_config": shoes_filters,
                "raw_screenshot_name": "warehouse-100060-top10-shoes.png",
                "note": "S13 第二行右侧，Top Category=SHOES，Top Store=CKC-VIP。",
            },
        ],
        "assets": [
            {
                "slide_code": "S13",
                "slide": 13,
                "shape_name": "图片 4",
                "shape_id": 5,
                "asset_name": "s13_top",
                "capture_id": "warehouse_discount_top",
                "crop": {"left": 10, "top": 150, "width": 840, "height": 1030},
                "source_view": "Performance by Discount",
                "note": "S13 第一行 Performance by Discount__年份-全部。",
            },
            {
                "slide_code": "S13",
                "slide": 13,
                "shape_name": "图片 5",
                "shape_id": 6,
                "asset_name": "s13_left",
                "capture_id": "warehouse_top10_bags",
                "crop": {"left": 860, "top": 1640, "width": 530, "height": 1670},
                "source_view": "TOP10 by Category",
                "note": "S13 第二行左侧，TOP10 by Category___年份-全部（BAGS）。",
            },
            {
                "slide_code": "S13",
                "slide": 13,
                "shape_name": "图片 9",
                "shape_id": 10,
                "asset_name": "s13_mid",
                "capture_id": "warehouse_top10_shoes",
                "crop": {"left": 860, "top": 1640, "width": 530, "height": 1670},
                "source_view": "TOP10 by Category",
                "note": "S13 第二行右侧，TOP10 by Category___年份-全部（SHOES）。",
            },
        ],
        "report_period": {
            "month_label": report_month_label,
            "month_number": report_month_number,
            "year": report_year,
            "compare_year": compare_year,
        },
    }


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)
    creationflags = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) if os.name == "nt" else 0
    process = subprocess.Popen(
        args,
        cwd=str(cwd),
        env=env,
        text=True,
        encoding="utf-8",
        errors="replace",
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        creationflags=creationflags,
    )
    try:
        stdout, stderr = process.communicate(timeout=timeout)
    except subprocess.TimeoutExpired as exc:
        if process.poll() is None:
            if os.name == "nt":
                subprocess.run(
                    ["taskkill", "/F", "/T", "/PID", str(process.pid)],
                    stdout=subprocess.DEVNULL,
                    stderr=subprocess.DEVNULL,
                    check=False,
                )
            else:
                process.kill()
        stdout, stderr = process.communicate()
        exc.stdout = stdout
        exc.stderr = stderr
        raise

    result = subprocess.CompletedProcess(args, process.returncode, stdout, stderr)
    if check and process.returncode != 0:
        raise subprocess.CalledProcessError(
            process.returncode,
            args,
            output=stdout,
            stderr=stderr,
        )
    return result


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.waitForTimeout(3000);
  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};

  function getVisibleFrame() {{
    return page.frames().find(
      (candidate) => candidate !== page.mainFrame() && candidate.url().includes(spec.inner_frame_fragment)
    );
  }}

  const deadline = Date.now() + 120000;
  let frame = null;
  while (Date.now() < deadline) {{
    frame = getVisibleFrame();
    if (frame) {{
      break;
    }}
    await page.waitForTimeout(500);
  }}
  if (!frame) {{
    throw new Error(`Frame not found for ${{spec.capture_id}}`);
  }}

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

    const waitFor = async (predicate, timeoutMs, errorMessage) => {{
      const startedAt = Date.now();
      while (Date.now() - startedAt < timeoutMs) {{
        const value = predicate();
        if (value) {{
          return value;
        }}
        await sleep(250);
      }}
      throw new Error(errorMessage);
    }};

    const removeErrorDialog = () => {{
      document.querySelector('[data-tb-test-id="detailedErrorDialog-Dialog-Glass-Root"]')?.remove();
      document.querySelector('[data-tb-test-id="detailedErrorDialog-Dialog-Glass"]')?.remove();
      const closeButton = document.querySelector('[data-tb-test-id="detailedErrorDialog-Dialog-CloseButton"]');
      if (closeButton) {{
        try {{
          closeButton.click();
        }} catch (error) {{
        }}
      }}
    }};

    const getWidgetMap = () => {{
      const panels = Array.from(document.querySelectorAll('.QuickFilterPanel'));
      return new Map(
        panels
          .map((panel) => panel.querySelector('.CategoricalFilter')?.id)
          .filter(Boolean)
          .map((id) => {{
            const widget = window.dijit?.byId?.(id);
            return widget ? [widget.get_oFilter().caption, widget] : null;
          }})
          .filter(Boolean)
      );
    }};

    const applyAll = (widget) => {{
      window.tab.FilterClientCommands.setCategoricalQuickFilterToAll(
        window.tab.CancelDialog.getCustomWaitHandlerForReturnFocus(),
        widget.get_session().get_visualId(),
        widget.get_identifier()
      );
    }};

    const applyByLabels = (widget, values) => {{
      const tuples = widget.get_oFilter().table?.tuples || [];
      const desired = new Set(
        (values || []).map((value) => String(value ?? '').trim().toLowerCase())
      );
      const indices = tuples
        .map((tuple, index) => {{
          const label = tuple.d ?? tuple.t?.map((item) => item.v).join('|') ?? '';
          const normalizedLabel = String(label ?? '').trim().toLowerCase();
          return desired.has(normalizedLabel) ? index : -1;
        }})
        .filter((index) => index >= 0);
      if (!indices.length) {{
        throw new Error(`No matching values for ${{widget.get_oFilter().caption}}: ${{values.join(', ')}}`);
      }}
      window.tab.FilterClientCommands.setCategoricalFilterValuesByIndex(
        window.tab.CancelDialog.getCustomWaitHandlerForReturnFocus(),
        widget.get_session().get_visualId(),
        widget.get_identifier(),
        indices
      );
    }};

    await waitFor(
      () => window.dijit && document.querySelectorAll('.QuickFilterPanel').length >= 13,
      120000,
      'Quick filter widgets not ready'
    );

    removeErrorDialog();

    const widgetMap = getWidgetMap();
    const applied = [];
    for (const filter of config.filter_config) {{
      const widget = widgetMap.get(filter.caption);
      if (!widget) {{
        throw new Error(`Quick filter not found: ${{filter.caption}}`);
      }}

      if (filter.action === 'all') {{
        applyAll(widget);
      }} else if (filter.action === 'replace') {{
        applyByLabels(widget, filter.values || []);
      }} else {{
        throw new Error(`Unsupported filter action: ${{filter.action}}`);
      }}

      applied.push({{
        caption: filter.caption,
        action: filter.action,
        values: filter.values || [],
      }});
      removeErrorDialog();
      await sleep(2500);
    }}

    await sleep(6000);
    removeErrorDialog();

    return {{
      url: location.href,
      applied,
      summaries: Array.from(getWidgetMap().entries()).map(([caption, widget]) => ({{
        caption,
        summary: widget.get_oFilter().summary,
      }})),
    }};
  }}, spec);
}}
"""


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.waitForTimeout(3000);
  const frame = page.frames().find(
    (candidate) => candidate !== page.mainFrame() && candidate.url().includes(spec.inner_frame_fragment)
  );
  const target = frame || page;
  if (frame) {{
    await frame.evaluate(() => {{
      document.querySelector('[data-tb-test-id="detailedErrorDialog-Dialog-Glass-Root"]')?.remove();
      document.querySelector('[data-tb-test-id="detailedErrorDialog-Dialog-Glass"]')?.remove();
    }});
  }}
  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)],
        cwd=cwd,
        timeout=timeout,
    )


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
    try:
        run_playwright(
            ["--session", session, "state-load", str(state_path)],
            cwd=cwd,
            timeout=60,
        )
    except subprocess.CalledProcessError:
        # Playwright CLI on this machine only allows reading state files under the workdir root.
        pass


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 wait_for_image_ready(path: Path, *, timeout_seconds: int = 20) -> Path:
    """Helper."""
    deadline = time.monotonic() + timeout_seconds
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        if path.exists() and path.stat().st_size > 0:
            try:
                with Image.open(path) as image:
                    image.load()
                return path
            except OSError as exc:
                last_error = exc
        time.sleep(1)
    if last_error is not None:
        raise last_error
    raise FileNotFoundError(f"Image file was not ready in time: {path}")


def parse_args() -> argparse.Namespace:
    """Helper."""
    parser = argparse.ArgumentParser(description="Capture S13 Tableau assets for VIP report.")
    parser.add_argument(
        "--config",
        default=r"C:\Users\niuniu\.codex\vip-report\config.yaml",
        help="Path to VIP report config.yaml",
    )
    parser.add_argument(
        "--slides",
        default="S13",
        help="Comma-separated slide codes to generate, default: S13",
    )
    parser.add_argument(
        "--report-month",
        default="",
        help="Report month label for Tableau filter, e.g. 一月 / 1",
    )
    parser.add_argument(
        "--report-year",
        type=int,
        default=0,
        help="Report year for Tableau filter, e.g. 2026",
    )
    parser.add_argument(
        "--compare-year",
        type=int,
        default=0,
        help="Comparison year placeholder, default report year - 1",
    )
    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 collect_required_capture_ids(filtered_assets: list[dict[str, Any]]) -> set[str]:
    """Helper."""
    return {asset["capture_id"] for asset in filtered_assets}


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)
    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,
    )

    configure_script = write_js(
        workdir,
        f"tmp-{capture_spec['capture_id']}-{time.time_ns()}-configure.js",
        build_configure_view_js(capture_spec),
    )
    capture_script = write_js(
        workdir,
        f"tmp-{capture_spec['capture_id']}-{time.time_ns()}-capture.js",
        build_export_image_js(
            capture_spec["inner_frame_fragment"],
            str(preferred_capture_path.resolve()),
            viewport=VIEWPORT,
        ),
    )

    try:
        max_attempts = 3
        last_error: str | None = None
        for attempt in range(1, max_attempts + 1):
            configure_output = run_code(session, configure_script, cwd=workdir, timeout=420).strip()
            if "### Error" in configure_output:
                last_error = configure_output
                if attempt < max_attempts:
                    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,
                    )
                    time.sleep(3)
                    continue
                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 attempt < max_attempts:
                    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,
                    )
                    time.sleep(3)
                    continue
                raise RuntimeError(
                    f"Failed to capture Tableau view for {capture_spec['capture_id']}: {last_error}"
                )
            break
    finally:
        configure_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,
    )
    wait_for_image_ready(captured_path)
    normalize_image_size(captured_path, {"width": 1400, "height": 3360})
    return captured_path


def resolve_config_path(requested_path: str) -> Path:
    """Helper."""
    path = Path(requested_path)
    if path.exists():
        return path
    local_fallback = Path(__file__).resolve().parents[1] / "config.yaml"
    if local_fallback.exists():
        return local_fallback
    return path


def main() -> None:
    """Helper."""
    script_started_at = time.monotonic()
    args = parse_args()
    config_path = resolve_config_path(args.config)
    config = yaml.safe_load(config_path.read_text(encoding="utf-8"))
    report_month_label, report_month_number, report_year, compare_year = resolve_report_period(args, config)
    specs = build_specs(report_month_label, report_month_number, report_year, compare_year)

    requested = {item.strip().upper() for item in args.slides.split(",") if item.strip()}
    filtered_assets = [item for item in specs["assets"] if item["slide_code"] in requested]
    if not filtered_assets:
        raise SystemExit("No matching slides requested.")

    vip_workdir = Path(args.output_dir).resolve() if args.output_dir else Path(config["paths"]["workdir"]).resolve()
    workspace_root = vip_workdir if args.output_dir else vip_workdir.parents[1]
    asset_dir = vip_workdir / "assets" / "warehouse-100060"
    data_dir = vip_workdir / "data" / "warehouse-100060"
    asset_dir.mkdir(parents=True, exist_ok=True)
    data_dir.mkdir(parents=True, exist_ok=True)

    captures_by_id = {item["capture_id"]: item for item in specs["captures"]}
    required_capture_ids = collect_required_capture_ids(filtered_assets)
    capture_slide_codes = build_capture_slide_codes(filtered_assets)
    use_template_lock = should_use_template_lock(report_month_label, report_year, compare_year)
    log_progress("stage", f"start warehouse-100060 slides={','.join(sorted(requested))}")

    raw_screenshots: dict[str, Path] = {}
    if not use_template_lock:
        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_capture = captures_by_id[next(iter(sorted(required_capture_ids)))]
        first_target_url = f"{base_url}{first_capture['hash_url']}"
        run_playwright(["--session", session, "goto", first_target_url], cwd=vip_workdir, timeout=120)

        login_script = write_js(
            vip_workdir,
            f"tmp-warehouse-login-{time.time_ns()}.js",
            build_login_js(config["tableau"]["username"], config["tableau"]["password"]),
        )
        try:
            run_code(session, login_script, cwd=vip_workdir, timeout=120)
        finally:
            login_script.unlink(missing_ok=True)

        for capture_id in sorted(required_capture_ids):
            capture_spec = captures_by_id[capture_id]
            page_label = describe_capture(capture_id, capture_spec, capture_slide_codes)
            page_started_at = time.monotonic()
            log_progress("page", f"start {page_label}")
            raw_screenshots[capture_id] = capture_tableau_view(
                capture_spec,
                base_url=base_url,
                session=session,
                workdir=vip_workdir,
                workspace_root=workspace_root,
            )
            log_timing(f"page {page_label}", page_started_at)

        save_state(session, state_path, cwd=vip_workdir)

    operations = {"replace_text": [], "replace_images": []}
    manifest_items: list[dict[str, Any]] = []
    assets_started_at = time.monotonic()
    for asset in filtered_assets:
        asset_label = f"{asset['slide_code']} {asset['asset_name']}"
        asset_started_at = time.monotonic()
        log_progress("asset", f"start {asset_label}")
        target = asset_dir / f"{asset['asset_name']}.png"
        if use_template_lock:
            source_path = TEMPLATE_LOCK_DIR / f"s{asset['slide']}_shape{asset['shape_id']}.png"
            if not source_path.exists():
                raise FileNotFoundError(f"Template lock image not found: {source_path}")
            shutil.copyfile(source_path, target)
        else:
            source_path = raw_screenshots[asset["capture_id"]]
            crop_image(
                source_path,
                target,
                asset["crop"],
                resize_to=asset.get("resize_to"),
            )
            operations["replace_images"].append(
                {
                    "slide": asset["slide"],
                    "shape_id": asset["shape_id"],
                    "shape_name": asset["shape_name"],
                    "image_path": str(target),
                }
            )

        manifest_item = {
            "slide_code": asset["slide_code"],
            "slide": asset["slide"],
            "shape_id": asset["shape_id"],
            "shape_name": asset["shape_name"],
            "asset_name": asset["asset_name"],
            "asset_path": str(target),
            "source_capture_id": "template-lock" if use_template_lock else asset["capture_id"],
            "source_view": "template-lock" if use_template_lock else asset["source_view"],
            "note": asset["note"],
            "raw_screenshot": str(source_path),
            "crop": asset["crop"],
        }
        if asset.get("resize_to"):
            manifest_item["resize_to"] = asset["resize_to"]
        manifest_items.append(manifest_item)
        log_timing(f"asset {asset_label}", asset_started_at)
    log_timing(f"assets warehouse-100060 {','.join(sorted(requested))}", assets_started_at)

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

    manifest_path = data_dir / "warehouse-100060-assets.live.json"
    manifest_path.write_text(
        json.dumps(
            {
                "source": {
                    "captures": [
                        {
                            "capture_id": capture_id,
                            "hash_url": captures_by_id[capture_id]["hash_url"],
                            "activate_sheet": captures_by_id[capture_id]["activate_sheet"],
                            "filter_config": captures_by_id[capture_id]["filter_config"],
                            "raw_screenshot": str(raw_screenshots[capture_id]),
                            "note": captures_by_id[capture_id]["note"],
                        }
                        for capture_id in sorted(raw_screenshots)
                    ],
                    "config_path": str(config_path),
                    "report_period": specs["report_period"],
                },
                "assets": manifest_items,
                "operations_path": str(operations_path),
            },
            ensure_ascii=False,
            indent=2,
        ),
        encoding="utf-8",
    )

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


# 关键行注释：脚本直接运行时，从这里进入主流程。
if __name__ == "__main__":
    main()
