"""阿里巴巴询盘列表页抓取模块。"""

from __future__ import annotations

import asyncio
import re
import warnings
from datetime import date, datetime, timedelta
from urllib.parse import urlparse

from playwright.async_api import ElementHandle, Page

from cleaner import is_today_time
from config import MAX_PAGES, SELECTOR_TIMEOUT_MS, USER_LEVEL_ICON_MAP, WAIT_AFTER_NAV_MS
from page_selectors import SEL, build_url


TIME_PATTERN = r"\d{4}-\d{1,2}-\d{1,2}|\d{1,2}:\d{2}"
TRANSIENT_NAVIGATION_ERRORS = (
    "Execution context was destroyed",
    "Cannot find context with specified id",
    "Frame was detached",
    "Navigation interrupted by another one",
)
BROWSER_CLOSED_ERROR_MARKERS = (
    "Target page, context or browser has been closed",
    "Target closed",
    "Browser has been closed",
    "Connection closed",
    "Page crashed",
)
LOGIN_REQUIRED_MARKER = "LOGIN_REQUIRED"
LOGIN_REQUIRED_MESSAGE = (
    "LOGIN_REQUIRED: 已跳转到 login.alibaba.com，当前浏览器未登录阿里国际站。"
    "已取消抓取；请在新打开的浏览器里完成登录后重新运行查询。"
)


class ListScrapeError(Exception):
    """List page scraping error."""


class LoginRequiredError(ListScrapeError):
    """Raised when Alibaba redirects the message center to login."""


def is_login_url(url: str) -> bool:
    try:
        host = urlparse(str(url or "")).hostname or ""
    except ValueError:
        return False
    return host == "login.alibaba.com" or host.endswith(".login.alibaba.com")


async def assert_not_login_redirect(page: Page) -> None:
    current_url = str(getattr(page, "url", "") or "")
    if is_login_url(current_url):
        raise LoginRequiredError(f"{LOGIN_REQUIRED_MESSAGE} 当前地址: {current_url}")


def normalize_time_range(
    start_date: datetime,
    end_date: datetime,
    *,
    clamp_future_end: bool = True,
) -> tuple[datetime, datetime]:
    """Normalize inputs to a [start, end) window while preserving explicit times."""
    start_dt = start_date.replace(microsecond=0)
    end_dt = end_date.replace(microsecond=0)
    tomorrow_start = datetime.combine(date.today(), datetime.min.time()) + timedelta(days=1)

    if clamp_future_end and end_dt > tomorrow_start:
        end_dt = tomorrow_start
    if start_dt >= end_dt:
        raise ListScrapeError(
            "--end-time must be later than --start-time (end is exclusive). "
            "For one day like 2026-04-14, use --end-time 2026-04-15."
        )
    return start_dt, end_dt


def build_time_range_timestamps(
    start_date: datetime,
    end_date: datetime,
    *,
    clamp_future_end: bool = True,
) -> tuple[int, int, datetime, datetime]:
    start_dt, end_dt = normalize_time_range(start_date, end_date, clamp_future_end=clamp_future_end)
    return int(start_dt.timestamp() * 1000), int(end_dt.timestamp() * 1000), start_dt, end_dt


def parse_item_time_window(filter_value: str) -> tuple[datetime, datetime] | None:
    try:
        if re.match(r"^\d{4}-\d{1,2}-\d{1,2}$", filter_value or ""):
            start = datetime.strptime(filter_value, "%Y-%m-%d")
            return start, start + timedelta(days=1)
        if re.match(r"^\d{1,2}:\d{2}$", filter_value or ""):
            parsed_time = datetime.strptime(filter_value, "%H:%M").time()
            start = datetime.combine(date.today(), parsed_time)
            return start, start + timedelta(minutes=1)
    except (TypeError, ValueError):
        return None
    return None


def parse_item_date(filter_value: str) -> datetime | None:
    window = parse_item_time_window(filter_value)
    return window[0] if window else None


def item_time_overlaps(filter_value: str, range_start: datetime, range_end: datetime) -> bool | None:
    item_window = parse_item_time_window(filter_value)
    if item_window is None:
        return None
    item_start, item_end = item_window
    return item_start < range_end and item_end > range_start


