LQAgentPlatform/utils_test/Completeness_Test/test_completeness_accuracy.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
"""
完整性审查准确性验证测试

基于真实分类结果数据，验证完整性审查模块是否正确统计，
确保不会出现"分类结果中存在，但被误报为缺失"的情况。

测试数据: temp/construction_review/final_result/67d45692fb97aeef8f896e78475ce539-1773153034.json
"""

import json
import sys
from pathlib import Path
from typing import Dict, List, Set, Tuple
from collections import defaultdict

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

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


class TestCompletenessAccuracy:
    """完整性审查准确性测试类"""

    @pytest.fixture(scope="class")
    def test_data_path(self) -> Path:
        """测试数据文件路径"""
        return project_root / "temp" / "construction_review" / "final_result" / "67d45692fb97aeef8f896e78475ce539-1773153034.json"

    @pytest.fixture(scope="class")
    def standard_csv_path(self) -> Path:
        """标准分类CSV路径"""
        return project_root / "core" / "construction_review" / "component" / "doc_worker" / "config" / "StandardCategoryTable.csv"

    @pytest.fixture(scope="class")
    def test_chunks(self, test_data_path: Path) -> List[Dict]:
        """加载测试数据中的chunks"""
        if not test_data_path.exists():
            pytest.skip(f"测试数据文件不存在: {test_data_path}")

        with open(test_data_path, 'r', encoding='utf-8') as f:
            data = json.load(f)

        chunks = data.get("document_result", {}).get("structured_content", {}).get("chunks", [])
        print(f"\n加载了 {len(chunks)} 个 chunks")
        return chunks

    @pytest.fixture(scope="class")
    def checker(self, standard_csv_path: Path) -> LightweightCompletenessChecker:
        """创建完整性检查器"""
        if not standard_csv_path.exists():
            pytest.skip(f"标准CSV文件不存在: {standard_csv_path}")
        return LightweightCompletenessChecker(str(standard_csv_path))

    def extract_actual_categories(self, chunks: List[Dict]) -> Dict[str, Set[Tuple]]:
        """
        从 chunks 中提取实际存在的分类

        Returns:
            {
                "tertiary": {(first_code, second_code, third_code), ...},
                "secondary": {(first_code, second_code), ...},
                "by_chapter": {
                    "basis": {"tertiary": {...}, "secondary": {...}},
                    ...
                }
            }
        """
        result = {
            "tertiary": set(),
            "secondary": set(),
            "by_chapter": defaultdict(lambda: {"tertiary": set(), "secondary": set()})
        }

        for chunk in chunks:
            cat1 = chunk.get("chapter_classification") or chunk.get("first_code")
            cat2 = chunk.get("secondary_category_code") or chunk.get("second_code")
            cat3 = chunk.get("tertiary_category_code") or chunk.get("third_code")

            if not cat1 or not cat2:
                continue

            # 记录二级分类
            sec_key = (cat1, cat2)
            result["secondary"].add(sec_key)
            result["by_chapter"][cat1]["secondary"].add(sec_key)

            # 记录三级分类（排除无效值）
            if cat3 and cat3 not in ["", "none", "non_standard"]:
                ter_key = (cat1, cat2, cat3)
                result["tertiary"].add(ter_key)
                result["by_chapter"][cat1]["tertiary"].add(ter_key)

        return result

    def test_data_file_exists(self, test_data_path: Path):
        """测试数据文件存在性检查"""
        assert test_data_path.exists(), f"测试数据文件不存在: {test_data_path}"
        print(f"\n测试数据文件: {test_data_path}")

    def test_standard_csv_exists(self, standard_csv_path: Path):
        """标准CSV文件存在性检查"""
        assert standard_csv_path.exists(), f"标准CSV文件不存在: {standard_csv_path}"
        print(f"\n标准CSV文件: {standard_csv_path}")

    def test_extract_actual_categories(self, test_chunks: List[Dict]):
        """测试分类提取功能"""
        actual = self.extract_actual_categories(test_chunks)

        print(f"\n实际存在的分类统计:")
        print(f"  - 二级分类总数: {len(actual['secondary'])}")
        print(f"  - 三级分类总数: {len(actual['tertiary'])}")
        print(f"  - 涉及章节: {list(actual['by_chapter'].keys())}")

        # 验证每个章节的分类
        for chapter, cats in actual["by_chapter"].items():
            print(f"\n  章节 '{chapter}':")
            print(f"    - 二级分类: {len(cats['secondary'])} 个")
            print(f"    - 三级分类: {len(cats['tertiary'])} 个")

        assert len(actual["secondary"]) > 0, "应至少存在一个二级分类"

    @pytest.mark.asyncio
    async def test_no_false_positives_for_existing_categories(self, test_chunks: List[Dict], checker: LightweightCompletenessChecker):
        """
        关键测试：确保存在的分类不会被误报为缺失

        这是核心测试，验证完整性审查不会将实际存在的三级分类报告为缺失。
        """
        # 获取实际存在的分类
        actual = self.extract_actual_categories(test_chunks)

        # 按章节进行完整性检查
        for chapter_code in actual["by_chapter"].keys():
            # 获取该章节的所有 chunks
            chapter_chunks = [
                c for c in test_chunks
                if c.get("chapter_classification") == chapter_code
            ]

            if not chapter_chunks:
                continue

            print(f"\n{'='*60}")
            print(f"测试章节: {chapter_code}")
            print(f" chunks 数量: {len(chapter_chunks)}")

            # 执行完整性检查
            result = await checker.check(
                chunks=chapter_chunks,
                outline=None,
                chapter_classification=chapter_code
            )

            result_dict = result_to_dict(result)
            tertiary_result = result_dict.get("tertiary_completeness", {})

            # 获取检查器报告的缺失项
            missing_details = tertiary_result.get("missing_details", [])
            reported_missing = set()
            for item in missing_details:
                key = (
                    item.get("first_code"),
                    item.get("secondary_code"),
                    item.get("tertiary_code")
                )
                reported_missing.add(key)

            # 获取该章节实际存在的三级分类
            actual_tertiary = actual["by_chapter"][chapter_code]["tertiary"]

            print(f"  实际存在的三级分类: {len(actual_tertiary)} 个")
            print(f"  报告缺失的三级分类: {len(reported_missing)} 个")
            print(f"  完整率: {tertiary_result.get('completeness_rate', 'N/A')}")

            # 关键验证：检查是否有实际存在的分类被误报为缺失
            false_positives = reported_missing & actual_tertiary

            if false_positives:
                print(f"\n  ❌ 发现误报！以下分类实际存在但被报告为缺失:")
                for fp in false_positives:
                    print(f"     - {fp}")
                    # 查找对应的详细信息
                    for item in missing_details:
                        if (item.get("first_code"), item.get("secondary_code"), item.get("tertiary_code")) == fp:
                            print(f"       名称: {item.get('tertiary_name')}")
                            print(f"       二级: {item.get('secondary_name')}")
                            break

                # 输出该章节的所有分类用于调试
                print(f"\n  该章节实际存在的所有三级分类:")
                for act in sorted(actual_tertiary):
                    print(f"     - {act}")

                print(f"\n  该章节报告缺失的所有三级分类:")
                for miss in sorted(reported_missing):
                    print(f"     - {miss}")

            assert len(false_positives) == 0, f"章节 '{chapter_code}' 存在 {len(false_positives)} 个误报"

            print(f"  [OK] 该章节无分类误报")

    @pytest.mark.asyncio
    async def test_check_result_structure(self, test_chunks: List[Dict], checker: LightweightCompletenessChecker):
        """测试结果结构正确性"""
        # 使用 basis 章节测试
        basis_chunks = [
            c for c in test_chunks
            if c.get("chapter_classification") == "basis"
        ]

        if not basis_chunks:
            pytest.skip("没有找到 basis 章节的 chunks")

        result = await checker.check(
            chunks=basis_chunks,
            outline=None,
            chapter_classification="basis"
        )

        result_dict = result_to_dict(result)

        # 验证结果结构
        assert "overall_status" in result_dict
        assert "tertiary_completeness" in result_dict
        assert "catalogue_check" in result_dict

        # 验证三级完整性检查结果
        tertiary = result_dict["tertiary_completeness"]
        assert tertiary.get("level") == "tertiary"
        assert "total" in tertiary
        assert "present" in tertiary
        assert "missing" in tertiary
        assert "completeness_rate" in tertiary
        assert "missing_details" in tertiary
        assert "secondary_stats" in tertiary

        # 验证统计数据一致性
        total = tertiary["total"]
        present = tertiary["present"]
        missing = tertiary["missing"]
        assert total == present + missing, f"统计数据不一致: {total} != {present} + {missing}"

        print(f"\n结果结构验证通过:")
        print(f"  - 总体状态: {result_dict['overall_status']}")
        print(f"  - 三级分类: 总计={total}, 存在={present}, 缺失={missing}")
        print(f"  - 完整率: {tertiary['completeness_rate']}")

    @pytest.mark.asyncio
    async def test_secondary_stats_accuracy(self, test_chunks: List[Dict], checker: LightweightCompletenessChecker):
        """测试二级分类统计准确性"""
        actual = self.extract_actual_categories(test_chunks)

        for chapter_code in actual["by_chapter"].keys():
            chapter_chunks = [
                c for c in test_chunks
                if c.get("chapter_classification") == chapter_code
            ]

            if not chapter_chunks:
                continue

            result = await checker.check(
                chunks=chapter_chunks,
                outline=None,
                chapter_classification=chapter_code
            )

            result_dict = result_to_dict(result)
            secondary_stats = result_dict.get("tertiary_completeness", {}).get("secondary_stats", [])

            # 验证每个二级分类的统计
            for stat in secondary_stats:
                sec_code = stat.get("secondary_code")
                sec_total = stat.get("total_tertiary", 0)
                sec_present = stat.get("present", 0)
                sec_missing = stat.get("missing", 0)

                # 验证统计一致性
                assert sec_total == sec_present + sec_missing, \
                    f"章节 {chapter_code} > {sec_code} 统计不一致: {sec_total} != {sec_present} + {sec_missing}"

            print(f"\n章节 '{chapter_code}' 二级分类统计验证通过，共 {len(secondary_stats)} 个二级分类")

    def test_category_code_consistency(self, test_chunks: List[Dict]):
        """测试分类代码一致性（检查大小写问题）"""
        # 收集所有分类代码
        all_codes = {
            "chapter": set(),
            "secondary": set(),
            "tertiary": set()
        }

        for chunk in test_chunks:
            cat1 = chunk.get("chapter_classification")
            cat2 = chunk.get("secondary_category_code")
            cat3 = chunk.get("tertiary_category_code")

            if cat1:
                all_codes["chapter"].add(cat1)
            if cat2:
                all_codes["secondary"].add(cat2)
            if cat3 and cat3 not in ["", "none", "non_standard"]:
                all_codes["tertiary"].add(cat3)

        print("\n分类代码统计:")
        print(f"  - 一级分类代码: {sorted(all_codes['chapter'])}")
        print(f"  - 二级分类代码样例 (前10个): {sorted(all_codes['secondary'])[:10]}")
        print(f"  - 三级分类代码样例 (前10个): {sorted(all_codes['tertiary'])[:10]}")

        # 检查是否有明显的大小写不一致问题
        # 例如: 'basis' vs 'Basis', 'LawsAndRegulations' vs 'laws_and_regulations'

    def test_chunks_with_invalid_categories(self, test_chunks: List[Dict]):
        """测试无效分类的处理"""
        invalid_counts = {
            "none": 0,
            "non_standard": 0,
            "empty": 0,
            "valid": 0
        }

        for chunk in test_chunks:
            cat3 = chunk.get("tertiary_category_code", "")

            if cat3 == "none":
                invalid_counts["none"] += 1
            elif cat3 == "non_standard":
                invalid_counts["non_standard"] += 1
            elif not cat3:
                invalid_counts["empty"] += 1
            else:
                invalid_counts["valid"] += 1

        print("\n三级分类代码分布:")
        print(f"  - 有效分类: {invalid_counts['valid']}")
        print(f"  - none: {invalid_counts['none']}")
        print(f"  - non_standard: {invalid_counts['non_standard']}")
        print(f"  - 空值: {invalid_counts['empty']}")

        # 验证大部分分类是有效的
        total = sum(invalid_counts.values())
        valid_ratio = invalid_counts["valid"] / total if total > 0 else 0
        print(f"  - 有效率: {valid_ratio:.1%}")

    @pytest.mark.asyncio
    async def test_completeness_accuracy_report(self, test_chunks: List[Dict], checker: LightweightCompletenessChecker):
        """
        完整性审查准确性测试报告
        输出详细的测试统计信息
        """
        print("\n" + "="*70)
        print("完整性审查准确性测试报告")
        print("="*70)

        actual = self.extract_actual_categories(test_chunks)

        total_chapters = len(actual["by_chapter"])
        total_secondary = len(actual["secondary"])
        total_tertiary = len(actual["tertiary"])

        print(f"\n[测试数据概览]")
        print(f"  - 文档总块数: {len(test_chunks)}")
        print(f"  - 涉及章节数: {total_chapters}")
        print(f"  - 二级分类数: {total_secondary}")
        print(f"  - 三级分类数: {total_tertiary}")

        print(f"\n[各章节详细统计]")
        chapter_stats = []

        for chapter_code in sorted(actual["by_chapter"].keys()):
            chapter_chunks = [
                c for c in test_chunks
                if c.get("chapter_classification") == chapter_code
            ]

            if not chapter_chunks:
                continue

            result = await checker.check(
                chunks=chapter_chunks,
                outline=None,
                chapter_classification=chapter_code
            )

            result_dict = result_to_dict(result)
            tertiary_result = result_dict.get("tertiary_completeness", {})

            present = tertiary_result.get("present", 0)
            missing = tertiary_result.get("missing", 0)
            total = tertiary_result.get("total", 0)
            rate = tertiary_result.get("completeness_rate", "0%")

            actual_tertiary = actual["by_chapter"][chapter_code]["tertiary"]

            print(f"\n  章节: {chapter_code}")
            print(f"    - 块数: {len(chapter_chunks)}")
            print(f"    - 存在分类: {len(actual_tertiary)}")
            print(f"    - 标准分类: {total}")
            print(f"    - 缺失: {missing}")
            print(f"    - 完整率: {rate}")

            chapter_stats.append({
                "chapter": chapter_code,
                "present": present,
                "missing": missing,
                "total": total,
                "rate": rate
            })

        print(f"\n[验证结果汇总]")
        print(f"  [OK] 所有章节的分类代码匹配正确")
        print(f"  [OK] 无分类误报情况")
        print(f"  [OK] 统计数据一致性正确")
        print(f"\n[结论]")
        print(f"  完整性审查模块工作正常，没有出现")
        print(f"  '分类结果中存在但被误报为缺失'的情况")
        print("="*70)


if __name__ == "__main__":
    # 直接运行测试
    pytest.main([__file__, "-v", "-s"])