"""根据 inquiries.json 生成固定格式报告。

Accio Work 不应临时编写 gen_report.py；统一走这个脚本，避免表格重复、
详情缺失时崩溃，以及历史范围 partial 数据被误生成正式报告。
"""

from __future__ import annotations

import argparse
import html
import json
import os
import re
import secrets
import socket
import struct
import sys
import uuid
from collections import Counter, OrderedDict
from datetime import datetime, timedelta, timezone
from pathlib import Path
from urllib.parse import quote
from zoneinfo import ZoneInfo


SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
PROJECT_DIR = os.path.dirname(SCRIPT_DIR)
if SCRIPT_DIR not in sys.path:
    sys.path.insert(0, SCRIPT_DIR)

from accio_scrape_manager import validate_inquiries
from config import DEFAULT_EXPORT_MARKDOWN, DEFAULT_EXPORT_PDF, EXPORT_MARKDOWN_ENV, EXPORT_PDF_ENV
from output import build_metrics, format_display_date_window


DEFAULT_INPUT = os.path.join(PROJECT_DIR, "inquiries.json")
DEFAULT_TEMPLATE = os.path.join(PROJECT_DIR, "references", "report_template.md")
RUNTIME_DIR = os.path.join(PROJECT_DIR, "outputs", "runtime")
FROZEN_BUSINESS_OPPORTUNITY_LABEL = "已冻结"
REPORT_TEXT_STDOUT_LIMIT = 12000
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 = timezone(timedelta(hours=8))
LA_TZ = ZoneInfo("America/Los_Angeles")


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


def read_text(path):
    with open(path, encoding="utf-8") as handle:
        return handle.read()


def write_text(path, text):
    parent = os.path.dirname(os.path.abspath(path))
    if parent:
        os.makedirs(parent, exist_ok=True)
    with open(path, "w", encoding="utf-8") as handle:
        handle.write(text)


def report_text_path_for(data, output_path):
    stem = Path(output_path).stem or "report"
    return str(Path(RUNTIME_DIR) / f"{stem}_report_text.txt")


def attach_report_text_payload(payload, report, data, output_path):
    text_path = report_text_path_for(data, output_path)
    write_text(text_path, report)
    payload["report_text_path"] = text_path
    if len(report) <= REPORT_TEXT_STDOUT_LIMIT:
        payload["report_text"] = report
    else:
        payload["report_text_omitted"] = True
        payload["report_text_bytes"] = len(report.encode("utf-8"))
        payload["report_text_message"] = "报告正文较长，为避免 Accio Work 工具输出截断，请读取 report_text_path 后基于完整正文回复用户。"


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


def windows_documents_dir():
    if os.name != "nt":
        return None
    try:
        import ctypes
        from ctypes import wintypes
    except Exception:
        return None

    class GUID(ctypes.Structure):
        _fields_ = [
            ("Data1", wintypes.DWORD),
            ("Data2", wintypes.WORD),
            ("Data3", wintypes.WORD),
            ("Data4", ctypes.c_ubyte * 8),
        ]

        def __init__(self, value):
            parsed = uuid.UUID(value)
            data4 = (ctypes.c_ubyte * 8)(
                parsed.clock_seq_hi_variant,
                parsed.clock_seq_low,
                *parsed.node.to_bytes(6, "big"),
            )
            super().__init__(
                parsed.time_low,
                parsed.time_mid,
                parsed.time_hi_version,
                data4,
            )

    folder_id_documents = GUID("FDD39AD0-238F-46AF-ADB4-6C85480369C7")
    path_ptr = wintypes.LPWSTR()
    try:
        shell32 = ctypes.windll.shell32
        shell32.SHGetKnownFolderPath.argtypes = [
            ctypes.POINTER(GUID),
            wintypes.DWORD,
            wintypes.HANDLE,
            ctypes.POINTER(wintypes.LPWSTR),
        ]
        shell32.SHGetKnownFolderPath.restype = ctypes.c_long
        result = shell32.SHGetKnownFolderPath(
            ctypes.byref(folder_id_documents),
            0,
            None,
            ctypes.byref(path_ptr),
        )
        if result != 0 or not path_ptr.value:
            return None
        return Path(path_ptr.value)
    except Exception:
        return None
    finally:
        try:
            if path_ptr.value:
                ctypes.windll.ole32.CoTaskMemFree(path_ptr)
        except Exception:
            pass


def documents_dir():
    path = windows_documents_dir() or (Path.home() / "Documents")
    path.mkdir(parents=True, exist_ok=True)
    return path


def random_four_digits():
    return f"{secrets.randbelow(10000):04d}"


def unique_pdf_path(markdown_path, target_dir=None):
    target = Path(target_dir) if target_dir is not None else documents_dir()
    target.mkdir(parents=True, exist_ok=True)
    stem = Path(markdown_path).stem or "report"
    candidate = target / f"{stem}.pdf"
    if not candidate.exists():
        return candidate
    for _ in range(100):
        candidate = target / f"{stem}_{random_four_digits()}.pdf"
        if not candidate.exists():
            return candidate
    raise FileExistsError(f"Unable to find a free PDF path for {stem}")


def render_inline_markdown(text):
    escaped = html.escape(str(text or ""))
    escaped = re.sub(r"`([^`]+)`", r"<code>\1</code>", escaped)
    escaped = re.sub(r"\*\*(.+?)\*\*", r"<strong>\1</strong>", escaped)
    return escaped


def split_markdown_table_row(line):
    text = str(line or "").strip()
    if text.startswith("|"):
        text = text[1:]
    if text.endswith("|"):
        text = text[:-1]

    cells = []
    cell = []
    escaped = False
    for char in text:
        if escaped:
            cell.append(char)
            escaped = False
            continue
        if char == "\\":
            escaped = True
            continue
        if char == "|":
            cells.append("".join(cell).strip())
            cell = []
            continue
        cell.append(char)
    if escaped:
        cell.append("\\")
    cells.append("".join(cell).strip())
    return cells


