"""
Alibaba inquiry scraper.

Public behavior:
- keep `--mode new` as the only public mode
- default to today''s new inquiries when no explicit window is provided
- use `--start-time` / `--end-time` for explicit historical windows
- keep buyer-name search as a recent-year search when no explicit window is provided
"""
import sys
import os
import argparse
import asyncio
import gc
import platform
import json
import random
import re
import shutil
import subprocess
import time
from urllib.error import URLError
from urllib.parse import unquote
from urllib.request import urlopen
from collections import defaultdict

try:
    if hasattr(sys.stdout, "reconfigure"):
        sys.stdout.reconfigure(encoding="utf-8")
    if hasattr(sys.stderr, "reconfigure"):
        sys.stderr.reconfigure(encoding="utf-8")
except Exception:
    pass

# 添加脚本目录到 sys.path，支持直接运行
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_DIR = os.path.dirname(SCRIPT_DIR)
OUTPUT_JSON_PATH = os.path.join(PROJECT_DIR, "inquiries.json")
BUYER_OUTPUT_PREFIX = "buyer_inquiries"
ASSIGNEES_JSON_PATH = os.path.join(PROJECT_DIR, "references", "assignees.json")
if SCRIPT_DIR not in sys.path:
    sys.path.insert(0, SCRIPT_DIR)

from playwright.async_api import async_playwright

from cookie_manager import clear_cookies, load_cookies, save_cookies
from config import (
    MAX_INQUIRIES,
    DEFAULT_MAX_DETAIL,
)
from list_scraper import (
    scrape_list,
    ListScrapeError,
    LoginRequiredError,
    assert_not_login_redirect,
    LOGIN_REQUIRED_MESSAGE,
    navigate_to_url,
)
from detail_scraper import scrape_detail, ScrapeError
from output import (
    format_output,
    format_empty_output,
    format_error_output,
    build_output_data,
)
from batch_aggregator import (
    build_chunk_windows,
    month_key_for_window,
    batch_root_paths,
    raw_shard_path,
    write_json_atomic,
    read_json,
    merge_inquiries_by_id,
    aggregate_skipped_stats,
    build_batch_meta,
    build_manifest,
)
from page_selectors import build_url
from browser_pages import new_background_page
from datetime import datetime, timedelta


CDP_ENDPOINTS = (
    "http://127.0.0.1:9222",
    "http://localhost:9222",
)

DEFAULT_RESERVED_MEMORY_BYTES = int(1.5 * 1024 ** 3)
# Windows 上 Chrome/Playwright 波动更明显，自动 tab 数计算时多预留 2.3GB。
WINDOWS_RESERVED_MEMORY_BYTES = int(2.3 * 1024 ** 3)
MEMORY_PER_TAB_BYTES = 512 * 1024 ** 2
MAX_TOTAL_TABS = 10
LIST_PAGE_RESERVED_TABS = 1
MAX_DETAIL_TABS = MAX_TOTAL_TABS - LIST_PAGE_RESERVED_TABS
DEFAULT_TABS = 2
CDP_READY_TIMEOUT_SECONDS = 10
CDP_READY_POLL_SECONDS = 0.5
MAC_DEFAULT_CHROME_BOOT_WAIT_SECONDS = 1
DETAIL_MAX_ATTEMPTS = 3
EMPTY_DETAIL_ERROR_MESSAGE = "DOM extraction failed: no messages found"
DETAIL_EMPTY_REPAIR_DELAY_SECONDS = 60
LONG_ADVANCED_WINDOW_DAYS = 5
TAB_OPEN_PACE_MIN_SECONDS = 0.4
TAB_OPEN_PACE_MAX_SECONDS = 0.6
# 高级时间 URL 分片已启用；命中 advanced_time 且窗口超过 7 天时按 chunk-days 拆分。
ENABLE_ADVANCED_TIME_BATCH = True
# owner_id 与 startTime/endTime 合并到同一个 search JSON 不稳定，默认关闭。
ENABLE_ADVANCED_TIME_OWNER_SEARCH = False
MAX_DETAIL_NOT_SET = "not set"
LOGIN_REQUIRED_EXIT_CODE = 3


def login_required_error_output(exc):
    return format_error_output("LOGIN_REQUIRED", str(exc) or LOGIN_REQUIRED_MESSAGE)


async def abort_for_login_required(exc, pw):
    print(login_required_error_output(exc))
    if pw:
        await pw.stop()
    sys.exit(LOGIN_REQUIRED_EXIT_CODE)


def available_memory_bytes():
    system = platform.system()

    try:
        if system == "Windows":
            import ctypes

            class MemoryStatus(ctypes.Structure):
                _fields_ = [
                    ("dwLength", ctypes.c_ulong),
                    ("dwMemoryLoad", ctypes.c_ulong),
                    ("ullTotalPhys", ctypes.c_ulonglong),
                    ("ullAvailPhys", ctypes.c_ulonglong),
                    ("ullTotalPageFile", ctypes.c_ulonglong),
                    ("ullAvailPageFile", ctypes.c_ulonglong),
                    ("ullTotalVirtual", ctypes.c_ulonglong),
                    ("ullAvailVirtual", ctypes.c_ulonglong),
                    ("ullAvailExtendedVirtual", ctypes.c_ulonglong),
                ]

            status = MemoryStatus()
            status.dwLength = ctypes.sizeof(status)
            ctypes.windll.kernel32.GlobalMemoryStatusEx(ctypes.byref(status))
            return status.ullAvailPhys

        if system == "Linux":
            with open("/proc/meminfo", encoding="utf-8") as f:
                for line in f:
                    if line.startswith("MemAvailable:"):
                        return int(line.split()[1]) * 1024

        if system == "Darwin":
            import re
            import subprocess

            vm_stat = subprocess.check_output(["vm_stat"], text=True)
            page_size = int(re.search(r"page size of (\d+) bytes", vm_stat).group(1))
            pages = 0
            for name in ("Pages free", "Pages inactive", "Pages speculative"):
                match = re.search(rf"{name}:\s+(\d+)\.", vm_stat)
                if match:
                    pages += int(match.group(1))
            return pages * page_size
    except Exception:
        return None

    return None


def auto_tabs():
    available = available_memory_bytes()
    if not available:
        return DEFAULT_TABS
    reserved = WINDOWS_RESERVED_MEMORY_BYTES if platform.system() == "Windows" else DEFAULT_RESERVED_MEMORY_BYTES
    tabs = (available - reserved) // MEMORY_PER_TAB_BYTES
    return max(1, min(MAX_DETAIL_TABS, int(tabs)))


try:
    TABS = auto_tabs()
except Exception:
    TABS = DEFAULT_TABS


def cdp_is_ready(endpoint):
    try:
        with urlopen(f"{endpoint}/json/version", timeout=2) as response:
            return 200 <= response.status < 300
    except (OSError, URLError):
        return False


def wait_for_cdp(timeout_seconds=CDP_READY_TIMEOUT_SECONDS):
    deadline = time.time() + timeout_seconds
    while time.time() < deadline:
        for endpoint in CDP_ENDPOINTS:
            if cdp_is_ready(endpoint):
                return True
        time.sleep(CDP_READY_POLL_SECONDS)
    return False


def chrome_profile_path():
    return os.path.expanduser("~/.chrome-cdp-profile")


def windows_chrome_path():
    candidates = [
        os.path.join(os.environ.get("PROGRAMFILES", ""), "Google", "Chrome", "Application", "chrome.exe"),
        os.path.join(os.environ.get("PROGRAMFILES(X86)", ""), "Google", "Chrome", "Application", "chrome.exe"),
        os.path.join(os.environ.get("LOCALAPPDATA", ""), "Google", "Chrome", "Application", "chrome.exe"),
    ]
    for path in candidates:
        if path and os.path.exists(path):
            return path
    return shutil.which("chrome.exe") or shutil.which("chrome")


