"""
alibaba-inquiry-analyst / detail_scraper.py
===========================================
详情页抓取模块 - DOM提取唯一路径。

核心原则：
- 仅使用DOM结构提取消息（准确率100%）
- DOM提取失败直接抛出异常，无fallback
"""

from __future__ import annotations

from html import unescape
import re
import time
from urllib.parse import parse_qs, unquote, urlparse

from playwright.async_api import BrowserContext, Page, TimeoutError as PlaywrightTimeout

from browser_pages import new_background_page

from config import (
    DETAIL_TIMEOUT_MS,
    WAIT_DETAIL_RENDER_MS,
    DEPTH_CONFIG,
)
from cleaner import SPEAKER_BUYER, SPEAKER_SELLER, SPEAKER_BOT, compute_stats


class ScrapeError(Exception):
    """抓取错误"""
    pass


AUTO_REPLY_MARKERS = (
    "\u81ea\u52a8\u63a5\u5f85\u53d1\u9001",
    "\u4e0a\u6761\u6d88\u606f\u662f\u81ea\u52a8\u56de\u590d",
    "\u81ea\u52a8\u63a5\u5f85\u5c06\u4e0d\u518d\u89e6\u53d1",
    "\u81ea\u52a8\u63a5\u5f85\u5bf9\u8bdd\u8f6e\u6b21\u5df2\u8fbe\u4e0a\u9650",
    "\u5f53\u524d\u4f1a\u8bdd\u81ea\u52a8\u63a5\u5f85\u5df2\u505c\u6b62",
)

PRODUCT_CARD_TYPES = {"3", "6", "2000", "2028"}
HISTORY_WAIT_AFTER_SCROLL_MS = 3500
HISTORY_STABLE_ROUNDS_TO_STOP = 2
HISTORY_MAX_ROUNDS = 30
HISTORY_MAX_SECONDS = 180


# ── DOM提取消息 ────────────────────────────────────────────────────────────

def clean_message_content(text: str, buyer_name: str = "") -> str:
    """
    清理消息内容中的状态标记和噪音。
    """
    lines = text.split('\n')
    cleaned = []

    for line in lines:
        line_stripped = line.strip()

        # 跳过状态标记
        if line_stripped in ['已读', '未读', '反馈', '自动接待发送']:
            continue

        # 跳过翻译标记
        if line_stripped in ['由阿里翻译提供', '阿里国际Marco大模型翻译', '阿里翻译']:
            continue

        # 跳过纯时间戳行
        if re.match(r'^20\d{2}-\d{2}-\d{2} \d{2}:\d{2}$', line_stripped):
            continue

        # 移除买家名称前缀
        if buyer_name and line_stripped.startswith(buyer_name):
            line = line.replace(buyer_name, '', 1).strip()
            line_stripped = line

        if line_stripped:
            cleaned.append(line)

    return '\n'.join(cleaned).strip()


def is_auto_reply_message(text: str) -> bool:
    """Return True when a right-side message looks like an auto-reply."""
    normalized = re.sub(r"\s+", "", text or "")
    return any(marker in normalized for marker in AUTO_REPLY_MARKERS)


def speaker_from_class(class_attr: str, raw_text: str = "") -> str | None:
    if "item-left" in (class_attr or ""):
        return SPEAKER_BUYER
    if "item-right" in (class_attr or ""):
        return SPEAKER_BOT if is_auto_reply_message(raw_text) else SPEAKER_SELLER
    return None


def extract_timestamp(text: str) -> str:
    time_match = re.search(r"(20\d{2}-\d{2}-\d{2} \d{2}:\d{2})", text or "")
    return time_match.group(1) if time_match else ""


def dedupe_preserve_order(values: list[str]) -> list[str]:
    seen = set()
    result = []
    for value in values:
        text = str(value or "").strip()
        if not text or text in seen:
            continue
        seen.add(text)
        result.append(text)
    return result


def card_type_from_url(raw_url: str) -> str:
    try:
        query = parse_qs(urlparse(unquote(unescape(raw_url or ""))).query)
    except Exception:
        return ""
    return (query.get("type") or [""])[0]


def _append_workspace_card_candidates(candidates: list[dict], raw_url: str, source_prefix: str = "workspace_card") -> None:
    normalized_url = unquote(unescape(raw_url or ""))
    card_type = card_type_from_url(normalized_url)
    if card_type not in PRODUCT_CARD_TYPES:
        return

    try:
        query = parse_qs(urlparse(normalized_url).query)
    except Exception:
        query = {}

    for key in ("id", "ids", "productId", "product_id", "offerId", "offer_id", "itemId", "item_id"):
        for value in query.get(key, []):
            for product_id in re.findall(r"\d{8,}", str(value)):
                candidates.append({
                    "product_id": product_id,
                    "source": f"{source_prefix}:{key}",
                    "card_type": card_type,
                    "raw_url": normalized_url,
                })


