"""文字、扫描、混合 PDF 路由及缓存隔离；测试数据不属于现场报告。"""
import json
from datetime import date
from pathlib import Path

from PIL import Image
import pypdfium2 as pdfium
import pytest

from extensions.main_fan import documents
from db import main_fan_store


def native_pdf(target: Path):
    lines = ["SYNTHETIC NATIVE TEXT TEST - NOT A REAL MINE REPORT",
             "Main ventilation fan system performance report",
             "Report ID: NATIVE-TEST-001", "Inspected mine: Synthetic Native Test Mine",
             "Equipment: 1# main ventilation fan", "Inspection date: 2026-01-02",
             "Recommended next inspection: 2029-01-01", "Model: SYNTHETIC-FAN Serial: TEST-001",
             "30 Hz measured operating condition: Flow 60 m3/s; static pressure 500 Pa;",
             "Shaft power 50 kW; electric input power 55 kW; static efficiency 60 percent.",
             "Original conclusion: Synthetic data for text extraction test only."]
    content = b"BT /F1 10 Tf 50 780 Td "
    for index, line in enumerate(lines):
        if index: content += b"0 -18 Td "
        content += b"(" + line.encode("ascii") + b") Tj\n"
    content += b"ET\n"
    objects = [b"<< /Type /Catalog /Pages 2 0 R >>", b"<< /Type /Pages /Kids [3 0 R] /Count 1 >>",
        b"<< /Type /Page /Parent 2 0 R /MediaBox [0 0 595 842] /Resources << /Font << /F1 4 0 R >> >> /Contents 5 0 R >>",
        b"<< /Type /Font /Subtype /Type1 /BaseFont /Helvetica >>",
        b"<< /Length " + str(len(content)).encode() + b" >>\nstream\n"+content+b"endstream"]
    value, offsets = b"%PDF-1.4\n", [0]
    for i, obj in enumerate(objects, 1):
        offsets.append(len(value)); value += str(i).encode()+b" 0 obj\n"+obj+b"\nendobj\n"
    xref = len(value)
    value += f"xref\n0 {len(objects)+1}\n0000000000 65535 f \n".encode()
    value += b"".join(f"{offset:010d} 00000 n \n".encode() for offset in offsets[1:])
    value += f"trailer\n<< /Size {len(objects)+1} /Root 1 0 R >>\nstartxref\n{xref}\n%%EOF\n".encode()
    target.write_bytes(value)


def report(page=1, mine="Auto Extracted Mine"):
    return {"reports": [{"report_id": "AUTO-001", "mine_name": mine, "fan_name": "1号通风机",
                         "evidence": {"page": page, "excerpt": "Automatically supplied test response"}}]}


def test_native_scan_and_mixed_pdf_routes(tmp_path):
    native, scanned, mixed = [tmp_path/name for name in ("native.pdf", "scan.pdf", "mixed.pdf")]
    native_pdf(native)
    Image.new("RGB", (120, 160), "white").save(scanned, "PDF")
    with pdfium.PdfDocument.new() as combined:
        with pdfium.PdfDocument(native) as first: combined.import_pages(first)
        with pdfium.PdfDocument(scanned) as second: combined.import_pages(second)
        combined.save(mixed)
    for source, expected in [(native, ["text"]), (scanned, ["vision"]), (mixed, ["text", "vision"])]:
        folder = tmp_path/source.stem; folder.mkdir()
        pages = documents._prepare(source, folder)
        assert [p["route"] for p in pages] == expected
        if source == native:
            assert "NATIVE-TEST-001" in pages[0]["text"]
            assert pages[0]["image"] is None and not list(folder.glob("*.png"))