def chrome_path_for_platform():
    system = platform.system()
    if system == "Windows":
        return windows_chrome_path()
    if system == "Darwin":
        path = "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"
        return path if os.path.exists(path) else shutil.which("google-chrome") or shutil.which("chrome")
    return (
        shutil.which("google-chrome")
        or shutil.which("google-chrome-stable")
        or shutil.which("chromium")
        or shutil.which("chromium-browser")
    )


def manual_cdp_hint():
    profile = chrome_profile_path()
    system = platform.system()
    if system == "Windows":
        return (
            'PowerShell: & "$env:ProgramFiles\\Google\\Chrome\\Application\\chrome.exe" '
            f'--remote-debugging-port=9222 --user-data-dir="{profile}" '
            "--no-first-run --no-default-browser-check"
        )
    if system == "Darwin":
        return (
            "open -a 'Google Chrome' && open -na 'Google Chrome' --args "
            f'--remote-debugging-port=9222 --user-data-dir="{profile}" '
            "--no-first-run --no-default-browser-check &"
        )
    return (
        "google-chrome --remote-debugging-port=9222 "
        f'--user-data-dir="{profile}" --no-first-run --no-default-browser-check &'
    )


def mac_chrome_is_running():
    try:
        result = subprocess.run(
            ["osascript", "-e", 'application "Google Chrome" is running'],
            capture_output=True,
            text=True,
            timeout=3,
            check=False,
        )
        return result.stdout.strip().lower() == "true"
    except Exception:
        return False


def launch_chrome_cdp():
    chrome_path = chrome_path_for_platform()
    if not chrome_path:
        raise RuntimeError(f"未找到 Chrome 可执行文件。手动启动示例：{manual_cdp_hint()}")

    profile = chrome_profile_path()
    os.makedirs(profile, exist_ok=True)
    chrome_args = [
        "--remote-debugging-port=9222",
        f"--user-data-dir={profile}",
        "--no-first-run",
        "--no-default-browser-check",
        "--start-minimized",
    ]
    if platform.system() == "Darwin":
        if not mac_chrome_is_running():
            subprocess.Popen(
                ["open", "-a", "Google Chrome"],
                stdout=subprocess.DEVNULL,
                stderr=subprocess.DEVNULL,
                close_fds=True,
            )
            time.sleep(MAC_DEFAULT_CHROME_BOOT_WAIT_SECONDS)
        args = ["open", "-na", "Google Chrome", "--args", *chrome_args]
    else:
        args = [chrome_path, *chrome_args]
    subprocess.Popen(
        args,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        close_fds=True,
    )


def ensure_cdp_ready():
    if wait_for_cdp(timeout_seconds=2):
        return False
    launch_chrome_cdp()
    if not wait_for_cdp():
        raise RuntimeError(f"Chrome CDP 启动后仍不可用。手动启动示例：{manual_cdp_hint()}")
    return True


async def connect_cdp(pw):
    errors = []
    for endpoint in CDP_ENDPOINTS:
        try:
            return await pw.chromium.connect_over_cdp(endpoint)
        except Exception as e:
            message = str(e).splitlines()[0] if str(e) else repr(e)
            errors.append(f"{endpoint}: {message}")
    raise RuntimeError("; ".join(errors))


async def maybe_close_browser(browser, launched_by_us):
    if launched_by_us and browser:
        try:
            await browser.close()
        except Exception:
            pass


def normalize_assignee_name(name):
    return " ".join((name or "").lower().split())


def load_assignee_cache_payload(path=ASSIGNEES_JSON_PATH):
    if not os.path.exists(path):
        return {}
    try:
        with open(path, encoding="utf-8") as handle:
            payload = json.load(handle)
            return payload if isinstance(payload, dict) else {}
    except (OSError, json.JSONDecodeError):
        return {}


def load_assignee_cache(path=ASSIGNEES_JSON_PATH):
    owners = load_assignee_cache_payload(path).get("owners", [])
    return owners if isinstance(owners, list) else []


def clear_assignee_cache(path=ASSIGNEES_JSON_PATH):
    try:
        os.remove(path)
    except FileNotFoundError:
        return "missing"
    return "cleared"


def assignee_cache_is_fresh_today(path=ASSIGNEES_JSON_PATH, today=None):
    payload = load_assignee_cache_payload(path)
    owners = payload.get("owners", [])
    if not owners:
        return False

    saved_at = str(payload.get("saved_at", "")).strip()
    if not saved_at:
        return False

    try:
        saved_date = datetime.strptime(saved_at[:10], "%Y-%m-%d").date()
    except ValueError:
        return False
    return saved_date == (today or datetime.now().date())


async def get_assignees_for_scrape(page, cache_path=ASSIGNEES_JSON_PATH):
    cached = load_assignee_cache(cache_path)
    if assignee_cache_is_fresh_today(cache_path):
        return cached, None

    try:
        return await refresh_assignee_cache(page, cache_path=cache_path), None
    except Exception as e:
        if cached:
            return cached, f"负责人缓存刷新失败，已使用本地缓存: {e}"
        return [], f"负责人缓存刷新失败，且没有可用本地缓存: {e}"


def save_assignee_cache(owners, path=ASSIGNEES_JSON_PATH):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    data = {
        "version": 1,
        "saved_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "owners": owners,
    }
    with open(path, "w", encoding="utf-8") as handle:
        json.dump(data, handle, ensure_ascii=False, indent=2)
        handle.write("\n")
        handle.flush()
        os.fsync(handle.fileno())


def find_assignee_owner_id(assignee, owners):
    needle = normalize_assignee_name(assignee)
    if not needle:
        return None
    matches = [
        owner for owner in owners
        if needle == normalize_assignee_name(owner.get("name", ""))
        or needle in normalize_assignee_name(owner.get("name", ""))
    ]
    if len(matches) != 1:
        return None
    return matches[0].get("owner_id")


def owner_id_from_url(url):
    if "search=" not in url:
        return None
    raw_search = url.split("search=", 1)[1].split("&", 1)[0]
    try:
        owner_id = json.loads(unquote(raw_search)).get("owner")
        return int(owner_id) if owner_id is not None else None
    except (TypeError, ValueError, json.JSONDecodeError):
        return None


def normalize_owner_id(owner_id):
    try:
        return int(owner_id)
    except (TypeError, ValueError):
        return None


async def visible_assignee_names(page):
    """
    读取下拉菜单中全部业务员名称，兼容懒加载/虚拟滚动。
    在滚动过程中持续收集名称，避免顶部项被卸载后遗漏。
    """
    menu = page.locator(".aui-dropdown-menu:visible").filter(has_text="All").first
    collected = []
    previous_count = -1
    for _ in range(20):
        names = await menu.locator(".aui-menu-item").all_inner_texts()
        for name in names:
            n = name.strip()
            if n and n not in collected and n.lower() != "all":
                collected.append(n)
        if len(collected) == previous_count:
            break
        previous_count = len(collected)
        menu_el = await menu.element_handle()
        if menu_el:
            await page.evaluate("""
                el => {
                    let scrollable = el;
                    if (scrollable.scrollHeight <= scrollable.clientHeight) {
                        const all = scrollable.querySelectorAll('*');
                        for (const child of all) {
                            if (child.scrollHeight > child.clientHeight) {
                                scrollable = child;
                                break;
                            }
                        }
                    }
                    scrollable.scrollTop = scrollable.scrollHeight;
                }
            """, menu_el)
        await page.wait_for_timeout(300)
    return collected