async def has_no_data_notice(list_page: Page) -> bool:
    try:
        if await list_page.locator(SEL["no_data_notice"]).count() > 0:
            return True
    except Exception:
        pass
    try:
        body_text = await list_page.locator("body").inner_text(timeout=1000)
        return any(marker in body_text for marker in ("暂无数据", "No data", "No Data"))
    except Exception:
        return False


async def goto_advanced_search_page(list_page: Page, page_num: int) -> bool:
    clicked = await list_page.evaluate(
        """
        (targetPage) => {
            const target = Number(targetPage);
            const pages = document.querySelector('.ui2-pagination-pages');
            if (!pages) return false;

            const direct = pages.querySelector(`a[data-page="${target}"]`);
            if (direct) {
                direct.click();
                return true;
            }

            const next = pages.querySelector('a.next[data-role="next"]');
            const current = pages.querySelector('a.current');
            const currentPage = Number(current?.getAttribute('data-page') || current?.textContent || 0);
            if (next && currentPage + 1 === target) {
                next.click();
                return true;
            }

            const input = document.querySelector('.ui2-pagination-goto');
            const go = document.querySelector('.ui2-pagination-go');
            if (input && go) {
                input.focus();
                input.value = String(target);
                input.dispatchEvent(new Event('input', { bubbles: true }));
                input.dispatchEvent(new Event('change', { bubbles: true }));
                go.click();
                return true;
            }
            return false;
        }
        """,
        page_num,
    )
    return bool(clicked)


def is_transient_navigation_error(exc: Exception) -> bool:
    message = str(exc)
    return any(marker in message for marker in TRANSIENT_NAVIGATION_ERRORS)


def is_browser_closed_error(exc: Exception) -> bool:
    message = str(exc)
    return any(marker in message for marker in BROWSER_CLOSED_ERROR_MARKERS)


def browser_closed_error(exc: Exception) -> ListScrapeError:
    return ListScrapeError(
        "Chrome/CDP 浏览器异常关闭，可能是浏览器 OOM、Chrome 崩溃、远程调试端口失联或页面被手动关闭。"
        f"原始错误: {type(exc).__name__}: {exc}"
    )


async def navigate_to_url(list_page: Page, url: str, timeout_ms: int = SELECTOR_TIMEOUT_MS) -> None:
    timeout_seconds = max(1.0, timeout_ms / 1000) + 5
    try:
        await asyncio.wait_for(
            list_page.goto(url, wait_until="domcontentloaded", timeout=timeout_ms),
            timeout=timeout_seconds,
        )
    except asyncio.TimeoutError as exc:
        raise ListScrapeError(
            f"List page navigation timed out after {timeout_ms}ms. "
            "Chrome tab may be crashed or unresponsive."
        ) from exc
    except Exception as exc:
        if is_browser_closed_error(exc):
            raise browser_closed_error(exc) from exc
        raise ListScrapeError(
            f"List page navigation failed: {type(exc).__name__}: {exc}"
        ) from exc


async def wait_for_list_ready(
    list_page: Page,
    *,
    prev_count: int = 0,
    require_growth: bool = False,
) -> tuple[int, bool, bool]:
    waited_ms = 0
    poll_interval_ms = 500
    current_count = 0

    while waited_ms < SELECTOR_TIMEOUT_MS:
        await assert_not_login_redirect(list_page)
        try:
            if await has_no_data_notice(list_page):
                return len(await list_page.query_selector_all(SEL["list_item"])), True, True

            current_count = len(await list_page.query_selector_all(SEL["list_item"]))
            if current_count > 0 and not require_growth:
                return current_count, True, False
            if require_growth and current_count > prev_count:
                return current_count, True, False

            loading = await list_page.query_selector("[class*='loading'], .loading-mask, .spin")
        except Exception as exc:
            if is_browser_closed_error(exc):
                raise browser_closed_error(exc) from exc
            if not is_transient_navigation_error(exc):
                raise
            await list_page.wait_for_timeout(poll_interval_ms)
            waited_ms += poll_interval_ms
            continue

        await list_page.wait_for_timeout(poll_interval_ms)
        waited_ms += poll_interval_ms

        if require_growth and loading is None and waited_ms >= 3000:
            try:
                current_count = len(await list_page.query_selector_all(SEL["list_item"]))
            except Exception as exc:
                if is_browser_closed_error(exc):
                    raise browser_closed_error(exc) from exc
                if not is_transient_navigation_error(exc):
                    raise
                continue
            if current_count >= prev_count:
                return current_count, True, False

    return current_count, False, False


