import argparse
from pathlib import Path

import pytest

from gyxx_flow.modules.content_marketing import bilibili_scraper as bili
from gyxx_flow.modules.content_marketing import collection_completeness as cc
from gyxx_flow.modules.content_marketing import pgy_xhs_scraper_v2 as pgy
from gyxx_flow.modules.content_marketing import run_all
from gyxx_flow.modules.content_marketing import self_douyin_scraper as self_dy
from gyxx_flow.modules.content_marketing import xingtu_scraper_v2 as xingtu
from gyxx_flow.modules.content_marketing.serialized_page_action import (
    is_transient_browser_error,
    run_serialized_page_action,
)


def test_normalize_title_handles_nfkc_case_and_zero_width():
    assert cc.normalize_title("ＭａｃＢｏｏｋ\u200b 通勤包！") == "macbook通勤包"


def test_coerce_url_unwraps_feishu_markdown_links():
    value = "[查看笔记](https://www.xiaohongshu.com/discovery/item/abc123?x=1)"
    assert cc.coerce_url(value) == "https://www.xiaohongshu.com/discovery/item/abc123?x=1"


def test_match_tasks_prefers_note_id_over_changed_title():
    tasks = [{
        "record_id": "r1",
        "target_title": "飞书中的旧标题",
        "note_url": "https://www.xiaohongshu.com/explore/abc123",
        "note_id": "abc123",
    }]
    cards = [
        {"title": "完全不同的新标题", "href": "https://www.xiaohongshu.com/explore/abc123", "note_id": "abc123"},
        {"title": "飞书中的旧标题", "href": "https://www.xiaohongshu.com/explore/other", "note_id": "other"},
    ]

    matched = cc.match_tasks_to_cards(tasks, cards, platform="xhs")

    assert matched["r1"]["note_id"] == "abc123"
    assert matched["r1"]["match_method"] == "content_id"


def test_match_tasks_accepts_unique_truncated_title_but_rejects_ambiguous_title():
    task = {"record_id": "r1", "target_title": "男生长期主义通勤双肩包分享"}
    unique = [
        {"title": "男生长期主义通勤双肩包"},
        {"title": "夏日轻量斜挎包"},
    ]
    ambiguous = [
        {"title": "男生长期主义通勤双肩包"},
        {"title": "男生长期主义通勤双肩包"},
    ]

    assert cc.match_tasks_to_cards([task], unique, platform="douyin")["r1"]
    assert "r1" not in cc.match_tasks_to_cards([task], ambiguous, platform="douyin")


def test_title_similarity_ignores_episode_prefix_and_hashtag_suffix():
    target = "第20集：Pocket4拍旋焦，无后期也能出片？"
    platform_title = (
        "Pocket4拍旋焦，无后期也能出片？ #旋焦 #pocket4 #摄影装备 "
        "无论Pocket3还是Pocket4都可以轻松拍出旋焦效果！"
    )

    assert cc.title_similarity(platform_title, target) >= 0.98


def test_title_similarity_rejects_different_episode_numbers():
    assert cc.title_similarity(
        "第19集：Pocket4拍旋焦，无后期也能出片？",
        "第20集：Pocket4拍旋焦，无后期也能出片？",
    ) == 0.0


def test_match_tasks_rejects_duplicate_exact_titles_with_different_ids():
    task = {"record_id": "r1", "target_title": "完全相同的目标标题"}
    cards = [
        {"title": "完全相同的目标标题", "note_id": "video-1"},
        {"title": "完全相同的目标标题", "note_id": "video-2"},
    ]

    assert "r1" not in cc.match_tasks_to_cards([task], cards, platform="douyin")


def test_one_card_is_not_reused_for_two_different_creator_tasks():
    tasks = [
        {
            "record_id": "loose",
            "target_title": "commuter backpack review today",
            "note_url": "https://v.douyin.com/loose-source/",
        },
        {
            "record_id": "exact",
            "target_title": "commuter backpack review",
            "note_url": "https://v.douyin.com/exact-source/",
        },
    ]
    cards = [{"title": "commuter backpack review", "note_id": "observed-1"}]

    matched = cc.match_tasks_to_cards(tasks, cards, platform="douyin")

    assert set(matched) == {"exact"}


def test_same_source_url_can_share_one_card_across_styles():
    tasks = [
        {
            "record_id": "style-1",
            "target_title": "同一篇跨款式笔记",
            "note_url": "https://v.douyin.com/same-short-link/",
        },
        {
            "record_id": "style-2",
            "target_title": "同一篇跨款式笔记",
            "note_url": "https://v.douyin.com/same-short-link/",
        },
    ]
    cards = [{"title": "同一篇跨款式笔记", "note_id": "observed-1"}]

    assert set(cc.match_tasks_to_cards(tasks, cards, platform="douyin")) == {
        "style-1", "style-2",
    }


def test_known_mismatched_content_ids_cannot_fall_back_to_same_title():
    task = {
        "record_id": "r1",
        "target_title": "完全相同标题",
        "note_id": "video-a",
    }
    card = {"title": "完全相同标题", "note_id": "video-b"}

    assert cc.match_tasks_to_cards([task], [card], platform="douyin") == {}


def test_xingtu_can_use_strong_title_when_share_id_is_canonicalized():
    task = {
        "record_id": "r1",
        "target_title": "这家包颜值和实用性真挺高",
        "note_id": "7665297310758210854",
    }
    card = {
        "title": "这家包颜值和实用性真挺高 #潮流运动 #光影行星",
        "note_id": "canonical- Xingtu-item-id",
    }

    matched = cc.match_tasks_to_cards(
        [task],
        [card],
        platform="douyin",
        allow_mismatched_content_id_title=True,
    )

    assert matched["r1"]["match_method"] == "title"
    assert matched["r1"]["match_score"] >= 0.9