async def resolve_target_owner_id(page, assignee_name):
    """
    当缓存中找不到 owner_id 时，直接在页面内点击目标业务员并从 URL 读取 owner_id。
    """
    if not assignee_name:
        return None
    try:
        base_url = build_url(1)
        await navigate_to_url(page, base_url)
        await page.wait_for_selector(".filter-owner", timeout=10000)
        await page.wait_for_timeout(800)

        await page.locator(".filter-owner").first.click(force=True)
        await page.wait_for_timeout(1000)
        menu = page.locator(".aui-dropdown-menu:visible").filter(has_text="All").first

        target = menu.get_by_text(assignee_name, exact=True)
        if await target.count() == 0:
            needle = assignee_name.lower()
            items = await menu.locator(".aui-menu-item").all_inner_texts()
            for item in items:
                if needle in item.lower():
                    target = menu.get_by_text(item.strip(), exact=True)
                    break
            else:
                await page.mouse.click(0, 0)
                await page.wait_for_timeout(300)
                return None

        await target.click(timeout=10000)
        await page.wait_for_timeout(800)
        owner_id = owner_id_from_url(page.url)
        try:
            await page.mouse.click(0, 0)
            await page.wait_for_timeout(300)
        except Exception:
            pass
        return owner_id
    except Exception:
        try:
            await page.mouse.click(0, 0)
            await page.wait_for_timeout(300)
        except Exception:
            pass
        return None


async def validate_owner_id_for_url_filter(
    page,
    assignee_name,
    cached_owner_id,
    cache_path=ASSIGNEES_JSON_PATH,
):
    cached_owner_id = normalize_owner_id(cached_owner_id)
    if not assignee_name or cached_owner_id is None:
        return cached_owner_id, None

    page_owner_id = await resolve_target_owner_id(page, assignee_name)
    if page_owner_id == cached_owner_id:
        return cached_owner_id, None

    if page_owner_id is not None:
        try:
            refreshed = await refresh_assignee_cache(page, cache_path=cache_path)
            refreshed_owner_id = normalize_owner_id(find_assignee_owner_id(assignee_name, refreshed))
            if refreshed_owner_id is not None:
                return refreshed_owner_id, None
        except LoginRequiredError:
            raise
        except Exception as exc:
            return (
                page_owner_id,
                f"Assignee owner_id cache may be stale; refresh failed, using page owner_id: {exc}",
            )

        return (
            page_owner_id,
            "Assignee owner_id cache may be stale; refreshed cache did not contain the target, "
            "using page owner_id.",
        )

    try:
        refreshed = await refresh_assignee_cache(page, cache_path=cache_path)
        refreshed_owner_id = normalize_owner_id(find_assignee_owner_id(assignee_name, refreshed))
        if refreshed_owner_id is not None:
            return refreshed_owner_id, None
    except LoginRequiredError:
        raise
    except Exception as exc:
        return (
            None,
            f"Assignee owner_id could not be validated and cache refresh failed; using local assignee filtering: {exc}",
        )

    return (
        None,
        "Assignee owner_id could not be validated from page or refreshed cache; using local assignee filtering.",
    )


async def refresh_assignee_cache(page, cache_path=ASSIGNEES_JSON_PATH):
    base_url = build_url(1)
    owners = []
    warnings_list = []
    await navigate_to_url(page, base_url)
    await page.wait_for_selector(".filter-owner", timeout=10000)
    await page.wait_for_timeout(800)
    await page.locator(".filter-owner").first.click(force=True)
    await page.wait_for_timeout(300)
    names = await visible_assignee_names(page)
    await page.mouse.click(0, 0)
    await page.wait_for_timeout(500)
    if not names:
        raise RuntimeError("未读取到负责人名单")

    for name in names:
        try:
            await page.locator(".filter-owner").first.click(force=True)
            await page.wait_for_timeout(1000)
            menu = page.locator(".aui-dropdown-menu:visible").filter(has_text="All").first

            # 优先精确匹配，失败则降级为部分匹配（兼容 DOM 中多余空白）
            target = menu.get_by_text(name, exact=True)
            if await target.count() == 0:
                needle = name.lower()
                items = await menu.locator(".aui-menu-item").all_inner_texts()
                for item in items:
                    if needle in item.lower():
                        target = menu.get_by_text(item.strip(), exact=True)
                        break
                else:
                    warnings_list.append(f"负责人 '{name}' 在菜单中未找到匹配项")
                    try:
                        await page.mouse.click(0, 0)
                        await page.wait_for_timeout(300)
                    except Exception:
                        pass
                    continue

            await target.click(timeout=10000)
            await page.wait_for_timeout(800)
            owner_id = owner_id_from_url(page.url)
            if owner_id is not None:
                owners.append({"name": name, "owner_id": owner_id})
            else:
                warnings_list.append(f"负责人 '{name}' 点击后 URL 中未解析到 owner_id")
        except Exception as e:
            warnings_list.append(f"负责人 '{name}' 缓存刷新失败: {e}")
            try:
                await page.mouse.click(0, 0)
                await page.wait_for_timeout(300)
            except Exception:
                pass

    save_assignee_cache(owners, path=cache_path)
    if warnings_list:
        import warnings as py_warnings
        for w in warnings_list:
            py_warnings.warn(w, RuntimeWarning)
    return owners


_page_open_count = 0


def reset_throttle():
    global _page_open_count
    _page_open_count = 0


async def throttled_new_page(ctx):
    global _page_open_count
    _page_open_count += 1
    await pace_before_tab_open()
    return await new_background_page(ctx)


def is_retryable_detail_error(error):
    message = str(error)
    retryable_messages = (
        EMPTY_DETAIL_ERROR_MESSAGE,
        "page load timeout",
    )
    return any(text in message for text in retryable_messages)


def is_empty_detail_error(error):
    return EMPTY_DETAIL_ERROR_MESSAGE in str(error or "")


def is_repairable_detail_failure(inquiry):
    if not inquiry.get("detail_skipped"):
        return False
    return str(inquiry.get("scrape_status", "")).startswith("error:")


def is_empty_detail_failure(inquiry):
    if not is_repairable_detail_failure(inquiry):
        return False
    return (
        is_empty_detail_error(inquiry.get("detail_skip_reason"))
        or is_empty_detail_error(inquiry.get("scrape_status"))
    )


async def pace_before_tab_open(pace_lock=None):
    if pace_lock is None:
        await asyncio.sleep(random.uniform(
            TAB_OPEN_PACE_MIN_SECONDS,
            TAB_OPEN_PACE_MAX_SECONDS,
        ))
        return

    async with pace_lock:
        await pace_before_tab_open()


async def scrape_detail_item(
    ctx,
    inq,
    depth,
    semaphore,
    *,
    detail_tab_pace_lock=None,
    long_window_cleanup=False,
):
    async with semaphore:
        last_error = None
        for attempt in range(DETAIL_MAX_ATTEMPTS):
            try:
                await pace_before_tab_open(detail_tab_pace_lock)
                detail = await scrape_detail(
                    ctx,
                    inq.get("detail_url", ""),
                    depth=depth,
                    buyer_name=inq.get("buyer_name", ""),
                    request_page_gc=long_window_cleanup,
                )
                inq["detail"] = detail
                inq["detail_skipped"] = False
                inq["scrape_status"] = "success"
                inq.pop("detail_skip_reason", None)
                # 提取业务员首次回复时间到顶层，方便报告直接使用
                inq["first_reply_time"] = detail.get("stats", {}).get("first_seller_response")
                return
            except ScrapeError as e:
                last_error = e
                if not is_retryable_detail_error(e) or attempt == DETAIL_MAX_ATTEMPTS - 1:
                    break
                await asyncio.sleep(1)
            except Exception as e:
                last_error = e
                break
            finally:
                if long_window_cleanup:
                    gc.collect()

        inq["detail"] = None
        inq["detail_skipped"] = True
        inq["first_reply_time"] = None
        if isinstance(last_error, ScrapeError):
            inq["detail_skip_reason"] = str(last_error)
        else:
            inq["detail_skip_reason"] = f"unexpected: {last_error}"
        inq["scrape_status"] = f"error:{last_error}"


