"""Read the shipped DOM observers without navigation, persistence or proofs."""
from copy import deepcopy
import json
from pathlib import Path
import subprocess

import pytest

from backend.browser import tools


URL = 'https://www.xiaohongshu.com/explore/' + 'a' * 24
FIXTURE = Path(__file__).with_name('reply_dom_fixture_d41.cjs')


def node(identity='', classes='', text='', *children, tag='div', **attrs):
    return {'tag': tag, 'attrs': {'id': identity, 'class': classes, **attrs}, 'text': text, 'children': list(children)}


def content(identity, text='PRIVATE-QUOTE', *children):
    return node(identity+'-content', 'content', '', node(identity+'-text', 'note-text', text, *children))


def speaker(identity, person, own=None, *, kind='reply', extra=(), **attrs):
    author = node(identity+'-author', 'name', 'PRIVATE-SAME-NAME', tag='a',
                  href='https://www.xiaohongshu.com/user/profile/' + person*24)
    return node(identity, 'comment-item' if kind == 'comment' else 'comment-item-sub', '',
                node('', 'author-wrapper', '', author), own or content(identity), *extra,
                **{'data-comment-id': identity+'-key', **attrs})


def target(*, href=True, identity='c', user_attribute=False):
    attrs = {'href': 'https://www.xiaohongshu.com/user/profile/'+identity*24} if href else {}
    if user_attribute:
        attrs['data-target-user-id'] = identity*24
    return content('child', '回复 ', node('target', 'user-name', 'PRIVATE-SAME-NAME',
                   tag='a' if href else 'span', **attrs), node('', '', ': PRIVATE-CHILD-QUOTE'))


def thread(*replies, identity='thread'):
    return node(identity, 'parent-comment', '', speaker(identity+'-root', 'b', kind='comment'),
                node(identity+'-replies', 'reply-container', '', *replies))


def tree(*groups):
    return node('noteContainer', '', '', node('detail-desc', '', 'PRIVATE-NOTE'), *groups)


def find(value, identity):
    if value.get('attrs', {}).get('id') == identity:
        return value
    for child in value.get('children', []):
        found = find(child, identity)
        if found:
            return found
    return None


class Page:
    def __init__(self, value, mutate=None):
        self.tree, self.mutate = deepcopy(value), mutate
        self.url = URL
        self.calls = []
        self.source_calls = {}

    async def get_url(self):
        return self.url

    async def evaluate(self, code, arg):
        assert code in {tools._INSPECT_SOURCE_CANDIDATES, tools._OBSERVE_SOURCE, tools._OBSERVE_RELATION_RECHECK}
        self.calls.append((code, arg))
        if code == tools._OBSERVE_SOURCE:
            assert isinstance(arg, str), 'preflight must never supply a native proof identity'
            self.source_calls[arg] = self.source_calls.get(arg, 0)+1
        if self.mutate:
            self.mutate(self, code, arg)
        response = subprocess.run(['node', str(FIXTURE)], text=True, capture_output=True, check=True, timeout=5,
            input=json.dumps({'code': code, 'arg': arg, 'tree': self.tree}, ensure_ascii=False))
        return response.stdout


class Session:
    def __init__(self, page):
        self.page = page

    async def get_current_page(self):
        return self.page


@pytest.fixture
def isolated(monkeypatch, tmp_path):
    task_id = 'reply_preflight_d44'
    tools.set_prospecting_context(task_id, artifacts={
        'messages': [{'role': 'user', 'text': '只预检第一组讨论，不保存、不点击'}],
        'sources': [{'source_id': 'old-note', 'source_type': 'note', 'note_id': 'a'*24,
                     'source_selector': '#detail-desc', 'content': 'PRIVATE-NOTE'},
                    {'source_id': 'old-child', 'source_type': 'reply', 'note_id': 'a'*24,
                     'source_selector': '#child-content', 'content': 'PRIVATE-CHILD-QUOTE', 'parent_source_id': None}],
        'source_observations': [{'operation': 'old-observation'}], 'operation_evidence': [{'id': 'old-fact'}],
        'knowledge_event_ids': ['old-knowledge'],
    })
    tools._reply_target_proofs[task_id] = {'child-key': {'run_id': 'old-run', 'signature': 'old', 'profile_id': 'c'*24}}
    tools._prospecting_run_context[task_id] = {'run_id': 'new-run', 'goal_version': 2}
    def forbidden(*_args, **_kwargs):
        raise AssertionError('read-only preflight must not write facts, navigate or use proofs')
    for name in ('_persist_evidence', '_operation_evidence', '_save_source', '_write_json_atomic', '_apply_reply_target_proof', 'task_doc_dir'):
        monkeypatch.setattr(tools, name, forbidden)
    before = deepcopy((tools._task_artifacts, tools._reply_target_proofs, tools._prospecting_run_context,
                       tools._pending_blocker, tools._pending_question))
    yield task_id
    assert before == (tools._task_artifacts, tools._reply_target_proofs, tools._prospecting_run_context,
                      tools._pending_blocker, tools._pending_question)
    assert list(tmp_path.iterdir()) == []
    tools.clear_task_context(task_id)


