LQAgentPlatform/utils_test/Completeness_Enhanced_Test/test_completeness_integration.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
完整性审查集成测试

测试流程：
1. 加载现有的 final_result.json 文件
2. 模拟三级分类结果（添加 tertiary_classification_details）
3. 运行完整性审查
4. 验证结果
"""

import asyncio
import json
import sys
from pathlib import Path

# 添加项目路径
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))

from core.construction_review.component.reviewers.completeness_reviewer import (
    LightweightCompletenessChecker,
    TertiarySpecLoader
)


def load_final_result(file_path: str) -> dict:
    """加载最终结果文件"""
    with open(file_path, 'r', encoding='utf-8') as f:
        return json.load(f)


def simulate_tertiary_classification(chunks: list) -> list:
    """
    模拟三级分类结果
    为每个二级分类下的 chunk 添加 tertiary_classification_details
    """
    # 三级分类标准映射（基于 StandardCategoryTable.csv）
    tertiary_mapping = {
        # 编制依据
        "LawsAndRegulations": [
            {"third_category_code": "NationalLawsAndRegulations", "third_category_name": "国家政府发布的法律法规与规章制度"},
            {"third_category_code": "ProvincialLawsAndRegulationsOfProjectLocation", "third_category_name": "工程所在地省级政府发布的法律法规与规章制度"}
        ],
        "StandardsAndSpecifications": [
            {"third_category_code": "IndustryStandards", "third_category_name": "行业标准"},
            {"third_category_code": "TechnicalRegulations", "third_category_name": "技术规程"}
        ],
        "DocumentSystems": [
            {"third_category_code": "SichuanRoadAndBridgeDocumentSystemsAndmanagementProcedures", "third_category_name": "四川路桥下发的文件制度和管理程序文件"},
            {"third_category_code": "RoadAndBridgeGroupDocumentSystemsAndmanagementProcedures", "third_category_name": "路桥集团下发的文件制度和管理程序文件"},
            {"third_category_code": "BridgeCompanyDocumentSystemsAndmanagementProcedures", "third_category_name": "桥梁公司下发的文件制度和管理程序文件"},
            {"third_category_code": "ConstructionUnitDocumentSystemsAndmanagementProcedures", "third_category_name": "建设单位下发的文件制度和管理程序文件"}
        ],
        "CompilationPrinciples": [
            {"third_category_code": "NationalPoliciesStandardsAndDesignDocument", "third_category_name": "国家方针、政策、标准和设计文件"},
            {"third_category_code": "BasicConstructionProcedures", "third_category_name": "基本建设程序"},
            {"third_category_code": "ProjectFunctionImplementation", "third_category_name": "工程项目功能实现"},
            {"third_category_code": "ContractPerformance", "third_category_name": "合同履约"},
            {"third_category_code": "ConstructionForceConcentration", "third_category_name": "施工力量集中"},
            {"third_category_code": "ProcessControl", "third_category_name": "工序控制"}
        ],
        "CompilationScope": [
            {"third_category_code": "ProjectCoverage", "third_category_name": "填写完整涵盖本方案包含的所有工程"},
            {"third_category_code": "ConstructionTechnology", "third_category_name": "部分工程可简要说明采取的施工工艺"}
        ],
        # 工程概况
        "DesignSummary": [
            {"third_category_code": "ProjectIntroduction", "third_category_name": "工程简介"},
            {"third_category_code": "MainTechnicalStandards", "third_category_name": "主要技术标准"}
        ],
        "GeologyWeather": [
            {"third_category_code": "HydrologicalConditions", "third_category_name": "水文状况"},
            {"third_category_code": "ClimaticConditions", "third_category_name": "气候条件"}
        ],
        "Surroundings": [
            {"third_category_code": "PositionalRelationship", "third_category_name": "位置关系"},
            {"third_category_code": "StructuralDimensions", "third_category_name": "结构尺寸"}
        ],
        "LayoutPlan": [
            {"third_category_code": "TemporaryFacilityLocation", "third_category_name": "临时设施位置"},
            {"third_category_code": "ConstructionWorkPlatform", "third_category_name": "施工作业平台与便道参数"},
            {"third_category_code": "TemporaryWaterAndElectricityArrangement", "third_category_name": "临时水电布置"}
        ],
        "RequirementsTech": [
            {"third_category_code": "DurationTarget", "third_category_name": "工期目标"},
            {"third_category_code": "QualityTarget", "third_category_name": "质量目标"},
            {"third_category_code": "SecurityGoals", "third_category_name": "安全目标"},
            {"third_category_code": "EnvironmentalGoals", "third_category_name": "环境目标"}
        ],
        "RiskLevel": [
            {"third_category_code": "DangerSource", "third_category_name": "危险源"},
            {"third_category_code": "ClassificationAndResponseMeasures", "third_category_name": "分级与应对措施"}
        ],
        "Stakeholders": [
            {"third_category_code": "UnitType", "third_category_name": "单位类型"}
        ],
        # 施工计划
        "Schedule": [
            {"third_category_code": "ProcessOperationTimeAnalysis", "third_category_name": "工序作业时间分析"},
            {"third_category_code": "KeyProjectNodeArrangement", "third_category_name": "关键工程（工序）节点安排"},
            {"third_category_code": "ConstructionScheduleGanttChart", "third_category_name": "施工进度计划横道图等"}
        ],
        "Materials": [
            {"third_category_code": "ListOfConstructionMeasuresAndMaterials", "third_category_name": "施工措施材料清单"}
        ],
        "Equipment": [
            {"third_category_code": "MainConstructionMachineryAndEquipment", "third_category_name": "主要施工机械设备"}
        ],
        "Workforce": [
            {"third_category_code": "WorkforceAllocationPlan", "third_category_name": "劳动力配置计划"},
            {"third_category_code": "StageLaborDemand", "third_category_name": "阶段劳动力需求"}
        ],
        "SafetyCost": [
            {"third_category_code": "CategoryOfSafetyProductionExpenses", "third_category_name": "安全生产费用类别"},
            {"third_category_code": "SecurityFeeName", "third_category_name": "安全费用名称"},
            {"third_category_code": "SingleInvestmentAmount", "third_category_name": "单项投入金额"},
            {"third_category_code": "TotalSafetyProductionExpenses", "third_category_name": "安全生产费用总额"}
        ],
        # 施工工艺技术
        "MethodsOverview": [
            {"third_category_code": "ConstructionTechnologySelection", "third_category_name": "施工工艺选择"},
            {"third_category_code": "MainConstructionMethods", "third_category_name": "主要施工方法"},
            {"third_category_code": "TemplateConfigurationQuantity", "third_category_name": "模板配置数量"}
        ],
        "TechParams": [
            {"third_category_code": "MaterialType", "third_category_name": "材料类型"},
            {"third_category_code": "MaterialSpecifications", "third_category_name": "材料规格"},
            {"third_category_code": "DeviceName", "third_category_name": "设备名称"},
            {"third_category_code": "DeviceModel", "third_category_name": "设备型号"},
            {"third_category_code": "EquipmentManufacturingTime", "third_category_name": "设备出厂时间"},
            {"third_category_code": "EquipmentPerformanceParameters", "third_category_name": "设备性能参数"},
            {"third_category_code": "EquipmentWeight", "third_category_name": "设备自重"}
        ],
        "Process": [
            {"third_category_code": "ConstructionProcess", "third_category_name": "施工工序"},
            {"third_category_code": "ProcessSequence", "third_category_name": "工艺顺序"},
            {"third_category_code": "ProcessFlowDiagram", "third_category_name": "工艺流程框图"}
        ],
        "PrepWork": [
            {"third_category_code": "MeasurementAndStakeout", "third_category_name": "测量放样"},
            {"third_category_code": "TemporaryWaterAndElectricityConsumption", "third_category_name": "临时水电用量"},
            {"third_category_code": "TheSiteIsFlat", "third_category_name": "场地平整"},
            {"third_category_code": "Staffing", "third_category_name": "人员配置"},
            {"third_category_code": "EquipmentEntry", "third_category_name": "设备进场"},
            {"third_category_code": "SafetyProtectionFacilities", "third_category_name": "安全防护措施"},
            {"third_category_code": "PersonnelAccess", "third_category_name": "人员上下通道"}
        ],
        "Operations": [
            {"third_category_code": "ConstructionProcessOperations", "third_category_name": "施工工序描述操作"},
            {"third_category_code": "ConstructionPoints", "third_category_name": "施工要点"},
            {"third_category_code": "FAQPrevention", "third_category_name": "常见问题及预防"},
            {"third_category_code": "ProblemSolvingMeasures", "third_category_name": "问题处理措施"}
        ],
        "Inspection": [
            {"third_category_code": "MaterialInspectionUponArrival", "third_category_name": "材料进场质量检验"},
            {"third_category_code": "RandomInspectionOfIncomingComponents", "third_category_name": "构配件进场质量抽查"},
            {"third_category_code": "ProcessInspectionContent", "third_category_name": "工序检查内容"},
            {"third_category_code": "ProcessInspectionStandards", "third_category_name": "工序检查标准"}
        ]
    }

    # 按二级分类分组
    secondary_groups = {}
    for chunk in chunks:
        sec_code = chunk.get("secondary_category_code", "")
        if sec_code and sec_code not in ("none", "None", ""):
            if sec_code not in secondary_groups:
                secondary_groups[sec_code] = []
            secondary_groups[sec_code].append(chunk)

    # 为每个二级分类添加三级分类详情
    updated_chunks = []
    for sec_code, group_chunks in secondary_groups.items():
        # 获取该二级分类对应的三级分类列表
        tertiary_list = tertiary_mapping.get(sec_code, [])

        for chunk in group_chunks:
            # 模拟三级分类结果
            chunk["tertiary_classification_details"] = tertiary_list

            # 设置第一个三级分类为主分类（向后兼容）
            if tertiary_list:
                chunk["tertiary_category_code"] = tertiary_list[0]["third_category_code"]
                chunk["tertiary_category_cn"] = tertiary_list[0]["third_category_name"]

            updated_chunks.append(chunk)

    return updated_chunks


async def test_completeness_check():
    """测试完整性审查"""

    print("=" * 60)
    print("完整性审查集成测试")
    print("=" * 60)

    # 1. 加载测试数据
    test_file = project_root / "temp/construction_review/final_result/4148f6019f89e061b15679666f646893-1773993108.json"

    if not test_file.exists():
        print(f"错误: 测试文件不存在: {test_file}")
        return

    print(f"\n1. 加载测试数据: {test_file.name}")
    data = load_final_result(str(test_file))

    chunks = data.get('document_result', {}).get('structured_content', {}).get('chunks', [])
    print(f"   原始 chunks 数量: {len(chunks)}")

    # 2. 检查原始数据
    print("\n2. 检查原始数据结构:")
    sample_chunk = chunks[0] if chunks else {}
    print(f"   chunk keys: {list(sample_chunk.keys())}")
    print(f"   有 tertiary_classification_details: {'tertiary_classification_details' in sample_chunk}")

    # 3. 模拟三级分类结果
    print("\n3. 模拟三级分类结果...")
    chunks_with_tertiary = simulate_tertiary_classification(chunks)

    # 统计三级分类情况
    tertiary_counts = {}
    for chunk in chunks_with_tertiary:
        sec_code = chunk.get("secondary_category_code", "")
        details = chunk.get("tertiary_classification_details", [])
        if sec_code:
            tertiary_counts[sec_code] = len(details)

    print(f"   已添加三级分类详情的 chunks: {len(chunks_with_tertiary)}")
    print(f"\n   各二级分类的三级分类数量:")
    for sec_code, count in sorted(tertiary_counts.items()):
        print(f"     {sec_code}: {count} 个三级分类")

    # 4. 运行完整性审查
    print("\n4. 运行完整性审查...")

    csv_path = str(project_root / "core/construction_review/component/doc_worker/config/StandardCategoryTable.csv")
    checker = LightweightCompletenessChecker(csv_path)

    result = await checker.check(
        chunks=chunks_with_tertiary,
        outline=None,
        chapter_classification="basis"  # 只测试编制依据章节
    )

    # 5. 输出结果
    print("\n5. 审查结果:")
    print(f"   总体状态: {result.overall_status}")

    # 检查三级完整性
    tertiary_result = result.tertiary_completeness
    print(f"\n   三级完整性:")
    print(f"     总数: {tertiary_result.get('total', 0)}")
    print(f"     已有: {tertiary_result.get('present', 0)}")
    print(f"     缺失: {tertiary_result.get('missing', 0)}")
    print(f"     完整率: {tertiary_result.get('completeness_rate', '0%')}")

    # 显示缺失详情
    missing_details = tertiary_result.get('missing_details', [])
    if missing_details:
        print(f"\n   缺失的三级分类 ({len(missing_details)} 个):")
        for item in missing_details[:10]:  # 只显示前10个
            print(f"     - {item.get('secondary_name', '')} > {item.get('tertiary_name', '')}")
    else:
        print(f"\n   [OK] 无缺失的三级分类!")

    # 6. 验证编制原则
    print("\n6. 验证 '编制原则' 二级分类:")

    # 查找编制原则相关的 chunks
    principle_chunks = [c for c in chunks_with_tertiary
                       if "CompilationPrinciples" in c.get("secondary_category_code", "")]

    if principle_chunks:
        chunk = principle_chunks[0]
        details = chunk.get("tertiary_classification_details", [])
        print(f"   找到编制原则 chunk")
        print(f"   三级分类详情数量: {len(details)}")
        print(f"   三级分类列表:")
        for d in details:
            print(f"     - {d.get('third_category_code')}: {d.get('third_category_name')}")

        # 检查完整性审查是否识别到这些三级分类
        actual_tertiary = set()
        for item in tertiary_result.get('secondary_stats', []):
            if item.get('secondary_code') == 'CompilationPrinciples':
                print(f"\n   完整性审查统计:")
                print(f"     总数: {item.get('total_tertiary', 0)}")
                print(f"     已有: {item.get('present', 0)}")
                print(f"     缺失: {item.get('missing', 0)}")
    else:
        print("   未找到编制原则相关的 chunks")

    print("\n" + "=" * 60)
    print("测试完成")
    print("=" * 60)


if __name__ == "__main__":
    asyncio.run(test_completeness_check())