async def repair_failed_detail_items(
    ctx,
    inquiries,
    depth,
    semaphore,
    *,
    detail_tab_pace_lock=None,
    long_window_cleanup=False,
):
    repair_candidates = [inq for inq in inquiries if is_repairable_detail_failure(inq)]
    if not repair_candidates:
        return

    repair_tasks = [
        asyncio.create_task(scrape_detail_item(
            ctx,
            inq,
            depth,
            semaphore,
            detail_tab_pace_lock=detail_tab_pace_lock,
            long_window_cleanup=long_window_cleanup,
        ))
        for inq in repair_candidates
    ]
    await asyncio.gather(*repair_tasks)


async def repair_empty_detail_items_after_delay(
    ctx,
    inquiries,
    depth,
    semaphore,
    *,
    detail_tab_pace_lock=None,
    long_window_cleanup=False,
    delay_seconds=None,
):
    repair_candidates = [inq for inq in inquiries if is_empty_detail_failure(inq)]
    if not repair_candidates:
        return

    if delay_seconds is None:
        delay_seconds = DETAIL_EMPTY_REPAIR_DELAY_SECONDS
    if delay_seconds > 0:
        await asyncio.sleep(delay_seconds)

    repair_candidates = [inq for inq in repair_candidates if is_empty_detail_failure(inq)]
    if not repair_candidates:
        return

    repair_tasks = [
        asyncio.create_task(scrape_detail_item(
            ctx,
            inq,
            depth,
            semaphore,
            detail_tab_pace_lock=detail_tab_pace_lock,
            long_window_cleanup=long_window_cleanup,
        ))
        for inq in repair_candidates
    ]
    await asyncio.gather(*repair_tasks)


def sanitize_filename_part(value, fallback="buyer"):
    text = " ".join(str(value or "").split())
    text = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", text)
    text = re.sub(r"\s+", "_", text)
    text = re.sub(r"_+", "_", text).strip(" ._")
    if not text:
        text = fallback
    return text[:80].rstrip(" ._") or fallback


def resolve_output_json_path(buyer_name="", project_dir=PROJECT_DIR):
    buyer_name = str(buyer_name or "").strip()
    if not buyer_name:
        return os.path.join(project_dir, "inquiries.json")
    buyer_slug = sanitize_filename_part(buyer_name)
    return os.path.join(project_dir, f"{BUYER_OUTPUT_PREFIX}_{buyer_slug}.json")


def write_output_json(output_text, output_path=OUTPUT_JSON_PATH):
    with open(output_path, "w", encoding="utf-8") as handle:
        handle.write(output_text)
        handle.write("\n")


def output_summary(output_text, output_path=OUTPUT_JSON_PATH):
    data = json.loads(output_text)
    summary = {
        "status": data.get("status"),
        "saved_to": output_path,
        "date": data.get("date"),
        "generated_at": data.get("generated_at"),
        "total": data.get("total"),
        "detail_count": data.get("detail_count", 0),
        "detail_skipped": data.get("detail_skipped", 0),
        "failed": data.get("failed", []),
    }
    print(json.dumps(summary, ensure_ascii=False, indent=2))


def emit_success_output(output_text, stdout_json, output_path=OUTPUT_JSON_PATH):
    if stdout_json:
        print(output_text)
        return
    write_output_json(output_text, output_path=output_path)
    output_summary(output_text, output_path=output_path)


def parse_mode_arg(value):
    if value == "new":
        return value
    raise argparse.ArgumentTypeError(
        f"未知模式: {value}。当前只支持 new。"
    )


def build_parser():
    parser = argparse.ArgumentParser(description="阿里巴巴国际站询盘抓取脚本")
    parser.add_argument(
        "--mode",
        type=parse_mode_arg,
        metavar="new",
        default="new",
        help="运行模式：new=今日新建询盘（默认）",
    )
    parser.add_argument(
        "--depth",
        choices=["quick", "normal", "full"],
        default=None,
        help="详情深度：quick=不翻页；normal=翻3次；full=翻到底。默认：今日主线为 full，显式时间窗为 normal",
    )
    parser.add_argument(
        "--max-detail",
        type=int,
        default=None,
        help="最多抓取详情条数，0=只抓列表；未设置时输出标记为 not set，显式时间窗默认不设详情上限",
    )
    parser.add_argument(
        "--start-time",
        type=str,
        default="",
        help="显式时间窗开始（YYYY-MM-DD、YYYY-MM-DD HH:MM:SS、10 位秒时间戳或 13 位毫秒时间戳）",
    )
    parser.add_argument(
        "--end-time",
        type=str,
        default="",
        help="显式时间窗结束，排他边界（YYYY-MM-DD、YYYY-MM-DD HH:MM:SS、10 位秒时间戳或 13 位毫秒时间戳）",
    )
    parser.add_argument(
        "--max",
        type=int,
        default=None,
        help=f"最多抓取询盘条数；未设置时默认 {MAX_INQUIRIES}，显式时间窗默认不设列表上限",
    )
    parser.add_argument(
        "--batch-root",
        type=str,
        default=os.path.join(PROJECT_DIR, "outputs", "inquiries"),
        help="分片落盘根目录（仅长时间窗批处理使用）",
    )
    parser.add_argument(
        "--chunk-days",
        type=int,
        default=1,
        help="批处理分片天数（默认按天切片）",
    )
    parser.add_argument(
        "--resume",
        action="store_true",
        help="批处理时跳过已存在的分片文件",
    )
    parser.add_argument("--category", type=str, default="", help="过滤分类，例如 TM")
    parser.add_argument("--assignee", type=str, default="", help="过滤业务员名称（模糊匹配）")
    parser.add_argument(
        "--buyer-name",
        type=str,
        default="",
        help="按买家名称搜索；未传显式时间窗时使用后台默认最近一年范围",
    )
    parser.add_argument(
        "--stdout-json",
        action="store_true",
        help="完整 JSON 输出到 stdout，不写入 inquiries.json",
    )
    parser.add_argument(
        "--clear-login-state",
        action="store_true",
        help="删除已保存的自动登录 cookie 后退出，便于切换账号",
    )
    parser.add_argument(
        "--list-assignees",
        action="store_true",
        help="只输出当前负责人名单并刷新缓存，不抓取询盘",
    )
    return parser


def parse_time_arg(value, flag_name):
    if value is None or str(value).strip() == "":
        return None
    text = str(value).strip()
    if text.isdigit():
        if len(text) == 13:
            return datetime.fromtimestamp(int(text) / 1000)
        if len(text) == 10:
            return datetime.fromtimestamp(int(text))
        raise ValueError(f"{flag_name} has invalid timestamp length: {text}")
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            return datetime.strptime(text, fmt)
        except ValueError:
            continue
    raise ValueError(f"{flag_name} format error: {text}")


def day_start(value):
    return datetime.combine(value.date(), datetime.min.time())


def format_window_bound(value):
    if value.time() == datetime.min.time():
        return value.strftime("%Y-%m-%d")
    return value.strftime("%Y-%m-%d %H:%M:%S")


def format_window_datetime(value):
    if value is None:
        return None
    return value.strftime("%Y-%m-%d %H:%M:%S")


def resolve_window_basis(start_arg, end_arg):
    if not start_arg and not end_arg:
        return None
    start_text = str(start_arg or "").strip()
    end_text = str(end_arg or "").strip()
    if (
        start_text.isdigit()
        and end_text.isdigit()
        and len(start_text) in (10, 13)
        and len(end_text) in (10, 13)
    ):
        return "timestamp"
    return "local_datetime_string"


def build_window_meta(start_arg, end_arg, time_window):
    basis = resolve_window_basis(start_arg, end_arg)
    if not basis:
        return None
    return {
        "requested_start_time": str(start_arg or ""),
        "requested_end_time": str(end_arg or ""),
        "actual_start_time": format_window_datetime(time_window.get("window_start")),
        "actual_end_time": format_window_datetime(time_window.get("window_end")),
        "actual_window_basis": basis,
    }


