"""Accio Work 历史日期抓取管理入口。

这个脚本专门给 Accio Work 调用：长时间抓取只启动一次，后续只轮询状态，
避免 bash 工具超时后重复启动多个 scraper，导致 Chrome tab 越积越多。
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
import platform
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import time
import struct
from datetime import datetime, timedelta, timezone
from urllib.parse import quote
from urllib.request import urlopen
from zoneinfo import ZoneInfo

from config import DEFAULT_EXPORT_MARKDOWN, DEFAULT_EXPORT_PDF, EXPORT_MARKDOWN_ENV, EXPORT_PDF_ENV


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")
RUNTIME_DIR = os.path.join(PROJECT_DIR, "outputs", "runtime")
LOG_DIR = os.path.join(RUNTIME_DIR, "logs")
STATE_PATH = os.path.join(RUNTIME_DIR, "accio_scrape_state.json")
START_LOCK_NAME = "accio_scrape_manager.lock"
REPORT_FINALIZE_LOCK_NAME = "accio_report_finalize.lock"
REPORT_OUTPUT_SUBDIR = "alibaba-inquiry-analyst-output"
REPORT_OUTPUT_DIR_ENV = "ACCIO_REPORT_OUTPUT_DIR"
ACCIO_PROJECT_DIR_ENV = "ACCIO_PROJECT_DIR"
APPLE_TIME_SERVER = "time.apple.com"
NTP_PORT = 123
NTP_PACKET_SIZE = 48
NTP_EPOCH_DELTA_SECONDS = 2208988800
NTP_TIMEOUT_SECONDS = 1.5
CHINA_TZ = ZoneInfo("Asia/Shanghai")
LA_TZ = ZoneInfo("America/Los_Angeles")
AUTO_COMPARE_TZS = (
    ("America/Los_Angeles", LA_TZ),
    ("Asia/Shanghai", CHINA_TZ),
)
CDP_ENDPOINTS = (
    "http://127.0.0.1:9222",
    "http://localhost:9222",
)
DEFAULT_PAGE_SIZE = 20
DEFAULT_ESTIMATED_INQUIRIES_PER_DAY = 20
DEFAULT_WATCH_INTERVAL_SECONDS = 60
DEFAULT_STATUS_POLL_SECONDS = 60
DEFAULT_STATUS_RUNNING_TAIL_LINES = 1
DEFAULT_WAIT_MAX_SECONDS = 90
WAIT_TIMEOUT_BUFFER_SECONDS = 30
DEFAULT_WAIT_POLL_INTERVAL_SECONDS = 5
MIN_DYNAMIC_WATCH_MINUTES = 20
MAX_DYNAMIC_WATCH_MINUTES = 240
WATCH_BUFFER_MINUTES = 10
SHARD_TIME_SAFETY_FACTOR = 1.5
DEGRADED_MISSING_DETAIL_LIMIT = 15
MODE_VALUES = ("new",)
LOGIN_REQUIRED_MARKER = "LOGIN_REQUIRED"
LOGIN_REQUIRED_MESSAGE = (
    "已跳转到 login.alibaba.com，当前浏览器未登录阿里国际站。"
    "已取消抓取；请在新打开的浏览器里完成登录后重新运行查询。"
)
LOGIN_REQUIRED_NEXT_STEP = "login_in_browser_then_retry"
SHORT_TIMEOUT_SECONDS = 120
SHORT_TIMEOUT_ENV_NAMES = (
    "ACCIO_BASH_TIMEOUT_SECONDS",
    "ACCIO_TOOL_TIMEOUT_SECONDS",
    "AIOS_TOOL_TIMEOUT_SECONDS",
)
WATCH_TOOL_TIMEOUT_BUFFER_SECONDS = 60
REPORT_FINALIZE_LOCK_WAIT_SECONDS = 5
REPORT_FINALIZE_LOCK_POLL_SECONDS = 0.5
REPORT_TEXT_RESULT_LIMIT = 12000
REPORT_TEXT_TTL_SECONDS = 72 * 60 * 60
FATAL_RUNTIME_LOG_MARKERS = (
    "FATAL ERROR",
    "JavaScript heap out of memory",
    "Allocation failed - JavaScript heap out of memory",
    "Connection closed while reading from the driver",
)
ERROR_LOG_MARKERS = (
    *FATAL_RUNTIME_LOG_MARKERS,
    "Traceback (most recent call last)",
    "TargetClosedError",
    "Target page, context or browser has been closed",
    "Page crashed",
    "browser has been closed",
    "Browser closed",
    "Connection closed",
    "Chrome/CDP",
    "Chrome tab may be crashed",
    "List page navigation timed out",
    "List page navigation failed",
    "OOM",
    "out of memory",
    LOGIN_REQUIRED_MARKER,
    "login.alibaba.com",
    '"status": "error"',
)


def now_text():
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")


def fetch_apple_ntp_utc(timeout=NTP_TIMEOUT_SECONDS):
    packet = b"\x1b" + (b"\0" * (NTP_PACKET_SIZE - 1))
    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as sock:
        sock.settimeout(timeout)
        sock.sendto(packet, (APPLE_TIME_SERVER, NTP_PORT))
        data, _ = sock.recvfrom(NTP_PACKET_SIZE)
    if len(data) < NTP_PACKET_SIZE:
        raise OSError("incomplete NTP response")
    seconds = struct.unpack("!I", data[40:44])[0]
    return datetime.fromtimestamp(seconds - NTP_EPOCH_DELTA_SECONDS, timezone.utc)


def current_utc_now():
    try:
        return fetch_apple_ntp_utc()
    except OSError:
        return datetime.now(timezone.utc)


def safe_name(value):
    return str(value).replace("-", "").replace(":", "").replace(" ", "_")


def print_json(payload, exit_code=0):
    print(json.dumps(payload, ensure_ascii=False, indent=2))
    raise SystemExit(exit_code)


def parse_bool_env(value):
    if value is None:
        return None
    text = str(value).strip().lower()
    if text in ("1", "true", "yes", "y", "on"):
        return True
    if text in ("0", "false", "no", "n", "off"):
        return False
    return None


def resolve_export_pdf_enabled(environ=None):
    env_value = parse_bool_env((environ or os.environ).get(EXPORT_PDF_ENV))
    if env_value is not None:
        return env_value
    return bool(DEFAULT_EXPORT_PDF)


def resolve_export_markdown_enabled(environ=None):
    env_value = parse_bool_env((environ or os.environ).get(EXPORT_MARKDOWN_ENV))
    if env_value is not None:
        return env_value
    return bool(DEFAULT_EXPORT_MARKDOWN)


def ensure_runtime_dirs():
    os.makedirs(LOG_DIR, exist_ok=True)


def cleanup_old_report_text_files(now=None, ttl_seconds=REPORT_TEXT_TTL_SECONDS):
    try:
        if not os.path.isdir(RUNTIME_DIR):
            return {"deleted": 0, "errors": []}
        cutoff = time.time() if now is None else now
        cutoff -= ttl_seconds
        deleted = 0
        errors = []
        for name in os.listdir(RUNTIME_DIR):
            if not name.endswith("_report_text.txt"):
                continue
            path = os.path.join(RUNTIME_DIR, name)
            try:
                if os.path.isfile(path) and os.path.getmtime(path) < cutoff:
                    os.remove(path)
                    deleted += 1
            except OSError as exc:
                errors.append({"path": path, "error": str(exc)})
        return {"deleted": deleted, "errors": errors}
    except Exception:
        return {"deleted": 0, "errors": ["report_text_cleanup_failed"]}


def normalize_output_dir(path):
    text = str(path or "").strip()
    if not text:
        return ""
    return os.path.abspath(os.path.expandvars(os.path.expanduser(text)))


def resolve_report_output_dir(args=None, state=None, environ=None):
    if environ is None:
        environ = os.environ
    cli_path = normalize_output_dir(getattr(args, "report_output_dir", "") if args else "")
    if cli_path:
        return {"path": cli_path, "source": "cli"}

    state_path = normalize_output_dir((state or {}).get("report_output_dir"))
    if state_path:
        return {
            "path": state_path,
            "source": (state or {}).get("report_output_dir_source") or "state",
        }

    env_report_dir = normalize_output_dir(environ.get(REPORT_OUTPUT_DIR_ENV))
    if env_report_dir:
        return {"path": env_report_dir, "source": "env_accio_report_output_dir"}

    env_project_dir = normalize_output_dir(environ.get(ACCIO_PROJECT_DIR_ENV))
    if env_project_dir:
        return {
            "path": os.path.join(env_project_dir, REPORT_OUTPUT_SUBDIR),
            "source": "env_accio_project_dir",
        }

    return {"path": PROJECT_DIR, "source": "fallback_project_dir"}


def path_inside_dir(path, directory):
    if not path or not directory:
        return False
    try:
        candidate = os.path.normcase(os.path.abspath(path))
        target = os.path.normcase(os.path.abspath(directory))
        return os.path.commonpath([candidate, target]) == target
    except (OSError, ValueError):
        return False


def read_json(path):
    with open(path, encoding="utf-8") as handle:
        return json.load(handle)


def read_text_if_exists(path):
    if not path:
        return None
    try:
        with open(path, encoding="utf-8") as handle:
            return handle.read()
    except OSError:
        return None


def parse_date(value):
    return parse_time_bound(value)


def parse_time_bound(value):
    text = str(value or "").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"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"time format error: {text}")


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


def time_bounds_match(actual, expected):
    if actual == expected:
        return True
    try:
        return parse_time_bound(actual) == parse_time_bound(expected)
    except (TypeError, ValueError):
        return False


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

def parse_datetime_text(value):
    if not value:
        return None
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
        try:
            return datetime.strptime(str(value), fmt)
        except ValueError:
            continue
    return None


def write_json_atomic(path, payload):
    ensure_runtime_dirs()
    tmp_path = f"{path}.tmp"
    with open(tmp_path, "w", encoding="utf-8") as handle:
        json.dump(payload, handle, ensure_ascii=False, indent=2)
        handle.write("\n")
        handle.flush()
        os.fsync(handle.fileno())
    os.replace(tmp_path, path)


def load_state():
    if not os.path.exists(STATE_PATH):
        return None
    try:
        return read_json(STATE_PATH)
    except Exception as exc:
        return {"status": "state_error", "error": str(exc), "state_path": STATE_PATH}


def start_lock_path():
    return os.path.join(os.path.dirname(STATE_PATH), START_LOCK_NAME)


def read_start_lock(path):
    try:
        return read_json(path)
    except Exception as exc:
        return {"status": "lock_error", "error": str(exc), "lock_path": path}


def process_running(pid):
    if not pid:
        return False
    try:
        pid = int(pid)
    except (TypeError, ValueError):
        return False
    if pid <= 0:
        return False

    if platform.system() == "Windows":
        result = subprocess.run(
            ["tasklist", "/FI", f"PID eq {pid}", "/FO", "CSV", "/NH"],
            capture_output=True,
            text=True,
            errors="ignore",
        )
        return result.returncode == 0 and str(pid) in result.stdout

    try:
        os.kill(pid, 0)
        return True
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except OSError:
        return False


def acquire_start_lock(key):
    path = start_lock_path()
    os.makedirs(os.path.dirname(path), exist_ok=True)
    payload = {
        "pid": os.getpid(),
        "created_at": now_text(),
        "key": key,
    }

    for _ in range(2):
        try:
            fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        except FileExistsError:
            existing = read_start_lock(path)
            existing_pid = existing.get("pid") if isinstance(existing, dict) else None
            if existing_pid and not process_running(existing_pid):
                try:
                    os.remove(path)
                except FileNotFoundError:
                    pass
                except OSError:
                    return {"acquired": False, "path": path, "lock": existing}
                continue
            return {"acquired": False, "path": path, "lock": existing}

        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(payload, handle, ensure_ascii=False, indent=2)
            handle.write("\n")
            handle.flush()
            os.fsync(handle.fileno())
        return {"acquired": True, "path": path, "lock": payload}

    return {"acquired": False, "path": path, "lock": read_start_lock(path)}


def release_start_lock(lock):
    if not lock or not lock.get("acquired"):
        return
    path = lock.get("path")
    if not path or not os.path.exists(path):
        return
    existing = read_start_lock(path)
    if isinstance(existing, dict) and existing.get("pid") == os.getpid():
        try:
            os.remove(path)
        except FileNotFoundError:
            pass


def project_lock_hash():
    source = os.path.abspath(STATE_PATH or PROJECT_DIR)
    return hashlib.sha1(source.encode("utf-8", errors="ignore")).hexdigest()[:12]


def report_finalize_lock_paths():
    runtime_path = os.path.join(os.path.dirname(STATE_PATH), REPORT_FINALIZE_LOCK_NAME)
    temp_path = os.path.join(
        tempfile.gettempdir(),
        "accio-inquiry-analyst",
        f"report-finalize-{project_lock_hash()}.lock",
    )
    return [
        {"scope": "runtime", "path": runtime_path},
        {"scope": "temp", "path": temp_path},
    ]


def report_finalize_lock_path():
    return report_finalize_lock_paths()[0]["path"]


def read_report_finalize_lock(path):
    try:
        return read_json(path)
    except Exception as exc:
        return {"status": "lock_error", "error": str(exc), "lock_path": path}


def try_acquire_report_finalize_lock(path, scope):
    try:
        os.makedirs(os.path.dirname(path), exist_ok=True)
    except OSError as exc:
        return {
            "acquired": False,
            "path": path,
            "scope": scope,
            "retryable": True,
            "lock": {"status": "lock_error", "error": str(exc), "lock_path": path, "scope": scope},
        }
    payload = {
        "pid": os.getpid(),
        "created_at": now_text(),
        "scope": scope,
    }

    for _ in range(2):
        try:
            fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
        except FileExistsError:
            existing = read_report_finalize_lock(path)
            existing_pid = existing.get("pid") if isinstance(existing, dict) else None
            if existing_pid and not process_running(existing_pid):
                try:
                    os.remove(path)
                except FileNotFoundError:
                    pass
                except OSError:
                    return {"acquired": False, "path": path, "scope": scope, "retryable": True, "lock": existing}
                continue
            if isinstance(existing, dict):
                existing.setdefault("scope", scope)
            return {"acquired": False, "path": path, "scope": scope, "retryable": False, "lock": existing}
        except OSError as exc:
            return {
                "acquired": False,
                "path": path,
                "scope": scope,
                "retryable": True,
                "lock": {"status": "lock_error", "error": str(exc), "lock_path": path, "scope": scope},
            }

        try:
            with os.fdopen(fd, "w", encoding="utf-8") as handle:
                json.dump(payload, handle, ensure_ascii=False, indent=2)
                handle.write("\n")
                handle.flush()
                os.fsync(handle.fileno())
        except OSError as exc:
            try:
                os.remove(path)
            except OSError:
                pass
            return {
                "acquired": False,
                "path": path,
                "scope": scope,
                "retryable": True,
                "lock": {"status": "lock_error", "error": str(exc), "lock_path": path, "scope": scope},
            }
        return {"acquired": True, "path": path, "scope": scope, "lock": payload}

    existing = read_report_finalize_lock(path)
    if isinstance(existing, dict):
        existing.setdefault("scope", scope)
    return {"acquired": False, "path": path, "scope": scope, "retryable": False, "lock": existing}


def acquire_report_finalize_lock():
    errors = []
    for candidate in report_finalize_lock_paths():
        result = try_acquire_report_finalize_lock(candidate["path"], candidate["scope"])
        if result.get("acquired"):
            return result
        if not result.get("retryable"):
            return result
        errors.append({
            "scope": result.get("scope"),
            "path": result.get("path"),
            "lock": result.get("lock"),
        })
    return {
        "acquired": False,
        "path": None,
        "scope": "all",
        "retryable": True,
        "lock": {
            "status": "lock_error",
            "error": "all finalize lock paths failed",
            "errors": errors,
        },
    }


def release_report_finalize_lock(lock):
    if not lock or not lock.get("acquired"):
        return
    path = lock.get("path")
    if not path or not os.path.exists(path):
        return
    existing = read_report_finalize_lock(path)
    if isinstance(existing, dict) and existing.get("pid") == os.getpid():
        try:
            os.remove(path)
        except FileNotFoundError:
            pass
        except OSError:
            pass


def detect_short_timeout_environment(environ=None):
    environ = environ or os.environ
    for name in SHORT_TIMEOUT_ENV_NAMES:
        value = environ.get(name)
        if value is None:
            continue
        try:
            seconds = float(value)
        except (TypeError, ValueError):
            continue
        return {
            "short_timeout": seconds <= SHORT_TIMEOUT_SECONDS,
            "reason": f"{name}={value}",
            "timeout_seconds": seconds,
        }

    if environ.get("AIOS_SHELL") == "1":
        return {"short_timeout": True, "reason": "AIOS_SHELL=1"}
    if environ.get("AIOS_RUNTIME_ORIGINATOR") == "AIOS Desktop Runtime":
        return {"short_timeout": True, "reason": "AIOS_RUNTIME_ORIGINATOR=AIOS Desktop Runtime"}
    return {"short_timeout": False, "reason": "not_detected"}


def explicit_tool_timeout(environ=None):
    environ = environ or os.environ
    for name in SHORT_TIMEOUT_ENV_NAMES:
        value = environ.get(name)
        if value is None:
            continue
        try:
            seconds = float(value)
        except (TypeError, ValueError):
            return {
                "known": False,
                "reason": "invalid_timeout_env",
                "name": name,
                "value": value,
            }
        if seconds <= 0:
            return {
                "known": False,
                "reason": "invalid_timeout_env",
                "name": name,
                "value": value,
            }
        return {
            "known": True,
            "reason": f"{name}={value}",
            "name": name,
            "value": value,
            "timeout_seconds": seconds,
        }

    return {"known": False, "reason": "timeout_env_missing"}


def required_watch_timeout_seconds(max_minutes):
    return int(float(max_minutes) * 60 + WATCH_TOOL_TIMEOUT_BUFFER_SECONDS)


def resolve_wait_max_seconds(args, environ=None):
    requested = getattr(args, "max_wait_seconds", None)
    if requested is not None:
        return max(1, int(requested))

    timeout = explicit_tool_timeout(environ)
    if timeout.get("known"):
        safe_seconds = int(float(timeout["timeout_seconds"]) - WAIT_TIMEOUT_BUFFER_SECONDS)
        return max(5, safe_seconds)
    return DEFAULT_WAIT_MAX_SECONDS


def terminate_process(pid):
    if not process_running(pid):
        return False
    pid = int(pid)
    if platform.system() == "Windows":
        subprocess.run(
            ["taskkill", "/PID", str(pid), "/T", "/F"],
            capture_output=True,
            text=True,
            errors="ignore",
        )
        return True

    killpg = getattr(os, "killpg", None)
    if killpg:
        try:
            # Managed POSIX scraper processes are launched with start_new_session=True,
            # so the child PID is also the process-group id.
            killpg(pid, signal.SIGTERM)
            return True
        except ProcessLookupError:
            return False
        except PermissionError:
            return False
        except OSError:
            pass

    os.kill(pid, signal.SIGTERM)
    return True


def is_alibaba_message_tab(url):
    text = str(url or "")
    if "message.alibaba.com" not in text:
        return False
    return "/message/" in text or "maDetail.htm" in text or "default.htm#feedback" in text


def close_alibaba_message_tabs():
    closed = []
    errors = []
    for base in CDP_ENDPOINTS:
        try:
            with urlopen(f"{base}/json", timeout=2) as response:
                targets = json.loads(response.read().decode("utf-8"))
        except Exception as exc:
            errors.append(f"{base}: {exc}")
            continue

        for target in targets:
            url = target.get("url", "")
            target_id = target.get("id")
            if not target_id or not is_alibaba_message_tab(url):
                continue
            try:
                with urlopen(f"{base}/json/close/{quote(str(target_id), safe='')}", timeout=2):
                    pass
                closed.append({"id": target_id, "url": url})
            except Exception as exc:
                errors.append(f"close {target_id}: {exc}")
        break
    return {"closed_count": len(closed), "closed": closed, "errors": errors}


def request_from_args(args):
    return {
        "mode": args.mode,
        "start_time": args.start_time,
        "end_time": args.end_time,
        "depth": args.depth,
        "category": args.category or "",
        "assignee": args.assignee or "",
        "buyer_name": getattr(args, "buyer_name", "") or "",
        "max_detail": args.max_detail,
        "max": args.max,
        "chunk_days": args.chunk_days,
        "resume": bool(args.resume),
    }


def canonical_key(request):
    parts = [
        ("mode", request.get("mode")),
        ("start", request.get("start_time")),
        ("end", request.get("end_time")),
        ("depth", request.get("depth")),
        ("category", request.get("category") or ""),
        ("assignee", request.get("assignee") or ""),
        ("buyer_name", request.get("buyer_name") or ""),
        ("max_detail", request.get("max_detail")),
        ("max", request.get("max")),
        ("chunk_days", request.get("chunk_days")),
        ("resume", bool(request.get("resume"))),
    ]
    return "|".join(f"{key}={value}" for key, value in parts)


def build_scrape_command(request):
    command = [
        sys.executable,
        "-X",
        "utf8",
        os.path.join("scripts", "scrape_inquiries.py"),
        "--mode",
        request["mode"],
        "--chunk-days",
        str(request["chunk_days"]),
    ]
    if request.get("start_time"):
        command.extend(["--start-time", request["start_time"]])
    if request.get("end_time"):
        command.extend(["--end-time", request["end_time"]])
    if request.get("depth"):
        command.extend(["--depth", request["depth"]])
    if request.get("category"):
        command.extend(["--category", request["category"]])
    if request.get("assignee"):
        command.extend(["--assignee", request["assignee"]])
    if request.get("buyer_name"):
        command.extend(["--buyer-name", request["buyer_name"]])
    if request.get("max_detail") is not None:
        command.extend(["--max-detail", str(request["max_detail"])])
    if request.get("max") is not None:
        command.extend(["--max", str(request["max"])])
    if request.get("resume"):
        command.append("--resume")
    return command


def launch_scrape_process(request, log_prefix="scrape"):
    ensure_runtime_dirs()
    log_name = "{prefix}_{start}_{end}_{stamp}.log".format(
        prefix=log_prefix,
        start=safe_name(request["start_time"]),
        end=safe_name(request["end_time"]),
        stamp=datetime.now().strftime("%Y%m%d_%H%M%S"),
    )
    log_path = os.path.join(LOG_DIR, log_name)
    command = build_scrape_command(request)
    env = os.environ.copy()
    env["PYTHONUTF8"] = "1"

    with open(log_path, "w", encoding="utf-8") as log_handle:
        log_handle.write(f"[{now_text()}] start: {' '.join(command)}\n")
        log_handle.flush()
        popen_kwargs = {}
        if platform.system() == "Windows":
            popen_kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
        else:
            # POSIX (macOS/Darwin, Linux): setsid() creates a new session so the
            # scraper survives the parent terminal lifecycle.
            popen_kwargs["start_new_session"] = True
        proc = subprocess.Popen(
            command,
            cwd=PROJECT_DIR,
            stdout=log_handle,
            stderr=subprocess.STDOUT,
            env=env,
            **popen_kwargs,
        )
    return {"proc": proc, "log_path": log_path, "command": command}


def epoch_ms(value):
    return int(value.timestamp() * 1000)


def parse_request_epoch_ms(value):
    text = str(value or "").strip()
    if not text:
        return None
    if text.isdigit():
        if len(text) == 13:
            return int(text)
        if len(text) == 10:
            return int(text) * 1000
        return None
    try:
        parsed = parse_time_bound(text)
    except ValueError:
        return None
    return epoch_ms(parsed.replace(tzinfo=CHINA_TZ))


def format_timestamp_ms(value):
    return str(int(value))


def auto_compare_snapshot_path(kind, request):
    source = canonical_key(request)
    digest = hashlib.sha1(source.encode("utf-8", errors="ignore")).hexdigest()[:12]
    return os.path.join(RUNTIME_DIR, f"auto_compare_{kind}_{digest}.json")


def copy_output_snapshot(target_path):
    ensure_runtime_dirs()
    shutil.copyfile(OUTPUT_JSON_PATH, target_path)
    return target_path


def restore_output_from_snapshot(source_path):
    if not source_path:
        return None
    shutil.copyfile(source_path, OUTPUT_JSON_PATH)
    return OUTPUT_JSON_PATH


def auto_compare_plan_for_request(request, utc_now=None):
    if not request.get("start_time") or not request.get("end_time"):
        return None
    if request.get("category") or request.get("assignee") or request.get("buyer_name"):
        return None
    if request.get("max") is not None:
        return None

    start_ms = parse_request_epoch_ms(request.get("start_time"))
    end_ms = parse_request_epoch_ms(request.get("end_time"))
    if start_ms is None or end_ms is None:
        return None

    utc_now = utc_now or current_utc_now()
    one_day_ms = 24 * 60 * 60 * 1000
    for tz_name, tz in AUTO_COMPARE_TZS:
        today = utc_now.astimezone(tz).replace(hour=0, minute=0, second=0, microsecond=0)
        for label, start in (("today", today), ("yesterday", today - timedelta(days=1))):
            end = start + timedelta(days=1)
            candidate_start_ms = epoch_ms(start)
            candidate_end_ms = epoch_ms(end)
            if start_ms == candidate_start_ms and end_ms == candidate_end_ms:
                compare_request = dict(request)
                compare_request["start_time"] = format_timestamp_ms(start_ms - one_day_ms)
                compare_request["end_time"] = format_timestamp_ms(end_ms - one_day_ms)
                compare_request["depth"] = "quick"
                compare_request["resume"] = False
                return {
                    "status": "pending",
                    "timezone": tz_name,
                    "matched_window": label,
                    "main_request": dict(request),
                    "compare_request": compare_request,
                    "main_snapshot_path": auto_compare_snapshot_path("main", request),
                    "compare_snapshot_path": auto_compare_snapshot_path("previous", compare_request),
                }
    return None


def expected_windows_for_request(request):
    start = parse_date(request["start_time"])
    end = parse_date(request["end_time"])
    chunk_days = int(request.get("chunk_days") or 1)
    windows = []
    cursor = start
    index = 0
    while cursor < end:
        month_boundary = (cursor.replace(day=1) + timedelta(days=32)).replace(day=1)
        window_end = min(cursor + timedelta(days=chunk_days), end, month_boundary)
        if chunk_days == 1:
            window_id = cursor.strftime("%Y-%m-%d")
        else:
            window_id = f"{cursor.strftime('%Y-%m-%d')}_to_{window_end.strftime('%Y-%m-%d')}"
        windows.append({
            "window_id": window_id,
            "start": cursor.strftime("%Y-%m-%d"),
            "end": window_end.strftime("%Y-%m-%d"),
            "month_key": cursor.strftime("%Y-%m"),
            "index": index,
        })
        cursor = window_end
        index += 1
    return windows


def raw_shard_path_for_window(window):
    return os.path.join(
        PROJECT_DIR,
        "outputs",
        "inquiries",
        "raw",
        window["month_key"],
        f"{window['window_id']}.json",
    )


def request_day_count(request):
    duration_seconds = (parse_date(request["end_time"]) - parse_date(request["start_time"])).total_seconds()
    return max(1, int((duration_seconds + 86399) // 86400))


def collect_shard_progress(state):
    request = (state or {}).get("request") or {}
    if not request.get("start_time") or not request.get("end_time"):
        return {
            "expected_windows": 0,
            "completed_windows": 0,
            "observed_total": 0,
            "estimated_total": 0,
            "estimated_pages": 0,
            "longest_shard_seconds": None,
            "remaining_windows": 0,
            "completed_shards": [],
        }

    windows = expected_windows_for_request(request)
    created_at = parse_datetime_text((state or {}).get("created_at"))
    resume = bool(request.get("resume"))
    completed = []
    completion_times = []
    observed_total = 0
    observed_detail_count = 0

    for window in windows:
        path = raw_shard_path_for_window(window)
        if not os.path.exists(path):
            continue
        mtime = datetime.fromtimestamp(os.path.getmtime(path))
        if created_at and not resume and mtime < created_at - timedelta(seconds=5):
            continue
        try:
            payload = read_json(path)
        except Exception:
            continue
        total = int(payload.get("total") or 0)
        detail_count = int(payload.get("detail_count") or 0)
        observed_total += total
        observed_detail_count += detail_count
        completion_times.append(mtime)
        completed.append({
            "window_id": window["window_id"],
            "path": path,
            "total": total,
            "detail_count": detail_count,
            "detail_skipped": int(payload.get("detail_skipped") or 0),
            "mtime": mtime.strftime("%Y-%m-%d %H:%M:%S"),
        })

    expected_count = len(windows)
    completed_count = len(completed)
    if completed_count:
        estimated_total = int(round(observed_total / completed_count * expected_count))
    else:
        estimated_total = request_day_count(request) * DEFAULT_ESTIMATED_INQUIRIES_PER_DAY

    longest_shard_seconds = None
    if created_at and completion_times:
        previous = created_at
        durations = []
        for mtime in sorted(completion_times):
            duration = (mtime - previous).total_seconds()
            if duration > 0:
                durations.append(duration)
            previous = mtime
        if durations:
            longest_shard_seconds = max(durations)

    return {
        "expected_windows": expected_count,
        "completed_windows": completed_count,
        "remaining_windows": max(0, expected_count - completed_count),
        "observed_total": observed_total,
        "observed_detail_count": observed_detail_count,
        "estimated_total": estimated_total,
        "estimated_pages": max(1, (estimated_total + DEFAULT_PAGE_SIZE - 1) // DEFAULT_PAGE_SIZE),
        "longest_shard_seconds": longest_shard_seconds,
        "completed_shards": completed,
    }


def elapsed_seconds_from_state(state):
    created_at = parse_datetime_text((state or {}).get("created_at"))
    if not created_at:
        return 0
    return max(0, int((datetime.now() - created_at).total_seconds()))


def dynamic_watch_minutes(state, progress):
    request = (state or {}).get("request") or {}
    days = request_day_count(request) if request.get("start_time") and request.get("end_time") else 1
    elapsed_minutes = elapsed_seconds_from_state(state) / 60

    longest = progress.get("longest_shard_seconds")
    if longest and progress.get("completed_windows"):
        remaining_minutes = progress.get("remaining_windows", 0) * longest * SHARD_TIME_SAFETY_FACTOR / 60
        minutes = elapsed_minutes + remaining_minutes + WATCH_BUFFER_MINUTES
    else:
        estimated_total = progress.get("estimated_total") or days * DEFAULT_ESTIMATED_INQUIRIES_PER_DAY
        estimated_pages = max(1, (estimated_total + DEFAULT_PAGE_SIZE - 1) // DEFAULT_PAGE_SIZE)
        minutes = WATCH_BUFFER_MINUTES + days * 6 + estimated_pages * 1.5

    return int(max(MIN_DYNAMIC_WATCH_MINUTES, min(MAX_DYNAMIC_WATCH_MINUTES, round(minutes + 0.5))))


def resolve_watch_minutes(raw_value, state, progress):
    if raw_value is None or str(raw_value).strip().lower() in {"", "auto", "dynamic", "0"}:
        return dynamic_watch_minutes(state, progress), True
    return int(raw_value), False


def inquiry_summary(data):
    inquiries = data.get("inquiries") or []
    return {
        "status": data.get("status"),
        "mode": data.get("mode"),
        "scope_desc": data.get("scope_desc"),
        "start_time": data.get("start_time"),
        "end_time": data.get("end_time"),
        "total": data.get("total", len(inquiries)),
        "detail_count": data.get("detail_count", 0),
        "detail_skipped": data.get("detail_skipped", 0),
        "failed_count": len(data.get("failed") or []),
        "partial": bool(data.get("partial")),
        "inquiry_count": len(inquiries),
    }


def validation_result(
    *,
    summary,
    complete,
    reportable,
    state,
    errors=None,
    blocking_errors=None,
    soft_errors=None,
    degraded=False,
    missing_detail_count=0,
):
    blocking_errors = list(blocking_errors or [])
    soft_errors = list(soft_errors or [])
    combined_errors = list(errors) if errors is not None else blocking_errors + soft_errors
    return {
        "complete": complete,
        "reportable": reportable,
        "degraded": degraded,
        "state": state,
        "errors": combined_errors,
        "blocking_errors": blocking_errors,
        "soft_errors": soft_errors,
        "missing_detail_count": missing_detail_count,
        "summary": summary,
    }


def detail_failure_key(item, index):
    inquiry_id = str(item.get("inquiry_id") or "").strip()
    return inquiry_id or f"row-{index}"


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


def summarize_missing_detail_ids(keys):
    if not keys:
        return ""
    preview = ", ".join(keys[:10])
    if len(keys) > 10:
        return f"{preview} ..."
    return preview


def validate_inquiries(request=None, path=OUTPUT_JSON_PATH):
    if not os.path.exists(path):
        return validation_result(
            summary={},
            complete=False,
            reportable=False,
            degraded=False,
            state="missing",
            blocking_errors=[f"未找到输出文件: {path}"],
        )

    try:
        data = read_json(path)
    except Exception as exc:
        return validation_result(
            summary={},
            complete=False,
            reportable=False,
            degraded=False,
            state="invalid_json",
            blocking_errors=[str(exc)],
        )

    blocking_errors = []
    soft_errors = []
    inquiries = data.get("inquiries")
    total = data.get("total")
    if data.get("status") != "ok":
        blocking_errors.append(f"status 不是 ok: {data.get('status')}")
    if request:
        if data.get("mode") != request.get("mode"):
            blocking_errors.append(f"mode 不匹配: {data.get('mode')} != {request.get('mode')}")
        if request.get("start_time") and not time_bounds_match(data.get("start_time"), request.get("start_time")):
            blocking_errors.append(
                f"start_time 不匹配: {data.get('start_time')} != {format_time_bound(request.get('start_time'))}"
            )
        if request.get("end_time") and not time_bounds_match(data.get("end_time"), request.get("end_time")):
            blocking_errors.append(
                f"end_time 不匹配: {data.get('end_time')} != {format_time_bound(request.get('end_time'))}"
            )
        if (data.get("filter_buyer_name") or "") != (request.get("buyer_name") or ""):
            blocking_errors.append(
                f"filter_buyer_name 不匹配: {data.get('filter_buyer_name')} != {request.get('buyer_name') or ''}"
            )
    if data.get("partial"):
        blocking_errors.append("partial == true")
    if not isinstance(inquiries, list):
        blocking_errors.append("inquiries 不是数组")
        inquiries = []
    if isinstance(total, int) and total != len(inquiries):
        blocking_errors.append(f"total 与 inquiries 数量不一致: {total} != {len(inquiries)}")

    ids = [str(item.get("inquiry_id") or "").strip() for item in inquiries if item.get("inquiry_id")]
    id_counts = {}
    for item_id in ids:
        id_counts[item_id] = id_counts.get(item_id, 0) + 1
    duplicate_ids = sorted(item_id for item_id, count in id_counts.items() if count > 1)
    if duplicate_ids:
        blocking_errors.append("inquiry_id 重复: " + ", ".join(duplicate_ids[:10]))

    actual_detail_count = sum(1 for item in inquiries if not item.get("detail_skipped"))
    actual_detail_skipped = sum(1 for item in inquiries if item.get("detail_skipped"))
    detail_count = data.get("detail_count", 0)
    detail_skipped = data.get("detail_skipped", 0)
    if detail_count != actual_detail_count:
        blocking_errors.append(f"detail_count 与 inquiries 明细不一致: {detail_count} != {actual_detail_count}")
    if detail_skipped != actual_detail_skipped:
        blocking_errors.append(f"detail_skipped 与 inquiries 明细不一致: {detail_skipped} != {actual_detail_skipped}")

    repairable_missing_keys = []
    non_repairable_missing = []
    for index, item in enumerate(inquiries):
        if not item.get("detail_skipped"):
            continue
        key = detail_failure_key(item, index)
        if is_repairable_detail_failure(item):
            if key not in repairable_missing_keys:
                repairable_missing_keys.append(key)
        else:
            non_repairable_missing.append(key)

    if non_repairable_missing:
        blocking_errors.append(
            "detail_skipped 包含显式跳过或截断详情: "
            + summarize_missing_detail_ids(non_repairable_missing)
        )

    missing_detail_count = len(repairable_missing_keys)
    if missing_detail_count > DEGRADED_MISSING_DETAIL_LIMIT:
        blocking_errors.append(
            f"补跑后仍缺失详情超过容错阈值: {missing_detail_count} > {DEGRADED_MISSING_DETAIL_LIMIT}"
        )
    elif missing_detail_count > 0:
        soft_errors.append(
            "补跑后仍缺失详情 {count} 条: {ids}".format(
                count=missing_detail_count,
                ids=summarize_missing_detail_ids(repairable_missing_keys),
            )
        )

    failed = data.get("failed") or []
    failed_ids = sorted({
        str(item).strip()
        for item in failed
        if str(item).strip()
    })
    if failed_ids:
        if failed_ids != sorted(repairable_missing_keys):
            blocking_errors.append(
                "failed 与残余错误详情不一致: " + summarize_missing_detail_ids(failed_ids)
            )
    elif missing_detail_count > 0:
        blocking_errors.append("failed 为空，但 inquiries 中仍有错误详情缺失")

    batch = data.get("batch") or {}
    if batch.get("role") == "batch-run" and not inquiries:
        blocking_errors.append("batch-run 索引没有 inquiries 明细")

    reportable = not blocking_errors
    degraded = reportable and bool(soft_errors)
    complete = reportable and not soft_errors
    state = "complete" if complete else "degraded" if degraded else "incomplete"
    return validation_result(
        summary=inquiry_summary(data),
        complete=complete,
        reportable=reportable,
        degraded=degraded,
        state=state,
        blocking_errors=blocking_errors,
        soft_errors=soft_errors,
        missing_detail_count=missing_detail_count,
    )


def log_tail(path, lines=40):
    if not path or not os.path.exists(path):
        return []
    if lines is None or int(lines) <= 0:
        return []
    with open(path, encoding="utf-8", errors="replace") as handle:
        return handle.readlines()[-lines:]


def log_contains_marker(path, markers, lines=200):
    return any(
        any(marker in line for marker in markers)
        for line in log_tail(path, lines)
    )


def log_has_error(path):
    return log_contains_marker(path, ERROR_LOG_MARKERS)


def log_has_fatal_runtime_error(path):
    return log_contains_marker(path, FATAL_RUNTIME_LOG_MARKERS)


def status_tail_lines(args, default=0):
    explicit_tail = getattr(args, "tail", None)
    tail = default if explicit_tail is None else int(explicit_tail or 0)
    if getattr(args, "verbose", False) and tail <= 0:
        return 40
    return tail


def slim_progress(progress):
    progress = progress or {}
    keys = (
        "expected_windows",
        "completed_windows",
        "remaining_windows",
        "observed_total",
        "observed_detail_count",
        "estimated_total",
        "estimated_pages",
    )
    return {key: progress.get(key) for key in keys if key in progress}


def slim_validation(validation):
    validation = validation or {}
    result = {
        "complete": validation.get("complete"),
        "reportable": validation.get("reportable"),
        "degraded": validation.get("degraded"),
        "state": validation.get("state"),
        "missing_detail_count": validation.get("missing_detail_count", 0),
        "summary": validation.get("summary") or {},
    }
    errors = validation.get("errors") or []
    if errors:
        result["errors"] = errors
    blocking_errors = validation.get("blocking_errors") or []
    if blocking_errors:
        result["blocking_errors"] = blocking_errors
    soft_errors = validation.get("soft_errors") or []
    if soft_errors:
        result["soft_errors"] = soft_errors
    return result


def add_status_log_fields(payload, state, args, include_error=False, default_tail=0):
    tail = status_tail_lines(args, default=default_tail)
    if tail > 0:
        payload["log_tail"] = log_tail(state.get("log_path"), tail)
    if include_error or log_has_error(state.get("log_path")):
        payload["error_log_tail"] = log_tail(state.get("log_path"), max(tail, 40))
    return payload


def log_has_login_required(path):
    return any(
        LOGIN_REQUIRED_MARKER in line or "login.alibaba.com" in line
        for line in log_tail(path, 200)
    )


def state_has_login_required(state):
    return (
        bool((state or {}).get("login_required"))
        or (state or {}).get("status") == "login_required"
        or log_has_login_required((state or {}).get("log_path"))
    )


def update_state_fatal_runtime_error(state, progress=None):
    state = dict(state or {})
    updated_at = now_text()
    state.update({
        "status": "error",
        "reason": "browser_runtime_oom_or_crash",
        "message": "检测到 Chrome/Playwright driver OOM 或连接崩溃；请 cleanup 后重新 start。",
        "updated_at": updated_at,
        "finished_at": updated_at,
    })
    if progress is not None:
        state["progress"] = progress
    write_json_atomic(STATE_PATH, state)
    return state


def fatal_runtime_error_payload(state, progress=None, tail=40, killed_process=False):
    tail_lines = max(int(tail or 0), 80)
    return {
        "status": "error",
        "reason": "browser_runtime_oom_or_crash",
        "message": "检测到 Chrome/Playwright driver OOM 或连接崩溃。请先运行 cleanup --force，再重新 start；不要继续 status/watch 或读取旧结果生成报告。",
        "state_path": STATE_PATH,
        "log_path": (state or {}).get("log_path"),
        "pid": (state or {}).get("pid"),
        "killed_process": bool(killed_process),
        "progress": slim_progress(progress),
        "error_log_tail": log_tail((state or {}).get("log_path"), tail_lines),
        "next_step": "cleanup_then_retry",
        "cleanup_command": "python -X utf8 scripts\\accio_scrape_manager.py cleanup --force",
    }


def update_state_login_required(state, progress=None):
    state = dict(state or {})
    updated_at = now_text()
    state.update({
        "status": "login_required",
        "login_required": True,
        "message": LOGIN_REQUIRED_MESSAGE,
        "updated_at": updated_at,
        "finished_at": updated_at,
    })
    if progress is not None:
        state["progress"] = progress
    write_json_atomic(STATE_PATH, state)
    return state


def login_required_payload(state, progress=None, tail=40, killed_process=False):
    payload = {
        "status": "login_required",
        "message": LOGIN_REQUIRED_MESSAGE,
        "state_path": STATE_PATH,
        "log_path": (state or {}).get("log_path"),
        "pid": (state or {}).get("pid"),
        "killed_process": bool(killed_process),
        "progress": progress,
        "error_log_tail": log_tail((state or {}).get("log_path"), max(int(tail or 0), 40)),
        "next_step": LOGIN_REQUIRED_NEXT_STEP,
    }
    if int(tail or 0) > 0:
        payload["log_tail"] = log_tail((state or {}).get("log_path"), tail)
    return payload


def update_state_from_validation(state, validation):
    state = dict(state or {})
    state["updated_at"] = now_text()
    state["validation"] = validation
    if validation["reportable"]:
        state["status"] = "completed"
        state["finished_at"] = state["updated_at"]
    else:
        state["status"] = "partial"
        state["finished_at"] = state["updated_at"]
    write_json_atomic(STATE_PATH, state)
    return state


def handle_start(args):
    request = request_from_args(args)
    key = canonical_key(request)
    lock = acquire_start_lock(key)
    if not lock.get("acquired"):
        print_json({
            "status": "blocked",
            "message": "已有 start 启动临界区锁，不能并发启动。请运行 status 查看当前任务。",
            "state_path": STATE_PATH,
            "lock_path": lock.get("path"),
            "lock": lock.get("lock"),
            "next_step": "status",
        }, exit_code=2)

    try:
        state = load_state()
        report_output = resolve_report_output_dir(args, state)
        if state and state.get("pid") and process_running(state.get("pid")):
            print_json({
                "status": "running",
                "message": "已有抓取任务正在运行，不能重复启动",
                "state_path": STATE_PATH,
                "pid": state.get("pid"),
                "key": state.get("key"),
                "same_request": state.get("key") == key,
                "log_path": state.get("log_path"),
            })

        validation = validate_inquiries(request)
        if state and state.get("key") == key and not args.force:
            if validation["reportable"]:
                state = update_state_from_validation(state, validation)
                print_json({
                    "status": "completed",
                    "message": "目标窗口已有可报告结果；如需重新抓取，先确认后使用 start --force",
                    "state": state,
                })
            if state.get("status") in {"running", "partial", "error", "stale", "completed"}:
                print_json({
                    "status": "stale" if state.get("status") == "running" else state.get("status", "existing"),
                    "message": "发现同一目标窗口的既有状态，默认不重复启动。需要重跑时先 cleanup，再由用户确认 start --force。",
                    "state_path": STATE_PATH,
                    "validation": validation,
                    "log_path": state.get("log_path"),
                })

        cleanup_result = close_alibaba_message_tabs()
        launch = launch_scrape_process(request)
        proc = launch["proc"]
        log_path = launch["log_path"]
        command = launch["command"]
        auto_compare = auto_compare_plan_for_request(request)

        state = {
            "status": "running",
            "created_at": now_text(),
            "updated_at": now_text(),
            "pid": proc.pid,
            "key": key,
            "request": request,
            "command": command,
            "log_path": log_path,
            "output_path": OUTPUT_JSON_PATH,
            "report_output_dir": report_output["path"],
            "report_output_dir_source": report_output["source"],
            "cleanup_before_start": cleanup_result,
        }
        if auto_compare:
            state["auto_compare"] = auto_compare
        write_json_atomic(STATE_PATH, state)
        print_json({
            "status": "running",
            "message": "抓取任务已启动。后续只运行 status 轮询，不要重新启动 scraper。",
            "pid": proc.pid,
            "state_path": STATE_PATH,
            "log_path": log_path,
            "output_path": OUTPUT_JSON_PATH,
            "report_output_dir": report_output["path"],
            "report_output_dir_source": report_output["source"],
            "cleanup_before_start": cleanup_result,
        })
    finally:
        release_start_lock(lock)


def handle_status(args):
    state = load_state()
    if not state:
        print_json({"status": "missing", "error": "没有找到 manager 状态文件", "state_path": STATE_PATH}, exit_code=1)
    progress = collect_shard_progress(state)
    if state_has_login_required(state):
        killed_process = False
        if state.get("pid") and process_running(state.get("pid")):
            killed_process = terminate_process(state.get("pid"))
        state = update_state_login_required(state, progress=progress)
        print_json(
            login_required_payload(
                state,
                progress=progress,
                tail=args.tail,
                killed_process=killed_process,
            ),
            exit_code=3,
        )

    if state.get("pid") and process_running(state.get("pid")):
        if log_has_fatal_runtime_error(state.get("log_path")):
            killed_process = terminate_process(state.get("pid"))
            state = update_state_fatal_runtime_error(state, progress=progress)
            print_json(
                fatal_runtime_error_payload(
                    state,
                    progress=progress,
                    tail=args.tail,
                    killed_process=killed_process,
                ),
                exit_code=2,
            )
        payload = {
            "status": "running",
            "pid": state.get("pid"),
            "elapsed_seconds": elapsed_seconds_from_state(state),
            "progress": slim_progress(progress),
            "suggested_after_seconds": DEFAULT_STATUS_POLL_SECONDS,
            "next_step": "status",
        }
        add_status_log_fields(payload, state, args, default_tail=DEFAULT_STATUS_RUNNING_TAIL_LINES)
        print_json(payload)

    validation = validate_inquiries(state.get("request"))
    state = update_state_from_validation(state, validation)

    next_step = "finish" if validation["reportable"] else "do_not_report"
    message = (
        "抓取已完成，可运行 finish 生成报告。不要再读日志、列目录或重复 status。"
        if validation["reportable"]
        else "抓取已结束，但结果不可报告。请按 errors 处理，不要读取旧文件生成报告。"
    )
    report_output_dir = state.get("report_output_dir")
    existing_report = {}
    include_markdown = resolve_export_markdown_enabled()
    include_pdf = resolve_export_pdf_enabled()
    if apply_existing_report(
        existing_report,
        state,
        report_output_dir,
        include_markdown=include_markdown,
        include_pdf=include_pdf,
    ):
        next_step = "present_report"
        message = "抓取与报告均已完成，可直接展示报告正文；如本轮启用了文件导出，再展示对应文件路径。不要再运行 finish。"

    payload = {
        "status": state["status"],
        "message": message,
        "output_path": state.get("output_path") or OUTPUT_JSON_PATH,
        "validation": slim_validation(validation),
        "next_step": next_step,
    }
    payload.update(existing_report)
    add_status_log_fields(payload, state, args, include_error=not validation["reportable"])
    print_json(payload, exit_code=0 if validation["reportable"] else 2)


def run_report(
    report_output_dir=None,
    input_path=None,
    compare_input=None,
    export_markdown=None,
    export_pdf=None,
):
    command = [sys.executable, "-X", "utf8", os.path.join("scripts", "accio_generate_report.py")]
    if input_path:
        command.extend(["--input", input_path])
    if compare_input:
        command.extend(["--compare-input", compare_input])
    if report_output_dir:
        command.extend(["--output-dir", report_output_dir])
    if export_markdown is True:
        command.append("--md")
    elif export_markdown is False:
        command.append("--no-md")
    if export_pdf is True:
        command.append("--pdf")
    elif export_pdf is False:
        command.append("--no-pdf")
    result = subprocess.run(
        command,
        cwd=PROJECT_DIR,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )
    parsed = None
    if result.stdout.strip():
        try:
            parsed = json.loads(result.stdout)
        except json.JSONDecodeError:
            parsed = None
    return {
        "returncode": result.returncode,
        "stdout": result.stdout,
        "stderr": result.stderr,
        "json": parsed,
    }


def report_source_for_cleanup_key(cleanup_state_key):
    if cleanup_state_key == "cleanup_after_watch":
        return "watch"
    if cleanup_state_key == "cleanup_after_finish":
        return "finish"
    return "unknown"


def existing_report_path(state, report_output_dir=None):
    path = (state or {}).get("report_path")
    if not path:
        report = (state or {}).get("report")
        if isinstance(report, dict):
            path = report.get("output")
    if not path:
        return None
    check_path = path if os.path.isabs(path) else os.path.join(PROJECT_DIR, path)
    if report_output_dir and not path_inside_dir(check_path, report_output_dir):
        return None
    return path if os.path.exists(check_path) else None


def existing_report_text(state):
    report = (state or {}).get("report")
    if isinstance(report, dict) and report.get("report_text"):
        return report["report_text"]
    if isinstance(report, dict) and report.get("report_text_path"):
        return read_text_if_exists(report.get("report_text_path"))
    if (state or {}).get("report_text_path"):
        return read_text_if_exists((state or {}).get("report_text_path"))
    return (state or {}).get("report_text")


def existing_report_text_path(state):
    report = (state or {}).get("report")
    if isinstance(report, dict) and report.get("report_text_path"):
        return report["report_text_path"]
    return (state or {}).get("report_text_path")


def apply_existing_report(result, state, report_output_dir=None, include_markdown=True, include_pdf=True):
    report_path = existing_report_path(state, report_output_dir) if include_markdown else None
    report_text = existing_report_text(state) if not include_markdown else None
    if not report_path and not report_text:
        return False
    report = (state or {}).get("report")
    if isinstance(report, dict):
        result["report"] = report
        if report.get("assignee_followup"):
            result["assignee_followup"] = report["assignee_followup"]
    if report_path:
        result["report_path"] = report_path
    report_text_path = existing_report_text_path(state)
    if report_text_path:
        result["report_text_path"] = report_text_path
    if report_text and len(report_text) <= REPORT_TEXT_RESULT_LIMIT:
        result["report_text"] = report_text
    elif report_text_path:
        result["report_text_omitted"] = True
        result["report_text_message"] = "报告正文较长，为避免工具输出截断，请读取 report_text_path 后基于完整正文回复用户。"
    pdf_report_path = (state or {}).get("pdf_report_path")
    if include_pdf and pdf_report_path:
        result["pdf_report_path"] = pdf_report_path
    if (state or {}).get("report_generated_at"):
        result["report_generated_at"] = state["report_generated_at"]
    if (state or {}).get("report_generated_by"):
        result["report_generated_by"] = state["report_generated_by"]
    return True


def persist_report_state(state, result, cleanup_state_key):
    state = dict(state or {})
    if result.get("report_output_dir"):
        state["report_output_dir"] = result["report_output_dir"]
    if result.get("report_output_dir_source"):
        state["report_output_dir_source"] = result["report_output_dir_source"]
    if result.get("report"):
        state["report"] = result["report"]
    if result.get("report_path"):
        state["report_path"] = result["report_path"]
    elif result.get("report"):
        state.pop("report_path", None)
    if result.get("report_text"):
        state["report_text"] = result["report_text"]
    elif result.get("report_text_path"):
        state.pop("report_text", None)
    elif result.get("report"):
        state.pop("report_text", None)
    if result.get("report_text_path"):
        state["report_text_path"] = result["report_text_path"]
    elif result.get("report") and not result.get("report_text"):
        state.pop("report_text_path", None)
    if result.get("pdf_report_path"):
        state["pdf_report_path"] = result["pdf_report_path"]
    elif result.get("report"):
        state.pop("pdf_report_path", None)
    generated_at = now_text()
    generated_by = report_source_for_cleanup_key(cleanup_state_key)
    state["report_generated_at"] = generated_at
    state["report_generated_by"] = generated_by
    state["updated_at"] = generated_at
    write_json_atomic(STATE_PATH, state)
    result["report_generated_at"] = generated_at
    result["report_generated_by"] = generated_by
    return state


def merge_report_fields(target, source):
    for key in (
        "report",
        "report_path",
        "report_text",
        "report_text_path",
        "report_text_omitted",
        "report_text_message",
        "pdf_report_path",
        "report_output_dir",
        "report_output_dir_source",
        "report_generated_at",
        "report_generated_by",
    ):
        if key in (source or {}) and key not in target:
            target[key] = source[key]
    return target


def existing_report_available(state, report_output_dir=None, include_markdown=True):
    if include_markdown:
        return bool(existing_report_path(state, report_output_dir))
    return bool(existing_report_text(state) or existing_report_text_path(state))


def wait_for_existing_report(report_output_dir=None, include_markdown=True):
    deadline = time.monotonic() + REPORT_FINALIZE_LOCK_WAIT_SECONDS
    while time.monotonic() < deadline:
        time.sleep(REPORT_FINALIZE_LOCK_POLL_SECONDS)
        state = load_state()
        if existing_report_available(state, report_output_dir, include_markdown=include_markdown):
            return state
    return load_state()


def start_auto_compare_if_needed(state, result, args):
    auto_compare = (state or {}).get("auto_compare")
    if not auto_compare or auto_compare.get("status") != "pending":
        return None

    main_snapshot = auto_compare["main_snapshot_path"]
    copy_output_snapshot(main_snapshot)
    compare_request = dict(auto_compare["compare_request"])
    launch = launch_scrape_process(compare_request, log_prefix="compare")
    proc = launch["proc"]

    auto_compare = dict(auto_compare)
    auto_compare.update({
        "status": "running",
        "main_snapshot_path": main_snapshot,
        "compare_started_at": now_text(),
        "compare_log_path": launch["log_path"],
        "compare_command": launch["command"],
    })
    state = dict(state or {})
    state.update({
        "status": "running",
        "updated_at": now_text(),
        "pid": proc.pid,
        "key": canonical_key(compare_request),
        "request": compare_request,
        "command": launch["command"],
        "log_path": launch["log_path"],
        "output_path": OUTPUT_JSON_PATH,
        "auto_compare": auto_compare,
    })
    write_json_atomic(STATE_PATH, state)
    result.update({
        "status": "running",
        "message": "主窗口抓取已完成，已启动前一日对比窗口抓取；请继续运行 wait。",
        "pid": proc.pid,
        "next_step": "wait",
        "auto_compare": auto_compare,
        "log_path": launch["log_path"],
    })
    return state


def complete_auto_compare_if_needed(state):
    auto_compare = (state or {}).get("auto_compare")
    if not auto_compare or auto_compare.get("status") != "running":
        return state

    compare_snapshot = auto_compare["compare_snapshot_path"]
    copy_output_snapshot(compare_snapshot)
    auto_compare = dict(auto_compare)
    auto_compare.update({
        "status": "completed",
        "compare_snapshot_path": compare_snapshot,
        "completed_at": now_text(),
    })
    main_snapshot = auto_compare.get("main_snapshot_path")
    if main_snapshot:
        restore_output_from_snapshot(main_snapshot)
    state = dict(state or {})
    state.update({
        "auto_compare": auto_compare,
        "request": auto_compare.get("main_request") or state.get("request"),
        "key": canonical_key(auto_compare.get("main_request") or state.get("request") or {}),
        "output_path": auto_compare.get("main_snapshot_path") or state.get("output_path"),
        "updated_at": now_text(),
    })
    write_json_atomic(STATE_PATH, state)
    return state


def report_inputs_for_state(state):
    auto_compare = (state or {}).get("auto_compare") or {}
    if auto_compare.get("status") != "completed":
        return {
            "input_path": None,
            "compare_input": None,
        }
    return {
        "input_path": auto_compare.get("main_snapshot_path"),
        "compare_input": auto_compare.get("compare_snapshot_path"),
    }


def emit_watch_progress(state, progress, max_minutes, dynamic):
    payload = {
        "status": "running",
        "pid": state.get("pid"),
        "elapsed_seconds": elapsed_seconds_from_state(state),
        "max_minutes": max_minutes,
        "dynamic_max_minutes": dynamic,
        "progress": {
            "expected_windows": progress.get("expected_windows"),
            "completed_windows": progress.get("completed_windows"),
            "remaining_windows": progress.get("remaining_windows"),
            "observed_total": progress.get("observed_total"),
            "estimated_total": progress.get("estimated_total"),
            "estimated_pages": progress.get("estimated_pages"),
            "longest_shard_seconds": progress.get("longest_shard_seconds"),
        },
    }
    print(json.dumps(payload, ensure_ascii=False), flush=True)


def finalize_completed_task(state, progress, args, cleanup_state_key):
    validation = validate_inquiries(state.get("request"))
    state = update_state_from_validation(state, validation)
    report_output = resolve_report_output_dir(args, state)
    include_markdown = resolve_export_markdown_enabled()
    include_pdf = resolve_export_pdf_enabled()
    state["report_output_dir"] = report_output["path"]
    state["report_output_dir_source"] = report_output["source"]
    result = {
        "status": state["status"],
        "state_path": STATE_PATH,
        "log_path": state.get("log_path"),
        "output_path": state.get("output_path") or OUTPUT_JSON_PATH,
        "report_output_dir": report_output["path"],
        "report_output_dir_source": report_output["source"],
        "validation": validation,
        "progress": progress,
        "log_tail": log_tail(state.get("log_path"), args.tail),
    }

    if not validation["reportable"]:
        result["next_step"] = "do_not_report"
        print_json(result, exit_code=2)

    if not args.no_report:
        if start_auto_compare_if_needed(state, result, args):
            print_json(result)
        state = complete_auto_compare_if_needed(state)
        if (state.get("auto_compare") or {}).get("status") == "completed":
            validation = validate_inquiries(
                state.get("request"),
                path=state.get("output_path") or OUTPUT_JSON_PATH,
            )
            state = update_state_from_validation(state, validation)
            result["status"] = state["status"]
            result["validation"] = validation
        result["output_path"] = state.get("output_path") or result["output_path"]
        if state.get("auto_compare"):
            result["auto_compare"] = state["auto_compare"]

    if not args.no_report:
        if not apply_existing_report(
            result,
            state,
            report_output["path"],
            include_markdown=include_markdown,
            include_pdf=include_pdf,
        ):
            lock = acquire_report_finalize_lock()
            if not lock.get("acquired"):
                latest_state = wait_for_existing_report(
                    report_output["path"],
                    include_markdown=include_markdown,
                )
                if apply_existing_report(
                    result,
                    latest_state,
                    report_output["path"],
                    include_markdown=include_markdown,
                    include_pdf=include_pdf,
                ):
                    state = dict(latest_state or state)
                else:
                    lock_unavailable = lock.get("scope") == "all"
                    result.update({
                        "status": "blocked",
                        "message": (
                            "无法建立报告收尾锁；为避免重复生成报告，已停止自动生成新报告。"
                            if lock_unavailable
                            else "报告生成收尾正在进行中，尚未发现可复用报告；请稍后重试 status 或 finish。"
                        ),
                        "reason": "finalize_lock_unavailable" if lock_unavailable else "finalize_lock_busy",
                        "lock_path": lock.get("path"),
                        "lock_scope": lock.get("scope"),
                        "lock": lock.get("lock"),
                        "next_step": "fix_lock_or_manual_report" if lock_unavailable else "finish",
                    })
                    print_json(result, exit_code=2)
            else:
                try:
                    result["finalize_lock_scope"] = lock.get("scope")
                    latest_state = load_state() or state
                    if apply_existing_report(
                        result,
                        latest_state,
                        report_output["path"],
                        include_markdown=include_markdown,
                        include_pdf=include_pdf,
                    ):
                        state = dict(latest_state)
                    else:
                        report_inputs = report_inputs_for_state(state)
                        if report_inputs.get("input_path") or report_inputs.get("compare_input"):
                            report = run_report(
                                report_output["path"],
                                input_path=report_inputs.get("input_path"),
                                compare_input=report_inputs.get("compare_input"),
                                export_markdown=include_markdown,
                                export_pdf=include_pdf,
                            )
                        else:
                            report = run_report(
                                report_output["path"],
                                export_markdown=include_markdown,
                                export_pdf=include_pdf,
                            )
                        result["report"] = report.get("json") or {
                            "returncode": report["returncode"],
                            "stdout": report["stdout"],
                            "stderr": report["stderr"],
                        }
                        if isinstance(result["report"], dict) and result["report"].get("output"):
                            result["report_path"] = result["report"]["output"]
                        if isinstance(result["report"], dict) and result["report"].get("report_text_path"):
                            result["report_text_path"] = result["report"]["report_text_path"]
                        if isinstance(result["report"], dict) and result["report"].get("report_text"):
                            report_text = result["report"]["report_text"]
                            if len(report_text) <= REPORT_TEXT_RESULT_LIMIT:
                                result["report_text"] = report_text
                            elif result.get("report_text_path"):
                                result["report_text_omitted"] = True
                                result["report_text_message"] = "报告正文较长，为避免工具输出截断，请读取 report_text_path 后基于完整正文回复用户。"
                        if isinstance(result["report"], dict) and result["report"].get("report_text_omitted"):
                            result["report_text_omitted"] = True
                            result["report_text_message"] = result["report"].get(
                                "report_text_message",
                                "报告正文较长，为避免工具输出截断，请读取 report_text_path 后基于完整正文回复用户。",
                            )
                        if isinstance(result["report"], dict) and result["report"].get("assignee_followup"):
                            result["assignee_followup"] = result["report"]["assignee_followup"]
                        if isinstance(result["report"], dict) and result["report"].get("pdf_output"):
                            result["pdf_report_path"] = result["report"]["pdf_output"]
                        if report["returncode"] != 0:
                            print_json(result, exit_code=report["returncode"])
                        state = persist_report_state(state, result, cleanup_state_key)
                finally:
                    release_report_finalize_lock(lock)

    if not args.no_cleanup:
        cleanup_result = close_alibaba_message_tabs()
        state = merge_report_fields(dict(load_state() or state), state)
        state[cleanup_state_key] = cleanup_result
        state["updated_at"] = now_text()
        write_json_atomic(STATE_PATH, state)
        result["cleanup"] = cleanup_result

    print_json(result)


def block_watch_in_short_timeout_environment(state, args, max_minutes, dynamic):
    timeout = explicit_tool_timeout()
    required_seconds = required_watch_timeout_seconds(max_minutes)
    if timeout.get("known") and timeout.get("timeout_seconds", 0) >= required_seconds:
        return

    pid = state.get("pid")
    running = bool(pid and process_running(pid))
    if timeout.get("known"):
        reason = "watch_would_exceed_tool_timeout"
        message = (
            "当前工具 timeout 不足以运行 watch，即使带 --force 也不会长时间等待。"
            "请改用 status 短轮询；status 显示可报告后运行 finish。"
        )
    else:
        reason = "watch_timeout_unknown"
        message = (
            "当前工具调用没有提供可信的超时上限，watch 不会长时间等待，即使带 --force。"
            "请改用 status 短轮询；status 显示可报告后运行 finish。"
        )
    print_json({
        "status": "blocked",
        "reason": reason,
        "message": message,
        "state_path": STATE_PATH,
        "log_path": state.get("log_path"),
        "pid": pid,
        "running": running,
        "force": bool(getattr(args, "force", False)),
        "tool_timeout": timeout,
        "required_timeout_seconds": required_seconds,
        "watch_estimate": {
            "max_minutes": max_minutes,
            "dynamic": dynamic,
            "buffer_seconds": WATCH_TOOL_TIMEOUT_BUFFER_SECONDS,
        },
        "next_step": "status",
        "log_tail": log_tail(state.get("log_path"), args.tail),
    }, exit_code=2)


def handle_watch(args):
    state = load_state()
    if not state:
        print_json({"status": "missing", "error": "没有找到 manager 状态文件", "state_path": STATE_PATH}, exit_code=1)
    progress = collect_shard_progress(state)
    if state_has_login_required(state):
        killed_process = False
        if state.get("pid") and process_running(state.get("pid")):
            killed_process = terminate_process(state.get("pid"))
        state = update_state_login_required(state, progress=progress)
        print_json(
            login_required_payload(
                state,
                progress=progress,
                tail=args.tail,
                killed_process=killed_process,
            ),
            exit_code=3,
        )
    if log_has_fatal_runtime_error(state.get("log_path")):
        killed_process = False
        if state.get("pid") and process_running(state.get("pid")):
            killed_process = terminate_process(state.get("pid"))
        state = update_state_fatal_runtime_error(state, progress=progress)
        print_json(
            fatal_runtime_error_payload(
                state,
                progress=progress,
                tail=args.tail,
                killed_process=killed_process,
            ),
            exit_code=2,
        )
    max_minutes, dynamic = resolve_watch_minutes(args.max_minutes, state, progress)
    block_watch_in_short_timeout_environment(state, args, max_minutes, dynamic)

    while True:
        state = load_state()
        if not state:
            print_json({"status": "missing", "error": "状态文件在 watch 期间消失", "state_path": STATE_PATH}, exit_code=1)

        progress = collect_shard_progress(state)
        max_minutes, dynamic = resolve_watch_minutes(args.max_minutes, state, progress)
        elapsed_seconds = elapsed_seconds_from_state(state)

        if state_has_login_required(state):
            killed_process = False
            if state.get("pid") and process_running(state.get("pid")):
                killed_process = terminate_process(state.get("pid"))
            state = update_state_login_required(state, progress=progress)
            print_json(
                login_required_payload(
                    state,
                    progress=progress,
                    tail=args.tail,
                    killed_process=killed_process,
                ),
                exit_code=3,
            )

        if log_has_fatal_runtime_error(state.get("log_path")):
            killed_process = False
            if state.get("pid") and process_running(state.get("pid")):
                killed_process = terminate_process(state.get("pid"))
            state = update_state_fatal_runtime_error(state, progress=progress)
            print_json(
                fatal_runtime_error_payload(
                    state,
                    progress=progress,
                    tail=args.tail,
                    killed_process=killed_process,
                ),
                exit_code=2,
            )

        if state.get("pid") and process_running(state.get("pid")):
            if elapsed_seconds > max_minutes * 60:
                state = dict(state)
                state["status"] = "watch_timeout"
                state["updated_at"] = now_text()
                state["watch"] = {
                    "max_minutes": max_minutes,
                    "dynamic": dynamic,
                    "elapsed_seconds": elapsed_seconds,
                    "progress": progress,
                }
                write_json_atomic(STATE_PATH, state)
                print_json({
                    "status": "watch_timeout",
                    "message": "抓取进程仍在运行，watch 已达到动态等待上限；未终止抓取进程。后续只需重新运行 watch 或 status，不要重新 start。",
                    "state_path": STATE_PATH,
                    "log_path": state.get("log_path"),
                    "elapsed_seconds": elapsed_seconds,
                    "max_minutes": max_minutes,
                    "dynamic_max_minutes": dynamic,
                    "progress": progress,
                    "log_tail": log_tail(state.get("log_path"), args.tail),
                }, exit_code=2)

            emit_watch_progress(state, progress, max_minutes, dynamic)
            time.sleep(max(1, int(args.interval_seconds)))
            continue

        finalize_completed_task(state, progress, args, "cleanup_after_watch")


def handle_wait(args):
    state = load_state()
    if not state:
        print_json({"status": "missing", "error": "没有找到 manager 状态文件", "state_path": STATE_PATH}, exit_code=1)

    progress = collect_shard_progress(state)
    if state_has_login_required(state):
        killed_process = False
        if state.get("pid") and process_running(state.get("pid")):
            killed_process = terminate_process(state.get("pid"))
        state = update_state_login_required(state, progress=progress)
        print_json(
            login_required_payload(
                state,
                progress=progress,
                tail=args.tail,
                killed_process=killed_process,
            ),
            exit_code=3,
        )

    if log_has_fatal_runtime_error(state.get("log_path")):
        killed_process = False
        if state.get("pid") and process_running(state.get("pid")):
            killed_process = terminate_process(state.get("pid"))
        state = update_state_fatal_runtime_error(state, progress=progress)
        print_json(
            fatal_runtime_error_payload(
                state,
                progress=progress,
                tail=args.tail,
                killed_process=killed_process,
            ),
            exit_code=2,
        )

    if not (state.get("pid") and process_running(state.get("pid"))):
        finalize_completed_task(state, progress, args, "cleanup_after_wait")

    max_wait_seconds = resolve_wait_max_seconds(args)
    wait_deadline = time.monotonic() + max_wait_seconds

    while time.monotonic() < wait_deadline:
        time.sleep(DEFAULT_WAIT_POLL_INTERVAL_SECONDS)
        state = load_state()
        if not state:
            print_json({"status": "missing", "error": "状态文件在 wait 期间消失", "state_path": STATE_PATH}, exit_code=1)

        progress = collect_shard_progress(state)
        if state_has_login_required(state):
            killed_process = False
            if state.get("pid") and process_running(state.get("pid")):
                killed_process = terminate_process(state.get("pid"))
            state = update_state_login_required(state, progress=progress)
            print_json(
                login_required_payload(
                    state,
                    progress=progress,
                    tail=args.tail,
                    killed_process=killed_process,
                ),
                exit_code=3,
            )

        if log_has_fatal_runtime_error(state.get("log_path")):
            killed_process = False
            if state.get("pid") and process_running(state.get("pid")):
                killed_process = terminate_process(state.get("pid"))
            state = update_state_fatal_runtime_error(state, progress=progress)
            print_json(
                fatal_runtime_error_payload(
                    state,
                    progress=progress,
                    tail=args.tail,
                    killed_process=killed_process,
                ),
                exit_code=2,
            )

        if not (state.get("pid") and process_running(state.get("pid"))):
            finalize_completed_task(state, progress, args, "cleanup_after_wait")

    payload = {
        "status": "running",
        "message": f"wait 已等待 {max_wait_seconds} 秒，抓取进程仍在运行。可再次运行 wait 继续等待。",
        "state_path": STATE_PATH,
        "log_path": state.get("log_path"),
        "pid": state.get("pid"),
        "elapsed_seconds": elapsed_seconds_from_state(state),
        "progress": slim_progress(progress),
        "suggested_after_seconds": 0,
        "next_step": "wait",
    }
    add_status_log_fields(payload, state, args, default_tail=DEFAULT_STATUS_RUNNING_TAIL_LINES)
    print_json(payload)


def handle_finish(args):
    state = load_state()
    if not state:
        print_json({"status": "missing", "error": "没有找到 manager 状态文件", "state_path": STATE_PATH}, exit_code=1)

    progress = collect_shard_progress(state)
    if state_has_login_required(state):
        killed_process = False
        if state.get("pid") and process_running(state.get("pid")):
            killed_process = terminate_process(state.get("pid"))
        state = update_state_login_required(state, progress=progress)
        print_json(
            login_required_payload(
                state,
                progress=progress,
                tail=args.tail,
                killed_process=killed_process,
            ),
            exit_code=3,
        )

    if state.get("pid") and process_running(state.get("pid")):
        watch_minutes, dynamic = resolve_watch_minutes("auto", state, progress)
        print_json({
            "status": "running",
            "message": "抓取进程仍在运行，finish 不会等待。请继续运行 status，完成后再运行 finish。",
            "pid": state.get("pid"),
            "started_at": state.get("created_at"),
            "state_path": STATE_PATH,
            "log_path": state.get("log_path"),
            "progress": progress,
            "watch_estimate": {
                "max_minutes": watch_minutes,
                "dynamic": dynamic,
                "elapsed_seconds": elapsed_seconds_from_state(state),
            },
            "log_tail": log_tail(state.get("log_path"), args.tail),
            "next_step": "status",
        }, exit_code=2)

    finalize_completed_task(state, progress, args, "cleanup_after_finish")


def handle_cleanup(args):
    state = load_state()
    killed = False
    if state and state.get("pid") and process_running(state.get("pid")):
        if not args.force:
            print_json({
                "status": "blocked",
                "message": "抓取进程仍在运行，默认不清理。确实要终止时使用 cleanup --force。",
                "pid": state.get("pid"),
                "state_path": STATE_PATH,
            }, exit_code=2)
        killed = terminate_process(state.get("pid"))

    cleanup_result = close_alibaba_message_tabs()
    state = dict(state or {})
    state.update({
        "status": "cleaned",
        "updated_at": now_text(),
        "killed_process": killed,
        "cleanup": cleanup_result,
    })
    write_json_atomic(STATE_PATH, state)
    print_json({
        "status": "cleaned",
        "killed_process": killed,
        "cleanup": cleanup_result,
        "state_path": STATE_PATH,
    })


def build_parser():
    parser = argparse.ArgumentParser(description="Accio Work 历史询盘抓取管理器")
    sub = parser.add_subparsers(dest="command", required=True)

    start = sub.add_parser("start", help="启动历史日期抓取，但不等待抓完")
    start.add_argument(
        "--mode",
        type=parse_mode_arg,
        metavar="new",
        default="new",
        help="抓取模式：new=今日新建询盘（默认）",
    )
    start.add_argument("--start-time", default="")
    start.add_argument("--end-time", default="")
    start.add_argument("--depth", choices=["quick", "normal", "full"], default=None)
    start.add_argument("--category", default="")
    start.add_argument("--assignee", default="")
    start.add_argument("--buyer-name", default="")
    start.add_argument("--max-detail", type=int, default=None)
    start.add_argument("--max", type=int, default=None)
    start.add_argument("--chunk-days", type=int, default=1)
    start.add_argument("--resume", action="store_true")
    start.add_argument("--report-output-dir", default="")
    start.add_argument("--force", action="store_true", help="用户确认后才允许重新启动同一目标窗口")

    status = sub.add_parser("status", help="查看抓取状态")
    status.add_argument("--tail", type=int, default=None, help="诊断时返回日志尾巴行数；默认 running 返回 1 行，completed 不输出普通日志")
    status.add_argument("--verbose", action="store_true", help="诊断模式；未指定 --tail 时默认返回 40 行日志")

    watch = sub.add_parser("watch", help="持续等待抓取完成，完成后自动生成报告并清理 tab")
    watch.add_argument("--interval-seconds", type=int, default=DEFAULT_WATCH_INTERVAL_SECONDS)
    watch.add_argument("--report-output-dir", default="")
    watch.add_argument("--max-minutes", default="auto", help="等待上限。默认 auto 会根据分片进度动态估算")
    watch.add_argument("--tail", type=int, default=40)
    watch.add_argument("--no-report", action="store_true", help="只等待和校验，不自动生成报告")
    watch.add_argument("--no-cleanup", action="store_true", help="完成后不自动清理询盘 tab")
    watch.add_argument("--force", action="store_true", help="确认当前工具调用允许长时间运行时，绕过短超时保护")

    wait = sub.add_parser("wait", help="安全阻塞等待抓取进程，完成后自动生成报告并清理 tab")
    wait.add_argument("--max-wait-seconds", type=int, default=None)
    wait.add_argument("--tail", type=int, default=40)
    wait.add_argument("--report-output-dir", default="")
    wait.add_argument("--no-report", action="store_true", help="只等待和校验，不自动生成报告")
    wait.add_argument("--no-cleanup", action="store_true", help="完成后不自动清理询盘 tab")

    finish = sub.add_parser("finish", help="短收尾：不等待运行中的任务，只在完成后生成报告并清理 tab")
    finish.add_argument("--tail", type=int, default=40)
    finish.add_argument("--report-output-dir", default="")
    finish.add_argument("--no-report", action="store_true", help="只校验和清理，不自动生成报告")
    finish.add_argument("--no-cleanup", action="store_true", help="完成后不自动清理询盘 tab")

    cleanup = sub.add_parser("cleanup", help="清理已结束任务遗留的阿里询盘 tab")
    cleanup.add_argument("--force", action="store_true", help="同时终止仍在运行的抓取进程")

    return parser


def main():
    args = build_parser().parse_args()
    cleanup_old_report_text_files()
    if args.command == "start":
        handle_start(args)
    if args.command == "status":
        handle_status(args)
    if args.command == "watch":
        handle_watch(args)
    if args.command == "wait":
        handle_wait(args)
    if args.command == "finish":
        handle_finish(args)
    if args.command == "cleanup":
        handle_cleanup(args)


if __name__ == "__main__":
    main()