async def load_list_page(
    list_page: Page,
    page_num: int,
    owner_id: int | None = None,
    start_time_ms: int | None = None,
    end_time_ms: int | None = None,
    combine_owner_time_search: bool = False,
    buyer_name: str = "",
) -> tuple[int, bool, bool]:
    advanced_time = start_time_ms is not None and end_time_ms is not None
    search_url_mode = advanced_time or bool(buyer_name)
    url = build_url(
        page_num,
        owner_id=owner_id,
        start_time_ms=start_time_ms,
        end_time_ms=end_time_ms,
        include_owner_with_time=combine_owner_time_search,
        buyer_name=buyer_name,
    )
    try:
        prev_count = len(await list_page.query_selector_all(SEL["list_item"]))
    except Exception as exc:
        if is_transient_navigation_error(exc):
            prev_count = 0
        elif is_browser_closed_error(exc):
            raise browser_closed_error(exc) from exc
        else:
            raise

    if search_url_mode and page_num > 1:
        jumped = await goto_advanced_search_page(list_page, page_num)
        if not jumped:
            await list_page.wait_for_timeout(WAIT_AFTER_NAV_MS)
            try:
                return len(await list_page.query_selector_all(SEL["list_item"])), False, False
            except Exception as exc:
                if is_browser_closed_error(exc):
                    raise browser_closed_error(exc) from exc
                raise
    else:
        await navigate_to_url(list_page, url)
        await list_page.wait_for_timeout(500)
        await assert_not_login_redirect(list_page)

    if search_url_mode and page_num == 1:
        for _ in range(4):
            await list_page.wait_for_timeout(350)
            await assert_not_login_redirect(list_page)
            if "isAdvanceSearch=true" in (list_page.url or ""):
                await navigate_to_url(list_page, url)
                await list_page.wait_for_timeout(350)
                await assert_not_login_redirect(list_page)
                break

    try:
        count, success, reached_end = await wait_for_list_ready(
            list_page,
            prev_count=prev_count,
            require_growth=(page_num > 1 and not advanced_time),
        )
    except LoginRequiredError:
        raise
    except Exception as exc:
        warnings.warn(
            f"List page load exception (page={page_num}): {type(exc).__name__}: {exc}",
            RuntimeWarning,
        )
        count, success, reached_end = prev_count, False, False

    await list_page.wait_for_timeout(WAIT_AFTER_NAV_MS)
    await assert_not_login_redirect(list_page)
    return count, success, reached_end


def parse_times_from_text(text: str) -> tuple[str, str]:
    text = (text or "").replace("\u00a0", " ")
    update_match = re.search(rf"更新时间\s*[:：]?\s*({TIME_PATTERN})", text)
    create_match = re.search(rf"创建时间\s*[:：]?\s*({TIME_PATTERN})", text)

    if update_match or create_match:
        return (
            update_match.group(1) if update_match else "",
            create_match.group(1) if create_match else "",
        )

    times = re.findall(TIME_PATTERN, text)
    if not times:
        return "", ""
    if len(times) == 1:
        return times[0], ""
    return times[0], times[1]


async def extract_times(item_el: ElementHandle) -> tuple[str, str]:
    try:
        time_el = await item_el.query_selector(SEL["time_row"])
        text = (await time_el.inner_text()).strip() if time_el else await item_el.inner_text()
        return parse_times_from_text(text)
    except Exception:
        return "", ""


