novel-skill/scripts/quality_tuner.py

#!/usr/bin/env python3
"""
quality_tuner.py — 质量阈值自动调优器
调优结果写入 evolution/thresholds.json，不修改 SKILL.md。
"""

import sys, os, re, json, math
from pathlib import Path
from typing import Any

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from utils import (
    find_project_root, get_evolution_dir,
    load_evolution, save_evolution, parse_quality_log,
    extract_quality_features, log_evolution_event,
)

DEFAULT_THRESHOLDS = {
    "pass": 80, "revise_low": 60, "rewrite_low": 40,
    "max_revision_rounds": 5,
    "weights": {"P1": 0.35, "P2": 0.25, "P3": 0.15, "P4": 0.15, "P5": 0.10},
    "stale_hook_limit": 10, "burst_hook_limit": 5,
    "no_advance_limit": 5,
    "payoff_interval_min": 2000, "payoff_interval_max": 5000,
    "weak_hook_consecutive_max": 3,
}


class QualityTuner:
    def __init__(self, thresholds: dict | None = None):
        self.thresholds = thresholds or load_evolution("thresholds.json") or DEFAULT_THRESHOLDS.copy()
        self.history: list[dict] = []
        self.adjustments: list[dict] = []

    def load_history(self, project: Path) -> None:
        self.history = parse_quality_log(project)

    def analyze(self) -> dict:
        if not self.history:
            return {"status": "no_data"}

        features = extract_quality_features(self.history)
        suggestions = []

        recent = [r for r in self.history[-10:] if r.get("score")]
        recent_scores = [r["score"] for r in recent]
        pass_rate = sum(1 for s in recent_scores if s >= self.thresholds["pass"]) / max(len(recent_scores), 1)

        if pass_rate > 0.9 and len(recent_scores) >= 5:
            suggestions.append({
                "type": "pass_threshold_raise",
                "current": self.thresholds["pass"],
                "suggested": min(88, self.thresholds["pass"] + 3),
                "reason": f"近期通过率 {pass_rate:.0%}，阈值可能偏低",
                "confidence": 0.7,
            })
        elif pass_rate < 0.3 and len(recent_scores) >= 5:
            suggestions.append({
                "type": "pass_threshold_lower",
                "current": self.thresholds["pass"],
                "suggested": max(70, self.thresholds["pass"] - 3),
                "reason": f"近期通过率 {pass_rate:.0%}，阈值可能偏高",
                "confidence": 0.6,
            })

        pass_weights = ["P1", "P2", "P3", "P4", "P5"]
        pass_scores = {}
        for p in pass_weights:
            scores = [r.get(f"score_{p.lower()}", 0) for r in self.history if f"score_{p.lower()}" in r]
            if scores:
                pass_scores[p] = sum(scores) / len(scores)

        if pass_scores:
            weakest = min(pass_scores, key=pass_scores.get)
            strongest = max(pass_scores, key=pass_scores.get)
            if pass_scores[strongest] - pass_scores[weakest] > 15:
                current_weight = self.thresholds["weights"][weakest]
                new_weight = min(0.40, current_weight + 0.05)
                suggestions.append({
                    "type": "weight_adjust",
                    "target_pass": weakest,
                    "current_weight": current_weight,
                    "suggested_weight": new_weight,
                    "reason": f"{weakest} 最弱 ({pass_scores[weakest]:.0f})，增权",
                    "confidence": 0.75,
                    "redistribute_from": strongest,
                })

        rounds = [r.get("revision_rounds", 1) for r in self.history if r.get("revision_rounds")]
        avg_rounds = sum(rounds) / len(rounds) if rounds else 1.0
        if avg_rounds > 3:
            suggestions.append({
                "type": "root_cause_warning",
                "reason": f"平均修订 {avg_rounds:.1f} 轮，建议向上追溯大纲/设定问题",
                "confidence": 0.8,
            })

        s1_recent = sum(r.get("s1_count", 0) for r in recent)
        if s1_recent > 0:
            suggestions.append({
                "type": "quality_alert", "level": "S1", "count": s1_recent,
                "reason": f"近10章有 {s1_recent} 个S1致命问题",
                "confidence": 0.9,
            })

        return {
            "status": "ok", "features": features, "pass_rate": pass_rate,
            "recent_scores": recent_scores, "pass_scores": pass_scores,
            "suggestions": suggestions, "avg_revision_rounds": avg_rounds,
        }

    def apply(self, suggestions: list[dict], auto_apply: bool = False) -> list[dict]:
        applied = []
        for s in suggestions:
            if s.get("type") == "quality_alert":
                applied.append({**s, "applied": False})
                continue
            if not auto_apply and s.get("confidence", 0) < 0.7:
                applied.append({**s, "applied": False})
                continue
            if s["type"] in ("pass_threshold_raise", "pass_threshold_lower"):
                old = self.thresholds["pass"]
                self.thresholds["pass"] = s["suggested"]
                self.adjustments.append({"what": "pass", "from": old, "to": s["suggested"]})
                applied.append({**s, "applied": True})
            elif s["type"] == "weight_adjust":
                target = s["target_pass"]
                redistribute = s.get("redistribute_from")
                old_w = self.thresholds["weights"][target]
                self.thresholds["weights"][target] = s["suggested_weight"]
                if redistribute:
                    self.thresholds["weights"][redistribute] -= (s["suggested_weight"] - old_w)
                self.adjustments.append({"what": f"weight_{target}", "from": old_w, "to": s["suggested_weight"]})
                applied.append({**s, "applied": True})
        return applied

    def sync_to_evolution(self) -> None:
        """将当前阈值写入 evolution/thresholds.json"""
        save_evolution("thresholds.json", self.thresholds)

    def sync_to_genre_evolution(self, genre_id: str) -> None:
        """将当前阈值写入 evolution/genres/{genre_id}/thresholds.json"""
        from utils import save_genre_evolution
        save_genre_evolution(genre_id, "thresholds.json", self.thresholds)