async def inspect(value, task_id, mutate=None, maximum=5):
    page = Page(value, mutate)
    result = await tools._inspect_reply_relations(Session(page), maximum, task_id)
    serialized = json.dumps(result)
    assert 'PRIVATE' not in serialized and 'https://' not in serialized and 'profile_id' not in serialized
    assert result['note_id'] == 'a'*24
    for identity in ('a'*24, 'b'*24, 'c'*24, 'd'*24, 'e'*24):
        assert identity not in json.dumps({key: value for key, value in result.items() if key != 'note_id'})
    return result, page


def child_row(result):
    return next(row for row in result['candidates'] if row['source_selector'] == '#child-content')


@pytest.mark.parametrize('mode', ['href', 'user_attribute', 'comment_id'])
async def test_real_unique_identity_or_explicit_comment_ref_qualifies_without_saving(mode, isolated):
    own = target(href=mode == 'href', user_attribute=mode == 'user_attribute')
    child = speaker('child', 'd', own, **({'data-parent-comment-id': 'parent-key'} if mode == 'comment_id' else {}))
    result, _ = await inspect(tree(thread(speaker('parent', 'c'), child)), isolated)
    row = child_row(result)
    assert row['classification'] == 'unique_dom_deep_reply' and row['eligible'] is True
    assert row['parent_is_reply'] is True and row['parent_relation_observed'] is True
    assert row['parent_source_selector'] == '#parent-content' and row['root_source_selector'] == '#thread-root-content'
    assert row['native_navigation_confirmed'] is False
    assert row['parent_resolution'] == ('explicit_comment_id' if mode == 'comment_id' else 'unique_target_profile')


async def test_actual_nested_parent_reply_has_structural_evidence(isolated):
    child = speaker('child', 'd')
    parent = speaker('parent', 'c', extra=[child])
    root = speaker('root', 'b', kind='comment', extra=[parent])
    result, _ = await inspect(tree(root), isolated)
    row = child_row(result)
    assert row['classification'] == 'unique_dom_deep_reply'
    assert row['parent_resolution'] == 'structural_parent' and row['parent_is_reply'] is True


async def test_identical_names_and_two_actual_comments_by_target_identity_remain_ambiguous(isolated):
    result, _ = await inspect(tree(thread(speaker('parent', 'c'), speaker('another', 'c'), speaker('child', 'd', target()))), isolated)
    row = child_row(result)
    assert row['classification'] == 'ambiguous' and not row['eligible']
    assert row['relation_diagnostics']['matching_parent_comments'] == 2
    assert row['parent_source_selector'] == ''


async def test_plain_span_is_unproved_even_with_an_existing_native_proof_cache(isolated):
    result, _ = await inspect(tree(thread(speaker('parent', 'c'), speaker('child', 'd', target(href=False)))), isolated)
    row = child_row(result)
    assert row['classification'] == 'needs_native_identity' and row['observation_stable'] is True
    assert row['relation_diagnostics']['target_identities'] == 0
    assert row['relation_diagnostics']['target_profile_links'] == 0
    assert not row['eligible'] and not row['native_navigation_confirmed']


@pytest.mark.parametrize('change', ['source_key', 'quote', 'rendering', 'identity', 'url'])
async def test_source_key_quote_rendering_identity_or_url_change_is_unknown(change, isolated):
    def mutate(page, code, selector):
        if code != tools._OBSERVE_SOURCE or selector != '#child-content' or page.source_calls[selector] != 2:
            return
        if change == 'source_key':
            find(page.tree, 'child')['attrs']['data-comment-id'] = 'changed-key'
        elif change == 'quote':
            find(page.tree, 'child-text')['text'] = 'CHANGED-QUOTE'
        elif change == 'rendering':
            find(page.tree, 'child-content')['rect'] = {'left': 0, 'right': 300, 'top': 700, 'bottom': 735, 'width': 300, 'height': 35}
        elif change == 'identity':
            # Same nickname, target counts, parent key and selector; the actual
            # target data-* identity changes together with its observed author.
            find(page.tree, 'target')['attrs']['data-target-user-id'] = 'e'*24
            find(page.tree, 'parent-author')['attrs']['href'] = 'https://www.xiaohongshu.com/user/profile/'+'e'*24
        else:
            page.url = URL.replace('a'*24, 'f'*24)
    result, _ = await inspect(tree(thread(speaker('parent', 'c'), speaker('child', 'd', target(href=False, user_attribute=True)))), isolated, mutate)
    row = child_row(result)
    assert row['classification'] == 'unknown' and row['observation_stable'] is False and not row['eligible']