def test_xingtu_api_and_dom_duplicate_are_merged_before_matching():
    cards = [
        {
            "title": "同一条视频",
            "play_count": 19481,
            "note_id": "video-1",
            "source": "show_items_api",
        },
        {
            "title": "同一条视频",
            "play_count": "1.9w",
            "note_id": "",
            "page": 1,
        },
    ]

    deduped = cc.deduplicate_observed_cards(cards)
    matched = cc.match_tasks_to_cards(
        [{"record_id": "r1", "target_title": "同一条视频"}],
        deduped,
        platform="douyin",
    )

    assert len(deduped) == 1
    assert matched["r1"]["note_id"] == "video-1"


def test_merge_partial_summary_keeps_existing_success_rows():
    existing = {
        "style": "款式A", "index": 1, "total": 2, "matched": 1, "filled": 1,
        "results": [
            {"record_id": "ok", "status": "success", "matched": True, "write_ok": True},
            {"record_id": "retry", "status": "retryable_failure", "matched": False},
        ],
    }
    partial = {
        "style": "款式A", "index": 1, "total": 1, "matched": 1, "filled": 1,
        "results": [
            {"record_id": "retry", "status": "success", "matched": True, "write_ok": True},
        ],
    }

    merged = cc.merge_style_summary(existing, partial)

    assert merged["total"] == 2
    assert {row["record_id"] for row in merged["results"]} == {"ok", "retry"}
    assert merged["matched"] == 2
    assert merged["filled"] == 2
    assert merged["complete"] is True


def test_summary_contract_detects_missing_or_write_failure():
    summary = {
        "total": 2,
        "results": [
            {"record_id": "ok", "status": "success", "matched": True, "write_ok": True},
            {"record_id": "bad", "status": "write_failure", "matched": True, "write_ok": False},
        ],
    }
    finalized = cc.finalize_summary(summary, dry_run=False)
    assert finalized["complete"] is False
    assert finalized["unresolved"] == 1

    missing = cc.finalize_summary({"total": 2, "results": summary["results"][:1]}, dry_run=False)
    assert missing["complete"] is False
    assert missing["missing_results"] == 1


def _style_for_platform(platform: str):
    return {
        "index": 1,
        "name": "款式A",
        "base_token": "base",
        "table_id": "table",
        "field_map": {
            "creator_name": {"field_id": "name"},
            "creator_id": {"field_id": "creator_id"},
            "note_title": {"field_id": "title"},
            "publish_time": {"field_id": "pub"},
            "note_url": {"field_id": "url"},
            "platform": {"field_id": "platform"},
            "read_count_7d": {"field_id": "read7"},
        },
    }


def test_extractors_preserve_note_url_and_content_id(monkeypatch):
    xhs_row = {
        "record_id": "xhs-r", "name": "达人", "creator_id": "xhs-id",
        "title": "标题", "pub": "2026-07-20", "platform": "小红书",
        "url": "https://www.xiaohongshu.com/explore/abc123", "read7": None,
    }
    dy_row = {
        "record_id": "dy-r", "name": "达人", "creator_id": "dy-id",
        "title": "标题", "pub": "2026-07-20", "platform": "抖音",
        "url": "https://www.douyin.com/video/7654321", "read7": None,
    }
    monkeypatch.setattr(pgy, "list_records_by_table", lambda *_: [xhs_row])
    monkeypatch.setattr(xingtu, "list_records_by_table", lambda *_: [dy_row])

    xhs_task = pgy.extract_target_tasks(_style_for_platform("小红书"))[0]
    dy_task = xingtu.extract_target_tasks(_style_for_platform("抖音"))[0]

    assert xhs_task["note_url"].endswith("/abc123")
    assert xhs_task["note_id"] == "abc123"
    assert dy_task["note_url"].endswith("/7654321")
    assert dy_task["note_id"] == "7654321"


@pytest.mark.parametrize(
    ("module", "platform", "filled_url"),
    [
        (pgy, "小红书", "https://www.douyin.com/video/7654321"),
        (xingtu, "抖音", "http://8.99 分享文案 https://v.douyin.com/abc123/"),
    ],
)
def test_filled_publish_link_triggers_collection_without_platform_validation(
    monkeypatch, module, platform, filled_url,
):
    row = {
        "record_id": "r1",
        "name": "达人",
        "creator_id": "creator-id",
        "title": "已发布标题",
        "pub": "2026-07-20",
        "platform": platform,
        "url": filled_url,
        "read7": None,
    }
    monkeypatch.setattr(module, "list_records_by_table", lambda *_: [row])

    task = module.extract_target_tasks(_style_for_platform(platform))[0]

    assert task["input_error"] is None


def test_extractors_can_collect_missing_title_when_direct_url_has_content_id(monkeypatch):
    xhs_style = _style_for_platform("小红书")
    dy_style = _style_for_platform("抖音")
    xhs_row = {
        "record_id": "xhs-r", "name": "达人", "creator_id": "xhs-id",
        "title": None, "pub": "2026-07-20", "platform": "小红书",
        "url": "https://www.xiaohongshu.com/explore/abc123", "read7": None,
    }
    dy_row = {
        "record_id": "dy-r", "name": "达人", "creator_id": "dy-id",
        "title": None, "pub": "2026-07-20", "platform": "抖音",
        "url": "https://www.douyin.com/video/7654321", "read7": None,
    }
    monkeypatch.setattr(pgy, "list_records_by_table", lambda *_: [xhs_row])
    monkeypatch.setattr(xingtu, "list_records_by_table", lambda *_: [dy_row])

    assert pgy.extract_target_tasks(xhs_style)[0]["note_id"] == "abc123"
    assert xingtu.extract_target_tasks(dy_style)[0]["note_id"] == "7654321"
    assert xhs_style["_extract_stats"]["skipped_incomplete"] == 0
    assert dy_style["_extract_stats"]["skipped_incomplete"] == 0