@pytest.mark.asyncio
async def test_native_pdf_needs_no_vision_setting_and_sends_no_images(tmp_path, monkeypatch):
    source = tmp_path/"native.pdf"; native_pdf(source)
    monkeypatch.setenv("MAIN_FAN_VISION_MODEL", "")
    calls = []
    class TextModel:
        async def ainvoke(self, messages, config):
            from langchain_core.messages import AIMessage
            calls.append(messages)
            return AIMessage(content=json.dumps(report()), response_metadata={"finish_reason": "stop"},
                             usage_metadata={"input_tokens": 5,"output_tokens": 5,"total_tokens":10})
    monkeypatch.setattr("agents.model_factory.build_model", lambda **kw: TextModel())
    facts = await documents.read_document(source,"/workspace/native.pdf",cache_policy="refresh")
    assert facts.reports[0].evidence.method == "text_extraction"
    assert facts.extraction_audit["pages"][0]["route"] == "text"
    assert facts.extraction_audit["cache_used"] is False
    assert all(block["type"] == "text" for block in calls[0][-1].content)


@pytest.mark.asyncio
async def test_default_reader_rejects_manually_seeded_cache(tmp_path):
    source=tmp_path/"scan.pdf"; Image.new("RGB", (120, 160), "white").save(source,"PDF")
    digest=documents.fingerprint(source)
    main_fan_store.save_document(digest,report(mine="Pretranscribed Mine"),"manual_transcription")
    seen=[]
    async def extractor(pages,folder):
        seen.extend(pages)
        return report(mine="Fresh Automatic Mine")
    facts=await documents.read_document(source,"/workspace/scan.pdf",extractor=extractor)
    assert seen and facts.reports[0].mine_name == "Fresh Automatic Mine"
    assert facts.extraction_audit["transcript_seed_loaded"] is False


@pytest.mark.asyncio
@pytest.mark.parametrize("policy",["refresh","off"])
async def test_refresh_and_off_bypass_existing_automatic_cache(tmp_path,policy):
    source=tmp_path/"scan.pdf"; Image.new("RGB", (120, 160), "white").save(source,"PDF")
    digest=documents.fingerprint(source)
    main_fan_store.save_document(digest,report(mine="Old Auto Mine"),"automatic_extraction")
    async def extractor(pages,folder): return report(mine="Fresh Auto Mine")
    facts=await documents.read_document(source,"/workspace/scan.pdf",extractor=extractor,cache_policy=policy)
    assert facts.reports[0].mine_name == "Fresh Auto Mine"
    assert facts.extraction_audit["cache_used"] is False
    cached=main_fan_store.document(digest)
    assert cached["reports"][0]["mine_name"] == ("Old Auto Mine" if policy=="off" else "Fresh Auto Mine")


@pytest.mark.asyncio
async def test_truncated_model_output_is_audited_but_never_cached(tmp_path,monkeypatch):
    source=tmp_path/"scan.pdf"; Image.new("RGB", (120, 160), "white").save(source,"PDF")
    monkeypatch.setenv("MAIN_FAN_VISION_MODEL","test-vision")
    class Truncated:
        async def ainvoke(self,*args,**kw):
            from langchain_core.messages import AIMessage
            return AIMessage(content='{"reports":[',response_metadata={"finish_reason":"length"})
    monkeypatch.setattr("agents.model_factory.build_model",lambda **kw:Truncated())
    with pytest.raises(ValueError,match="截断"):
        await documents.read_document(source,"/workspace/scan.pdf",audit_dir=tmp_path/"audit")
    assert main_fan_store.document(documents.fingerprint(source)) is None
    call=json.loads(next((tmp_path/"audit").glob('call-*.json')).read_text())
    assert call["finish_reason"] == "length" and call["route"] == "vision"


def test_multi_page_identity_normalizes_spacing_and_fan_notation():
    from algorithms.main_fan.models import DocumentFacts
    all_facts,reports=DocumentFacts(),{}
    first=DocumentFacts.model_validate(report())
    first.reports[0].report_id="ABC - 001"
    first.reports[0].fan_name="1#通风机"
    documents._merge_batch(all_facts,reports,first,[{"page":1,"route":"vision"}])
    second=DocumentFacts.model_validate(report(page=2))
    second.reports[0].report_id="ABC-001"
    documents._merge_batch(all_facts,reports,second,[{"page":2,"route":"text"}])
    assert len(reports)==1