def product_id_candidates(raw_url: str = "", html: str = "") -> list[dict]:
    """Extract product id candidates from product/TM/inquiry card payloads."""
    candidates = []
    normalized_url = unescape(raw_url or "")
    decoded_url = unquote(normalized_url)
    card_type = card_type_from_url(normalized_url)

    _append_workspace_card_candidates(candidates, normalized_url)

    text = unquote(unescape(f"{html or ''} {decoded_url}"))
    for match in re.finditer(r"https?://[^\"'\s<>]+", text):
        _append_workspace_card_candidates(candidates, match.group(0), source_prefix="dom_url")

    for match in re.finditer(r"(?:https?:)?//[^\"'\s<>]+?product-detail/[^\"'\s<>]*?-(\d{8,})\.html", text):
        candidates.append({
            "product_id": match.group(1),
            "source": "product_detail_url",
            "card_type": card_type,
            "raw_url": match.group(0),
        })

    for match in re.finditer(r"\bproduct-detail/[^\"'\s<>]*?-(\d{8,})\.html", text):
        candidates.append({
            "product_id": match.group(1),
            "source": "product_detail_url",
            "card_type": card_type,
            "raw_url": match.group(0),
        })

    for label, pattern in (
        ("clickContentId", r'["\']?clickContentId["\']?\s*[:=]\s*["\']?(\d{8,})["\']?'),
        ("productId", r'["\']?productId["\']?\s*[:=]\s*["\']?(\d{8,})["\']?'),
        ("offerId", r'["\']?offerId["\']?\s*[:=]\s*["\']?(\d{8,})["\']?'),
        ("itemId", r'["\']?itemId["\']?\s*[:=]\s*["\']?(\d{8,})["\']?'),
    ):
        for match in re.finditer(pattern, text, re.I):
            candidates.append({
                "product_id": match.group(1),
                "source": label,
                "card_type": card_type,
                "raw_url": normalized_url,
            })

    deduped = []
    seen = set()
    for candidate in candidates:
        product_id = candidate["product_id"]
        if product_id in seen:
            continue
        seen.add(product_id)
        deduped.append(candidate)
    return deduped


def looks_like_price_or_quantity(line: str) -> bool:
    text = (line or "").strip()
    if not text:
        return True
    if re.match(r"^[A-Z]$", text):
        return True
    if re.search(r"[$￥¥€£]|CN¥|\d+\s*-\s*\d+|-\d+%", text):
        return True
    return text in {
        "认证",
        "DOC认证",
        "询盘",
        "Inquiry from TM",
        "采购数量",
        "详细需求",
        "查看详情",
        "立即报价",
        "最小订购量:",
        "最小订购量",
        "Unit",
        "Set",
        "Piece",
    }


def extract_product_title_from_text(text: str, buyer_name: str = "") -> str:
    lines = []
    for line in (text or "").splitlines():
        value = line.strip()
        if not value:
            continue
        if buyer_name and value == buyer_name:
            continue
        if extract_timestamp(value) == value:
            continue
        if value in {"已读", "未读", "反馈", "系统自动发送"}:
            continue
        lines.append(value)

    markers = {"认证", "DOC认证", "询盘", "Inquiry from TM"}
    for index, line in enumerate(lines):
        if line in markers:
            for candidate in lines[index + 1:]:
                if not looks_like_price_or_quantity(candidate):
                    return candidate

    stop_words = ("采购数量", "详细需求", "ID:", "查看详情", "立即报价", "最小订购量")
    candidates = [
        line for line in lines
        if not looks_like_price_or_quantity(line)
        and not any(word in line for word in stop_words)
        and len(line) > 12
    ]
    return max(candidates, key=len) if candidates else ""