def test_missing_critical_mapping_is_an_incomplete_summary(monkeypatch):
    style = _style_for_platform("抖音")
    del style["field_map"]["note_url"]
    monkeypatch.setattr(
        xingtu,
        "list_records_by_table",
        lambda *_: (_ for _ in ()).throw(AssertionError("mapping should fail first")),
    )

    _, summaries = xingtu.collect_tasks_across_styles([style], None)
    finalized = cc.finalize_summary(summaries[1], dry_run=True)

    assert "note_url" in finalized["error"]
    assert finalized["complete"] is False
    assert finalized["unresolved"] >= 1


class _FakeInput:
    def __init__(self):
        self.filled = []

    def click(self):
        return None

    def fill(self, value):
        self.filled.append(value)

    def press(self, _key):
        return None


class _FakeLocator:
    def __init__(self, input_):
        self.first = input_

    def filter(self, **_kwargs):
        return self


class _FakeSearchPage:
    def __init__(self):
        self.input = _FakeInput()
        self.has_result_calls = 0

    def wait_for_timeout(self, _ms):
        return None

    def evaluate(self, script, arg=None):
        if "const rows" in script:
            self.has_result_calls += 1
            return self.has_result_calls >= 2
        if "const inputs" in script:
            return ""
        raise AssertionError(script)

    def locator(self, _selector):
        return _FakeLocator(self.input)

    def wait_for_url(self, *_args, **_kwargs):
        return None


def test_pgy_search_retry_keeps_creator_id_query():
    page = _FakeSearchPage()

    found = pgy._ensure_search_results(
        page, query="xhs-id-123", result_name="达人昵称", max_retries=2
    )

    assert found is True
    assert page.input.filled == ["xhs-id-123"]


def test_apply_matched_task_reports_write_failure(monkeypatch):
    task = {
        "record_id": "r1", "publish_time": object(),
        "style_context": {
            "index": 1, "base_token": "base", "table_id": "table", "field_map": {},
        },
    }
    summary = {"matched": 0, "filled": 0, "skipped_no_pubtime": 0, "results": []}
    monkeypatch.setattr(pgy, "pick_read_field", lambda *_: ("read_count_7d", "fid", "7天曝光量"))
    monkeypatch.setattr(pgy, "write_back", lambda *_: False)

    outcome = pgy.apply_matched_task(task, {"read_count": "123", "title": "标题"}, summary, dry_run=False)

    assert outcome["status"] == "write_failure"
    assert outcome["write_ok"] is False


def test_pgy_default_creator_batch_rotates_browser_before_memory_accumulates():
    assert pgy.DEFAULT_BATCH_SIZE == 4
    assert xingtu.DEFAULT_BATCH_SIZE == 0


@pytest.mark.parametrize("message", [
    "Page crashed: Out of Memory",
    "Target page, context or browser has been closed: not enough memory",
])
def test_pgy_treats_chromium_memory_exhaustion_as_a_recoverable_session_loss(message):
    assert pgy.is_browser_session_lost(RuntimeError(message)) is True


def test_repeatable_style_argument_and_selection_keep_every_requested_style():
    parser = argparse.ArgumentParser()
    cc.add_repeatable_style_argument(parser, "styles")
    args = parser.parse_args(["--style", "1", "--style", "3"])
    styles = [{"index": 1}, {"index": 2}, {"index": 3}]

    assert args.style == [1, 3]
    assert cc.select_requested_styles(styles, args.style) == [styles[0], styles[2]]
    with pytest.raises(ValueError, match="99"):
        cc.select_requested_styles(styles, [1, 99])


def test_pgy_stable_cards_retry_an_initial_empty_render(monkeypatch):
    sequence = [[], [{"title": "标题", "read_count": "1"}], [{"title": "标题", "read_count": "1"}]]
    monkeypatch.setattr(pgy, "parse_cards_on_page", lambda _page: sequence.pop(0))

    class Page:
        waits = 0

        def wait_for_timeout(self, _ms):
            self.waits += 1

    page = Page()
    assert pgy.parse_cards_stable(page) == [{"title": "标题", "read_count": "1"}]
    assert page.waits == 2


def test_pgy_stable_cards_waits_through_a_slow_initial_render(monkeypatch):
    card = {"title": "标题", "read_count": "1"}
    sequence = [[], [], [], [], [], [card], [card]]
    monkeypatch.setattr(pgy, "parse_cards_on_page", lambda _page: sequence.pop(0))

    class Page:
        waits = 0

        def wait_for_timeout(self, _ms):
            self.waits += 1

    page = Page()
    assert pgy.parse_cards_stable(page) == [card]
    assert page.waits == 6


def test_pgy_card_parser_recovers_note_id_from_internal_card_link():
    note_id = "0123456789abcdef01234567"

    class Page:
        def evaluate(self, _script):
            return [{
                "title": "标题",
                "read_count": "1",
                "href": "",
                "note_id": "",
                "_href_candidates": [
                    f"https://pgy.xiaohongshu.com/internal?source_note_id={note_id}"
                ],
                "_data_note_ids": [],
            }]

    cards = pgy.parse_cards_on_page(Page())

    assert cards == [{
        "title": "标题",
        "read_count": "1",
        "href": "",
        "note_id": note_id,
    }]


