"""固定工作流回归：PDF 分流、完整输入、单次 Qwen、标准报文及禁止推送。"""
import json
import shutil

import httpx
from PIL import Image
import pypdfium2 as pdfium
import pytest

import config
from db import business_store, main_fan_store
from extensions.main_fan import exploration, workflow, workflow_documents, workflow_json_repair, workflow_page_filter
from extensions.main_fan.batch import BatchRunner
from runtime.python_sandbox import digest, workspace_file
from runtime.result_models import PlatformResultPayload
from test_main_fan_batch import batch_workspace, bound
from test_main_fan_documents import native_pdf
from test_main_fan_preparation import FixedGateway


def conclusion(code='NORMAL'):
    return {'objectDetail': '测试甲矿主备风机（通风井归属待确认）',
        'dataDetail': '测试资料，开停及风量来自固定代码。',
        'judgment': {'resultCode': code, 'result': '现有数据未发现确凿异常。',
            'reason': '原件为模拟资料，不代入本矿实际结论；未提供准确切换间隔。', 'evidenceLink': ''}}


@pytest.mark.asyncio
async def test_mixed_pdf_uses_python_and_ocr_then_rebinds_raw_cache(batch_workspace, monkeypatch):
    native, scan, mixed = [batch_workspace.root / name for name in ('native.pdf', 'scan.pdf', 'mixed.pdf')]
    native_pdf(native)
    image = Image.new('RGB', (300, 400), 'white')
    image.save(scan, 'PDF')
    image.close()
    with pdfium.PdfDocument.new() as combined:
        with pdfium.PdfDocument(native) as doc:
            combined.import_pages(doc)
        with pdfium.PdfDocument(scan) as doc:
            combined.import_pages(doc)
        combined.save(mixed)
    calls = []
    async def respond(request):
        calls.append(json.loads(request.content))
        assert 'authorization' not in request.headers
        return httpx.Response(200, json={'choices': [{'finish_reason': 'stop',
            'message': {'content': '合成扫描页：风量 80.50 m³/s，功率 132.6 kW。'}}]})
    with bound(batch_workspace) as (scope, identifier):
        virtual = '/workspace/exploration/' + identifier + '/mixed.pdf'
        local = workspace_file(scope.user, scope.session, virtual)
        shutil.copyfile(mixed, local)
        material = {'path': virtual, 'filename': 'mixed.pdf', 'sha256': digest(local),
                    'page_count': 2, 'is_mock': True, 'binding': 'synthetic'}
        async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
            first = workflow_documents.PDFExtractor(scope.user, scope.session,
                '/workspace/exploration/' + identifier + '/first', scope.cache_root, client=client)
            reports = await first.extract([material])
            assert reports[0]['complete'] and not reports[0]['reused']
            assert [row['method'] for row in reports[0]['pages']] == ['python', 'glm_ocr']
            assert 'NATIVE-TEST-001' in reports[0]['pages'][0]['text']
            assert len(calls) == 1 and calls[0]['messages'][0]['content'][1]['text'] == 'Text Recognition:'
            second = workflow_documents.PDFExtractor(scope.user, scope.session,
                '/workspace/exploration/' + identifier + '/second', scope.cache_root, client=client)
            cached = await second.extract([material])
            assert cached[0]['reused'] and len(calls) == 1
            assert all('/second/' in row['text_path'] for row in cached[0]['pages'])


def test_watermark_or_large_embedded_scan_is_not_full_native_text():
    assert not workflow_documents.usable_text('契约锁\n扫描全能王')
    assert workflow_documents.usable_text('Main fan performance table report with readable values 80.5 132.6')
    class ImageObject:
        def get_bounds(self):
            return (0, 0, 595, 600)
    class Page:
        def get_size(self):
            return (595, 842)
        def get_objects(self, **kwargs):
            return iter([ImageObject()])
    assert workflow_documents.large_raster(Page()), '带可读页眉的扫描表格也必须送 OCR'


