from io import BytesIO import re from fastapi.testclient import TestClient from openpyxl import load_workbook from app.core.config import settings from app.db.repositories import document as document_repo from app.db.init_db import init_db from app.db.seed import seed_demo_data from app.db.session import SessionLocal from app.main import app from app.services.document_business_templates import resolve_document_template def test_seed_dashboard_diagnosis_and_evaluation(monkeypatch) -> None: init_db() with SessionLocal() as db: seed_demo_data(db) matched = resolve_document_template( db, title="上湾煤矿2025年7月份配风计划7.15(1)", ) assert matched.code == "monthly_air_distribution_plan" capacity = resolve_document_template( db, title="上湾煤矿2025年通风能力核定", ) assert capacity.code == "ventilation_capacity_verification_report" dust_plan = resolve_document_template( db, title="附件：2025年采掘作业场所粉尘分散度、游离二氧化硅含量监测", ) assert dust_plan.code == "dust_dispersion_free_silica_monitoring_ledger" air_door_record = resolve_document_template( db, title="上湾煤矿行车风门检查记录", ) assert air_door_record.code == "vehicle_air_door_inspection_record" with TestClient(app) as client: mines = client.get("/api/v1/mines") assert mines.status_code == 200 mine_id = mines.json()[0]["id"] dashboard = client.get("/api/v1/dashboard/overview", params={"mine_id": mine_id}) assert dashboard.status_code == 200 assert dashboard.json()["mine_name"] == "上湾煤矿" documents = client.get( "/api/v1/documents", params={"mine_id": mine_id, "page": 1, "page_size": 10}, ) assert documents.status_code == 200 document_id = documents.json()["items"][0]["id"] async def fake_process_document_run(run_id: str) -> None: return None monkeypatch.setattr( "app.api.v1.endpoints.documents.process_document_run", fake_process_document_run, ) template_id = client.get("/api/v1/documents/templates").json()[0]["id"] reprocess = client.post( f"/api/v1/documents/{document_id}/process", json={ "force": True, "parse_mode": "lightweight", "template_id": template_id, }, ) assert reprocess.status_code == 202 assert reprocess.json()["document_id"] == document_id assert reprocess.json()["status"] == "pending" processing_tasks = client.get( "/api/v1/documents/processing-tasks", params={"mine_id": mine_id, "limit": 10}, ) assert processing_tasks.status_code == 200 assert "processing_seconds" in processing_tasks.json()[0] assert "started_at" in processing_tasks.json()[0] assert any( item["document_id"] == document_id for item in processing_tasks.json() ) with SessionLocal() as db: document_repo.create_processing_run( db, { "document_id": document_id, "status": "completed", "stage": "completed", "progress": 100, "document_type": "", "parser_engine": "pytest", "parsed_markdown_path": "", "parsed_markdown": "PYTEST_FULLTEXT_TOKEN 用于验证解析全文检索。", "model_output": {}, "error": "", "metadata_json": {}, }, ) fulltext_search = client.get( "/api/v1/documents/search", params={"mine_id": mine_id, "q": "PYTEST_FULLTEXT_TOKEN"}, ) assert fulltext_search.status_code == 200 fulltext_search = fulltext_search.json() assert fulltext_search["total"] >= 1 assert fulltext_search["items"][0]["document"]["id"] == document_id assert any( snippet["source"] == "parsed_text" for snippet in fulltext_search["items"][0]["snippets"] ) export = client.get( "/api/v1/documents/export", params={"mine_id": mine_id}, ) assert export.status_code == 200 assert export.content.startswith(b"PK") assert "spreadsheetml" in export.headers["content-type"] assert "filename*=UTF-8''" in export.headers["content-disposition"] workbook = load_workbook(BytesIO(export.content), data_only=True) sheet = workbook["资料清单"] assert "上湾煤矿业务资料清单" in str(sheet["A1"].value) assert sheet["A3"].value == "序号" assert sheet["B3"].value == "资料名称" assert sheet["E3"].value == "资料子类" assert all( sheet.cell(row_index, 5).value for row_index in range(4, sheet.max_row + 1) ) document_types = [ sheet.cell(row_index, 6).value for row_index in range(4, sheet.max_row + 1) ] assert all( not value or re.fullmatch(r"[a-z0-9_]+", str(value)) is None for value in document_types ) workbook.close() preview_file = settings.upload_path / "pytest_preview_source.csv" preview_file.parent.mkdir(parents=True, exist_ok=True) preview_file.write_text("column_a,column_b\nvalue_1,value_2\n", encoding="utf-8") preview_document = client.post( "/api/v1/documents", json={ "mine_id": mine_id, "title": "pytest preview document", "category": "TEST", "file_name": "pytest_preview_source.csv", "file_path": str(preview_file), "file_ext": ".csv", "file_size": preview_file.stat().st_size, }, ) assert preview_document.status_code == 201 preview_document_id = preview_document.json()["id"] preview = client.get(f"/api/v1/documents/{preview_document_id}/preview") assert preview.status_code == 200 preview_data = preview.json() assert preview_data["preview_type"] == "text" assert preview_data["source_available"] is True assert preview_data["can_download"] is True assert "column_a,column_b" in preview_data["text"] inline_file = client.get(f"/api/v1/documents/{preview_document_id}/file") assert inline_file.status_code == 200 assert "inline" in inline_file.headers["content-disposition"] assert inline_file.content == preview_file.read_bytes() download_file = client.get( f"/api/v1/documents/{preview_document_id}/file", params={"download": "true"}, ) assert download_file.status_code == 200 assert "attachment" in download_file.headers["content-disposition"] assert client.delete(f"/api/v1/documents/{preview_document_id}").status_code == 200 statistics = client.get( "/api/v1/documents/statistics", params={"mine_id": mine_id}, ) assert statistics.status_code == 200 statistics_data = statistics.json() assert statistics_data["total_documents"] >= 1 assert statistics_data["extracted_fields"] >= 2 assert 0 <= statistics_data["field_completeness"] <= 100 assert 0 <= statistics_data["extraction_accuracy"] <= 100 assert statistics_data["pending_tasks"] >= 1 assert statistics_data["estimated_minutes"] >= 0 assert statistics_data["storage_used_bytes"] >= 0 assert statistics_data["storage_capacity_bytes"] > 0 assert 0 <= statistics_data["storage_usage_percent"] <= 100 assert statistics_data["expected_fields"] >= statistics_data["extracted_fields"] assert ( statistics_data["valid_fields"] + statistics_data["invalid_fields"] + statistics_data["missing_fields"] == statistics_data["expected_fields"] ) expected_completeness = round( statistics_data["valid_fields"] / max(statistics_data["expected_fields"], 1) * 100, 1, ) assert statistics_data["field_completeness"] == expected_completeness document_detail = client.get(f"/api/v1/documents/{document_id}") assert document_detail.status_code == 200 assert document_detail.json()["id"] == document_id metadata_update = client.patch( f"/api/v1/documents/{document_id}", json={ "title": "更新后的资料名称", "code": "UPDATED-001", "category": "瓦斯防治", }, ) assert metadata_update.status_code == 200 assert metadata_update.json()["title"] == "更新后的资料名称" assert metadata_update.json()["code"] == "UPDATED-001" assert metadata_update.json()["category"] == "瓦斯防治" extraction = client.post( f"/api/v1/documents/{document_id}/extractions", json={ "document_id": document_id, "field_key": "test_field", "field_name": "测试字段", "field_value": "原始值", "normalized_value": "原始值", }, ) assert extraction.status_code == 201 extraction = extraction.json() assert extraction["edited"] is False assert extraction["extraction_correct"] is True updated_extraction = client.patch( f"/api/v1/documents/{document_id}/extractions/{extraction['id']}", json={ "field_value": "前端编辑值", "normalized_value": "前端编辑值", }, ) assert updated_extraction.status_code == 200 updated_extraction = updated_extraction.json() assert updated_extraction["normalized_value"] == "前端编辑值" assert updated_extraction["edited"] is True assert updated_extraction["original_value"] == "原始值" assert updated_extraction["edit_judgement_status"] == "pending" assert updated_extraction["extraction_correct"] is True statistics_after_edit = client.get( "/api/v1/documents/statistics", params={"mine_id": mine_id}, ) assert statistics_after_edit.status_code == 200 statistics_after_edit = statistics_after_edit.json() assert statistics_after_edit["edited_fields"] >= 1 assert statistics_after_edit["incorrect_fields"] >= 1 reverted_extraction = client.patch( f"/api/v1/documents/{document_id}/extractions/{extraction['id']}", json={ "field_value": "原始值", "normalized_value": "原始值", }, ) assert reverted_extraction.status_code == 200 reverted_extraction = reverted_extraction.json() assert reverted_extraction["edited"] is False assert reverted_extraction["edit_judgement_status"] == "completed" assert reverted_extraction["edit_judgement_source"] == "reverted" assert reverted_extraction["extraction_correct"] is True statistics_after_revert = client.get( "/api/v1/documents/statistics", params={"mine_id": mine_id}, ) assert statistics_after_revert.status_code == 200 statistics_after_revert = statistics_after_revert.json() assert statistics_after_revert["edited_fields"] == 0 assert statistics_after_revert["incorrect_fields"] == 0 deleted = client.delete(f"/api/v1/documents/{document_id}") assert deleted.status_code == 200 assert client.get(f"/api/v1/documents/{document_id}").status_code == 404 templates = client.get("/api/v1/documents/templates") assert templates.status_code == 200 assert {item["code"] for item in templates.json()} == { "ground_main_fan_inspection_report", "main_fan_external_leakage_rate_report", "monthly_air_distribution_plan", "mine_ventilation_resistance_report", "ventilation_capacity_verification_report", "airflow_measurement_decadal_report", "gas_grade_appraisal_report", "coal_dust_explosion_spontaneous_combustion_report", "safety_monitoring_system_ledger", "dust_dispersion_free_silica_monitoring_ledger", "vehicle_air_door_inspection_record", } diagnosis = client.post( "/api/v1/diagnosis/run", json={"mine_id": mine_id, "scene": "all", "requested_by": "pytest"}, ) assert diagnosis.status_code == 201 assert diagnosis.json()["status"] == "completed" evaluation = client.post( "/api/v1/evaluations/run", json={"mine_id": mine_id, "requested_by": "pytest"}, ) assert evaluation.status_code == 201 assert evaluation.json()["grade"] in {"A", "B", "C", "D"}