gaokao-volunteer-system/docs/plans/T1-4-to-1-10-audit-completion.md

T1.4-T1.10 实施计划汇总

AI审核服务剩余任务详细计划

For Hermes: Use subagent-driven-development skill to implement this plan task-by-task.

---

Task 1.4.1: 集成规范检查器

Objective: 实现 checker_integration.py，复用现有规范检查器

Files:

• Create: skills/gaokao-audit/scripts/checker_integration.py
• Create: skills/gaokao-audit/tests/test_checker_integration.py

Step 1: 写测试

"""规范检查集成测试"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

import pytest
from skills.gaokao-audit.scripts.checker_integration import CheckerIntegration


def test_check_hunan_plan():
    """测试检查湖南方案"""
    checker = CheckerIntegration()
    plan_text = "湖南 578分 45个学校 院校专业组"
    result = checker.check(plan_text, province="湖南")

    assert "errors" in result
    assert "summary" in result
    # 应该检测出"45个学校"这个错误（致命）


def test_check_zhejiang_plan():
    """测试检查浙江方案"""
    checker = CheckerIntegration()
    plan_text = "浙江 620分 80个院校专业组"
    result = checker.check(plan_text, province="浙江")

    # 浙江是"专业+学校"模式，不是院校专业组
    # 应该检查模式是否正确
    assert "errors" in result


def test_check_unknown_province():
    """测试不存在的省份"""
    checker = CheckerIntegration()
    result = checker.check("test", province="不存在的省")

    # 应该优雅处理
    assert "errors" in result


def test_format_check_results():
    """测试结果格式化"""
    checker = CheckerIntegration()
    result = checker.check("test", province="湖南")
    formatted = checker.format_results(result)

    assert isinstance(formatted, dict)
    assert "policy_errors" in formatted
    assert "fatal_count" in formatted

Step 2: 实现 checker_integration.py

"""
规范检查集成

集成 gaokao-spec-checker 复用27省规则库。
"""
import sys
import os
from typing import Dict, Any

# 添加路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

from skills.gaokao-spec-checker.scripts.spec_checker_v2 import (
    GaokaoSpecCheckerV2,
    PROVINCE_RULES,
    detect_province,
)


class CheckerIntegration:
    """规范检查集成器"""

    def __init__(self):
        self.checker = GaokaoSpecCheckerV2()

    def check(self, plan_text: str, province: str = None) -> Dict[str, Any]:
        """执行规范检查

        Args:
            plan_text: 方案文本
            province: 省份（可选，自动检测）

        Returns:
            检查结果字典
        """
        # 自动检测省份
        if not province:
            province = detect_province(plan_text)

        # 执行检查
        if province:
            report = self.checker.auto_detect_and_check(plan_text)
        else:
            report = "未识别省份"

        # 解析报告
        errors = self._parse_report(report)

        return {
            "province": province,
            "errors": errors,
            "summary": self._summarize(errors),
            "raw_report": report,
        }

    def _parse_report(self, report: str) -> Dict[str, list]:
        """解析报告为结构化数据"""
        result = {"fatal": [], "warning": [], "info": []}

        current_section = None
        for line in report.split("\n"):
            line = line.strip()

            if "致命错误" in line:
                current_section = "fatal"
            elif "严重错误" in line or "严重警告" in line:
                current_section = "warning"
            elif "一般警告" in line or "一般提示" in line:
                current_section = "info"
            elif line and current_section and (line[0].isdigit() or line.startswith("•")):
                # 简化的错误提取
                result[current_section].append({
                    "description": line,
                })

        return result

    def _summarize(self, errors: Dict[str, list]) -> Dict[str, int]:
        """生成摘要"""
        return {
            "fatal_count": len(errors.get("fatal", [])),
            "warning_count": len(errors.get("warning", [])),
            "info_count": len(errors.get("info", [])),
        }

    def format_results(self, result: Dict[str, Any]) -> Dict[str, Any]:
        """格式化结果"""
        return {
            "policy_errors": result["errors"].get("fatal", []),
            "warnings": result["errors"].get("warning", []),
            "info": result["errors"].get("info", []),
            "fatal_count": result["summary"]["fatal_count"],
            "warning_count": result["summary"]["warning_count"],
            "info_count": result["summary"]["info_count"],
        }

Step 3: 提交

git add skills/gaokao-audit/scripts/checker_integration.py
git add skills/gaokao-audit/tests/test_checker_integration.py
git commit -m "feat(audit): 集成规范检查器 - T1.4.1"

---

Task 1.5.1: 实现扎堆检测器

Objective: 实现 crowd_detector.py，检测方案中院校的扎堆风险

Files:

• Create: skills/gaokao-audit/scripts/crowd_detector.py
• Create: skills/gaokao-audit/tests/test_crowd_detector.py

Step 1: 写测试

"""扎堆检测器测试"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