def test_full_monitoring_expands_existing_results_without_recalculation(batch_workspace):
    with bound(batch_workspace) as (scope, identifier):
        current = main_fan_store.workflow(identifier, scope.user, scope.session)
        prefix = '/workspace/exploration/' + identifier
        periods = [{'state': '开' if i % 2 else '停', 'observed_from': '2026-10-01T00:00:00+08:00',
            'observed_until': '2026-10-01T00:01:00+08:00', 'sample_count': 1} for i in range(25)]
        events = [{'event_id': 'S' + str(i), 'from_fan': '1号风机', 'to_fan': '2号风机',
            'after': '2026-10-01T00:00:00+08:00', 'by': '2026-10-01T00:01:00+08:00',
            'flow_context': [], 'flow_comparison': {'method': '已整理的单点对照',
                'relative_change_percent': None, 'invalid_reasons': ['flow_pair_not_unique']}} for i in range(25)]
        details = {'cutoff': scope.cutoff, 'lookback_days': 40, 'mine': scope.mine,
            'points': [{'node_id': 'state1', 'label': '1号一级', 'system': 'safety', 'role': 'fan_state',
                'state_intervals': periods, 'state_changes': [], 'state_value_semantics': '接口原标签'}],
            'preprocessing': {'switch_candidates': events}}
        detail_path, compact_path = prefix + '/details.json', prefix + '/compact.json'
        exploration._write(workspace_file(scope.user, scope.session, detail_path), details)
        exploration._write(workspace_file(scope.user, scope.session, compact_path), {'details_path': detail_path})
        context = current['context']
        context['prepared_data'] = {'agent_data_path': compact_path}
        context.setdefault('query_exports', {})[detail_path] = digest(workspace_file(scope.user, scope.session, detail_path))
        actual = workflow.full_monitoring(context, scope.user, scope.session)
        assert len(actual['states'][0]['periods']) == 25
        assert len(actual['switch_candidates']) == 25
        assert not actual['states'][0].get('truncated')
        assert actual['switch_candidates'][-1]['comparison']['change_percent'] is None


@pytest.mark.parametrize('raw', ['', '准备调用工具', '{"objectDetail":"a"}', json.dumps(conclusion(), ensure_ascii=False) + ' trailing'])
def test_invalid_model_reply_never_becomes_completed_result(raw):
    with pytest.raises(workflow.WorkflowError):
        workflow.parse_decision(raw)


def test_unescaped_quotes_report_json_position_without_disclosing_body():
    raw = json.dumps(conclusion(), ensure_ascii=False).replace('现有数据未发现确凿异常。', '核心项"性能一致性"未核实。')
    with pytest.raises(workflow.WorkflowError) as error:
        workflow.parse_decision(raw)
    assert error.value.code == 'qwen_invalid_json'
    assert error.value.details['error_type'] == 'JSONDecodeError'
    assert error.value.details['line'] == 1 and error.value.details['column'] > 1
    assert 'objectDetail' not in str(error.value)


def test_schema_failure_is_distinct_from_invalid_json():
    with pytest.raises(workflow.WorkflowError) as error:
        workflow.parse_decision('{}')
    assert error.value.code == 'qwen_schema_validation'
    assert error.value.details['errors'][0]['loc'] == ('objectDetail',)
    assert 'input' not in error.value.details['errors'][0]


def test_json_repair_can_fill_fixed_link_but_preserves_business_values():
    value = conclusion()
    value['dataDetail'] = '风量96.7 m³/s，静压495.58 Pa。'
    original = json.dumps(value, ensure_ascii=False)[:-1]
    assert workflow_json_repair.can_repair(original, 'qwen_invalid_json')
    candidate = workflow.WorkflowDecision.model_validate(value)
    assert not workflow_json_repair.preservation_errors(original, candidate)
    candidate.dataDetail = '风量97.7 m³/s，静压495.58 Pa。'
    assert 'numeric_content' in workflow_json_repair.preservation_errors(original, candidate)
    candidate.dataDetail = '风量96.7 m³/s，静压495.58 kPa。'
    assert 'identifiers_or_units' in workflow_json_repair.preservation_errors(original, candidate)
    missing_link = json.loads(json.dumps(value))
    missing_link['judgment'].pop('evidenceLink')
    assert workflow_json_repair.can_repair(json.dumps(missing_link), 'qwen_schema_validation')
    assert not workflow_json_repair.preservation_errors(json.dumps(missing_link),
        workflow.WorkflowDecision.model_validate(value))


def test_all_failed_pdf_pages_do_not_block_available_monitoring():
    reports = [{'path': '/workspace/report.pdf', 'pages': [
        {'page': number, 'status': 'failed', 'text': '', 'error_stage': 'python_text',
         'error': 'TypeError: unsupported context manager'} for number in (1, 2)]}]
    monitoring = {'states': [{'periods': [{'state': '开'}]}], 'flows': [], 'switch_candidates': []}
    availability = workflow.data_availability(reports, monitoring)
    assert availability['any_available'] and availability['pdf_readable_pages'] == 0
    assert availability['known_state_periods'] == 1
    workflow.validate_result_policy(workflow.WorkflowDecision.model_validate(conclusion()), availability)