def is_markdown_table_separator(cells):
    return bool(cells) and all(
        re.match(r"^:?-{3,}:?$", str(cell or "").strip())
        for cell in cells
    )


def render_html_table(lines):
    rows = [split_markdown_table_row(line) for line in lines]
    if len(rows) > 1 and is_markdown_table_separator(rows[1]):
        headers = rows[0]
        body_rows = rows[2:]
    else:
        headers = []
        body_rows = rows

    parts = ["<table>"]
    if headers:
        parts.append("<thead><tr>")
        for cell in headers:
            parts.append(f"<th>{render_inline_markdown(cell)}</th>")
        parts.append("</tr></thead>")
    parts.append("<tbody>")
    for row in body_rows:
        parts.append("<tr>")
        tag = "td"
        for cell in row:
            parts.append(f"<{tag}>{render_inline_markdown(cell)}</{tag}>")
        parts.append("</tr>")
    parts.append("</tbody></table>")
    return "".join(parts)


def markdown_to_html_body(markdown_text):
    lines = str(markdown_text or "").splitlines()
    parts = []
    paragraph = []
    index = 0

    def flush_paragraph():
        if not paragraph:
            return
        parts.append(f"<p>{render_inline_markdown(' '.join(paragraph))}</p>")
        paragraph.clear()

    while index < len(lines):
        line = lines[index]
        stripped = line.strip()

        if not stripped:
            flush_paragraph()
            index += 1
            continue

        if stripped.startswith("|") and stripped.count("|") >= 2:
            flush_paragraph()
            table_lines = []
            while index < len(lines):
                candidate = lines[index].strip()
                if not (candidate.startswith("|") and candidate.count("|") >= 2):
                    break
                table_lines.append(lines[index])
                index += 1
            parts.append(render_html_table(table_lines))
            continue

        heading = re.match(r"^(#{1,6})\s+(.+)$", stripped)
        if heading:
            flush_paragraph()
            level = len(heading.group(1))
            parts.append(f"<h{level}>{render_inline_markdown(heading.group(2))}</h{level}>")
            index += 1
            continue

        if stripped in ("---", "***", "___"):
            flush_paragraph()
            parts.append("<hr>")
            index += 1
            continue

        if stripped.startswith(">"):
            flush_paragraph()
            quote_lines = []
            while index < len(lines) and lines[index].strip().startswith(">"):
                quote_lines.append(lines[index].strip()[1:].strip())
                index += 1
            quote = "<br>".join(render_inline_markdown(value) for value in quote_lines)
            parts.append(f"<blockquote>{quote}</blockquote>")
            continue

        list_match = re.match(r"^([-*]|\d+\.)\s+(.+)$", stripped)
        if list_match:
            flush_paragraph()
            ordered = list_match.group(1).endswith(".")
            tag = "ol" if ordered else "ul"
            items = []
            while index < len(lines):
                item_stripped = lines[index].strip()
                item_match = re.match(r"^([-*]|\d+\.)\s+(.+)$", item_stripped)
                if not item_match:
                    break
                if item_match.group(1).endswith(".") != ordered:
                    break
                items.append(item_match.group(2))
                index += 1
            parts.append(
                f"<{tag}>"
                + "".join(f"<li>{render_inline_markdown(item)}</li>" for item in items)
                + f"</{tag}>"
            )
            continue

        paragraph.append(stripped)
        index += 1

    flush_paragraph()
    return "\n".join(parts)


def markdown_to_pdf_html(markdown_text):
    body = markdown_to_html_body(markdown_text)
    return f"""<!doctype html>
<html>
<head>
  <meta charset="utf-8">
  <style>
    @page {{ size: A4; margin: 14mm; }}
    body {{
      color: #1f2933;
      font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC", "Microsoft YaHei", Arial, sans-serif;
      font-size: 11.5px;
      line-height: 1.55;
    }}
    h1, h2, h3, h4, h5, h6 {{ color: #111827; line-height: 1.25; margin: 18px 0 8px; }}
    h1 {{ font-size: 22px; }}
    h2 {{ border-bottom: 1px solid #d8dee9; font-size: 17px; padding-bottom: 4px; }}
    h3 {{ font-size: 14px; }}
    p {{ margin: 7px 0; }}
    blockquote {{
      border-left: 3px solid #9aa5b1;
      color: #4b5563;
      margin: 8px 0;
      padding: 6px 10px;
      background: #f8fafc;
    }}
    table {{ border-collapse: collapse; font-size: 9.5px; margin: 10px 0 14px; width: 100%; }}
    tr {{ break-inside: avoid; page-break-inside: avoid; }}
    th, td {{ border: 1px solid #d8dee9; padding: 5px 6px; text-align: left; vertical-align: top; word-break: break-word; }}
    th {{ background: #eef2f7; color: #111827; font-weight: 600; }}
    ul, ol {{ margin: 7px 0 10px 18px; padding: 0; }}
    li {{ margin: 3px 0; }}
    code {{ background: #eef2f7; border-radius: 3px; font-family: Consolas, monospace; padding: 1px 3px; }}
    hr {{ border: 0; border-top: 1px solid #d8dee9; margin: 16px 0; }}
  </style>
</head>
<body>
{body}
</body>
</html>"""


def launch_pdf_browser(playwright):
    errors = []
    for options in (
        {"headless": True},
        {"headless": True, "channel": "chrome"},
        {"headless": True, "channel": "msedge"},
    ):
        try:
            return playwright.chromium.launch(**options)
        except Exception as exc:
            errors.append(exc)
    raise errors[0]


def convert_markdown_to_pdf(markdown_text, pdf_path):
    from playwright.sync_api import sync_playwright

    document = markdown_to_pdf_html(markdown_text)
    with sync_playwright() as playwright:
        browser = launch_pdf_browser(playwright)
        try:
            page = browser.new_page()
            page.set_content(document, wait_until="load")
            page.pdf(
                path=str(pdf_path),
                format="A4",
                print_background=True,
                prefer_css_page_size=True,
                margin={
                    "top": "14mm",
                    "right": "12mm",
                    "bottom": "14mm",
                    "left": "12mm",
                },
            )
        finally:
            browser.close()