import pytest
from skills.gaokao-audit.scripts.crowd_detector import CrowdDetector, CrowdRisk


@pytest.fixture
def detector():
    return CrowdDetector()


def test_detect_high_risk_school(detector):
    """测试检测高风险院校"""
    plan = [
        {"school": "长沙理工大学", "major": "会计学"},
    ]
    risks = detector.detect_risks(plan, province="湖南", score=575)

    assert len(risks) >= 1
    first = risks[0]
    assert "长沙理工" in first.school
    assert first.risk_level == "high"
    assert first.predicted_increase == 18


def test_detect_no_risk(detector):
    """测试没有扎堆风险"""
    plan = [
        {"school": "某某不知名学校", "major": "某专业"},
    ]
    risks = detector.detect_risks(plan, province="湖南", score=575)

    assert len(risks) == 0


def test_detect_multiple_risks(detector):
    """测试多个风险"""
    plan = [
        {"school": "长沙理工大学", "major": "会计学"},
        {"school": "江西财经大学", "major": "会计学"},
        {"school": "湖南工商大学", "major": "会计学"},
    ]
    risks = detector.detect_risks(plan, province="湖南", score=575)

    # 前两个有风险，第三个没有
    assert len(risks) == 2


def test_get_risk_label(detector):
    """测试风险等级标签"""
    assert detector.get_risk_label(4) == "🔴 高风险"
    assert detector.get_risk_label(3) == "🟡 中风险"
    assert detector.get_risk_label(1) == "🟢 低风险"


def test_alternatives_included(detector):
    """测试包含替代方案"""
    plan = [{"school": "长沙理工大学", "major": "会计学"}]
    risks = detector.detect_risks(plan, province="湖南", score=575)

    assert len(risks) >= 1
    assert len(risks[0].alternatives) > 0
    assert all("name" in a for a in risks[0].alternatives)


def test_format_risks_for_report(detector):
    """测试报告格式化"""
    plan = [{"school": "长沙理工大学", "major": "会计学"}]
    risks = detector.detect_risks(plan, province="湖南", score=575)

    formatted = detector.format_for_report(risks)
    assert isinstance(formatted, list)
    if formatted:
        assert "name" in formatted[0]
        assert "risk_level_label" in formatted[0]
        assert "predicted_increase" in formatted[0]

Step 2: 实现 crowd_detector.py

"""
扎堆检测器

检测方案中的院校是否被大厂AI高频推荐，存在扎堆风险。
"""
import sys
import os
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

from data.crowd_db.loader import CrowdDBLoader


@dataclass
class CrowdRisk:
    """扎堆风险"""
    school: str
    major: str
    frequency: int
    platforms: List[str]
    predicted_increase: int
    risk_level: str  # high/medium/low
    alternatives: List[Dict[str, Any]] = field(default_factory=list)

    @property
    def risk_level_label(self) -> str:
        """中文标签"""
        labels = {
            "high": "🔴 高风险",
            "medium": "🟡 中风险",
            "low": "🟢 低风险",
        }
        return labels.get(self.risk_level, self.risk_level)