def resolve_time_window(start_time, end_time, *, clamp_future_end=True):
    today_start = day_start(datetime.now())
    tomorrow_start = today_start + timedelta(days=1)

    if (start_time is None) != (end_time is None):
        raise ValueError("--start-time and --end-time must be provided together")

    if start_time is None:
        return {
            "window_start": today_start,
            "window_end": tomorrow_start,
            "output_start_time": None,
            "output_end_time": None,
            "is_default_today": True,
            "is_explicit_window": False,
        }

    window_start = start_time
    window_end = end_time
    if clamp_future_end and window_end > tomorrow_start:
        window_end = tomorrow_start
    if window_start >= window_end:
        raise ValueError("time window is invalid: start must be earlier than end")

    return {
        "window_start": window_start,
        "window_end": window_end,
        "output_start_time": format_window_bound(window_start),
        "output_end_time": format_window_bound(window_end),
        "is_default_today": False,
        "is_explicit_window": True,
    }


def resolve_query_scope(time_window, buyer_name=""):
    buyer_name = (buyer_name or "").strip()
    buyer_name_default_window = bool(buyer_name) and time_window["is_default_today"]

    if buyer_name_default_window:
        return {
            **time_window,
            "query_start": None,
            "query_end": None,
            "output_start_time": None,
            "output_end_time": None,
            "buyer_name_default_window": True,
        }

    return {
        **time_window,
        "query_start": time_window["window_start"],
        "query_end": time_window["window_end"],
        "buyer_name_default_window": False,
    }


def should_enable_batch_mode(query_scope, buyer_name=""):
    if buyer_name:
        return False
    if not ENABLE_ADVANCED_TIME_BATCH:
        return False

    window_start = query_scope.get("query_start")
    window_end = query_scope.get("query_end")
    if window_start is None or window_end is None:
        return False
    if query_scope.get("is_default_today"):
        return False
    return (window_end - window_start).days > 7


def should_pace_detail_tabs(query_scope):
    if query_scope.get("is_default_today"):
        return False
    window_start = query_scope.get("query_start")
    window_end = query_scope.get("query_end")
    if window_start is None or window_end is None:
        return False
    return (window_end - window_start).days > LONG_ADVANCED_WINDOW_DAYS


def resolve_max_detail_settings(raw_max_detail, explicit_window):
    if raw_max_detail is None:
        if explicit_window:
            return None, MAX_DETAIL_NOT_SET
        return DEFAULT_MAX_DETAIL, MAX_DETAIL_NOT_SET
    return raw_max_detail, raw_max_detail


def resolve_max_items_setting(raw_max, explicit_window):
    if raw_max is None:
        if explicit_window:
            return None
        return MAX_INQUIRIES
    return raw_max


def advanced_month_time_bounds(window_states):
    starts = [str(w.get("start")) for w in window_states if w.get("start")]
    ends = [str(w.get("end")) for w in window_states if w.get("end")]
    if not starts or not ends:
        return None, None
    return min(starts), max(ends)


def merge_list_metadata(payloads):
    warnings = []
    seen = set()
    pages_loaded = 0
    pages_checked = 0
    partial = False

    for payload in payloads:
        if payload.get("partial"):
            partial = True
        if isinstance(payload.get("pages_loaded"), int):
            pages_loaded += payload["pages_loaded"]
        if isinstance(payload.get("pages_checked"), int):
            pages_checked += payload["pages_checked"]
        for warning in payload.get("warnings", []) or []:
            if warning not in seen:
                seen.add(warning)
                warnings.append(warning)

    merged = {}
    if partial:
        merged["partial"] = True
    if pages_loaded:
        merged["pages_loaded"] = pages_loaded
    if pages_checked:
        merged["pages_checked"] = pages_checked
    if warnings:
        merged["warnings"] = warnings
    return merged


async def get_assignees_for_scrape(page, cache_path=ASSIGNEES_JSON_PATH):
    cached = load_assignee_cache(cache_path)
    if assignee_cache_is_fresh_today(cache_path):
        return cached, None

    try:
        return await refresh_assignee_cache(page, cache_path=cache_path), None
    except LoginRequiredError:
        raise
    except Exception as exc:
        if cached:
            return cached, f"负责人缓存刷新失败，已使用本地缓存: {exc}"
        return [], f"负责人缓存刷新失败，且没有可用本地缓存: {exc}"


async def resolve_target_owner_id(page, assignee_name):
    if not assignee_name:
        return None
    try:
        base_url = build_url(1)
        await navigate_to_url(page, base_url)
        await page.wait_for_timeout(500)
        await assert_not_login_redirect(page)
        await page.wait_for_selector(".filter-owner", timeout=10000)
        await page.wait_for_timeout(800)
        await assert_not_login_redirect(page)

        await page.locator(".filter-owner").first.click(force=True)
        await page.wait_for_timeout(1000)
        menu = page.locator(".aui-dropdown-menu:visible").filter(has_text="All").first

        target = menu.get_by_text(assignee_name, exact=True)
        if await target.count() == 0:
            needle = assignee_name.lower()
            items = await menu.locator(".aui-menu-item").all_inner_texts()
            for item in items:
                if needle in item.lower():
                    target = menu.get_by_text(item.strip(), exact=True)
                    break
            else:
                await page.mouse.click(0, 0)
                await page.wait_for_timeout(300)
                return None

        await target.click(timeout=10000)
        await page.wait_for_timeout(800)
        await assert_not_login_redirect(page)
        owner_id = owner_id_from_url(page.url)
        try:
            await page.mouse.click(0, 0)
            await page.wait_for_timeout(300)
        except Exception:
            pass
        return owner_id
    except LoginRequiredError:
        raise
    except Exception:
        try:
            await page.mouse.click(0, 0)
            await page.wait_for_timeout(300)
        except Exception:
            pass
        return None


async def validate_owner_id_for_url_filter(
    page,
    assignee_name,
    cached_owner_id,
    cache_path=ASSIGNEES_JSON_PATH,
):
    cached_owner_id = normalize_owner_id(cached_owner_id)
    if not assignee_name or cached_owner_id is None:
        return cached_owner_id, None

    page_owner_id = await resolve_target_owner_id(page, assignee_name)
    if page_owner_id == cached_owner_id:
        return cached_owner_id, None

    if page_owner_id is not None:
        try:
            refreshed = await refresh_assignee_cache(page, cache_path=cache_path)
            refreshed_owner_id = normalize_owner_id(find_assignee_owner_id(assignee_name, refreshed))
            if refreshed_owner_id is not None:
                return refreshed_owner_id, None
        except LoginRequiredError:
            raise
        except Exception as exc:
            return (
                page_owner_id,
                f"Assignee owner_id cache may be stale; refresh failed, using page owner_id: {exc}",
            )

        return (
            page_owner_id,
            "Assignee owner_id cache may be stale; refreshed cache did not contain the target, using page owner_id.",
        )

    try:
        refreshed = await refresh_assignee_cache(page, cache_path=cache_path)
        refreshed_owner_id = normalize_owner_id(find_assignee_owner_id(assignee_name, refreshed))
        if refreshed_owner_id is not None:
            return refreshed_owner_id, None
    except LoginRequiredError:
        raise
    except Exception as exc:
        return (
            None,
            f"Assignee owner_id could not be validated and cache refresh failed; using local assignee filtering: {exc}",
        )

    return (
        None,
        "Assignee owner_id could not be validated from page or refreshed cache; using local assignee filtering.",
    )