@pytest.mark.parametrize('monitoring', [
    {'flows': [{'samples': 1, 'range': [0, 0]}]},
    {'switch_candidates': [{'comparison': {'before_value': 100, 'after_value': 0, 'change_percent': 100}}]},
    {'switch_candidates': [{'flow': [{'before': None, 'after': ['2026-10-22 00:00:00', 0]}]}]},
])
def test_zero_flow_and_partial_reference_values_count_as_available_data(monitoring):
    availability = workflow.data_availability([], monitoring)
    assert availability['any_available']
    codes = workflow.decision_schema(availability)['$defs']['WorkflowJudgment']['properties']['resultCode']['enum']
    assert codes == ['NORMAL', 'WARNING', 'ABNORMAL']
    with pytest.raises(workflow.WorkflowError) as error:
        workflow.validate_result_policy(workflow.WorkflowDecision.model_validate(conclusion('UNDETERMINED')), availability)
    assert error.value.code == 'workflow_result_policy'


def test_undetermined_only_when_all_three_sources_have_no_usable_data():
    reports = [{'is_mock': True, 'pages': [{'status': 'completed', 'text': '合成报告'}]}]
    monitoring = {'states': [{'periods': [{'state': '未知'}]}], 'flows': [{'samples': 0}],
        'switch_candidates': [{'duration_minutes': None, 'comparison': {'change_percent': None}}]}
    availability = workflow.data_availability(reports, monitoring)
    assert not availability['any_available']
    codes = workflow.response_format(availability)['json_schema']['schema']['$defs']['WorkflowJudgment']['properties']['resultCode']['enum']
    assert codes == ['UNDETERMINED']
    decision = workflow.WorkflowDecision.model_validate(conclusion('UNDETERMINED'))
    workflow.validate_result_policy(decision, availability)
    with pytest.raises(workflow.WorkflowError):
        workflow.validate_result_policy(workflow.WorkflowDecision.model_validate(conclusion()), availability)


def test_readable_real_report_counts_when_monitoring_is_missing():
    reports = [{'is_mock': False, 'pages': [{'status': 'completed', 'text': '测试甲矿通风机性能报告'}]}]
    availability = workflow.data_availability(reports, {})
    assert availability['any_available'] and availability['pdf_readable_pages'] == 1
    workflow.validate_result_policy(workflow.WorkflowDecision.model_validate(conclusion()), availability)


def test_page_filter_only_removes_known_irrelevant_pages_and_preserves_source():
    pages = [{'page': number, 'status': 'completed', 'text': text} for number, text in enumerate([
        '报告封面：测试甲矿，编号R-01', '声 明\n本报告未经批准不得复制。',
        '检测检验用仪器设备一览表\n记录包含风量，保守保留', '性能测定数据\n频率45Hz，叶片角度0°',
        '', '', '资质证书\n附有本矿通风机编号与额定功率', '',
    ], 1)]
    pages[5]['status'] = 'failed'
    pages[7]['blank'] = True
    source = {'path': '/workspace/report.pdf', 'page_count': 8, 'complete': False, 'pages': pages}
    selected, manifest = workflow_page_filter.select_pages([source])
    assert [p['page'] for p in selected[0]['pages']] == [1, 3, 4, 5, 6, 7]
    assert [p['page'] for p in source['pages']] == list(range(1, 9))
    assert 'page_selection' not in source and selected[0]['page_count'] == 8
    assert manifest[0]['excluded_pages'] == [{'page': 2, 'reason': '声明页'}, {'page': 8, 'reason': '空白页'}]