class CrowdDetector:
    """扎堆检测器"""

    def __init__(self, loader: Optional[CrowdDBLoader] = None):
        self.loader = loader or CrowdDBLoader()

    def detect_risks(
        self,
        volunteers: List[Dict[str, str]],
        province: str,
        score: int
    ) -> List[CrowdRisk]:
        """检测扎堆风险

        Args:
            volunteers: 志愿列表 [{"school": "xxx", "major": "yyy"}]
            province: 省份
            score: 用户分数

        Returns:
            CrowdRisk列表
        """
        risks = []

        for vol in volunteers:
            school = vol.get("school", "")
            major = vol.get("major", "")

            # 在大厂AI推荐库中查找
            rec = self.loader.find_recommendation_by_school(province, school)
            if not rec:
                continue

            # 检查分数段匹配
            score_match = self._is_in_score_range(province, score, school)
            if not score_match:
                continue

            # 创建风险对象
            risk = CrowdRisk(
                school=rec["name"],
                major=major or rec.get("major", ""),
                frequency=rec["frequency"],
                platforms=rec.get("platforms", []),
                predicted_increase=rec["predicted_increase"],
                risk_level=self._get_risk_level(rec["frequency"]),
                alternatives=rec.get("alternatives", []),
            )
            risks.append(risk)

        return risks

    def _is_in_score_range(self, province: str, score: int, school: str) -> bool:
        """检查学校是否在用户的分数段推荐中"""
        recs = self.loader.find_recommendations(province, score)
        return any(r["name"] in school or school in r["name"] for r in recs)

    def _get_risk_level(self, frequency: int) -> str:
        """根据频次计算风险等级"""
        if frequency >= 4:
            return "high"
        elif frequency >= 2:
            return "medium"
        else:
            return "low"

    def get_risk_label(self, frequency: int) -> str:
        """获取风险等级标签（供测试）"""
        level = self._get_risk_level(frequency)
        labels = {
            "high": "🔴 高风险",
            "medium": "🟡 中风险",
            "low": "🟢 低风险",
        }
        return labels.get(level, level)

    def format_for_report(self, risks: List[CrowdRisk]) -> List[Dict[str, Any]]:
        """格式化为报告格式"""
        return [
            {
                "name": r.school,
                "major": r.major,
                "frequency": r.frequency,
                "predicted_increase": r.predicted_increase,
                "risk_level": r.risk_level,
                "risk_level_label": r.risk_level_label,
                "platforms": r.platforms,
                "alternatives": r.alternatives,
            }
            for r in risks
        ]

Step 3: 运行测试

python3 -m pytest skills/gaokao-audit/tests/test_crowd_detector.py -v

Step 4: 提交

git add skills/gaokao-audit/scripts/crowd_detector.py
git add skills/gaokao-audit/tests/test_crowd_detector.py
git commit -m "feat(audit): 实现扎堆检测器 - T1.5.1"

---

Task 1.6.1: 实现审核服务主类

Objective: 实现 audit_service.py 主入口

Files:

• Create: skills/gaokao-audit/scripts/audit_service.py
• Create: skills/gaokao-audit/tests/test_audit_service.py

Step 1: 写测试

"""审核服务测试"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

import pytest
from skills.gaokao-audit.scripts.audit_service import AuditService, AuditResult


@pytest.fixture
def service():
    return AuditService()


@pytest.fixture
def sample_plan_text():
    return """
百度AI志愿助手为您推荐

考生信息
省份：湖南
高考分数：578
位次：约26800
选科：物理+化学+生物