async def extract_inquiry_id(item_el: ElementHandle) -> str:
    try:
        id_el = await item_el.query_selector(SEL["inquiry_id"])
        if id_el:
            match = re.search(r"\d{8,}", await id_el.inner_text())
            if match:
                return match.group(0)
        trade_id = await item_el.get_attribute("data-trade-id") or ""
        if trade_id and re.match(r"\d{6,}", trade_id):
            return trade_id
    except Exception:
        pass
    return ""


async def extract_detail_url(item_el: ElementHandle) -> str:
    try:
        url = await item_el.get_attribute("href")
        if url and "maDetail.htm" in url:
            return url
        link = await item_el.query_selector("a[href*='maDetail']")
        if link:
            return await link.get_attribute("href") or ""
    except Exception:
        pass
    return ""


async def safe_text(item_el: ElementHandle, selector: str) -> str:
    try:
        el = await item_el.query_selector(selector)
        return (await el.inner_text()).strip() if el else ""
    except Exception:
        return ""


async def extract_country(item_el: ElementHandle) -> str:
    try:
        flag_el = await item_el.query_selector(SEL["country_flag"])
        if not flag_el:
            return ""
        title = await flag_el.get_attribute("title") or ""
        return title.split("[")[0].strip() if "[" in title else title.strip()
    except Exception:
        return ""


def normalize_user_level(icon_src: str = "") -> str:
    for signature, level in USER_LEVEL_ICON_MAP.items():
        if signature in (icon_src or ""):
            return level
    return ""


async def extract_user_level(item_el: ElementHandle) -> tuple[str, str]:
    try:
        icon_el = await item_el.query_selector(SEL["user_level_icon"])
        if not icon_el:
            return "L0", ""
        icon_src = await icon_el.get_attribute("src") or ""
        return normalize_user_level(icon_src), icon_src
    except Exception:
        return "L0", ""