async def extract_product_cards_from_dom(page: Page, buyer_name: str = "") -> list[dict]:
    """Extract product cards from loaded message DOM."""
    cards = []
    msg_elements = await page.query_selector_all(".message-item-wrapper")

    for index, el in enumerate(msg_elements):
        class_attr = await el.get_attribute("class") or ""
        raw_text = await el.inner_text()
        raw_url = await el.get_attribute("data-original") or ""
        dx_source = await el.get_attribute("data-dx-event-datasource") or ""
        clk_info = await el.get_attribute("data-clkinfo") or ""
        exp_info = await el.get_attribute("data-expinfo") or ""
        try:
            html = await el.inner_html()
        except Exception:
            html = ""
        candidates = product_id_candidates(raw_url, f"{dx_source} {clk_info} {exp_info} {html}")
        if not candidates:
            continue

        speaker = speaker_from_class(class_attr, raw_text) or "system"
        title = extract_product_title_from_text(raw_text, buyer_name=buyer_name)
        timestamp = extract_timestamp(raw_text)
        for candidate in candidates:
            cards.append({
                "timestamp": timestamp,
                "speaker": speaker,
                "product_id": candidate["product_id"],
                "product_title": title,
                "card_type": candidate.get("card_type") or "",
                "source": candidate.get("source") or "",
                "raw_url": candidate.get("raw_url") or raw_url,
                "message_index": index,
            })

    return cards


async def extract_messages_from_dom(page: Page, buyer_name: str = "") -> list[dict]:
    """
    从页面DOM结构提取消息。

    DOM结构特征：
    - item-left: 买家消息（左侧）
    - item-right: 卖家消息（右侧）
    """
    messages = []
    msg_elements = await page.query_selector_all('.message-item-wrapper')

    for el in msg_elements:
        class_attr = await el.get_attribute('class') or ''
        raw_text = await el.inner_text()

        speaker = speaker_from_class(class_attr, raw_text)
        if speaker == SPEAKER_SELLER:
            auto_reply = await el.query_selector('.auto-reply-footer')
            speaker = SPEAKER_BOT if auto_reply or is_auto_reply_message(raw_text) else SPEAKER_SELLER
        if speaker is None:
            continue

        text = clean_message_content(raw_text, buyer_name)

        timestamp = extract_timestamp(raw_text)

        if text:
            messages.append({
                "timestamp": timestamp,
                "text": text,
                "speaker": speaker
            })

    return messages


# ── 历史消息加载 ────────────────────────────────────────────────────────────

async def load_all_history(page: Page, depth: str = "normal", return_meta: bool = False):
    """根据深度加载历史消息，返回时间戳数量或加载元数据。"""
    chat = await page.query_selector(".common-load-more")
    if not chat:
        meta = {
            "timestamp_count": 0,
            "message_count": 0,
            "rounds": 0,
            "stable_rounds": 0,
            "partial": False,
            "stop_reason": "no_chat_container",
        }
        return meta if return_meta else 0

    cfg = DEPTH_CONFIG[depth]
    max_scroll = cfg["max_scroll"]

    # quick keeps the no-scroll behavior; normal/full rely on the stability gate
    # so long TM histories can keep loading without stopping at an arbitrary 3 rounds.
    max_scroll = 0 if max_scroll == 0 else HISTORY_MAX_ROUNDS

    async def collect_state():
        try:
            return await page.evaluate(
                """
                () => {
                    const body = document.body?.innerText || "";
                    const messages = [...document.querySelectorAll(".message-item-wrapper")];
                    const messageIds = messages.map((el) => {
                        const expInfo = el.getAttribute("data-expinfo") || el.getAttribute("data-clkinfo") || "";
                        try {
                            const parsed = JSON.parse(expInfo);
                            if (parsed.messageId) return String(parsed.messageId);
                        } catch (_) {}
                        const original = el.getAttribute("data-original") || "";
                        const text = (el.innerText || "").replace(/\\s+/g, " ").slice(0, 160);
                        return `${text}|${original.slice(0, 160)}`;
                    }).filter(Boolean);
                    const chat = document.querySelector(".common-load-more");
                    return {
                        timestamp_count: (body.match(/20\\d\\d-\\d{2}-\\d{2} \\d{2}:\\d{2}/g) || []).length,
                        message_count: messages.length,
                        message_ids: [...new Set(messageIds)],
                        scroll_top: chat ? chat.scrollTop : null,
                        scroll_height: chat ? chat.scrollHeight : null,
                        client_height: chat ? chat.clientHeight : null,
                    };
                }
                """
            )
        except Exception:
            return {
                "timestamp_count": 0,
                "message_count": 0,
                "message_ids": [],
                "scroll_top": None,
                "scroll_height": None,
                "client_height": None,
            }

    if max_scroll <= 0:
        state = await collect_state()
        meta = {
            **state,
            "rounds": 0,
            "stable_rounds": 0,
            "partial": False,
            "stop_reason": "depth_quick",
        }
        return meta if return_meta else meta["timestamp_count"]

    state = await collect_state()
    seen_ids = set(state.get("message_ids") or [])
    stable_rounds = 0
    rounds = 0
    stop_reason = "max_rounds"
    started_at = time.monotonic()

    for rounds in range(1, max_scroll + 1):
        await page.evaluate(
            """
            (el) => {
                el.scrollTop = 0;
                el.dispatchEvent(new Event("scroll", {bubbles: true}));
                el.dispatchEvent(new WheelEvent("wheel", {
                    deltaY: -1600,
                    bubbles: true,
                    cancelable: true,
                    view: window,
                }));
            }
            """,
            chat,
        )
        await page.wait_for_timeout(HISTORY_WAIT_AFTER_SCROLL_MS)
        new_state = await collect_state()
        new_ids = set(new_state.get("message_ids") or [])
        added_ids = new_ids - seen_ids
        grew = (
            bool(added_ids)
            or new_state.get("message_count", 0) > state.get("message_count", 0)
            or new_state.get("timestamp_count", 0) > state.get("timestamp_count", 0)
        )
        if grew:
            stable_rounds = 0
        else:
            stable_rounds += 1
        seen_ids |= new_ids
        state = new_state

        if stable_rounds >= HISTORY_STABLE_ROUNDS_TO_STOP:
            stop_reason = "stable"
            break
        if time.monotonic() - started_at >= HISTORY_MAX_SECONDS:
            stop_reason = "max_seconds"
            break

    partial = stop_reason in {"max_rounds", "max_seconds"}
    meta = {
        **state,
        "rounds": rounds,
        "stable_rounds": stable_rounds,
        "partial": partial,
        "stop_reason": stop_reason,
    }
    if partial:
        meta["warning"] = f"history_load_partial:{stop_reason}"
    return meta if return_meta else meta["timestamp_count"]