推荐院校
1. 长沙理工大学 - 会计学
2. 湖南师范大学 - 会计学
3. 江西财经大学 - 会计学
4. 湘潭大学 - 工商管理
5. 湖南工商大学 - 财务管理
"""


def test_audit_plan_basic(service, sample_plan_text):
    """测试审核基本功能"""
    result = service.audit(sample_plan_text, format="text")

    assert isinstance(result, AuditResult)
    assert result.province == "湖南"
    assert result.candidate_score == 578
    assert len(result.volunteers) >= 3


def test_audit_detects_crowd_risk(service, sample_plan_text):
    """测试审核能检测扎堆风险"""
    result = service.audit(sample_plan_text, format="text")

    # 长沙理工大学会计学应该是高风险
    assert len(result.crowd_risks) >= 1
    high_risks = [r for r in result.crowd_risks if r.risk_level == "high"]
    assert len(high_risks) >= 1


def test_audit_calculates_score(service, sample_plan_text):
    """测试审核计算综合评分"""
    result = service.audit(sample_plan_text, format="text")

    assert 0 <= result.overall_score <= 100
    # 有扎堆风险的方案分数应该不高
    assert result.overall_score < 90


def test_audit_generates_suggestions(service, sample_plan_text):
    """测试审核生成建议"""
    result = service.audit(sample_plan_text, format="text")

    assert len(result.suggestions) > 0


def test_audit_to_dict(service, sample_plan_text):
    """测试审核结果序列化"""
    result = service.audit(sample_plan_text, format="text")
    d = result.to_dict()

    assert isinstance(d, dict)
    assert "province" in d
    assert "overall_score" in d
    assert "crowd_risks" in d
    assert "policy_errors" in d
    assert "suggestions" in d


def test_audit_data_trace(service, sample_plan_text):
    """测试数据溯源检查"""
    result = service.audit(sample_plan_text, format="text")

    # 数据溯源是警告级别
    assert hasattr(result, 'data_issues')

Step 2: 实现 audit_service.py

"""
审核服务主类