REPORT_FILE_LINKS_PLACEHOLDER = "{report_file_links}"


def export_pdf_report(markdown_path, markdown_text, pdf_path=None):
    pdf_path = Path(pdf_path) if pdf_path else unique_pdf_path(markdown_path)
    convert_markdown_to_pdf(markdown_text, pdf_path)
    return str(pdf_path)


def _to_accio_file_uri(path):
    """生成 Accio Work 可点击打开的 file URI。"""
    if not path:
        return ""
    p = str(path).strip()
    if not p:
        return ""

    norm = p.replace("\\", "/")
    encoded = quote(norm, safe="/:")
    if re.match(r"^[a-zA-Z]:/", norm):
        return "file:///" + encoded
    if norm.startswith("//"):
        return "file:" + encoded
    if norm.startswith("/"):
        return "file://" + encoded
    return encoded


def _to_accio_file_link(label, path):
    uri = _to_accio_file_uri(path)
    if not uri:
        return label
    needs_wrap = any(ch in uri for ch in " ()[]<>")
    target = f"<{uri}>" if needs_wrap else uri
    return f"[{label}]({target})"


def build_report_file_links(markdown_path, pdf_path=None, include_markdown=True, include_pdf=None):
    md_label = "📄 打开 Markdown 报告"
    pdf_label = "📕 打开 PDF 报告"
    if include_pdf is None:
        include_pdf = bool(pdf_path)
    if not include_markdown and not include_pdf:
        return ""

    md_link = _to_accio_file_link(md_label, markdown_path) if markdown_path else "Markdown: 未生成"

    lines = [
        "",
        "## 报告文件",
        "",
    ]
    raw_paths = []
    if include_markdown:
        lines.append(f"- {md_link}")
        raw_paths.append(f"MD: {markdown_path}")
    if include_pdf and pdf_path:
        pdf_link = _to_accio_file_link(pdf_label, pdf_path)
        lines.append(f"- {pdf_link}")
        raw_paths.append(f"PDF: {pdf_path}")
    if raw_paths:
        lines.extend([
            "",
            "<details><summary>查看原始路径（便于复制）</summary>",
            "",
            "```text",
        ])
        lines.extend(raw_paths)
        lines.extend([
            "```",
            "",
            "</details>",
        ])
    return "\n".join(lines).strip()


def append_report_file_paths(report, markdown_path, pdf_path=None, include_markdown=True, include_pdf=None):
    links = build_report_file_links(
        markdown_path,
        pdf_path,
        include_markdown=include_markdown,
        include_pdf=include_pdf,
    )
    if not links:
        return report.replace(REPORT_FILE_LINKS_PLACEHOLDER, "").rstrip() + "\n"
    if REPORT_FILE_LINKS_PLACEHOLDER in report:
        return report.replace(REPORT_FILE_LINKS_PLACEHOLDER, links)
    return report.rstrip() + "\n\n" + links + "\n"


def render_report_with_file_links(
    data,
    template,
    markdown_path,
    pdf_path=None,
    compare_data=None,
    include_markdown=True,
    include_pdf=None,
):
    links = build_report_file_links(
        markdown_path,
        pdf_path,
        include_markdown=include_markdown,
        include_pdf=include_pdf,
    )
    report = render_report(data, template, report_file_links=links, compare_data=compare_data)
    if REPORT_FILE_LINKS_PLACEHOLDER not in template:
        report = append_report_file_paths(
            report,
            markdown_path,
            pdf_path,
            include_markdown=include_markdown,
            include_pdf=include_pdf,
        )
    return report


def markdown_cell(value):
    text = "" if value is None else str(value)
    text = text.replace("\r", " ").replace("\n", " ")
    text = text.replace("|", "\\|")
    return text.strip() or "-"


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]
    unix_seconds = seconds - NTP_EPOCH_DELTA_SECONDS
    return datetime.fromtimestamp(unix_seconds, timezone.utc)


def local_utc_now():
    return datetime.now(timezone.utc)


def current_china_time():
    try:
        utc_now = fetch_apple_ntp_utc()
    except OSError:
        utc_now = local_utc_now()
    return utc_now.astimezone(CHINA_TZ)


def parse_china_datetime(value):
    text = str(value or "").strip()
    if not text:
        return None
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M", "%Y-%m-%d"):
        try:
            parsed = datetime.strptime(text, fmt)
            return parsed.replace(tzinfo=CHINA_TZ)
        except ValueError:
            continue
    return None


def format_notice_time(value):
    if value.second:
        return value.strftime("%Y-%m-%d %H:%M:%S")
    return value.strftime("%Y-%m-%d %H:%M")


def render_freshness_notice(data):
    end_time = parse_china_datetime(data.get("end_time"))
    if end_time is None:
        return ""
    now_china = current_china_time()
    if now_china >= end_time:
        return ""
    return (
        f"> 提示：当前中国时间 {format_notice_time(now_china)} "
        f"尚未到达本次统计窗口结束时间 {format_notice_time(end_time)}，"
        "以下数据为截至当前可抓取结果，窗口结束后可复核完整数据。\n"
    )


def render_actual_window_notice(data):
    start = parse_china_datetime(data.get("actual_start_time") or data.get("start_time"))
    end = parse_china_datetime(data.get("actual_end_time") or data.get("end_time"))
    if start is None or end is None:
        return ""

    basis = str(data.get("actual_window_basis") or "").strip()
    lines = [
        (
            f"> 本次实际抓取窗口：{format_notice_time(start)} 至 {format_notice_time(end)}"
            "（本地/北京时间，左闭右开）。"
        )
    ]
    if basis == "timestamp":
        la_start = start.astimezone(LA_TZ)
        la_end = end.astimezone(LA_TZ)
        lines.append(
            f"> 对应美西时间：{format_notice_time(la_start)} 至 {format_notice_time(la_end)}"
            "（America/Los_Angeles）。"
        )
        if (
            la_start.time() == datetime.min.time()
            and la_end.time() == datetime.min.time()
            and la_end.date() == la_start.date() + timedelta(days=1)
        ):
            lines.append(f"> 完整匹配美西 {la_start.strftime('%Y-%m-%d')}。")
    elif basis == "local_datetime_string":
        lines.append("> 该窗口来自日期/时间字符串输入，按本地/北京时间解释；不得直接改写为美西日期。")
    return "\n".join(lines) + "\n"