# ── 详情抓取主函数 ──────────────────────────────────────────────────────────

async def scrape_detail(
    ctx: BrowserContext,
    detail_url: str,
    depth: str = "normal",
    buyer_name: str = "",
    request_page_gc: bool = False,
) -> dict:
    """
    访问详情URL抓取对话。

    参数：
    - ctx: BrowserContext
    - detail_url: 详情页URL
    - depth: quick/normal/full
    - buyer_name: 买家名称（用于内容清理）
    - request_page_gc: 关闭详情页前调用 Playwright 官方页面 GC

    返回：
    {"messages": [...], "stats": {...}, "total_loaded": int}

    异常：
    - ScrapeError: DOM提取失败或URL无效
    - PlaywrightTimeout: 页面加载超时
    """
    if not detail_url:
        raise ScrapeError("detail_url is empty")

    # URL处理
    if detail_url.startswith("/"):
        detail_url = f"https://message.alibaba.com{detail_url}"
    elif not detail_url.startswith("http"):
        detail_url = f"https://message.alibaba.com/{detail_url}"

    # 安全校验
    from urllib.parse import urlparse
    if urlparse(detail_url).hostname != "message.alibaba.com":
        raise ScrapeError(f"invalid host: {urlparse(detail_url).hostname}")

    detail_page = None
    try:
        detail_page = await new_background_page(ctx)
        await detail_page.goto(detail_url, timeout=DETAIL_TIMEOUT_MS)
        await detail_page.wait_for_load_state("domcontentloaded", timeout=DETAIL_TIMEOUT_MS)
        await detail_page.wait_for_timeout(WAIT_DETAIL_RENDER_MS)

        history_load = await load_all_history(detail_page, depth=depth, return_meta=True)

        # DOM提取（唯一路径）
        messages = await extract_messages_from_dom(detail_page, buyer_name)
        product_cards = await extract_product_cards_from_dom(detail_page, buyer_name)

        if not messages:
            raise ScrapeError("DOM extraction failed: no messages found")

        history_summary = {
            "rounds": history_load.get("rounds", 0),
            "stable_rounds": history_load.get("stable_rounds", 0),
            "partial": history_load.get("partial", False),
            "stop_reason": history_load.get("stop_reason", ""),
            "message_count": history_load.get("message_count", 0),
            "timestamp_count": history_load.get("timestamp_count", 0),
        }
        if history_load.get("warning"):
            history_summary["warning"] = history_load["warning"]

        return {
            "messages": messages,
            "stats": compute_stats(messages),
            "total_loaded": history_load.get("timestamp_count", 0),
            "history_load": history_summary,
            "product_cards": product_cards,
        }

    except PlaywrightTimeout:
        raise ScrapeError("page load timeout")
    finally:
        if detail_page:
            if request_page_gc:
                try:
                    page_gc = getattr(detail_page, "request_gc", None)
                    if callable(page_gc):
                        await page_gc()
                except Exception:
                    pass
            try:
                await detail_page.close()
            except Exception:
                pass