async def refresh_assignee_cache(page, cache_path=ASSIGNEES_JSON_PATH):
    base_url = build_url(1)
    owners = []
    warnings_list = []
    await navigate_to_url(page, base_url)
    await page.wait_for_timeout(500)
    await assert_not_login_redirect(page)
    await page.wait_for_selector(".filter-owner", timeout=10000)
    await page.wait_for_timeout(800)
    await assert_not_login_redirect(page)
    await page.locator(".filter-owner").first.click(force=True)
    await page.wait_for_timeout(300)
    names = await visible_assignee_names(page)
    await page.mouse.click(0, 0)
    await page.wait_for_timeout(500)
    if not names:
        raise RuntimeError("未读取到负责人名单")

    for name in names:
        try:
            await page.locator(".filter-owner").first.click(force=True)
            await page.wait_for_timeout(1000)
            menu = page.locator(".aui-dropdown-menu:visible").filter(has_text="All").first

            target = menu.get_by_text(name, exact=True)
            if await target.count() == 0:
                needle = name.lower()
                items = await menu.locator(".aui-menu-item").all_inner_texts()
                for item in items:
                    if needle in item.lower():
                        target = menu.get_by_text(item.strip(), exact=True)
                        break
                else:
                    warnings_list.append(f"负责人 '{name}' 在菜单中未找到匹配项")
                    try:
                        await page.mouse.click(0, 0)
                        await page.wait_for_timeout(300)
                    except Exception:
                        pass
                    continue

            await target.click(timeout=10000)
            await page.wait_for_timeout(800)
            await assert_not_login_redirect(page)
            owner_id = owner_id_from_url(page.url)
            if owner_id is not None:
                owners.append({"name": name, "owner_id": owner_id})
            else:
                warnings_list.append(f"负责人 '{name}' 点击后 URL 中未解析到 owner_id")
        except LoginRequiredError:
            raise
        except Exception as exc:
            warnings_list.append(f"负责人 '{name}' 缓存刷新失败: {exc}")
            try:
                await page.mouse.click(0, 0)
                await page.wait_for_timeout(300)
            except Exception:
                pass

    save_assignee_cache(owners, path=cache_path)
    if warnings_list:
        import warnings as py_warnings

        for warning in warnings_list:
            py_warnings.warn(warning, RuntimeWarning)
    return owners


def build_batch_range_output(
    *,
    monthly_outputs,
    windows,
    query_scope,
    mode,
    depth,
    max_detail,
    category,
    assignee,
    buyer_name,
    batch_root,
    chunk_days,
    resume,
):
    if len(monthly_outputs) == 1:
        return monthly_outputs[0]["payload"]

    month_payloads = [item["payload"] for item in monthly_outputs]
    range_inquiries = []
    for payload in month_payloads:
        range_inquiries.extend(payload.get("inquiries", []))

    merged_inquiries, aggregation = merge_inquiries_by_id(range_inquiries)
    skipped_stats = aggregate_skipped_stats(month_payloads)
    merged_metadata = merge_list_metadata(month_payloads)
    partial_months = [
        item["month_key"]
        for item in monthly_outputs
        if item.get("status") != "ok" or item.get("payload", {}).get("partial")
    ]
    if partial_months:
        merged_metadata["partial"] = True
        merged_metadata.setdefault("warnings", []).append(
            "跨月聚合包含部分结果月份: " + ", ".join(partial_months)
        )

    range_meta = build_batch_meta(
        role="range",
        batch_root=batch_root,
        chunk_days=chunk_days,
        resume=resume,
        source_shards=[item["monthly_path"] for item in monthly_outputs],
        aggregation=aggregation,
    )
    range_meta["window_count"] = len(windows)
    range_meta["month_count"] = len(monthly_outputs)
    range_meta["months"] = [
        {
            "month_key": item["month_key"],
            "status": item["status"],
            "monthly_path": item["monthly_path"],
            "manifest_path": item["manifest_path"],
        }
        for item in monthly_outputs
    ]

    return build_window_output_payload(
        inquiries=merged_inquiries,
        skipped_stats=skipped_stats,
        mode=mode,
        depth=depth,
        max_detail=max_detail,
        start_time=query_scope["output_start_time"],
        end_time=query_scope["output_end_time"],
        category=category,
        assignee=assignee,
        buyer_name=buyer_name,
        list_metadata=merged_metadata,
        scope_desc=None,
        batch_meta=range_meta,
    )


async def collect_window_inquiries(
    ctx,
    list_page,
    *,
    start_date,
    end_date,
    combine_owner_time_search,
    max_items,
    filter_category,
    filter_assignee,
    filter_buyer_name,
    owner_id,
    max_detail,
    depth,
    no_detail,
    pre_list_warnings,
    long_advanced_window_mode=False,
    clamp_future_end=True,
):
    inquiries, skipped_stats, list_metadata = await scrape_list(
        list_page,
        max_items=max_items,
        start_date=start_date,
        end_date=end_date,
        filter_category=filter_category,
        filter_assignee=filter_assignee,
        filter_buyer_name=filter_buyer_name,
        owner_id=owner_id,
        combine_owner_time_search=combine_owner_time_search,
        clamp_future_end=clamp_future_end,
    )
    if pre_list_warnings:
        list_metadata.setdefault("warnings", []).extend(pre_list_warnings)

    if not inquiries:
        return inquiries, skipped_stats, list_metadata

    detail_tasks = []
    detail_semaphore = asyncio.Semaphore(TABS)
    detail_tab_pace_lock = asyncio.Lock()
    for idx, inq in enumerate(inquiries):
        if no_detail:
            inq["detail"] = None
            inq["detail_skipped"] = True
            inq["detail_skip_reason"] = "max-detail=0, skipping detail"
            inq["scrape_status"] = "skipped"
            inq["first_reply_time"] = None
        elif max_detail is not None and idx >= max_detail:
            inq["detail"] = None
            inq["detail_skipped"] = True
            inq["detail_skip_reason"] = f"exceeded max-detail limit ({max_detail})"
            inq["scrape_status"] = "skipped"
            inq["first_reply_time"] = None
        else:
            detail_tasks.append(
                asyncio.create_task(
                    scrape_detail_item(
                        ctx,
                        inq,
                        depth,
                        detail_semaphore,
                        detail_tab_pace_lock=detail_tab_pace_lock,
                        long_window_cleanup=long_advanced_window_mode,
                    )
                )
            )

    if detail_tasks:
        await asyncio.gather(*detail_tasks)
        await repair_failed_detail_items(
            ctx,
            inquiries,
            depth,
            detail_semaphore,
            detail_tab_pace_lock=detail_tab_pace_lock,
            long_window_cleanup=long_advanced_window_mode,
        )
        await repair_empty_detail_items_after_delay(
            ctx,
            inquiries,
            depth,
            detail_semaphore,
            detail_tab_pace_lock=detail_tab_pace_lock,
            long_window_cleanup=long_advanced_window_mode,
        )

    return inquiries, skipped_stats, list_metadata


def build_window_output_payload(
    *,
    inquiries,
    skipped_stats,
    mode,
    depth,
    max_detail,
    category,
    assignee,
    buyer_name,
    list_metadata,
    scope_desc,
    batch_meta,
    start_time=None,
    end_time=None,
    window_meta=None,
):
    if inquiries:
        return build_output_data(
            inquiries=inquiries,
            skipped_stats=skipped_stats,
            mode=mode,
            depth=depth,
            max_detail=max_detail,
            start_time=start_time,
            end_time=end_time,
            category=category,
            assignee=assignee,
            buyer_name=buyer_name,
            list_metadata=list_metadata,
            scope_desc=scope_desc,
            batch_meta=batch_meta,
            window_meta=window_meta,
        )

    return json.loads(
        format_empty_output(
            skipped_stats=skipped_stats,
            list_metadata=list_metadata,
            start_time=start_time,
            end_time=end_time,
            category=category,
            assignee=assignee,
            buyer_name=buyer_name,
            mode=mode,
            scope_desc=scope_desc,
            batch_meta=batch_meta,
            window_meta=window_meta,
        )
    )