def dedupe_inquiries(inquiries):
    grouped = OrderedDict()
    for item in inquiries:
        inquiry_id = str(item.get("inquiry_id") or "")
        key = inquiry_id or f"row-{len(grouped)}"
        if key not in grouped:
            grouped[key] = item
            continue
        current = grouped[key]
        current_detail = current.get("detail") or {}
        new_detail = item.get("detail") or {}
        if len(new_detail.get("messages") or []) > len(current_detail.get("messages") or []):
            grouped[key] = item
    return list(grouped.values())


def first_reply_label(item):
    if item.get("detail_skipped"):
        return "-"
    detail = item.get("detail") or {}
    stats = detail.get("stats") or {}
    if stats.get("first_seller_response"):
        return stats["first_seller_response"]
    if stats.get("unreplied") is True:
        return "未回复"
    return "-"


def last_update_label(item):
    detail = item.get("detail") or {}
    stats = detail.get("stats") or {}
    if stats.get("unreplied") is True:
        return "未回复"
    if stats.get("last_msg_time"):
        return stats["last_msg_time"]
    return "-"


def user_level_label(value):
    level = str(value or "").strip()
    if not level:
        return "未识别"
    if level == "L0":
        return "L0（无等级）"
    return level


def detail_heading_label(item):
    buyer = markdown_cell(item.get("buyer_name"))
    level = markdown_cell(user_level_label(item.get("user_level")))
    country = markdown_cell(item.get("country"))
    meta_parts = [level]
    if country != "-":
        meta_parts.append(country)
    return f"{buyer}（{'｜'.join(meta_parts)}）"


def product_ids_label(item):
    ids = item.get("buyer_product_ids") or item.get("product_ids") or []
    if not ids:
        detail = item.get("detail") or {}
        ids = [
            card.get("product_id")
            for card in detail.get("product_cards") or []
            if card.get("product_id")
        ]
    ids = list(dict.fromkeys(str(value) for value in ids if value))
    return ", ".join(ids) if ids else "-"


def business_type_label(item):
    category = str(item.get("category") or "").strip()
    product = str(item.get("product") or "").strip()
    product_lower = product.lower()

    if product_lower == "inquiry from tm" or "from tm" in product_lower:
        return "TM 商机"
    if "product detail" in product_lower or "product details" in product_lower:
        return "询盘商机"
    if "询盘商机" in category:
        return "询盘商机"
    if category:
        return category
    return "-"


def has_valid_detail(item):
    if item.get("detail_skipped"):
        return False
    if not item.get("detail"):
        return False
    status = item.get("scrape_status", "")
    return not (isinstance(status, str) and status.startswith("error"))


def reply_status_label(item):
    if not has_valid_detail(item):
        return "未知"

    stats = (item.get("detail") or {}).get("stats") or {}
    if not stats:
        return "未知"
    if stats.get("unreplied") is True:
        return "未回复"
    last_speaker = str(stats.get("last_msg_speaker") or "").lower()
    if last_speaker == "buyer":
        return "待回复"
    if stats.get("first_seller_response") or last_speaker in {"seller", "bot"}:
        return "已回复"
    return "未知"


def normalize_inline_text(value):
    text = " ".join(str(value or "").replace("\r", " ").replace("\n", " ").split())
    return text or "-"


def speaker_label(value):
    speaker = str(value or "").strip().lower()
    if speaker == "buyer":
        return "买家"
    if speaker == "seller":
        return "卖家"
    if speaker == "bot":
        return "自动回复"
    if speaker:
        return speaker
    return "消息"


def parse_message_time(value):
    text = str(value or "").strip()
    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d %H:%M"):
        try:
            return datetime.strptime(text, fmt)
        except ValueError:
            continue
    return None


def ordered_messages(messages):
    items = []
    has_time = False
    for index, message in enumerate(messages or []):
        parsed = parse_message_time(message.get("timestamp"))
        has_time = has_time or parsed is not None
        items.append((index, parsed, message))
    if not has_time:
        return [message for _, _, message in items]
    items.sort(key=lambda item: (item[1] is None, item[1] or datetime.max, item[0]))
    return [message for _, _, message in items]


def conversation_preview(item, limit=5):
    detail = item.get("detail") or {}
    preview = []
    for message in ordered_messages(detail.get("messages") or []):
        body = normalize_inline_text(message.get("text"))
        if body == "-":
            continue
        timestamp = normalize_inline_text(message.get("timestamp"))
        preview.append({
            "speaker": speaker_label(message.get("speaker")),
            "timestamp": "" if timestamp == "-" else timestamp,
            "text": body,
        })
        if len(preview) >= limit:
            break
    if preview:
        return preview
    fallback = normalize_inline_text(item.get("last_message"))
    if fallback != "-":
        return [{"speaker": "消息", "timestamp": "", "text": fallback}]
    return []


def inquiry_content(item):
    detail = item.get("detail") or {}
    messages = detail.get("messages") or []
    buyer_messages = []
    for message in ordered_messages(messages):
        speaker = str(message.get("speaker") or "").lower()
        text = message.get("text") or ""
        if speaker == "buyer" or ("seller" not in speaker and "bot" not in speaker):
            timestamp = normalize_inline_text(message.get("timestamp"))
            body = normalize_inline_text(text)
            buyer_messages.append(f"{timestamp}: {body}" if timestamp != "-" else body)
    if buyer_messages:
        return "； ".join(buyer_messages)
    return normalize_inline_text(item.get("last_message"))


def report_date(data):
    if data.get("start_time") and data.get("end_time"):
        return format_display_date_window(
            str(data.get("start_time")),
            str(data.get("end_time")),
            separator=" 至 ",
        )
    return str(data.get("date") or "")