def test_source_units_and_input_power_cannot_create_performance_rows():
    from algorithms.main_fan.models import DocumentFacts
    facts=DocumentFacts.model_validate({'reports':[{'report_id':'SYN','mine_name':'测试矿','fan_name':'1号通风机','points':[
        {'label':'判定汇总','frequency_hz':30,'flow_m3_s':63.11/60,'static_pressure_pa':520.74,'shaft_power_kw':56.12,
         'evidence':{'page':1,'excerpt':'30Hz63.11m³/s；静压520.74Pa；电动机运行功率56.12kW'}}]}]})
    issues=documents._numeric_issues(facts)
    assert len(issues)==1
    assert any('风量原文单位' in reason for reason in issues[0]['reasons'])
    assert any('原文未明确轴功率' in reason for reason in issues[0]['reasons'])
    assert facts.reports[0].points[0].shaft_power_kw==56.12  # 校验只标记，不能按公式改数。


def test_curve_caption_must_match_table_flow_and_shaft_power_before_comparison():
    from algorithms.main_fan.models import DocumentFacts
    from algorithms.main_fan.analysis import screen
    from algorithms.main_fan.models import ScreeningInput
    facts=DocumentFacts.model_validate({'reports':[{'report_id':'SYN','mine_name':'测试矿','fan_name':'1号通风机',
        'points':[{'label':'运行工况','frequency_hz':30,'condition':'converted','flow_m3_s':105,
          'static_pressure_pa':1600,'shaft_power_kw':260}],
        'curves':[{'label':'图注','frequency_hz':30,'condition':'converted','flow_m3_s':63,
          'static_pressure_pa':510,'shaft_power_kw':51}]}]})
    analysis=screen(ScreeningInput(mine_code='synthetic',mine_name='测试矿',fan_name='1号通风机',reports=facts.reports))
    assert not any(f.code=='curve_table_disagreement' for f in analysis.findings)


@pytest.mark.parametrize("key,value,excerpt,valid",[
    ("inspected_on", date(2022,3,18), "检测检验日期：2022年03月18日", True),
    ("inspected_on", date(2022,3,18), "检测检验日期 2022 年 03 月 18 日", True),
    ("recommended_next_on", date(2025,3,17), "建议下次检测检验日期 2025 年 03 月 17 日", True),
    ("inspected_on", date(2017,8,1), "生产/出厂日期：2017年08月", False),
    ("inspected_on", date(2025,3,17), "建议下次检测检验日期：2025年03月17日", False),
    ("recommended_next_on", date(2025,3,17), "建议下次检测检验日期：2025年03月17日", True),
    ("inspected_on", date(2026,1,2), "Inspection date: 2026-01-02", True),
    ("inspected_on", date(2023,3,18), "检测检验日期：2022年03月18日", False)])
def test_date_requires_matching_field_label_and_year_month_day(key,value,excerpt,valid):
    assert documents._date_supported(value,key,excerpt) == valid


def test_missing_identity_page_can_only_inherit_a_matching_automatic_report():
    from algorithms.main_fan.models import DocumentFacts
    all_facts,reports=DocumentFacts(),{}
    first=DocumentFacts.model_validate(report())
    documents._merge_batch(all_facts,reports,first,[{"page":1,"route":"vision"}])
    next_report=DocumentFacts.model_validate(report())
    next_report.reports[0].evidence.page=None
    documents._merge_batch(all_facts,reports,next_report,[{"page":2,"route":"vision"}])
    assert reports['AUTO-001'].evidence.page==1
    wrong=DocumentFacts.model_validate(report(mine="Different Mine"))
    wrong.reports[0].evidence.page=None
    with pytest.raises(ValueError,match="原页"):
        documents._merge_batch(all_facts,reports,wrong,[{"page":2,"route":"vision"}])