@pytest.mark.asyncio
@pytest.mark.parametrize('primary_failure,fallback_failure', [('length', False), ('repetition', False), ('length', True)])
async def test_failed_glm_page_uses_qwen_once_and_retains_both_responses(batch_workspace, monkeypatch, primary_failure, fallback_failure):
    monkeypatch.setattr(config, 'MODEL', 'Qwen/Qwen3.5-35B-A3B-FP8')
    monkeypatch.setattr(config, 'OPENAI_BASE_URL', 'https://qwen.invalid/v1')
    monkeypatch.setattr(config, 'OPENAI_API_KEY', 'qwen-only-key')
    monkeypatch.setattr(config, 'THINKING_LEVEL', 'highest')
    calls, qwen_calls, usages = [], [], []
    async def respond(request):
        body = json.loads(request.content)
        calls.append(body)
        if body['model'] == 'glm-ocr':
            assert 'authorization' not in request.headers
            finish, text = ('stop', '1.0500 1.1429\n' * 20) if primary_failure == 'repetition' else ('length', '不完整页')
        else:
            assert request.headers['authorization'] == 'Bearer qwen-only-key'
            assert body['chat_template_kwargs']['enable_thinking'] is True
            finish, text = ('length', '仍未完整') if fallback_failure else ('stop', '测试甲矿2号风机45Hz性能表：风量96.7，静压495.58。')
        return httpx.Response(200, json={'choices': [{'finish_reason': finish, 'message': {'content': text}}],
            'usage': {'prompt_tokens': 10, 'completion_tokens': 20}})
    with bound(batch_workspace) as (scope, identifier):
        directory = workspace_file(scope.user, scope.session, '/workspace/exploration/' + identifier + '/page.png').parent
        image = Image.new('RGB', (30, 40), 'white')
        image.save(directory / 'page.png')
        image.close()
        row = {'page': 16, 'status': 'pending', 'method': 'glm_ocr', 'text': '', 'image_name': 'page.png'}
        async with httpx.AsyncClient(transport=httpx.MockTransport(respond)) as client:
            extractor = workflow_documents.PDFExtractor(scope.user, scope.session, '/workspace/exploration/' + identifier,
                scope.cache_root, client=client, on_qwen_call=lambda: qwen_calls.append(True), on_qwen_usage=usages.append)
            await extractor._ocr(row, directory, client)
        assert len(calls) == 2 and len(qwen_calls) == 1 and len(usages) == 1
        assert row['method'] == 'qwen_ocr' and row['fallback_used'] and len(row['attempts']) == 2
        assert row['status'] == ('failed' if fallback_failure else 'completed')
        assert bool(row['text']) is not fallback_failure
        assert (directory / 'page-0016.response.json').is_file()
        assert (directory / 'page-0016.qwen.response.json').is_file()


@pytest.mark.asyncio
@pytest.mark.parametrize('repair_outcome', ['success', 'invalid_json', 'changed_result'])
async def test_json_repair_is_bounded_does_not_resend_source_or_reassess(batch_workspace, monkeypatch, repair_outcome):
    from extensions.main_fan import gateway_queries, history
    gateway = FixedGateway(batch_workspace.mine)
    monkeypatch.setenv('SHANXI_GATEWAY_ENABLED', '1')
    monkeypatch.setenv('MAIN_FAN_WORKFLOW_PUBLISH_ENABLED', 'False')
    monkeypatch.setenv('MAIN_FAN_JSON_REPAIR_MAX_ATTEMPTS', '3')
    monkeypatch.setenv('BILLING_ENABLED', 'False')
    monkeypatch.setattr(config, 'MODEL', 'Qwen/Qwen3.5-35B-A3B-FP8')
    monkeypatch.setattr(config, 'THINKING_LEVEL', 'highest')
    monkeypatch.setattr(config, 'OPENAI_BASE_URL', 'https://qwen.invalid/v1')
    monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway)
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    original = json.dumps(conclusion(), ensure_ascii=False)[:-1]
    calls = []
    async def respond(request):
        body = json.loads(request.content)
        calls.append(body)
        if len(calls) == 1:
            assert 'pdf' in json.loads(body['messages'][1]['content'])
            reply = original
        else:
            inputs = json.loads(body['messages'][1]['content'])
            assert set(inputs) == {'invalid_json', 'required_schema', 'validation_error'}
            assert '原件为模拟资料' in inputs['invalid_json']
            reply = original if repair_outcome == 'invalid_json' else json.dumps(
                conclusion('WARNING' if repair_outcome == 'changed_result' else 'NORMAL'), ensure_ascii=False)
        return httpx.Response(200, json={'id': 'synthetic-' + str(len(calls)),
            'usage': {'prompt_tokens': 10, 'completion_tokens': 20},
            'choices': [{'finish_reason': 'stop', 'message': {'content': reply}}]})
    monkeypatch.setattr(workflow, 'qwen_http_client', lambda: httpx.AsyncClient(transport=httpx.MockTransport(respond)))
    output = batch_workspace.root / ('repair-' + repair_outcome)
    result = await BatchRunner(output, batch_workspace.pdf).run(
        {'dept_ids': ['1219'], 'cutoff': '2026-10-20 15:00:00', 'lookback_days': 1},
        snapshot={'snapshot_id': 'f' * 32, 'mines': [batch_workspace.mine]})
    assert len(calls) == (2 if repair_outcome == 'success' else 4)
    execution = json.loads((output / '1219/execution-unresolved-main.json').read_text())
    assert execution['diagnostics']['judgment_calls'] == 1
    assert execution['diagnostics']['json_repair_calls'] == len(calls) - 1
    assert not business_store.list_jobs('main_fan', 'main-fan-batch', 100)
    folder = output / '1219/unresolved-main'
    if repair_outcome == 'success':
        assert result['status'] == 'completed'
        first = json.loads((folder / 'Qwen原始响应.json').read_text())['choices'][0]['message']['content']
        assert first == original
        final = json.loads((folder / 'Qwen最终响应.json').read_text())['choices'][0]['message']['content']
        assert json.loads(final) == conclusion()
    else:
        assert result['status'] == 'partial'
        assert execution['error'] == 'qwen_json_repair_exhausted'
        assert not (folder / '待审阅报文.json').exists()