def safe_report_filename_part(value):
    text = str(value or "").strip().replace("-", "").replace(":", "").replace(" ", "_")
    text = re.sub(r'[<>"/\\|?*\x00-\x1f]+', "_", text)
    text = re.sub(r"_+", "_", text).strip("._ ")
    return text or "latest"


def output_path_for(data, explicit_output, output_dir=""):
    if explicit_output:
        return explicit_output
    base_dir = output_dir or PROJECT_DIR
    if data.get("start_time") and data.get("end_time"):
        display_window = format_display_date_window(
            str(data.get("start_time")),
            str(data.get("end_time")),
            separator="|",
        )
        if "|" in display_window:
            display_start, display_end = display_window.split("|", 1)
        else:
            display_start = display_window
            display_end = display_window
        start = safe_report_filename_part(display_start)
        end = safe_report_filename_part(display_end)
        return os.path.join(base_dir, f"report_{start}_{end}.md")
    date = safe_report_filename_part(data.get("date") or "latest")
    return os.path.join(base_dir, f"report_{date}.md")


def ensure_report_output_dir(path):
    if not path:
        return ""
    target = os.path.abspath(path)
    os.makedirs(target, exist_ok=True)
    probe = Path(target) / f".accio-write-test-{os.getpid()}-{random_four_digits()}.tmp"
    with open(probe, "w", encoding="utf-8") as handle:
        handle.write("ok")
    try:
        probe.unlink()
    except OSError:
        pass
    return target


def render_rows(inquiries):
    rows = []
    for idx, item in enumerate(inquiries, 1):
        rows.append(
            "| {idx} | {buyer} | {level} | {market} | {assignee} | {business_type} | {inquiry_id} | {product} | {product_ids} | {created} | {first_reply} | {last_update} | {reply_status} | {content} |".format(
                idx=idx,
                buyer=markdown_cell(item.get("buyer_name")),
                level=markdown_cell(user_level_label(item.get("user_level"))),
                market=markdown_cell(item.get("country")),
                assignee=markdown_cell(item.get("assignee")),
                business_type=markdown_cell(business_type_label(item)),
                inquiry_id=markdown_cell(item.get("inquiry_id")),
                product=markdown_cell(item.get("product")),
                product_ids=markdown_cell(product_ids_label(item)),
                created=markdown_cell(item.get("create_time")),
                first_reply=markdown_cell(first_reply_label(item)),
                last_update=markdown_cell(last_update_label(item)),
                reply_status=markdown_cell(reply_status_label(item)),
                content=markdown_cell(inquiry_content(item)),
            )
        )
    return "\n".join(rows) if rows else "| - | 暂无 | - | - | - | - | - | - | - | - | - | - | - | - |"


def render_summary_bullets(data, inquiries, top_countries, metrics=None):
    metrics = metrics or effective_metrics(data, inquiries)
    replied = sum(1 for item in inquiries if reply_status_label(item) == "已回复")
    pending = sum(1 for item in inquiries if reply_status_label(item) == "待回复")
    unreplied = sum(1 for item in inquiries if reply_status_label(item) == "未回复")
    unknown = sum(1 for item in inquiries if reply_status_label(item) == "未知")
    reply_parts = [
        f"已回复 {replied} 条",
        f"待回复 {pending} 条",
        f"未回复 {unreplied} 条",
    ]
    if unknown:
        reply_parts.append(f"另有 {unknown} 条因详情抓取受限显示为“未知”")
    return "\n".join([
        f"- 统计时段：{markdown_cell(str(data.get('scope_desc') or report_date(data)))}",
        f"- 询盘总量：{len(inquiries)} 条",
        "- 买家总数：{total} 人（询盘人数 {inquiry_buyers} / TM去重后 {tm}；询盘个数 {inquiry_count} / 原始TM {raw_tm}）".format(
            total=metrics.get("deduped_buyer_count")
            if metrics.get("deduped_buyer_count") is not None
            else metrics.get("inquiry_buyer_count", 0) + metrics.get("tm_buyer_count", 0),
            inquiry_buyers=metrics.get("inquiry_buyer_deduped_count", metrics.get("inquiry_buyer_count", 0)),
            inquiry_count=metrics.get("inquiry_count", 0),
            tm=metrics.get("tm_buyer_deduped_count", metrics.get("tm_buyer_count", 0)),
            raw_tm=metrics.get("tm_buyer_count", 0),
        ),
        f"- 主要市场：{markdown_cell(top_countries)}",
        f"- 回复情况：{'，'.join(reply_parts)}。",
    ])


def render_business_type_table(inquiries):
    total = len(inquiries)
    counts = Counter(business_type_label(item) for item in inquiries)
    rows = []
    for label, count in counts.most_common():
        ratio = f"{count / total:.0%}" if total else "0%"
        rows.append(f"| {markdown_cell(label)} | {count} | {ratio} |")
    return "\n".join(rows) if rows else "| - | 0 | 0% |"


def effective_metrics(data, inquiries):
    stored = data.get("metrics") or {}
    computed = build_metrics(inquiries)
    return {**stored, **computed}


def render_tm_dedup_note(metrics):
    raw_tm = int(metrics.get("tm_buyer_count") or 0)
    deduped_tm = int(metrics.get("tm_buyer_deduped_count", raw_tm) or 0)
    removed = int(metrics.get("tm_buyer_deduped_removed_count") or 0)
    if not removed:
        return f"- TM商机去重：无重复买家，去重后 TM {deduped_tm} 人"
    buyers = [
        str(item.get("buyer_name") or item.get("buyer_key") or "").strip()
        for item in metrics.get("tm_buyer_deduped_removed_buyers") or []
        if str(item.get("buyer_name") or item.get("buyer_key") or "").strip()
    ]
    buyer_label = "、".join(markdown_cell(name) for name in buyers[:5])
    if len(buyers) > 5:
        buyer_label += f" 等 {len(buyers)} 人"
    elif buyer_label:
        buyer_label = f"（{buyer_label}）"
    return f"- TM商机去重：原始 TM {raw_tm} 人，询盘覆盖重复买家 {removed} 人{buyer_label}，去重后 TM {deduped_tm} 人"