def test_conflicting_inspection_dates_do_not_become_valid_again_on_later_pages():
    from algorithms.main_fan.models import DocumentFacts
    all_facts,reports=DocumentFacts(),{}
    for page,year in [(1,2022),(2,2023),(3,2022)]:
        data=report(page=page)
        data['reports'][0]['inspected_on']=f'{year}-03-18'
        data['reports'][0]['evidence']['excerpt']=f'检测检验日期：{year}年03月18日'
        documents._merge_batch(all_facts,reports,DocumentFacts.model_validate(data),[{"page":page,"route":"vision"}])
    assert reports['AUTO-001'].inspected_on is None
    assert all_facts.extraction_audit['metadata_conflicts']


def test_server_rebinds_metadata_methods_from_real_page_route():
    from algorithms.main_fan.models import DocumentFacts
    raw=report()
    raw['reports'][0]['metadata_evidence']={'inspected_on':{'page':1,'excerpt':'检测日期 2022-03-18','method':'text_extraction'}}
    facts=DocumentFacts.model_validate(raw)
    facts.extraction_audit={'pages':[{'page':1,'route':'vision'}]}
    documents.rebind_sources(facts,'/workspace/original.pdf','synthetic-hash')
    evidence=facts.reports[0].metadata_evidence['inspected_on']
    assert evidence.source == '/workspace/original.pdf' and evidence.sha256 == 'synthetic-hash'
    assert evidence.method == 'model_extraction'


@pytest.mark.asyncio
async def test_incomplete_curve_caption_is_excluded_without_discarding_complete_rows(monkeypatch,tmp_path):
    monkeypatch.setenv('MAIN_FAN_VISION_MODEL','test-vision')
    raw=report()
    raw['reports'][0]['curves']=[{'label':'40Hz图','frequency_hz':40,'flow_m3_s':None,
        'static_pressure_pa':None,'shaft_power_kw':None,'evidence':{'page':1,'excerpt':'40Hz性能曲线'}}]
    class Model:
        async def ainvoke(self,*args,**kwargs):
            from langchain_core.messages import AIMessage
            return AIMessage(content=json.dumps(raw))
    monkeypatch.setattr('agents.model_factory.build_model',lambda **kw:Model())
    (tmp_path/'image.png').write_bytes(b'synthetic')
    facts=await documents.model_extract([{'page':1,'route':'vision','text':'','image':'image.png'}],tmp_path)
    assert facts.reports and not facts.reports[0].curves
    assert any('未纳入公式计算' in item for item in facts.warnings)


@pytest.mark.asyncio
async def test_report_warnings_are_preserved_at_document_level(monkeypatch,tmp_path):
    monkeypatch.setenv('MAIN_FAN_VISION_MODEL','test-vision')
    raw=report()
    raw['reports'][0]['warnings']=['电动机输入功率不可代替轴功率']
    class Model:
        async def ainvoke(self,*args,**kwargs):
            from langchain_core.messages import AIMessage
            return AIMessage(content=json.dumps(raw))
    monkeypatch.setattr('agents.model_factory.build_model',lambda **kw:Model())
    (tmp_path/'image.png').write_bytes(b'synthetic')
    facts=await documents.model_extract([{'page':1,'route':'vision','text':'','image':'image.png'}],tmp_path)
    assert facts.reports[0].report_id == 'AUTO-001'
    assert facts.warnings == ['报告 AUTO-001：电动机输入功率不可代替轴功率']