async def test_changed_candidate_excerpt_before_first_observation_is_unknown(isolated):
    def mutate(page, code, selector):
        if code == tools._OBSERVE_SOURCE and selector == '#child-content' and page.source_calls[selector] == 1:
            find(page.tree, 'child-text')['text'] = 'DIFFERENT-QUOTE'
    result, _ = await inspect(tree(thread(speaker('parent', 'c'), speaker('child', 'd', target()))), isolated, mutate)
    assert child_row(result)['classification'] == 'unknown'


async def test_parent_moved_to_another_thread_does_not_qualify_old_child_observation(isolated):
    def mutate(page, code, selector):
        if (code == tools._OBSERVE_SOURCE and selector == '#parent-content'
                and page.source_calls.get('#child-content') == 1):
            old = find(page.tree, 'thread-replies')['children']
            parent = old.pop(next(index for index, row in enumerate(old) if row['attrs']['id'] == 'parent'))
            find(page.tree, 'other-replies')['children'].append(parent)
    value = tree(thread(speaker('parent', 'c'), speaker('child', 'd', target())), thread(identity='other'))
    result, _ = await inspect(value, isolated, mutate)
    assert child_row(result)['classification'] == 'unknown' and not child_row(result)['eligible']


async def test_direct_root_reply_is_not_a_qualified_reply_to_reply_sample(isolated):
    result, _ = await inspect(tree(thread(speaker('child', 'd'))), isolated)
    row = child_row(result)
    assert row['parent_is_reply'] is False and row['parent_resolution'] == 'thread_root'
    assert not row['eligible'] and row['classification'] == 'unknown'


async def test_later_candidate_mutation_invalidates_an_earlier_qualified_row(isolated):
    def mutate(page, code, selector):
        if code == tools._OBSERVE_SOURCE and selector == '#later-content' and page.source_calls[selector] == 1:
            find(page.tree, 'child')['attrs']['data-comment-id'] = 'new-child-key'
    value = tree(thread(speaker('parent', 'c'), speaker('child', 'd', target()), speaker('later', 'e')))
    result, _ = await inspect(value, isolated, mutate)
    assert child_row(result)['classification'] == 'unknown' and not child_row(result)['eligible']


async def test_explicit_first_group_does_not_preflight_another_group(isolated):
    value = tree(thread(speaker('first', 'c')), thread(speaker('parent', 'c'), speaker('child', 'd', target()), identity='other'))
    result, _ = await inspect(value, isolated)
    assert all(row['source_selector'] != '#child-content' for row in result['candidates'])
    assert result['eligible_count'] == 0


@pytest.mark.parametrize('maximum', [0, 6, True, '2'])
async def test_batch_bound_is_strict_before_reading(maximum):
    with pytest.raises(ValueError, match='1–5'):
        await tools._inspect_reply_relations(object(), maximum)


async def test_ordinary_inspector_still_records_by_default_but_internal_no_record_does_not(monkeypatch):
    tid = 'preflight_inspector_compatibility'
    tools.set_prospecting_context(tid)
    writes = []
    monkeypatch.setattr(tools, '_persist_evidence', lambda task: writes.append(task))
    page = Page(tree(thread(speaker('child', 'd'))))
    try:
        pure = await tools._inspect_source_candidates(Session(page), task_id=tid, record_observation=False)
        assert 'source_observations' not in tools._task_artifacts[tid] and not writes
        normal = await tools._inspect_source_candidates(Session(page), task_id=tid)
        assert normal == pure
        assert writes == [tid] and tools._task_artifacts[tid]['source_observations'][0]['operation'] == 'inspect_source_candidates'
    finally:
        tools.clear_task_context(tid)


def test_registered_preflight_accepts_only_a_bounded_count_not_selectors_or_identity(monkeypatch):
    monkeypatch.setattr(tools.settings, 'reply_relation_preflight_enabled', True)
    registry = tools.build_prospecting_tools('preflight_registration').registry.registry.actions
    action = registry['inspect_reply_relations']
    assert set(action.param_model.model_fields) == {'max_replies'}
    assert '不自动调用resolver' in action.description
    assert 'unique_dom_deep_reply' in action.description