def render_business_type_bullets(inquiries, metrics=None):
    total = len(inquiries)
    counts = Counter(business_type_label(item) for item in inquiries)
    rows = []
    for label, count in counts.most_common():
        ratio = f"{count / total:.0%}" if total else "0%"
        rows.append(f"- {markdown_cell(label)}：{count} 条（{ratio}）")
    metrics = metrics or build_metrics(inquiries)
    rows.append(render_tm_dedup_note(metrics))
    return "\n".join(rows) if rows else "- 暂无"


def format_count_change(current, previous):
    delta = int(current or 0) - int(previous or 0)
    if previous:
        percent = f"{delta / previous:+.0%}"
    else:
        percent = "新增" if current else "0%"
    return delta, percent


def opportunity_type_counts(inquiries):
    counts = Counter(business_type_label(item) for item in inquiries)
    return {
        "TM 商机": counts.get("TM 商机", 0),
        "询盘商机": counts.get("询盘商机", 0),
    }


def level_share_counts(inquiries):
    total = len(inquiries)
    counts = Counter(user_level_label(item.get("user_level")) for item in inquiries)
    return {
        level: {
            "count": count,
            "share": (count / total) if total else 0,
        }
        for level, count in counts.items()
    }


def render_previous_day_comparison(current_inquiries, previous_inquiries):
    if previous_inquiries is None:
        return ""

    current_types = opportunity_type_counts(current_inquiries)
    previous_types = opportunity_type_counts(previous_inquiries)
    lines = [
        "## 前一日对比",
        "",
        "| 商机类型 | 当前数量 | 前一日数量 | 增减 | 增减百分比 |",
        "|---|---:|---:|---:|---:|",
    ]
    for label in ("TM 商机", "询盘商机"):
        current = current_types.get(label, 0)
        previous = previous_types.get(label, 0)
        delta, percent = format_count_change(current, previous)
        lines.append(f"| {markdown_cell(label)} | {current} | {previous} | {delta:+d} | {percent} |")

    lines.extend([
        "",
        "| 用户等级 | 当前占比 | 前一日占比 | 占比变化 |",
        "|---|---:|---:|---:|",
    ])
    current_levels = level_share_counts(current_inquiries)
    previous_levels = level_share_counts(previous_inquiries)
    for level in sorted(set(current_levels) | set(previous_levels)):
        current_share = current_levels.get(level, {}).get("share", 0)
        previous_share = previous_levels.get(level, {}).get("share", 0)
        diff_points = (current_share - previous_share) * 100
        lines.append(
            "| {level} | {current} | {previous} | {diff:+.0f} 个百分点 |".format(
                level=markdown_cell(level),
                current=f"{current_share:.0%}",
                previous=f"{previous_share:.0%}",
                diff=diff_points,
            )
        )
    return "\n".join(lines)


def assignee_followup_rows(inquiries):
    grouped = {}
    for item in inquiries:
        assignee = str(item.get("assignee") or "未分配").strip() or "未分配"
        if assignee not in grouped:
            grouped[assignee] = Counter()
        grouped[assignee]["total"] += 1
        grouped[assignee][reply_status_label(item)] += 1

    return [
        {
            "assignee": assignee,
            "total": grouped[assignee].get("total", 0),
            "replied": grouped[assignee].get("已回复", 0),
            "pending": grouped[assignee].get("待回复", 0),
            "unreplied": grouped[assignee].get("未回复", 0),
            "unknown": grouped[assignee].get("未知", 0),
        }
        for assignee in sorted(grouped)
    ]


def render_assignee_table(inquiries):
    summary_rows = assignee_followup_rows(inquiries)
    rows = []
    for row in summary_rows:
        rows.append(
            "| {assignee} | {total} | {replied} | {pending} | {unreplied} | {unknown} |".format(
                assignee=markdown_cell(row["assignee"]),
                total=row["total"],
                replied=row["replied"],
                pending=row["pending"],
                unreplied=row["unreplied"],
                unknown=row["unknown"],
            )
        )
    return "\n".join(rows) if rows else "| - | 0 | 0 | 0 | 0 | 0 |"


def render_followup_bullets(inquiries):
    rows = []
    for item in inquiries:
        status = reply_status_label(item)
        if status == "已回复":
            continue
        buyer = markdown_cell(item.get("buyer_name"))
        country = markdown_cell(item.get("country"))
        assignee = markdown_cell(item.get("assignee") or "未分配")
        content = markdown_cell(inquiry_content(item))
        if len(content) > 240:
            content = content[:220].rstrip() + "（内容较长，详见第5模块）"
        rows.append(f"- **{buyer}**（{country}，{assignee}）：{content}")
    return "\n".join(rows) if rows else "- 暂无待跟进重点。"


def render_followup_table(inquiries):
    rows = []
    for idx, item in enumerate(inquiries, 1):
        status = reply_status_label(item)
        if status == "已回复":
            continue
        rows.append(
            "| {idx} | {buyer} | {assignee} | {status} | {business_type} | {content} |".format(
                idx=idx,
                buyer=markdown_cell(item.get("buyer_name")),
                assignee=markdown_cell(item.get("assignee") or "未分配"),
                status=markdown_cell(status),
                business_type=markdown_cell(business_type_label(item)),
                content=markdown_cell(inquiry_content(item)),
            )
        )
    return "\n".join(rows) if rows else "| - | 暂无 | - | - | - | - |"


def render_preview_lines(item):
    preview = conversation_preview(item)
    if not preview:
        return "  1. 暂无可展示对话。"
    lines = []
    for index, message in enumerate(preview, 1):
        timestamp = f"{message['timestamp']}：" if message.get("timestamp") else ""
        lines.append(
            "  {index}. [{speaker}] {timestamp}{text}".format(
                index=index,
                speaker=markdown_cell(message.get("speaker")),
                timestamp=markdown_cell(timestamp),
                text=markdown_cell(message.get("text")),
            )
        )
    return "\n".join(lines)