def test_pgy_card_parser_prefers_strict_note_detail_id_over_generic_segment():
    note_id = "0123456789abcdef01234567"
    internal_href = f"https://example.invalid/note/detail/{note_id}"

    class Page:
        def evaluate(self, _script):
            return [{
                "title": "标题",
                "read_count": "1",
                "href": internal_href,
                # This is what the old generic /note/<segment> parser produced.
                "note_id": "detail",
                "_href_candidates": [internal_href],
                "_data_note_ids": [],
            }]

    cards = pgy.parse_cards_on_page(Page())

    assert cards[0]["note_id"] == note_id


def test_pgy_dom_parser_reads_root_data_id_and_rejects_arbitrary_href_fallback():
    captured = {}

    class Page:
        def evaluate(self, script):
            captured["script"] = script
            return []

    assert pgy.parse_cards_on_page(Page()) == []
    assert "const dataNoteNodes = [card, ...card.querySelectorAll" in captured["script"]
    assert "const href = link?.href || '';" in captured["script"]
    assert "hrefCandidates[0]" not in captured["script"]


def test_pgy_arbitrary_shared_anchor_does_not_collapse_distinct_cards():
    shared_anchor = "https://example.invalid/creator/profile"

    class Page:
        def evaluate(self, _script):
            return [
                {
                    "title": "第一篇笔记",
                    "read_count": "1",
                    "like_count": "1",
                    "collect_count": "1",
                    "publish_time": "2026-08-01",
                    "href": "",
                    "note_id": "",
                    "_href_candidates": [shared_anchor],
                    "_data_note_ids": [],
                },
                {
                    "title": "第二篇笔记",
                    "read_count": "2",
                    "like_count": "2",
                    "collect_count": "2",
                    "publish_time": "2026-08-02",
                    "href": "",
                    "note_id": "",
                    "_href_candidates": [shared_anchor],
                    "_data_note_ids": [],
                },
            ]

    cards = pgy.parse_cards_on_page(Page())

    assert [card["href"] for card in cards] == ["", ""]
    assert pgy._card_observation_key(cards[0]) != pgy._card_observation_key(cards[1])


def test_pgy_deduplicates_one_card_repeated_across_pages_before_matching(monkeypatch):
    card = {
        "title": "同一篇笔记",
        "read_count": "100",
        "like_count": "10",
        "collect_count": "5",
        "publish_time": "2026-08-01",
        "href": "",
        "note_id": "",
    }
    pages = [[dict(card)], [dict(card)], [dict(card)]]
    monkeypatch.setattr(pgy, "parse_cards_stable", lambda _page: pages.pop(0))
    monkeypatch.setattr(pgy, "go_next_page", lambda _page: bool(pages))

    found, candidates, page_limit_hit = pgy.find_notes_for_tasks(
        object(),
        [{"record_id": "r1", "target_title": "同一篇笔记"}],
        max_pages=4,
    )

    assert found["r1"]["read_count"] == "100"
    assert candidates == ["同一篇笔记"]
    assert page_limit_hit is False


def test_pgy_missing_creator_is_blocked_but_unmatched_note_stays_retryable():
    no_creator = pgy._missing_detail_outcome(saw_search_result=False)
    no_unique_note = pgy._unmatched_note_outcome(
        candidates=["平台上的其他笔记"],
        page_limit_hit=False,
    )
    completed = cc.finalize_summary({
        "total": 2,
        "results": [
            {"record_id": "creator", **no_creator},
            {"record_id": "note", **no_unique_note},
        ],
    })

    assert completed["blocked_input"] == 1
    assert completed["retryable_failures"] == 1
    assert completed["unresolved"] == 1
    assert completed["complete"] is False
    assert pgy._missing_detail_outcome(saw_search_result=True)["status"] == "retryable_failure"
    assert pgy._unmatched_note_outcome(
        candidates=[], page_limit_hit=False
    )["reason"] == "empty_detail_page"
    assert pgy._unmatched_note_outcome(
        candidates=["候选"], page_limit_hit=True
    )["status"] == "retryable_failure"


@pytest.mark.parametrize(
    "state",
    [
        {
            "nextFound": True,
            "nextDisabled": True,
            "activePage": None,
            "lastPage": None,
            "numericPageCount": 0,
        },
        {
            "nextFound": True,
            "nextDisabled": False,
            "activePage": 7,
            "lastPage": 7,
            "numericPageCount": 7,
        },
    ],
)
def test_pgy_known_last_page_stops_without_clicking(state):
    class Page:
        def evaluate(self, script):
            if "pgy-pagination-state" in script:
                return state
            if "pgy-pagination-click" in script:
                pytest.fail("terminal paginator must not be clicked")
            return "last-page-card-signature"

    assert pgy.go_next_page(Page()) is False


def test_pgy_clicked_pagination_without_card_change_uses_source_end_semantics():
    class Page:
        def evaluate(self, script):
            if "pgy-pagination-state" in script:
                return {
                    "nextFound": True,
                    "nextDisabled": False,
                    "activePage": 2,
                    "lastPage": 7,
                    "numericPageCount": 7,
                }
            if "pgy-pagination-click" in script:
                return {"clicked": True, "text": "next"}
            return "same-card-signature"

        def wait_for_timeout(self, _ms):
            return None

    assert pgy.go_next_page(Page()) is False


def test_xingtu_paginates_before_falling_back_to_search(monkeypatch):
    pages = [
        [{"title": "无关视频", "play_count": "10"}],
        [{"title": "目标视频标题", "play_count": "20"}],
    ]
    calls = {"next": 0}
    monkeypatch.setattr(xingtu, "is_blocked", lambda _page: False)
    monkeypatch.setattr(xingtu, "parse_videos_stable", lambda _page: pages.pop(0))

    def next_page(_page):
        calls["next"] += 1
        return True

    monkeypatch.setattr(xingtu, "go_next_page", next_page)
    monkeypatch.setattr(xingtu, "submit_video_search", lambda *_: False)

    found, _, _ = xingtu.find_videos_for_tasks(
        object(), [{"record_id": "r1", "target_title": "目标视频标题"}], max_pages=2,
    )

    assert found["r1"]["play_count"] == "20"
    assert calls["next"] == 1