组合各模块，提供端到端的审核服务。
"""
import sys
import os
from dataclasses import dataclass, field, asdict
from typing import List, Dict, Any

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

from skills.gaokao-audit.scripts.plan_parser import PlanParser, ParsedPlan
from skills.gaokao-audit.scripts.checker_integration import CheckerIntegration
from skills.gaokao-audit.scripts.crowd_detector import CrowdDetector, CrowdRisk


@dataclass
class AuditResult:
    """审核结果"""
    # 考生信息
    province: str = None
    candidate_score: int = None
    candidate_rank: int = None
    subjects: str = None
    source: str = None
    volunteers: List[Dict[str, Any]] = field(default_factory=list)

    # 检查结果
    policy_errors: List[Dict[str, Any]] = field(default_factory=list)
    crowd_risks: List[CrowdRisk] = field(default_factory=list)
    data_issues: List[str] = field(default_factory=list)
    suggestions: List[str] = field(default_factory=list)

    # 综合评分
    overall_score: int = 100

    def to_dict(self) -> Dict[str, Any]:
        d = asdict(self)
        # 转换CrowdRisk为dict
        d["crowd_risks"] = [
            {
                "school": r.school,
                "major": r.major,
                "frequency": r.frequency,
                "predicted_increase": r.predicted_increase,
                "risk_level": r.risk_level,
                "risk_level_label": r.risk_level_label,
                "platforms": r.platforms,
                "alternatives": r.alternatives,
            }
            for r in self.crowd_risks
        ]
        return d


class AuditService:
    """审核服务"""

    def __init__(self):
        self.parser = PlanParser()
        self.checker = CheckerIntegration()
        self.detector = CrowdDetector()

    def audit(self, plan_text: str, format: str = "text") -> AuditResult:
        """执行审核

        Args:
            plan_text: 方案文本
            format: 格式 'text' | 'pdf_text' | 'screenshot_ocr'

        Returns:
            AuditResult对象
        """
        # 1. 解析方案
        parsed = self.parser.parse_text(plan_text)

        # 2. 政策检查
        check_result = self.checker.check(plan_text, province=parsed.province)
        policy_errors = check_result["errors"]["fatal"]
        warnings = check_result["errors"]["warning"]

        # 3. 扎堆检测
        crowd_risks = []
        if parsed.province and parsed.score:
            crowd_risks = self.detector.detect_risks(
                parsed.volunteers,
                province=parsed.province,
                score=parsed.score,
            )

        # 4. 数据溯源检查
        data_issues = self._check_data_trace(parsed)

        # 5. 生成建议
        suggestions = self._generate_suggestions(
            policy_errors, crowd_risks, data_issues
        )

        # 6. 计算综合评分
        overall_score = self._calculate_score(
            policy_errors, crowd_risks, data_issues
        )

        return AuditResult(
            province=parsed.province,
            candidate_score=parsed.score,
            candidate_rank=parsed.rank,
            subjects=parsed.subjects,
            source=parsed.source,
            volunteers=parsed.volunteers,
            policy_errors=policy_errors,
            crowd_risks=crowd_risks,
            data_issues=data_issues,
            suggestions=suggestions,
            overall_score=overall_score,
        )

    def _check_data_trace(self, parsed: ParsedPlan) -> List[str]:
        """检查数据溯源"""
        issues = []

        # 检查是否标注数据来源
        if not parsed.source:
            issues.append("未明确标注AI来源（千问/元宝/百度/豆包）")

        # 检查是否有分数标注年份
        if parsed.score and "2025" not in parsed.raw_text and "2024" not in parsed.raw_text:
            issues.append("未明确数据年份（建议标注2025年参考位次）")

        return issues

    def _generate_suggestions(
        self,
        policy_errors: List[Dict],
        crowd_risks: List[CrowdRisk],
        data_issues: List[str],
    ) -> List[str]:
        """生成建议"""
        suggestions = []

        if crowd_risks:
            high_risks = [r for r in crowd_risks if r.risk_level == "high"]
            if high_risks:
                suggestions.append(
                    f"检测到 {len(high_risks)} 所高风险扎堆院校，"
                    f"建议替换为低风险替代方案"
                )

        if policy_errors:
            suggestions.append(
                f"存在 {len(policy_errors)} 个政策错误，"
                f"必须修正后才能使用该方案"
            )

        if data_issues:
            suggestions.append(
                "建议核实数据来源，确保使用的位次/分数数据真实可靠"
            )

        if not suggestions:
            suggestions.append("方案整体合理，建议结合自身情况微调")

        return suggestions

    def _calculate_score(
        self,
        policy_errors: List[Dict],
        crowd_risks: List[CrowdRisk],
        data_issues: List[str],
    ) -> int:
        """计算综合评分"""
        score = 100

        # 政策错误扣分
        score -= len(policy_errors) * 15

        # 扎堆风险扣分
        for risk in crowd_risks:
            if risk.risk_level == "high":
                score -= 10
            elif risk.risk_level == "medium":
                score -= 5

        # 数据问题扣分
        score -= len(data_issues) * 3

        return max(0, min(100, score))

Step 3: 运行测试

python3 -m pytest skills/gaokao-audit/tests/test_audit_service.py -v

Step 4: 提交

git add skills/gaokao-audit/scripts/audit_service.py
git add skills/gaokao-audit/tests/test_audit_service.py
git commit -m "feat(audit): 实现审核服务主类 - T1.6.1"

---

Task 1.7.1: 实现报告生成器

Objective: 实现 report_generator.py，生成PDF报告

Files:

• Create: skills/gaokao-audit/scripts/report_generator.py
• Create: skills/gaokao-audit/tests/test_report_generator.py

Step 1: 写测试

"""报告生成器测试"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

import pytest
from skills.gaokao-audit.scripts.audit_service import AuditResult
from skills.gaokao-audit.scripts.crowd_detector import CrowdRisk
from skills.gaokao-audit.scripts.report_generator import ReportGenerator


@pytest.fixture
def generator():
    return ReportGenerator()


@pytest.fixture
def sample_result():
    return AuditResult(
        province="湖南",
        candidate_score=578,
        candidate_rank=26800,
        subjects="物理+化学+生物",
        source="百度",
        volunteers=[
            {"index": 1, "school": "长沙理工大学", "major": "会计学"},
        ],
        policy_errors=[],
        crowd_risks=[
            CrowdRisk(
                school="长沙理工大学",
                major="会计学",
                frequency=4,
                platforms=["千问", "元宝", "百度", "豆包"],
                predicted_increase=18,
                risk_level="high",
                alternatives=[
                    {"name": "湖南工商大学", "major": "会计学"},
                ],
            ),
        ],
        data_issues=["未明确数据年份"],
        suggestions=["检测到1所高风险扎堆院校"],
        overall_score=75,
    )