def render_detail_cards(inquiries):
    cards = []
    for idx, item in enumerate(inquiries, 1):
        buyer = detail_heading_label(item)
        card = "\n".join([
            f"### {idx}. {buyer}",
            "- 基本信息：{market} / {assignee} / {business_type} / {reply_status} / 询盘ID：{inquiry_id}".format(
                market=markdown_cell(item.get("country")),
                assignee=markdown_cell(item.get("assignee") or "未分配"),
                business_type=markdown_cell(business_type_label(item)),
                reply_status=markdown_cell(reply_status_label(item)),
                inquiry_id=markdown_cell(item.get("inquiry_id")),
            ),
            "- 用户等级：{level}".format(
                level=markdown_cell(user_level_label(item.get("user_level"))),
            ),
            "- 产品信息：{product}；产品ID：{product_ids}".format(
                product=markdown_cell(item.get("product")),
                product_ids=markdown_cell(product_ids_label(item)),
            ),
            "- 时间信息（北京时间）：创建 {created}；第一次回复 {first_reply}；最后更新 {last_update}".format(
                created=markdown_cell(item.get("create_time")),
                first_reply=markdown_cell(first_reply_label(item)),
                last_update=markdown_cell(last_update_label(item)),
            ),
            "- 前五句对话：",
            render_preview_lines(item),
            "- 摘要素材：最终面向用户回复时，请基于以上对话和基本信息生成 1-2 句自然中文业务摘要。",
        ])
        cards.append(card)
    return "\n\n".join(cards) if cards else "暂无询盘明细。"


def failed_detail_label(item):
    buyer = markdown_cell(item.get("buyer_name"))
    inquiry_id = markdown_cell(item.get("inquiry_id"))
    if buyer != "-" and inquiry_id != "-":
        return f"{buyer} / {inquiry_id}"
    if buyer != "-":
        return buyer
    return inquiry_id


def failed_detail_items(data, inquiries):
    failed_ids = {
        str(value).strip()
        for value in data.get("failed") or []
        if str(value).strip()
    }
    items_by_id = {
        str(item.get("inquiry_id") or "").strip(): item
        for item in inquiries
        if str(item.get("inquiry_id") or "").strip()
    }
    items = []
    seen = set()
    for failed_id in sorted(failed_ids):
        item = items_by_id.get(failed_id)
        if item:
            items.append(item)
            seen.add(failed_id)
    for item in inquiries:
        inquiry_id = str(item.get("inquiry_id") or "").strip()
        status = str(item.get("scrape_status") or "")
        if item.get("detail_skipped") and status.startswith("error:") and inquiry_id not in seen:
            items.append(item)
            if inquiry_id:
                seen.add(inquiry_id)
    return items


def render_data_quality_notes(data, inquiries):
    notes = []
    if data.get("partial") is True:
        notes.append("当前结果标记为 partial，正式使用前需要复核。")
    failed_items = failed_detail_items(data, inquiries)
    failed_keys = {
        str(item.get("inquiry_id") or "").strip() or str(id(item))
        for item in failed_items
    }
    skipped_without_failure = [
        item for item in inquiries
        if item.get("detail_skipped")
        and (str(item.get("inquiry_id") or "").strip() or str(id(item))) not in failed_keys
    ]
    if skipped_without_failure:
        notes.append(f"{len(skipped_without_failure)} 条询盘详情被跳过，回复状态可能为未知。")
    if failed_items:
        labels = "，".join(failed_detail_label(item) for item in failed_items[:10])
        if len(failed_items) > 10:
            labels += " ..."
        notes.append(f"{len(failed_items)} 条详情补抓后仍失败，建议复核：{labels}")
    warnings = data.get("warnings") or []
    notes.extend(str(warning) for warning in warnings if warning)
    if not notes:
        notes.append("未发现影响报告生成的数据完整性问题。")
    return "\n".join(f"- {markdown_cell(note)}" for note in notes)