def test_xingtu_video_search_uses_visible_input_and_source_keyboard_events():
    events = []

    class Locator:
        first = None

        def __init__(self):
            self.first = self

        def count(self):
            return 1

        def wait_for(self, **kwargs):
            events.append(("wait", kwargs))

        def click(self):
            events.append(("click",))

        def press(self, key):
            events.append(("locator_press", key))

        def fill(self, _value):
            pytest.fail("Xingtu search must preserve source key-by-key input")

    class Keyboard:
        def type(self, value, **kwargs):
            events.append(("type", value, kwargs))

        def press(self, key):
            events.append(("keyboard_press", key))

    class Page:
        keyboard = Keyboard()

        def __init__(self):
            self.selector = None
            self.input = Locator()

        def locator(self, selector):
            self.selector = selector
            return self.input

    page = Page()

    assert xingtu.submit_video_search(page, "目标标题") is True
    assert ":visible" in page.selector
    assert events == [
        ("wait", {"state": "visible", "timeout": 5000}),
        ("click",),
        ("locator_press", "Control+A"),
        ("locator_press", "Backspace"),
        ("type", "目标标题", {"delay": 20}),
        ("keyboard_press", "Enter"),
    ]


def test_xingtu_only_confirmed_creator_miss_is_blocked():
    no_creator = xingtu._missing_detail_outcome(saw_search_result=False)
    no_unique_video = xingtu._unmatched_video_outcome(
        candidates=["平台上的其他视频"],
        page_limit_hit=False,
    )
    completed = cc.finalize_summary({
        "total": 2,
        "results": [
            {"record_id": "creator", **no_creator},
            {"record_id": "video", **no_unique_video},
        ],
    })

    assert completed["blocked_input"] == 1
    assert completed["retryable_failures"] == 1
    assert completed["unresolved"] == 1
    assert completed["complete"] is False
    assert no_unique_video["status"] == "retryable_failure"
    assert xingtu._missing_detail_outcome(
        saw_search_result=True
    )["status"] == "retryable_failure"
    assert xingtu._unmatched_video_outcome(
        candidates=[], page_limit_hit=False
    )["reason"] == "empty_detail_page"
    assert xingtu._unmatched_video_outcome(
        candidates=["候选"], page_limit_hit=True
    )["status"] == "retryable_failure"


def test_xingtu_missing_creator_result_has_distinct_business_exception():
    class Context:
        pages = []

    class Page:
        context = Context()

        def evaluate(self, _script, argument=None):
            if argument is not None:
                return {"found": False}
            return False

        def wait_for_timeout(self, _milliseconds):
            return None

    with pytest.raises(xingtu.CreatorNotFound, match="未找到精确达人"):
        xingtu.open_creator_detail(Page(), "不存在的达人", "missing-id")


def test_xingtu_clicked_pagination_without_card_change_is_retryable_timeout():
    class Page:
        def evaluate(self, script):
            if "const all" in script:
                return {"clicked": True, "picked": "pagination"}
            return "same-card-signature"

        def wait_for_timeout(self, _milliseconds):
            return None

    with pytest.raises(xingtu.BrowserTimeoutError, match="did not change"):
        xingtu.go_next_page(Page())


def test_xingtu_normalizes_show_items_api_video():
    card = xingtu.normalize_xingtu_api_item({
        "item_id": 7654321,
        "item_title": "目标视频标题",
        "play": 168000,
        "like": 1200,
        "comment": 34,
        "share": 56,
        "url": "https://www.douyin.com/video/7654321",
    })

    assert card == {
        "title": "目标视频标题",
        "play_count": 168000,
        "like_count": 1200,
        "comment_count": 34,
        "share_count": 56,
        "href": "https://www.douyin.com/video/7654321",
        "note_id": "7654321",
        "source": "show_items_api",
    }


def test_xingtu_show_items_capture_collects_and_deduplicates_cards():
    import gc
    import weakref

    class Response:
        url = "https://www.xingtu.cn/gw/api/author/get_author_show_items_v2"

        def json(self):
            return {
                "latest_item_info": [
                    {"item_id": "1", "title": "视频一", "play": 10},
                    {"item_id": "2", "item_title": "视频二", "play": 20},
                ],
                "latest_star_item_info": [
                    {"item_id": "2", "item_title": "视频二", "play": 20},
                ],
            }

    class Page:
        def on(self, event, callback):
            assert event == "response"
            self.callback = callback

    page = Page()
    cards = xingtu.setup_show_items_capture(page)
    response = Response()
    response_ref = weakref.ref(response)
    page.callback(response)
    del response
    gc.collect()

    assert [card["note_id"] for card in cards] == ["1", "2"]
    assert response_ref() is None


def test_xingtu_matches_api_cards_without_dom_pagination(monkeypatch):
    monkeypatch.setattr(
        xingtu,
        "parse_videos_stable",
        lambda _page: (_ for _ in ()).throw(AssertionError("DOM should not be needed")),
    )
    monkeypatch.setattr(
        xingtu,
        "is_blocked",
        lambda _page: (_ for _ in ()).throw(AssertionError("page should not be needed")),
    )
    tasks = [{
        "record_id": "r1",
        "target_title": "飞书旧标题",
        "note_url": "https://www.douyin.com/video/7654321",
        "note_id": "7654321",
    }]
    api_cards = [{
        "title": "平台上的新标题",
        "play_count": 168000,
        "note_id": "7654321",
        "source": "show_items_api",
    }]

    found, _, page_limit_hit = xingtu.find_videos_for_tasks(
        object(), tasks, max_pages=5, api_cards=api_cards,
    )

    assert found["r1"]["play_count"] == 168000
    assert found["r1"]["match_method"] == "content_id"
    assert page_limit_hit is False


