LQAgentPlatform/utils_test/Grammar_Check_Test/analyze_grammar_quality.py

"""
分析最新审查结果中词句语法审查的质量

检查项：
1. "将A改为A" 模式（修正前后相同）
2. suggestion/reason 中的自我辩论（犹豫措辞）
3. risk_level 为空
4. 技术操作规程越界审查
5. 重复问题
6. JSON 解析失败
7. suggestion 过长（>200字，可能包含推理过程）
"""

import json
import re
import sys
import os

project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, project_root)


def extract_correction_pairs(suggestion: str):
    """从 suggestion 中提取所有 '将X改为Y' 的 (X, Y) 对"""
    quote_chars = r"""['""''「」]"""
    pattern = rf"将{quote_chars}(.*?){quote_chars}\s*改为\s*{quote_chars}(.*?){quote_chars}"
    return re.findall(pattern, suggestion)


def check_hesitation_words(text: str):
    """检查文本中是否包含犹豫措辞"""
    hesitation_words = [
        '可能', '暂定', '不确定', '重新审视', '然而', '不过', '似乎',
        '但是', '其实', '实际上', '再细看', '再想想', '仔细想想',
        '反过来', '另一方面', '换个角度'
    ]
    found = [w for w in hesitation_words if w in text]
    return found


def is_technical_procedure(issue_point: str, reason: str):
    """检查是否为技术操作规程越界审查"""
    technical_keywords = [
        '操作步骤', '工艺参数', '施工顺序', '操作规程',
        '技术规范', '施工方案', '工艺流程'
    ]
    combined = issue_point + reason
    return [kw for kw in technical_keywords if kw in combined]


def analyze_grammar_check_results(result_file: str):
    """分析词句语法审查结果质量"""
    with open(result_file, encoding='utf-8') as f:
        data = json.load(f)

    issues = data.get('issues', [])
    grammar_items = []

    for issue_wrapper in issues:
        for issue_id, issue_detail in issue_wrapper.items():
            review_lists = issue_detail.get('review_lists', [])
            metadata = issue_detail.get('metadata', {})
            for item in review_lists:
                check_item = item.get('check_item', '')
                if check_item in ['sensitive_word_check', 'grammar_check']:
                    grammar_items.append({
                        'item': item,
                        'issue_id': issue_id,
                        'location_label': metadata.get('review_location_label', '')
                    })

    print(f"Total grammar_check items: {len(grammar_items)}")
    print()

    # Quality checks
    a_to_a_issues = []
    hesitation_issues = []
    empty_risk_issues = []
    technical_issues = []
    duplicate_issues = []
    parse_failures = []
    long_suggestion_issues = []

    seen_corrections = {}

    for i, entry in enumerate(grammar_items):
        item = entry['item']
        check_result = item.get('check_result', {})

        # STRING format = parse failure
        if isinstance(check_result, str):
            parse_failures.append({
                'index': i + 1,
                'raw': check_result[:200],
                'location_label': entry['location_label']
            })
            continue

        issue_point = check_result.get('issue_point', '')
        location = check_result.get('location', '')
        suggestion = check_result.get('suggestion', '')
        reason = check_result.get('reason', '')
        risk_level = check_result.get('risk_level', '')

        # Check 1: Empty risk_level
        if not risk_level or risk_level.strip() == '':
            empty_risk_issues.append({
                'index': i + 1,
                'issue_point': issue_point,
                'suggestion': suggestion[:80]
            })

        # Check 2: A→A pattern
        pairs = extract_correction_pairs(suggestion)
        for before, after in pairs:
            if before.strip() == after.strip():
                a_to_a_issues.append({
                    'index': i + 1,
                    'issue_point': issue_point,
                    'before': before,
                    'after': after
                })

        # Check 3: Hesitation words in suggestion
        sug_hesitation = check_hesitation_words(suggestion)
        if sug_hesitation:
            hesitation_issues.append({
                'index': i + 1,
                'field': 'suggestion',
                'words': sug_hesitation,
                'text': suggestion[:100]
            })

        # Check 4: Hesitation words in reason
        reason_hesitation = check_hesitation_words(reason)
        if reason_hesitation:
            hesitation_issues.append({
                'index': i + 1,
                'field': 'reason',
                'words': reason_hesitation,
                'text': reason[:100]
            })

        # Check 5: Technical procedure
        tech_kws = is_technical_procedure(issue_point, reason)
        if tech_kws:
            technical_issues.append({
                'index': i + 1,
                'issue_point': issue_point,
                'keywords': tech_kws
            })

        # Check 6: Long suggestion (>200 chars)
        if len(suggestion) > 200:
            long_suggestion_issues.append({
                'index': i + 1,
                'issue_point': issue_point,
                'length': len(suggestion),
                'text': suggestion[:100]
            })

        # Check 7: Duplicates (same correction key)
        if pairs:
            sorted_pairs = sorted(pairs)
            correction_key = ",".join(f"{a}→{b}" for a, b in sorted_pairs)
        else:
            correction_key = suggestion.strip()

        if correction_key in seen_corrections:
            duplicate_issues.append({
                'index': i + 1,
                'first_index': seen_corrections[correction_key],
                'correction_key': correction_key,
                'issue_point': issue_point
            })
        else:
            seen_corrections[correction_key] = i + 1

    # Print results
    print("=" * 60)
    print("QUALITY ANALYSIS RESULTS")
    print("=" * 60)

    sections = [
        ("A->A Pattern (will A change to A)", a_to_a_issues),
        ("Self-debate / Hesitation words", hesitation_issues),
        ("Empty risk_level", empty_risk_issues),
        ("Technical procedure (out of scope)", technical_issues),
        ("Duplicate corrections", duplicate_issues),
        ("JSON parse failures", parse_failures),
        ("Long suggestions (>200 chars)", long_suggestion_issues),
    ]

    total_problems = 0
    for title, items in sections:
        count = len(items)
        total_problems += count
        status = "PASS" if count == 0 else "FAIL"
        print(f"\n[{status}] {title}: {count}")
        if items:
            for item in items:
                print(f"  - #{item.get('index', '?')}: {json.dumps(item, ensure_ascii=False)[:150]}")

    print(f"\n{'=' * 60}")
    print(f"TOTAL: {len(grammar_items)} items, {total_problems} quality issues")
    print(f"Quality rate: {(len(grammar_items) - total_problems) / len(grammar_items) * 100:.1f}%")

    # Print valid items summary
    print(f"\n{'=' * 60}")
    print("VALID ITEMS SUMMARY")
    print("=" * 60)
    for i, entry in enumerate(grammar_items):
        item = entry['item']
        check_result = item.get('check_result', {})
        if isinstance(check_result, str):
            print(f"  [{i+1}] [PARSE_FAIL] {check_result[:60]}...")
            continue
        issue_point = check_result.get('issue_point', '')
        suggestion = check_result.get('suggestion', '')
        risk_level = check_result.get('risk_level', '')
        print(f"  [{i+1}] [{risk_level}] {issue_point}: {suggestion[:60]}...")


if __name__ == "__main__":
    result_file = os.path.join(
        project_root,
        "temp", "construction_review", "final_result",
        "67d45692fb97aeef8f896e78475ce539-1779785718.json"
    )
    analyze_grammar_check_results(result_file)