@pytest.mark.asyncio
@pytest.mark.parametrize('always_invalid',[False,True])
async def test_schema_retry_rereads_original_pages_and_is_bounded(monkeypatch,tmp_path,always_invalid):
    source=tmp_path/'native.pdf'; native_pdf(source)
    seen=[]
    class Model:
        async def ainvoke(self,messages,config):
            from langchain_core.messages import AIMessage
            seen.append(messages)
            raw=report()
            if len(seen)==1 or always_invalid:
                raw['reports'][0]['unrecognized_field']='invalid response shape'
            return AIMessage(content=json.dumps(raw),response_metadata={'finish_reason':'stop'})
    monkeypatch.setattr('agents.model_factory.build_model',lambda **kw:Model())
    if always_invalid:
        with pytest.raises(ValueError):
            await documents.read_document(source,'/workspace/native.pdf',audit_dir=tmp_path/'audit')
        assert main_fan_store.document(documents.fingerprint(source)) is None
    else:
        facts=await documents.read_document(source,'/workspace/native.pdf',audit_dir=tmp_path/'audit')
        assert facts.reports[0].report_id=='AUTO-001'
        assert facts.extraction_audit['calls'][0]['validation_error']=='ValidationError'
    assert len(seen)==2
    assert seen[0][-1].content == seen[1][-1].content
    assert 'NATIVE-TEST-001' in seen[1][-1].content[0]['text']
    assert '结构校验未通过' in seen[1][0].content


def test_date_label_can_be_stored_separately_from_its_original_value():
    from algorithms.main_fan.models import DocumentFacts
    raw=report()
    raw['reports'][0]['inspected_on']='2022-03-18'
    raw['reports'][0]['metadata_evidence']={'inspected_on':{'page':1,'label':'检测检验日期','excerpt':'2022 年 03 月 18 日'}}
    facts=DocumentFacts.model_validate(raw)
    reports={}
    documents._merge_batch(DocumentFacts(),reports,facts,[{'page':1,'route':'vision'}])
    assert reports['AUTO-001'].inspected_on == date(2022,3,18)


@pytest.mark.asyncio
@pytest.mark.parametrize('still_inconsistent',[False,True])
async def test_numeric_recheck_uses_original_image_crop_and_never_backcalculates(monkeypatch,tmp_path,still_inconsistent):
    source=tmp_path/'scan.pdf'; Image.new('RGB',(240,320),'white').save(source,'PDF')
    monkeypatch.setenv('MAIN_FAN_VISION_MODEL','test-vision')
    seen=[]
    class Model:
        async def ainvoke(self,messages,config):
            from langchain_core.messages import AIMessage
            seen.append(messages)
            raw=report()
            raw['reports'][0]['curves']=[{'label':'synthetic curve','flow_m3_s':60,'static_pressure_pa':500,
                'shaft_power_kw':60 if len(seen)==1 or still_inconsistent else 50,
                'reported_efficiency_pct':60,'evidence':{'page':1,'excerpt':'synthetic caption'}}]
            return AIMessage(content=json.dumps(raw),response_metadata={'finish_reason':'stop'})
    monkeypatch.setattr('agents.model_factory.build_model',lambda **kw:Model())
    facts=await documents.read_document(source,'/workspace/scan.pdf',cache_policy='refresh')
    assert len(seen)==2
    second_images=[block for block in seen[1][-1].content if block['type']=='image_url']
    assert len(second_images)==2
    assert second_images[0] == [block for block in seen[0][-1].content if block['type']=='image_url'][0]
    assert '不按公式反算修正' in seen[1][0].content
    assert facts.extraction_audit['calls'][1]['purpose']=='numeric_recheck'
    assert facts.extraction_audit['calls'][1]['detail_images']==1
    assert facts.reports[0].curves[0].shaft_power_kw == (60 if still_inconsistent else 50)
    assert any('放大复读后公式仍不一致' in w for w in facts.warnings) == still_inconsistent


def test_deepseek_off_is_explicit_but_other_provider_settings_remain_generic(monkeypatch):
    from agents import model_factory
    seen=[]
    monkeypatch.setattr(model_factory,'_instantiate',lambda name,common:seen.append(common) or object())
    model_factory.build_model(model_name='deepseek-flash',thinking='off')
    assert seen[-1]['extra_body']['thinking']['type']=='disabled'
    model_factory.build_model(model_name='generic-test',thinking='off',api_key='synthetic-vision-token',base_url='https://vision.invalid')
    assert 'extra_body' not in seen[-1]
    assert seen[-1]['api_key']=='synthetic-vision-token' and seen[-1]['base_url']=='https://vision.invalid'