def test_generate_html(generator, sample_result):
    """测试生成HTML"""
    html = generator.render_html(sample_result)

    assert isinstance(html, str)
    assert "湖南" in html
    assert "长沙理工大学" in html
    assert "578" in html


def test_generate_pdf(tmp_path, generator, sample_result):
    """测试生成PDF"""
    output_path = tmp_path / "audit_report.pdf"

    pdf_path = generator.generate_pdf(
        sample_result,
        str(output_path),
    )

    assert os.path.exists(pdf_path)
    assert os.path.getsize(pdf_path) > 1000  # 至少1KB

Step 2: 实现 report_generator.py

"""
审核报告生成器

将审核结果格式化为HTML或PDF。
"""
import os
import sys
from datetime import datetime
from typing import Optional

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

from skills.gaokao-audit.scripts.audit_service import AuditResult


class ReportGenerator:
    """报告生成器"""

    TEMPLATE_PATH = os.path.join(
        os.path.dirname(__file__), '..', 'templates', 'audit_report.html'
    )

    def render_html(self, result: AuditResult) -> str:
        """渲染HTML报告"""
        # 简单模板渲染（生产可使用jinja2）
        with open(self.TEMPLATE_PATH, "r", encoding="utf-8") as f:
            template = f.read()

        # 准备数据
        data = {
            "source": result.source or "未指定",
            "candidate_info": f"{result.province} {result.candidate_score}分 {result.subjects or ''}",
            "audit_time": datetime.now().strftime("%Y-%m-%d %H:%M"),
            "overall_score": result.overall_score,
            "fatal_count": len(result.policy_errors),
            "warning_count": len(result.crowd_risks),
            "info_count": len(result.data_issues),
            "policy_errors": result.policy_errors,
            "crowd_risks": [
                {
                    "name": r.school,
                    "major": r.major,
                    "frequency": r.frequency,
                    "predicted_increase": r.predicted_increase,
                    "risk_level": r.risk_level,
                    "risk_level_label": r.risk_level_label,
                    "alternatives": r.alternatives,
                }
                for r in result.crowd_risks
            ],
            "data_issues": result.data_issues,
            "suggestions": result.suggestions,
        }

        # 简单模板替换（生产用jinja2）
        html = template
        for key, value in data.items():
            if isinstance(value, str):
                html = html.replace("{{ " + key + " }}", value)
            elif isinstance(value, int):
                html = html.replace("{{ " + key + " }}", str(value))

        return html

    def generate_pdf(
        self,
        result: AuditResult,
        output_path: str,
    ) -> str:
        """生成PDF报告

        Args:
            result: 审核结果
            output_path: 输出文件路径

        Returns:
            实际写入的PDF路径
        """
        try:
            from weasyprint import HTML

            html_content = self.render_html(result)

            # 确保输出目录存在
            os.makedirs(os.path.dirname(output_path) or ".", exist_ok=True)

            # 生成PDF
            HTML(string=html_content).write_pdf(output_path)

            return output_path
        except ImportError:
            # weasyprint 未安装
            html_path = output_path.replace(".pdf", ".html")
            html_content = self.render_html(result)
            with open(html_path, "w", encoding="utf-8") as f:
                f.write(html_content)
            return html_path

Step 3: 运行测试

pip3 install --user --break-system-packages weasyprint
python3 -m pytest skills/gaokao-audit/tests/test_report_generator.py -v

Step 4: 提交

git add skills/gaokao-audit/scripts/report_generator.py
git add skills/gaokao-audit/tests/test_report_generator.py
git commit -m "feat(audit): 实现报告生成器 - T1.7.1"

---

Task 1.8.1: 创建命令行入口

Objective: 创建 gaokao-audit CLI 脚本

Files:

• Create: skills/gaokao-audit/scripts/audit_cli.py
• Create: scripts/gaokao-audit (可执行)

Step 1: 实现 CLI

Create file: skills/gaokao-audit/scripts/audit_cli.py

"""
AI方案审核命令行工具
"""
import sys
import os
import argparse
import json

sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', '..'))

from skills.gaokao-audit.scripts.audit_service import AuditService
from skills.gaokao-audit.scripts.report_generator import ReportGenerator