# ── CLI ─────────────────────────────────────────────────

def analyze_and_tune(project: Path, auto_apply: bool = False) -> dict:
    print(f"🔧 质量阈值调优: {project.name}")
    tuner = QualityTuner()
    tuner.load_history(project)
    analysis = tuner.analyze()

    if analysis["status"] == "no_data":
        print("  ⚠ 没有质检历史数据，使用当前阈值")
        return analysis

    features = analysis.get("features", {})
    avg = features.get('avg_score', 0)
    print(f"  平均得分: {avg:.1f}" if isinstance(avg, (int, float)) else f"  平均得分: N/A")
    print(f"  通过率: {analysis.get('pass_rate', 0):.0%}")
    print(f"  S1/S2: {features.get('total_s1', 0)}/{features.get('total_s2', 0)}")

    suggestions = analysis.get("suggestions", [])
    if suggestions:
        print(f"\n  {len(suggestions)} 条优化建议：")
        for s in suggestions:
            print(f"  [{s['type']}] {s.get('reason', '')}")
        applied = tuner.apply(suggestions, auto_apply=auto_apply)
        applied_count = sum(1 for a in applied if a.get("applied"))
        print(f"\n  应用了 {applied_count} 条，跳过 {len(applied) - applied_count} 条")
    else:
        print("  ✓ 阈值配置合理，无需调整")
        applied = []

    from utils import find_project_genre, map_genre_to_inkos_id
    genre_id = map_genre_to_inkos_id(find_project_genre(project))
    tuner.sync_to_evolution()
    tuner.sync_to_genre_evolution(genre_id)
    print(f"  ✓ 阈值已同步到 skill 级和题材级 evolution/genres/{genre_id}/")
    return {"analysis": analysis, "applied": applied}


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser(description="质量阈值调优器")
    parser.add_argument("action", choices=["analyze", "tune", "show"],
                        help="analyze=分析 | tune=分析并应用 | show=查看当前阈值")
    parser.add_argument("--project", "-p", default=".")
    parser.add_argument("--auto", action="store_true")
    args = parser.parse_args()

    if args.action == "show":
        thresholds = load_evolution("thresholds.json") or DEFAULT_THRESHOLDS
        print(json.dumps(thresholds, ensure_ascii=False, indent=2))

    elif args.action in ("analyze", "tune"):
        root = find_project_root(Path(args.project).resolve())
        if not root:
            print("错误: 未找到小说项目根目录")
            sys.exit(1)
        auto = args.auto or (args.action == "tune")
        result = analyze_and_tune(root, auto_apply=auto)
        if args.action == "tune" and result.get("applied"):
            log_evolution_event(root, "质量阈值调优", {
                "应用": len([a for a in result["applied"] if a.get("applied")]),
            })