def test_xingtu_api_fuzzy_candidate_does_not_preempt_later_exact_dom(monkeypatch):
    monkeypatch.setattr(xingtu, "is_blocked", lambda _page: False)
    monkeypatch.setattr(
        xingtu,
        "parse_videos_stable",
        lambda _page: [{"title": "commuter backpack review", "note_id": "exact-card"}],
    )
    monkeypatch.setattr(xingtu, "go_next_page", lambda _page: False)
    monkeypatch.setattr(xingtu, "submit_video_search", lambda *_: False)
    task = {"record_id": "r1", "target_title": "commuter backpack review"}
    api_cards = [{
        "title": "commuter backpack review today",
        "note_id": "fuzzy-card",
        "source": "show_items_api",
    }]

    found, _, _ = xingtu.find_videos_for_tasks(
        object(), [task], max_pages=2, api_cards=api_cards,
    )

    assert found["r1"]["note_id"] == "exact-card"


def test_browser_session_loss_detection_is_specific():
    closed = RuntimeError("Target page, context or browser has been closed")
    assert pgy.is_browser_session_lost(closed)
    assert xingtu.is_browser_session_lost(closed)
    assert pgy.is_browser_session_lost(RuntimeError("Page crashed"))
    assert xingtu.is_browser_session_lost(RuntimeError("Page crashed"))
    assert pgy.is_browser_session_lost(
        RuntimeError("Execution context was destroyed, most likely because of a navigation.")
    )
    assert not pgy.is_browser_session_lost(RuntimeError("ordinary parse failure"))


def test_transient_browser_error_excludes_business_timeout():
    assert is_transient_browser_error(
        pgy.BrowserTimeoutError("Page.goto: Timeout 60000ms exceeded")
    )
    assert is_transient_browser_error(
        RuntimeError("Page.goto: net::ERR_CONNECTION_RESET")
    )
    assert not is_transient_browser_error(
        TimeoutError("Login timed out after 300 seconds.")
    )


def _stub_creator_session_failure(monkeypatch, tmp_path, module, error_factory):
    style = {"index": 1, "name": "style"}
    task = {
        "record_id": "record-1",
        "creator_name": "creator",
        "creator_id": None,
        "target_title": "target",
        "style_context": {"index": 1},
    }
    summary = {
        "index": 1,
        "name": "style",
        "total": 1,
        "filled": 0,
        "results": [],
    }
    group = {
        "creator_name": "creator",
        "creator_id": None,
        "tasks": [task],
    }
    sessions = []

    class Session:
        def __init__(self, **_kwargs):
            sessions.append(self)

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return False

    def fail_page_action(*_args, **_kwargs):
        raise error_factory()

    monkeypatch.setattr(
        module,
        "collect_tasks_across_styles",
        lambda *_args, **_kwargs: ([task], {1: summary}),
    )
    monkeypatch.setattr(module, "group_tasks_by_creator", lambda _tasks: [group])
    monkeypatch.setattr(module, "chunk_creator_groups", lambda groups, _size: [groups])
    monkeypatch.setattr(module, "DynamicSession", Session)
    monkeypatch.setattr(module, "run_serialized_page_action", fail_page_action)
    monkeypatch.setattr(module, "_force_cleanup_session", lambda _session: None)
    monkeypatch.setattr(module, "PROFILE_DIR", tmp_path / module.__name__.split(".")[-1])
    return style, sessions


@pytest.mark.parametrize("module", [pgy, xingtu])
def test_creator_collectors_retry_browser_timeout_in_three_fresh_sessions(
    monkeypatch, tmp_path, module
):
    style, sessions = _stub_creator_session_failure(
        monkeypatch,
        tmp_path,
        module,
        lambda: module.BrowserTimeoutError("Page.goto: Timeout 60000ms exceeded"),
    )

    summaries = module.scrape_styles([style], 1, True, None, True)

    assert len(sessions) == 3
    assert summaries[0]["complete"] is False
    assert summaries[0]["retryable_failures"] == 1


@pytest.mark.parametrize("module", [pgy, xingtu])
def test_creator_collectors_do_not_retry_acceptance_terminal(
    monkeypatch, tmp_path, module
):
    style, sessions = _stub_creator_session_failure(
        monkeypatch,
        tmp_path,
        module,
        lambda: module.AcceptanceCookieSkip("cookie unavailable in acceptance"),
    )

    with pytest.raises(module.AcceptanceCookieSkip):
        module.scrape_styles([style], 1, True, None, True)

    assert len(sessions) == 1


@pytest.mark.parametrize("module", [pgy, xingtu])
def test_creator_collectors_do_not_retry_login_timeout_terminal(
    monkeypatch, tmp_path, module
):
    style, sessions = _stub_creator_session_failure(
        monkeypatch,
        tmp_path,
        module,
        lambda: TimeoutError("Login timed out after 300 seconds."),
    )

    with pytest.raises(TimeoutError, match="Login timed out"):
        module.scrape_styles([style], 1, True, None, True)

    assert len(sessions) == 1