def render_report(data, template, report_file_links="", compare_data=None):
    detail_inquiries = data.get("inquiries") or []
    inquiries = dedupe_inquiries(detail_inquiries)
    previous_inquiries = None
    if compare_data is not None:
        previous_inquiries = dedupe_inquiries(compare_data.get("inquiries") or [])
    countries = Counter(item.get("country") for item in inquiries if item.get("country"))
    assignees = sorted({item.get("assignee") for item in inquiries if item.get("assignee")})
    products = {item.get("product") for item in inquiries if item.get("product")}
    metrics = effective_metrics(data, inquiries)
    unreplied = sum(
        1
        for item in inquiries
        if (item.get("detail") or {}).get("stats", {}).get("unreplied") is True
    )
    replied = max(0, len(inquiries) - unreplied)
    top_countries = ", ".join(f"{country}({count})" for country, count in countries.most_common(5)) or "-"
    freshness_notice = render_freshness_notice(data) if "{freshness_notice}" in template else ""
    actual_window_notice = render_actual_window_notice(data) if "{actual_window_notice}" in template else ""

    replacements = {
        "{generated_at}": str(data.get("generated_at") or ""),
        "{report_date}": report_date(data),
        "{scope_desc}": str(data.get("scope_desc") or report_date(data)),
        "{freshness_notice}": freshness_notice,
        "{actual_window_notice}": actual_window_notice,
        "{total_count}": str(len(inquiries)),
        "{inquiry_buyer_count}": str(metrics.get("inquiry_buyer_count", 0)),
        "{inquiry_buyer_deduped_count}": str(metrics.get("inquiry_buyer_deduped_count", metrics.get("inquiry_buyer_count", 0))),
        "{inquiry_count}": str(metrics.get("inquiry_count", 0)),
        "{tm_buyer_count}": str(metrics.get("tm_buyer_count", 0)),
        "{tm_buyer_deduped_count}": str(metrics.get("tm_buyer_deduped_count", metrics.get("tm_buyer_count", 0))),
        "{tm_buyer_deduped_removed_count}": str(metrics.get("tm_buyer_deduped_removed_count", 0)),
        "{deduped_buyer_count}": str(metrics.get("deduped_buyer_count", metrics.get("inquiry_buyer_count", 0) + metrics.get("tm_buyer_count", 0))),
        "{business_opportunity_count}": FROZEN_BUSINESS_OPPORTUNITY_LABEL,
        "{replied_count}": str(replied),
        "{unreplied_count}": str(unreplied),
        "{product_types}": str(len(products)),
        "{country_count}": str(len(countries)),
        "{top_countries}": top_countries,
        "{assignees}": ", ".join(assignees) or "-",
        "{active_period}": "全天",
        "{summary_bullets}": render_summary_bullets(data, inquiries, top_countries, metrics),
        "{one_liner_table}": render_rows(inquiries),
        "{inquiry_detail_table}": render_rows(detail_inquiries),
        "{inquiry_detail_cards}": render_detail_cards(detail_inquiries),
        "{business_type_table}": render_business_type_table(inquiries),
        "{business_type_bullets}": render_business_type_bullets(inquiries, metrics),
        "{previous_day_comparison}": render_previous_day_comparison(inquiries, previous_inquiries),
        "{assignee_followup_table}": render_assignee_table(inquiries),
        "{followup_table}": render_followup_table(inquiries),
        "{followup_bullets}": render_followup_bullets(inquiries),
        "{data_quality_notes}": render_data_quality_notes(data, inquiries),
        REPORT_FILE_LINKS_PLACEHOLDER: report_file_links,
    }

    report = template
    for key, value in replacements.items():
        report = report.replace(key, value)
    report = report.replace(
        f"{FROZEN_BUSINESS_OPPORTUNITY_LABEL} 个",
        FROZEN_BUSINESS_OPPORTUNITY_LABEL,
    )
    report = report.replace(
        f"{FROZEN_BUSINESS_OPPORTUNITY_LABEL} 条",
        FROZEN_BUSINESS_OPPORTUNITY_LABEL,
    )
    return report


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(args, environ=None):
    if getattr(args, "no_pdf", False):
        return False
    if getattr(args, "pdf", False):
        return True
    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(args, environ=None):
    if getattr(args, "no_md", False):
        return False
    if getattr(args, "md", False):
        return True
    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 main():
    parser = argparse.ArgumentParser(description="生成阿里询盘报告")
    parser.add_argument("--input", default=DEFAULT_INPUT)
    parser.add_argument("--template", default=DEFAULT_TEMPLATE)
    parser.add_argument("--output", default="")
    parser.add_argument("--output-dir", default="")
    parser.add_argument("--compare-input", default="")
    parser.add_argument("--allow-partial", action="store_true")
    parser.add_argument("--md", "--markdown", dest="md", action="store_true", help="生成 Markdown 报告文件；默认由配置决定，当前默认关闭")
    parser.add_argument("--no-md", "--no-markdown", dest="no_md", action="store_true", help="禁用 Markdown 报告文件导出")
    parser.add_argument("--pdf", action="store_true", help="生成 PDF 报告；默认由配置决定，当前默认关闭")
    parser.add_argument("--no-pdf", action="store_true", help="禁用 PDF 报告生成")
    args = parser.parse_args()
    export_markdown = resolve_export_markdown(args)
    export_pdf = resolve_export_pdf(args)
    if args.output:
        export_markdown = True

    output_dir = ""
    if args.output_dir and (export_markdown or export_pdf) and not args.output:
        try:
            output_dir = ensure_report_output_dir(args.output_dir)
        except OSError as exc:
            print_json({
                "status": "error",
                "error": "report_output_dir_unwritable",
                "output_dir": args.output_dir,
                "message": str(exc),
            }, exit_code=2)

    validation = validate_inquiries(path=args.input)
    if not args.allow_partial and not validation["reportable"]:
        print_json({
            "status": "error",
            "error": "数据不完整，拒绝生成正式报告",
            "validation": validation,
        }, exit_code=2)

    data = read_json(args.input)
    compare_data = read_json(args.compare_input) if args.compare_input else None
    template = read_text(args.template)
    output_path = output_path_for(data, args.output, output_dir)
    payload = {
        "status": "ok",
        "total": len(dedupe_inquiries(data.get("inquiries") or [])),
        "validation": validation,
        "export_markdown": export_markdown,
        "export_pdf": export_pdf,
    }
    payload["assignee_followup"] = assignee_followup_rows(dedupe_inquiries(data.get("inquiries") or []))
    if export_markdown:
        payload["output"] = output_path
    pdf_target_dir = Path(output_path).parent if output_dir else None
    planned_pdf_path = str(unique_pdf_path(output_path, target_dir=pdf_target_dir)) if export_pdf else None
    report = render_report_with_file_links(
        data,
        template,
        output_path,
        planned_pdf_path,
        compare_data=compare_data,
        include_markdown=export_markdown,
        include_pdf=export_pdf,
    )
    if export_markdown:
        write_text(output_path, report)
    else:
        attach_report_text_payload(payload, report, data, output_path)
    if export_pdf:
        try:
            payload["pdf_output"] = export_pdf_report(output_path, report, planned_pdf_path)
        except Exception:
            report = render_report_with_file_links(
                data,
                template,
                output_path,
                None,
                compare_data=compare_data,
                include_markdown=export_markdown,
                include_pdf=False,
            )
            if export_markdown:
                write_text(output_path, report)
            else:
                attach_report_text_payload(payload, report, data, output_path)
        else:
            if payload["pdf_output"] != planned_pdf_path:
                report = render_report_with_file_links(
                    data,
                    template,
                    output_path,
                    payload["pdf_output"],
                    compare_data=compare_data,
                    include_markdown=export_markdown,
                    include_pdf=True,
                )
                if export_markdown:
                    write_text(output_path, report)
                else:
                    attach_report_text_payload(payload, report, data, output_path)
    print_json(payload)


if __name__ == "__main__":
    main()