@pytest.mark.asyncio
@pytest.mark.parametrize('invalid', [False, True])
async def test_production_batch_calls_qwen_once_archives_inputs_and_never_pushes(batch_workspace, monkeypatch, invalid):
    from extensions.main_fan import gateway_queries, history
    gateway = FixedGateway(batch_workspace.mine)
    monkeypatch.setenv('SHANXI_GATEWAY_ENABLED', '1')
    monkeypatch.setenv('MAIN_FAN_WORKFLOW_PUBLISH_ENABLED', 'False')
    monkeypatch.setenv('BILLING_ENABLED', 'False')
    monkeypatch.setattr(config, 'MODEL', 'Qwen/Qwen3.5-35B-A3B-FP8')
    monkeypatch.setattr(config, 'THINKING_LEVEL', 'highest')
    monkeypatch.setattr(config, 'OPENAI_BASE_URL', 'https://qwen.invalid/v1')
    monkeypatch.setattr(gateway_queries, 'get_service', lambda _: gateway)
    monkeypatch.setattr(history, 'get_service', lambda _: gateway)
    calls = []
    async def respond(request):
        body = json.loads(request.content)
        calls.append(body)
        assert 'tools' not in body and len(body['messages']) == 2
        assert body['chat_template_kwargs']['enable_thinking'] is True
        assert body['response_format']['type'] == 'json_schema'
        assert body['response_format']['json_schema']['strict'] is True
        inputs = json.loads(body['messages'][1]['content'])
        assert inputs['pdf'][0]['pages'][0]['method'] == 'python'
        assert len(inputs['airflow']['flows']) == 2 and len(inputs['switching']['states']) == 4
        return httpx.Response(200, json={'id': 'synthetic-qwen-response',
            'usage': {'prompt_tokens': 123, 'completion_tokens': 45, 'total_tokens': 168},
            'choices': [{'finish_reason': 'stop', 'message': {'content': '{}' if invalid else json.dumps(conclusion()),
                'reasoning': 'Synthetic provider reasoning retained only in raw response.'}}]})
    # 在默认工作流的 HTTP 边界替换网络，不注入另一个业务执行器。
    monkeypatch.setattr(workflow, 'qwen_http_client',
        lambda: httpx.AsyncClient(transport=httpx.MockTransport(respond)))
    output = batch_workspace.root / ('invalid-workflow' if invalid else 'workflow-output')
    result = await BatchRunner(output, batch_workspace.pdf).run(
        {'dept_ids': ['1219'], 'cutoff': '2026-10-20 15:00:00', 'lookback_days': 1},
        snapshot={'snapshot_id': 'e' * 32, 'mines': [batch_workspace.mine]})
    assert len(calls) == 1 and not business_store.list_jobs('main_fan', 'main-fan-batch', 100)
    assert result['publish'] is False
    folder = output / '1219/unresolved-main'
    if invalid:
        assert result['status'] == 'partial'
        assert not (folder / '待审阅报文.json').exists()
        assert 'qwen_schema_validation' in result['mines'][0]['shafts'][0]['error']
    else:
        assert result['status'] == 'completed', result
        payload = PlatformResultPayload.model_validate_json((folder / '待审阅报文.json').read_text())
        assert payload.mineCode == batch_workspace.mine['mine_id']
        assert payload.judgment.resultCode == 'NORMAL'
        assert (folder / '给Qwen的数据.json').is_file() and (folder / 'PDF提取结果.json').is_file()
        assert 'reasoning' in (folder / 'Qwen原始响应.json').read_text()
        assert json.loads((folder / '推送记录.json').read_text())['status'] == 'local_only'