async def scrape_list(
    list_page: Page,
    max_items: int | None,
    start_date: datetime | None = None,
    end_date: datetime | None = None,
    filter_category: str = "",
    filter_assignee: str = "",
    filter_buyer_name: str = "",
    owner_id: int | None = None,
    combine_owner_time_search: bool = False,
    clamp_future_end: bool = True,
) -> tuple[list, dict, dict]:
    inquiries = []
    skipped_by_date = 0
    skipped_by_category = 0
    skipped_by_assignee = 0
    skipped_by_buyer_name = 0
    skipped_by_parse = 0
    seen_ids = set()
    metadata = {
        "partial": False,
        "pages_loaded": 0,
        "pages_checked": 0,
        "warnings": [],
    }
    consecutive_time_miss_pages = 0

    advanced_time = start_date is not None and end_date is not None
    range_start = None
    range_end = None
    start_time_ms = None
    end_time_ms = None
    if advanced_time:
        start_time_ms, end_time_ms, range_start, range_end = build_time_range_timestamps(
            start_date,
            end_date,
            clamp_future_end=clamp_future_end,
        )
    server_start_time_ms = None if filter_buyer_name else start_time_ms
    server_end_time_ms = None if filter_buyer_name else end_time_ms
    skip_default_today_filter = bool(filter_buyer_name) and range_start is None

    count, success, reached_end = await load_list_page(
        list_page,
        1,
        owner_id=owner_id,
        start_time_ms=server_start_time_ms,
        end_time_ms=server_end_time_ms,
        combine_owner_time_search=combine_owner_time_search,
        buyer_name=filter_buyer_name,
    )
    if not success:
        raise ListScrapeError(
            "List page #1 load failed. Please check login status, page structure, and network connectivity."
        )
    metadata["pages_checked"] = 1
    if reached_end and count == 0:
        return inquiries, _skipped_stats(
            skipped_by_date,
            skipped_by_category,
            skipped_by_assignee,
            skipped_by_buyer_name,
            skipped_by_parse,
        ), metadata

    for page_num in range(1, MAX_PAGES + 1):
        if page_num > 1:
            count, success, reached_end = await load_list_page(
                list_page,
                page_num,
                owner_id=owner_id,
                start_time_ms=server_start_time_ms,
                end_time_ms=server_end_time_ms,
                combine_owner_time_search=combine_owner_time_search,
                buyer_name=filter_buyer_name,
            )
            if not success:
                metadata["partial"] = True
                metadata["warnings"].append(
                    f"Page {page_num} load failed; returning partial results "
                    f"(loaded {metadata['pages_loaded']} pages, checked {metadata['pages_checked']} pages)."
                )
                break
            metadata["pages_checked"] = page_num
            if reached_end and count == 0:
                break

        try:
            items = await list_page.query_selector_all(SEL["list_item"])
        except Exception as exc:
            if is_browser_closed_error(exc):
                raise browser_closed_error(exc) from exc
            raise
        time_matched_this_page = 0

        for item in items:
            if max_items is not None and len(inquiries) >= max_items:
                break

            inquiry_id = await extract_inquiry_id(item)
            if not inquiry_id or inquiry_id in seen_ids:
                continue
            seen_ids.add(inquiry_id)

            update_time, create_time = await extract_times(item)
            filter_value = create_time

            if range_start is not None and range_end is not None:
                overlaps = item_time_overlaps(filter_value, range_start, range_end)
                if overlaps is None:
                    skipped_by_parse += 1
                    continue
                if not overlaps:
                    skipped_by_date += 1
                    continue
            elif skip_default_today_filter:
                pass
            else:
                if not is_today_time(filter_value):
                    skipped_by_date += 1
                    continue

            if update_time and not create_time and is_today_time(update_time):
                # Alibaba sometimes omits create_time in edge cases; keep the raw update_time
                # in the item, but still treat filtering as create_time-first only.
                pass

            time_matched_this_page += 1

            buyer_name = await safe_text(item, SEL["buyer_name"])
            if filter_buyer_name and filter_buyer_name.lower() not in buyer_name.lower():
                skipped_by_buyer_name += 1
                continue

            category = await safe_text(item, SEL["category"])
            assignee = await safe_text(item, SEL["assignee"])

            if filter_category:
                if filter_category.lower() == "tm" and "TM" not in category:
                    skipped_by_category += 1
                    continue
                if filter_category.lower() not in category.lower():
                    skipped_by_category += 1
                    continue

            if filter_assignee and assignee and filter_assignee.lower() not in assignee.lower():
                skipped_by_assignee += 1
                continue

            user_level, user_level_icon = await extract_user_level(item)
            inquiry = {
                "inquiry_id": inquiry_id,
                "detail_url": await extract_detail_url(item),
                "buyer_name": buyer_name,
                "product": await safe_text(item, SEL["product_name"]),
                "last_message": await safe_text(item, SEL["last_msg"]),
                "category": category,
                "assignee": assignee,
                "country": await extract_country(item),
                "user_level_icon": user_level_icon,
                "update_time": update_time,
                "create_time": create_time,
                "scrape_status": "pending",
                "detail": None,
            }
            if user_level:
                inquiry["user_level"] = user_level
            inquiries.append(inquiry)

        if time_matched_this_page == 0:
            consecutive_time_miss_pages += 1
            if consecutive_time_miss_pages >= 2:
                break
        else:
            metadata["pages_loaded"] += 1
            consecutive_time_miss_pages = 0

        if max_items is not None and len(inquiries) >= max_items:
            break

    return inquiries, _skipped_stats(
        skipped_by_date,
        skipped_by_category,
        skipped_by_assignee,
        skipped_by_buyer_name,
        skipped_by_parse,
    ), metadata


def _skipped_stats(
    skipped_by_date: int,
    skipped_by_category: int,
    skipped_by_assignee: int,
    skipped_by_buyer_name: int,
    skipped_by_parse: int,
) -> dict:
    return {
        "by_date": skipped_by_date,
        "by_category": skipped_by_category,
        "by_assignee": skipped_by_assignee,
        "by_buyer_name": skipped_by_buyer_name,
        "by_parse": skipped_by_parse,
        "total": (
            skipped_by_date
            + skipped_by_category
            + skipped_by_assignee
            + skipped_by_buyer_name
            + skipped_by_parse
        ),
    }