@pytest.mark.parametrize("error_type", [xingtu.SearchQuotaExhausted, xingtu.CaptchaBlocked])
def test_xingtu_does_not_retry_explicit_business_terminal(
    monkeypatch, tmp_path, error_type
):
    style, sessions = _stub_creator_session_failure(
        monkeypatch,
        tmp_path,
        xingtu,
        lambda: error_type("explicit platform terminal"),
    )

    with pytest.raises(error_type):
        xingtu.scrape_styles([style], 1, True, None, True)

    assert len(sessions) == 1


@pytest.mark.parametrize("module", [pgy, xingtu])
def test_creator_collectors_with_no_tasks_do_not_open_browser(
    monkeypatch, module
):
    style = {"index": 1, "name": "style"}
    summary = {"index": 1, "name": "style", "total": 0, "results": []}
    monkeypatch.setattr(
        module,
        "collect_tasks_across_styles",
        lambda *_args, **_kwargs: ([], {1: summary}),
    )
    monkeypatch.setattr(
        module,
        "DynamicSession",
        lambda **_kwargs: pytest.fail("no browser session expected"),
    )

    assert module.scrape_styles([style], 1, True, None, True) == [summary]


def test_long_page_action_does_not_retain_navigation_response():
    import gc
    import weakref

    response_ref = None

    class Response:
        url = "https://example.test/final"
        status = 200
        headers = {"content-type": "text/html"}

    class Page:
        closed = False

        def set_default_timeout(self, _timeout):
            pass

        def set_default_navigation_timeout(self, _timeout):
            pass

        def goto(self, *_args, **_kwargs):
            nonlocal response_ref
            response = Response()
            response_ref = weakref.ref(response)
            return response

        def is_closed(self):
            return self.closed

        def close(self):
            self.closed = True

    page = Page()

    class Context:
        def new_page(self):
            return page

    class Session:
        context = Context()

    def action(_page):
        gc.collect()
        assert response_ref is not None
        assert response_ref() is None

    navigation = run_serialized_page_action(Session(), "https://example.test", action)

    assert navigation == {
        "url": "https://example.test/final",
        "status": 200,
        "headers": {"content-type": "text/html"},
    }
    assert page.closed is True


def test_xingtu_nested_session_loss_is_reraised():
    closed = RuntimeError("Target page, context or browser has been closed")
    with pytest.raises(RuntimeError, match="has been closed"):
        xingtu.reraise_if_browser_session_lost(closed)
    xingtu.reraise_if_browser_session_lost(RuntimeError("ordinary lookup miss"))


def test_xingtu_detail_and_video_search_propagate_closed_session():
    closed = RuntimeError("Target page, context or browser has been closed")

    class Detail:
        def goto(self, *_args, **_kwargs):
            raise closed

    class Context:
        pages = []

        def new_page(self):
            return Detail()

    class SearchPage:
        context = Context()

        def evaluate(self, _script, _arg=None):
            return {"found": True, "url": "https://www.xingtu.cn/ad/creator/author-homepage/douyin-video/1"}

    with pytest.raises(RuntimeError, match="has been closed"):
        xingtu.open_creator_detail(SearchPage(), "达人", "creator-id")

    class Locator:
        first = None

        def __init__(self):
            self.first = self

        def count(self):
            return 1

        def wait_for(self, **_kwargs):
            raise closed

    class VideoPage:
        def locator(self, _selector):
            return Locator()

    with pytest.raises(RuntimeError, match="has been closed"):
        xingtu.submit_video_search(VideoPage(), "标题")


def test_pgy_fallback_detail_click_propagates_closed_session():
    closed = RuntimeError("Target page, context or browser has been closed")

    class Clicker:
        first = None

        def __init__(self):
            self.first = self

        def click(self, **_kwargs):
            raise closed

    class Context:
        pages = []

    class Page:
        context = Context()

        def evaluate(self, *_args, **_kwargs):
            return {"ok": False, "reason": "no-row"}

        def get_by_text(self, *_args, **_kwargs):
            return Clicker()

    with pytest.raises(RuntimeError, match="has been closed"):
        pgy.open_blogger_detail(Page(), "达人", None)


def test_bilibili_triggered_blocked_inputs_are_visible_terminal_rows():
    summary = bili.finalize_bili_summary({
        "style": "款式A", "index": 1, "total_b_records": 1,
        "details": [{
            "record_id": "r1", "status": "blocked_input",
            "matched": False, "reason": "publish_time_missing", "ok": False,
        }],
    })

    assert summary["blocked_input"] == 1
    assert summary["unresolved"] == 0
    assert summary["complete"] is True


def test_bilibili_missing_url_does_not_trigger_collection(monkeypatch):
    style = {
        "style": "款式A",
        "name": "款式A",
        "index": 1,
        "base_token": "base",
        "table_id": "table",
        "field_map": {
            "platform": {"field_id": "platform"},
            "note_url": {"field_id": "url"},
            "creator_name": {"field_id": "creator"},
            "publish_time": {"field_id": "pub"},
            "read_count_7d": {"field_id": "s7"},
            "read_count_14d": {"field_id": "s14"},
            "read_count_21d": {"field_id": "s21"},
            "read_count_28d": {"field_id": "s28"},
            "month_end": {"field_id": "sm"},
        },
    }
    rows = [
        {
            "record_id": "no-link",
            "platform": "B站",
            "creator": "未发布达人",
            "url": "",
            "pub": "2026-07-20",
        },
        {
            "record_id": "published",
            "platform": "B站",
            "creator": "已发布达人",
            "url": "https://www.bilibili.com/video/BV1234567890",
            "pub": "2026-07-20",
        },
    ]
    monkeypatch.setattr(bili, "list_all_records", lambda *_args: rows)
    monkeypatch.setattr(bili, "fetch_play_count", lambda *_args: 321)
    monkeypatch.setattr(bili, "write_record", lambda *_args, **_kwargs: True)

    summary = bili.process_style(
        style,
        only_record_ids=None,
        dry_run=True,
        delay=0,
        session=object(),
        state={},
        force_today=bili.date(2026, 7, 23),
        first_run=False,
    )

    assert summary["source_b_records"] == 2
    assert summary["skipped_no_url"] == 1
    assert summary["total_b_records"] == 1
    assert summary["success"] == 1
    assert summary["blocked_input"] == 0
    assert [row["record_id"] for row in summary["details"]] == ["published"]