async def main():
    parser = build_parser()
    args = parser.parse_args()
    output_json_path = resolve_output_json_path(args.buyer_name)

    if args.clear_login_state:
        cleared = clear_cookies(chrome_profile_path())
        try:
            assignee_cache_status = clear_assignee_cache()
        except OSError as exc:
            print(format_error_output("Assignee cache clear failed", str(exc)))
            sys.exit(1)
        print(
            json.dumps(
                {
                    "status": "ok",
                    "login_state": "cleared" if cleared else "missing",
                    "assignee_cache": assignee_cache_status,
                },
                ensure_ascii=False,
                indent=2,
            )
        )
        return

    try:
        start_time = parse_time_arg(args.start_time, "--start-time")
        end_time = parse_time_arg(args.end_time, "--end-time")
        window_basis = resolve_window_basis(args.start_time, args.end_time)
        clamp_future_end = window_basis != "timestamp"
        time_window = resolve_time_window(start_time, end_time, clamp_future_end=clamp_future_end)
    except ValueError as exc:
        print(format_error_output(str(exc), "Use YYYY-MM-DD, 10-digit seconds, or 13-digit milliseconds."))
        sys.exit(1)

    window_meta = build_window_meta(args.start_time, args.end_time, time_window)
    args.buyer_name = (args.buyer_name or "").strip()
    query_scope = resolve_query_scope(time_window, args.buyer_name)

    if args.depth is None:
        within_7_days = False
        window_start = query_scope["window_start"]
        if args.mode == "new":
            days_diff = (datetime.now() - window_start).days
            within_7_days = 0 <= days_diff <= 7
        args.depth = "full" if within_7_days else "normal"

    if args.max is not None and args.max <= 0:
        print(format_error_output(f"--max 参数错误: {args.max}", "必须为正整数"))
        sys.exit(1)

    if args.max_detail is not None and args.max_detail < 0:
        print(format_error_output(f"--max-detail 参数错误: {args.max_detail}", "必须为非负整数"))
        sys.exit(1)

    effective_max_detail, output_max_detail = resolve_max_detail_settings(
        args.max_detail,
        query_scope["is_explicit_window"],
    )
    effective_max_items = resolve_max_items_setting(
        args.max,
        query_scope["is_explicit_window"],
    )

    if args.chunk_days <= 0:
        print(format_error_output(f"--chunk-days 参数错误: {args.chunk_days}", "必须为正整数"))
        sys.exit(1)

    no_detail = effective_max_detail == 0
    depth = None if no_detail else args.depth
    batch_mode = should_enable_batch_mode(query_scope, args.buyer_name)
    long_advanced_window_mode = should_pace_detail_tabs(query_scope)

    try:
        launched_by_us = ensure_cdp_ready()
    except Exception as exc:
        print(format_error_output(f"CDP自动启动失败 (localhost:9222): {exc}", manual_cdp_hint()))
        sys.exit(1)

    pw = await async_playwright().start()
    try:
        browser = await connect_cdp(pw)
    except Exception as exc:
        print(format_error_output(f"CDP连接失败 (localhost:9222): {exc}", manual_cdp_hint()))
        await pw.stop()
        sys.exit(1)

    contexts = browser.contexts
    if not contexts:
        try:
            await pace_before_tab_open()
            await browser.new_page()
            contexts = browser.contexts
            if not contexts:
                raise RuntimeError("自动创建页面后仍没有可用的浏览器上下文")
        except Exception as exc:
            print(format_error_output("Chrome 没有已打开的标签页", f"自动创建页面失败: {exc}"))
            await maybe_close_browser(browser, launched_by_us)
            await pw.stop()
            sys.exit(1)

    ctx = contexts[0]
    profile_path = chrome_profile_path()
    await load_cookies(ctx, profile_path)

    if args.list_assignees:
        list_page = await throttled_new_page(ctx)
        keep_browser_open = False
        try:
            assignees = await refresh_assignee_cache(list_page)
            names = [item["name"] for item in assignees]
            print(json.dumps({"status": "ok", "count": len(names), "assignees": names}, ensure_ascii=False, indent=2))
        except LoginRequiredError as exc:
            keep_browser_open = True
            await abort_for_login_required(exc, pw)
        except Exception as exc:
            assignees = load_assignee_cache()
            names = [item["name"] for item in assignees]
            print(
                json.dumps(
                    {
                        "status": "fallback",
                        "count": len(names),
                        "assignees": names,
                        "warning": f"缓存刷新失败，使用本地缓存: {exc}",
                    },
                    ensure_ascii=False,
                    indent=2,
                )
            )
        finally:
            if not keep_browser_open:
                await list_page.close()
                await maybe_close_browser(browser, launched_by_us)
                await pw.stop()
        return

    list_page = await throttled_new_page(ctx)
    pre_list_warnings = []

    try:
        assignees, assignee_cache_warning = await get_assignees_for_scrape(list_page)
    except LoginRequiredError as exc:
        await abort_for_login_required(exc, pw)
    if assignee_cache_warning:
        pre_list_warnings.append(assignee_cache_warning)

    owner_id = find_assignee_owner_id(args.assignee, assignees) if args.assignee else None
    if args.assignee and owner_id is None:
        try:
            owner_id = await resolve_target_owner_id(list_page, args.assignee)
        except LoginRequiredError as exc:
            await abort_for_login_required(exc, pw)
        if owner_id is None:
            pre_list_warnings.append(
                f"未能获取业务员 '{args.assignee}' 的 owner_id，将使用客户端文本过滤，结果可能不完整。"
            )

    time_window_search = query_scope["query_start"] is not None and query_scope["query_end"] is not None
    combine_owner_time_search = time_window_search and ENABLE_ADVANCED_TIME_OWNER_SEARCH
    use_owner_url_filter = owner_id is not None and not args.buyer_name and (
        not time_window_search or combine_owner_time_search
    )
    if use_owner_url_filter:
        try:
            owner_id, owner_validation_warning = await validate_owner_id_for_url_filter(
                list_page,
                args.assignee,
                owner_id,
            )
        except LoginRequiredError as exc:
            await abort_for_login_required(exc, pw)
        if owner_validation_warning:
            pre_list_warnings.append(owner_validation_warning)
        use_owner_url_filter = owner_id is not None and not args.buyer_name and (
            not time_window_search or combine_owner_time_search
        )

    list_owner_id = owner_id if use_owner_url_filter else None
    list_filter_assignee = "" if use_owner_url_filter else args.assignee
    if args.buyer_name and owner_id is not None:
        pre_list_warnings.append(
            "Buyer-name search does not combine owner in the URL; using local assignee filtering."
        )
    if time_window_search and owner_id is not None and not combine_owner_time_search:
        pre_list_warnings.append(
            "Time-window search does not combine owner in the URL by default; using local assignee filtering."
        )

    if batch_mode:
        windows = build_chunk_windows(query_scope["window_start"], query_scope["window_end"], args.chunk_days)
        windows_by_month = defaultdict(list)
        payloads_by_month = defaultdict(list)
        source_shards_by_month = defaultdict(list)
        failures_by_month = defaultdict(list)

        for window in windows:
            month_key = month_key_for_window(window)
            shard_path = raw_shard_path(args.batch_root, window)

            window_state = {
                "window_id": window.window_id,
                "start": window.start.strftime("%Y-%m-%d"),
                "end": window.end.strftime("%Y-%m-%d"),
                "status": "pending",
                "source_file": shard_path,
            }
            windows_by_month[month_key].append(window_state)

            if args.resume and os.path.exists(shard_path):
                try:
                    shard_payload = read_json(shard_path)
                    payloads_by_month[month_key].append(shard_payload)
                    source_shards_by_month[month_key].append(shard_path)
                    window_state["status"] = "ok"
                    window_state["resume_hit"] = True
                    window_state["total"] = shard_payload.get("total", 0)
                    continue
                except Exception as exc:
                    window_state["status"] = "missing"
                    window_state["error"] = f"resume read failed: {exc}"
                    failures_by_month[month_key].append({"window_id": window.window_id, "error": str(exc)})
                    continue

            try:
                inquiries, skipped_stats, list_metadata = await collect_window_inquiries(
                    ctx,
                    list_page,
                    start_date=window.start,
                    end_date=window.end,
                    combine_owner_time_search=combine_owner_time_search,
                    max_items=effective_max_items,
                    filter_category=args.category,
                    filter_assignee=list_filter_assignee,
                    filter_buyer_name=args.buyer_name,
                    owner_id=list_owner_id,
                    max_detail=effective_max_detail,
                    depth=depth,
                    no_detail=no_detail,
                    pre_list_warnings=pre_list_warnings,
                    long_advanced_window_mode=long_advanced_window_mode,
                    clamp_future_end=clamp_future_end,
                )
                shard_meta = build_batch_meta(
                    role="shard",
                    batch_root=args.batch_root,
                    window=window,
                    chunk_days=args.chunk_days,
                    resume=args.resume,
                )
                shard_payload = build_window_output_payload(
                    inquiries=inquiries,
                    skipped_stats=skipped_stats,
                    mode=args.mode,
                    depth=depth,
                    max_detail=output_max_detail,
                    start_time=window.start.strftime("%Y-%m-%d"),
                    end_time=window.end.strftime("%Y-%m-%d"),
                    category=args.category,
                    assignee=args.assignee,
                    buyer_name=args.buyer_name,
                    list_metadata=list_metadata,
                    scope_desc=window.scope_desc,
                    batch_meta=shard_meta,
                )
                write_json_atomic(shard_path, shard_payload)

                payloads_by_month[month_key].append(shard_payload)
                source_shards_by_month[month_key].append(shard_path)
                window_state["status"] = "ok"
                window_state["total"] = shard_payload.get("total", 0)
            except LoginRequiredError as exc:
                await abort_for_login_required(exc, pw)
            except ListScrapeError as exc:
                window_state["status"] = "error"
                window_state["error"] = str(exc)
                failures_by_month[month_key].append({"window_id": window.window_id, "error": str(exc)})
            except Exception as exc:
                window_state["status"] = "error"
                window_state["error"] = f"unexpected: {exc}"
                failures_by_month[month_key].append(
                    {"window_id": window.window_id, "error": f"unexpected: {exc}"}
                )

        await save_cookies(ctx, profile_path)
        await list_page.close()
        await maybe_close_browser(browser, launched_by_us)
        await pw.stop()

        monthly_outputs = []
        for month_key in sorted(windows_by_month.keys()):
            paths = batch_root_paths(args.batch_root, month_key)
            monthly_path = paths["monthly_path"]
            manifest_path = paths["manifest_path"]
            month_payloads = payloads_by_month.get(month_key, [])
            source_shards = source_shards_by_month.get(month_key, [])

            month_inquiries = []
            for payload in month_payloads:
                month_inquiries.extend(payload.get("inquiries", []))

            merged_inquiries, aggregation = merge_inquiries_by_id(month_inquiries)
            skipped_stats = aggregate_skipped_stats(month_payloads)
            merged_metadata = merge_list_metadata(month_payloads)
            month_failures = failures_by_month.get(month_key, [])
            if month_failures:
                merged_metadata.setdefault("warnings", []).append(
                    f"共有 {len(month_failures)} 个分片执行失败，月度聚合为部分结果。"
                )
                merged_metadata["partial"] = True

            monthly_meta = build_batch_meta(
                role="monthly",
                batch_root=args.batch_root,
                month_key=month_key,
                chunk_days=args.chunk_days,
                resume=args.resume,
                source_shards=source_shards,
                aggregation=aggregation,
                manifest_path=manifest_path,
                monthly_path=monthly_path,
            )
            month_start_time, month_end_time = advanced_month_time_bounds(windows_by_month[month_key])
            monthly_payload = build_window_output_payload(
                inquiries=merged_inquiries,
                skipped_stats=skipped_stats,
                mode=args.mode,
                depth=depth,
                max_detail=output_max_detail,
                start_time=month_start_time,
                end_time=month_end_time,
                category=args.category,
                assignee=args.assignee,
                buyer_name=args.buyer_name,
                list_metadata=merged_metadata,
                scope_desc=f"{month_key} 月度聚合",
                batch_meta=monthly_meta,
            )
            write_json_atomic(monthly_path, monthly_payload)

            manifest_status = "ok" if not month_failures else "partial"
            manifest = build_manifest(
                report_month=month_key,
                batch_root=args.batch_root,
                chunk_days=args.chunk_days,
                windows=windows_by_month[month_key],
                source_shards=source_shards,
                monthly_path=monthly_path,
                status=manifest_status,
                failures=month_failures or None,
                aggregation=aggregation,
            )
            write_json_atomic(manifest_path, manifest)
            monthly_outputs.append(
                {
                    "month_key": month_key,
                    "status": manifest_status,
                    "monthly_path": monthly_path,
                    "manifest_path": manifest_path,
                    "payload": monthly_payload,
                }
            )

        if not monthly_outputs:
            print(format_error_output("批处理未产出任何月聚合结果"))
            sys.exit(1)

        final_output_obj = build_batch_range_output(
            monthly_outputs=monthly_outputs,
            windows=windows,
            query_scope=query_scope,
            mode=args.mode,
            depth=depth,
            max_detail=output_max_detail,
            category=args.category,
            assignee=args.assignee,
            buyer_name=args.buyer_name,
            batch_root=args.batch_root,
            chunk_days=args.chunk_days,
            resume=args.resume,
        )
        emit_success_output(
            json.dumps(final_output_obj, ensure_ascii=False, indent=2),
            args.stdout_json,
            output_path=output_json_path,
        )
        return

    try:
        inquiries, skipped_stats, list_metadata = await collect_window_inquiries(
            ctx,
            list_page,
            start_date=query_scope["query_start"],
            end_date=query_scope["query_end"],
            combine_owner_time_search=combine_owner_time_search,
            max_items=effective_max_items,
            filter_category=args.category,
            filter_assignee=list_filter_assignee,
            filter_buyer_name=args.buyer_name,
            owner_id=list_owner_id,
            max_detail=effective_max_detail,
            depth=depth,
            no_detail=no_detail,
            pre_list_warnings=pre_list_warnings,
            long_advanced_window_mode=long_advanced_window_mode,
            clamp_future_end=clamp_future_end,
        )
    except LoginRequiredError as exc:
        await abort_for_login_required(exc, pw)
    except ListScrapeError as exc:
        print(format_error_output(str(exc)))
        await list_page.close()
        await maybe_close_browser(browser, launched_by_us)
        await pw.stop()
        sys.exit(1)

    await save_cookies(ctx, profile_path)

    if not inquiries:
        output_text = format_empty_output(
            skipped_stats,
            list_metadata,
            start_time=query_scope["output_start_time"],
            end_time=query_scope["output_end_time"],
            category=args.category,
            assignee=args.assignee,
            buyer_name=args.buyer_name,
            mode=args.mode,
            window_meta=window_meta,
        )
        await list_page.close()
        await maybe_close_browser(browser, launched_by_us)
        await pw.stop()
        emit_success_output(output_text, args.stdout_json, output_path=output_json_path)
        return

    await list_page.close()
    await maybe_close_browser(browser, launched_by_us)
    await pw.stop()

    output_text = format_output(
        inquiries=inquiries,
        skipped_stats=skipped_stats,
        mode=args.mode,
        depth=depth,
        max_detail=output_max_detail,
        start_time=query_scope["output_start_time"],
        end_time=query_scope["output_end_time"],
        category=args.category,
        assignee=args.assignee,
        buyer_name=args.buyer_name,
        list_metadata=list_metadata if inquiries else None,
        window_meta=window_meta,
    )
    emit_success_output(output_text, args.stdout_json, output_path=output_json_path)


if __name__ == "__main__":
    asyncio.run(main())