def main():
    parser = argparse.ArgumentParser(description="AI志愿方案审核工具")
    parser.add_argument("input", help="方案文件路径（txt格式）")
    parser.add_argument("-o", "--output", help="PDF输出路径")
    parser.add_argument("-f", "--format", default="text", choices=["text", "pdf_text"],
                        help="输入格式")
    parser.add_argument("--json", action="store_true", help="输出JSON结果")

    args = parser.parse_args()

    # 读取输入
    if not os.path.exists(args.input):
        print(f"❌ 文件不存在: {args.input}")
        sys.exit(1)

    with open(args.input, "r", encoding="utf-8") as f:
        plan_text = f.read()

    print(f"📄 读取文件: {args.input} ({len(plan_text)} 字符)")

    # 执行审核
    print("🔍 开始审核...")
    service = AuditService()
    result = service.audit(plan_text, format=args.format)

    # 输出结果
    if args.json:
        print(json.dumps(result.to_dict(), ensure_ascii=False, indent=2))
    else:
        print(f"\n📊 审核结果:")
        print(f"  省份: {result.province}")
        print(f"  分数: {result.candidate_score}")
        print(f"  来源: {result.source or '未指定'}")
        print(f"  综合评分: {result.overall_score}/100")
        print(f"  致命错误: {len(result.policy_errors)} 个")
        print(f"  扎堆风险: {len(result.crowd_risks)} 个")

        if result.crowd_risks:
            print(f"\n🔴 扎堆风险院校:")
            for r in result.crowd_risks:
                print(f"  - {r.school} {r.major} "
                      f"({r.frequency}/4个AI推荐, 预测+{r.predicted_increase}分) "
                      f"[{r.risk_level_label}]")

    # 生成PDF
    if args.output or not args.json:
        output_path = args.output or f"audit_report_{os.path.basename(args.input)}.pdf"
        generator = ReportGenerator()
        result_path = generator.generate_pdf(result, output_path)
        print(f"\n📄 报告已生成: {result_path}")


if __name__ == "__main__":
    main()

Step 2: 创建可执行脚本

cd /home/long/project/gaokao-volunteer-system

# 创建可执行包装脚本
cat > scripts/gaokao-audit << 'EOF'
#!/bin/bash
# AI方案审核CLI wrapper
exec python3 /home/long/project/gaokao-volunteer-system/skills/gaokao-audit/scripts/audit_cli.py "$@"
EOF

chmod +x scripts/gaokao-audit

Step 3: 测试CLI

# 用之前的测试样本
./scripts/gaokao-audit skills/gaokao-audit/tests/fixtures/sample_xianyu.txt

Expected: 输出审核结果和PDF报告

Step 4: 提交

git add skills/gaokao-audit/scripts/audit_cli.py
git add scripts/gaokao-audit
git commit -m "feat(audit): 创建命令行CLI入口 - T1.8.1"

---

总结

完成清单

• T1.4.1: 集成规范检查器
• T1.5.1: 扎堆检测器
• T1.6.1: 审核服务主类
• T1.7.1: 报告生成器
• T1.8.1: CLI入口

完整审核流程

用户上传方案 (text/pdf/screenshot)
    ↓
audit_cli.py
    ↓
AuditService.audit()
    ├─ PlanParser.parse_text()
    ├─ CheckerIntegration.check()
    ├─ CrowdDetector.detect_risks()
    └─ 计算评分 + 生成建议
    ↓
ReportGenerator.generate_pdf()
    ↓
PDF报告

产出文件

文件	说明
scripts/audit_cli.py	CLI入口
scripts/audit_service.py	主服务
scripts/checker_integration.py	规范检查集成
scripts/crowd_detector.py	扎堆检测
scripts/plan_parser.py	方案解析
scripts/report_generator.py	报告生成
scripts/gaokao-audit	可执行命令
templates/audit_report.html	报告模板
tests/*.py	单元测试

验证

• 所有单元测试通过
• CLI命令可用
• PDF生成成功
• 端到端流程跑通

---

下一步: 启动服务，承接大厂AI用户