@pytest.mark.parametrize(
    ("raw", "expected"),
    [
        ("2026-07-20T00:00:00.000+08:00", bili.date(2026, 7, 20)),
        ("2026-07-20T00:00:00+08:00", bili.date(2026, 7, 20)),
        ("2026-07-19T16:00:00.000Z", bili.date(2026, 7, 19)),
        ("2026-07-20", bili.date(2026, 7, 20)),
    ],
)
def test_bilibili_parse_pub_date_accepts_feishu_iso_values(raw, expected):
    assert bili.parse_pub_date(raw) == expected


def test_bilibili_partial_retry_preserves_old_success():
    existing = {
        "style": "款式A", "index": 1, "total_b_records": 2,
        "details": [
            {"record_id": "ok", "status": "success", "ok": True},
            {"record_id": "retry", "status": "retryable_failure", "ok": False},
        ],
    }
    partial = {
        "style": "款式A", "index": 1, "total_b_records": 1,
        "details": [{"record_id": "retry", "status": "success", "ok": True}],
    }

    merged = bili.merge_bili_summary(existing, partial)

    assert merged["total_b_records"] == 2
    assert {row["record_id"] for row in merged["details"]} == {"ok", "retry"}
    assert merged["complete"] is True


def test_bilibili_top_level_error_cannot_finalize_as_complete():
    finalized = bili.finalize_bili_summary({
        "style": "款式A",
        "index": 1,
        "total_b_records": 0,
        "details": [],
        "error": "missing platform/url field_id",
        "unresolved": 1,
        "retryable_failures": 1,
        "complete": False,
    })

    assert finalized["complete"] is False
    assert finalized["unresolved"] >= 1


def test_bilibili_missing_slot_mapping_is_system_error_not_blocked_input():
    summary = bili.process_style(
        {
            "style": "款式A",
            "name": "款式A",
            "index": 1,
            "base_token": "base",
            "table_id": "table",
            "field_map": {
                "platform": {"field_id": "platform"},
                "note_url": {"field_id": "url"},
            },
        },
        only_record_ids=None,
        dry_run=True,
        delay=0,
        session=object(),
        state={},
        force_today=None,
        first_run=False,
    )

    finalized = bili.finalize_bili_summary(summary, dry_run=True)
    assert "publish_time" in finalized["error"]
    assert "read_count_7d" in finalized["error"]
    assert finalized["complete"] is False


def test_self_douyin_missing_mapping_and_write_failure_are_not_complete(monkeypatch):
    missing = self_dy.scrape_one_style(
        {"name": "款式A", "index": 1, "base_token": "base", "table_id": "table", "field_map": {}},
        login_timeout=1,
        headless=True,
        only_record_ids=None,
        dry_run=False,
    )
    assert cc.finalize_summary(missing)["complete"] is False

    style = _style_for_platform("抖音")
    row = {
        "record_id": "r1",
        "platform": "抖音",
        "name": "自营达人",
        "title": "目标作品",
        "url": "https://www.douyin.com/video/1",
        "pub": "2026-07-20",
    }

    class Page:
        def goto(self, *_args, **_kwargs):
            return None

        def wait_for_timeout(self, _ms):
            return None

    class Session:
        def __init__(self, **_kwargs):
            pass

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return None

        def fetch(self, _url, page_action, wait):
            page_action(Page())

    monkeypatch.setattr(self_dy, "DynamicSession", Session)
    monkeypatch.setattr(self_dy, "list_records_by_table", lambda *_: [row])
    monkeypatch.setattr(self_dy, "dismiss_popups", lambda *_: None)
    monkeypatch.setattr(self_dy, "restore_cookies", lambda *_: None)
    monkeypatch.setattr(self_dy, "maybe_wait_for_login", lambda *_: None)
    monkeypatch.setattr(self_dy, "save_state", lambda *_: None)
    monkeypatch.setattr(
        self_dy,
        "scroll_and_collect_posts",
        lambda *_args, **_kwargs: [{"title": "目标作品", "play_count": 123}],
    )
    monkeypatch.setattr(self_dy, "pick_read_field", lambda *_: ("read_count_7d", "read7", "7天曝光量"))
    monkeypatch.setattr(self_dy, "write_back", lambda *_: False)

    failed = cc.finalize_summary(
        self_dy.scrape_one_style(
            style,
            login_timeout=1,
            headless=True,
            only_record_ids=None,
            dry_run=False,
        )
    )

    assert failed["results"][0]["status"] == "write_failure"
    assert failed["complete"] is False


def test_run_all_rejects_stale_or_incomplete_summary(tmp_path: Path):
    path = tmp_path / "summary.json"
    path.write_text("[]", encoding="utf-8")
    started_at = path.stat().st_mtime + 1

    assert run_all.validate_summary_payload([], {1}, path, started_at)[0] is False

    path.write_text('[{"index": 1, "total": 2, "results": [{"record_id": "a", "status": "success"}]}]', encoding="utf-8")
    assert run_all.validate_summary_payload(
        [{"index": 1, "total": 2, "results": [{"record_id": "a", "status": "success"}]}],
        {1}, path, path.stat().st_mtime - 1,
    )[0] is False