#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ IndexReadingMixin extracted from IndexManager. """ from __future__ import annotations import json import sys from datetime import datetime from typing import Any, Dict, List, Optional class IndexReadingMixin: def save_chapter_reading_power(self, meta: ChapterReadingPowerMeta): """保存章节追读力元数据""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( """ INSERT OR REPLACE INTO chapter_reading_power (chapter, hook_type, hook_strength, coolpoint_patterns, micropayoffs, hard_violations, soft_suggestions, is_transition, override_count, debt_balance) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( meta.chapter, meta.hook_type, meta.hook_strength, json.dumps(meta.coolpoint_patterns, ensure_ascii=False), json.dumps(meta.micropayoffs, ensure_ascii=False), json.dumps(meta.hard_violations, ensure_ascii=False), json.dumps(meta.soft_suggestions, ensure_ascii=False), 1 if meta.is_transition else 0, meta.override_count, meta.debt_balance, ), ) conn.commit() def get_chapter_reading_power(self, chapter: int) -> Optional[Dict]: """获取章节追读力元数据""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( "SELECT * FROM chapter_reading_power WHERE chapter = ?", (chapter,) ) row = cursor.fetchone() if row: return self._row_to_dict( row, parse_json=[ "coolpoint_patterns", "micropayoffs", "hard_violations", "soft_suggestions", ], ) return None def get_recent_reading_power(self, limit: int = 10) -> List[Dict]: """获取最近章节的追读力元数据""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( """ SELECT * FROM chapter_reading_power ORDER BY chapter DESC LIMIT ? """, (limit,), ) return [ self._row_to_dict( row, parse_json=[ "coolpoint_patterns", "micropayoffs", "hard_violations", "soft_suggestions", ], ) for row in cursor.fetchall() ] def get_pattern_usage_stats(self, last_n_chapters: int = 20) -> Dict[str, int]: """获取最近N章的爽点模式使用统计""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( """ SELECT coolpoint_patterns FROM chapter_reading_power ORDER BY chapter DESC LIMIT ? """, (last_n_chapters,), ) stats = {} for row in cursor.fetchall(): if row["coolpoint_patterns"]: try: patterns = json.loads(row["coolpoint_patterns"]) for p in patterns: stats[p] = stats.get(p, 0) + 1 except json.JSONDecodeError as exc: print( f"[index_manager] failed to parse JSON in chapter_reading_power.coolpoint_patterns: {exc}", file=sys.stderr, ) return stats def get_hook_type_stats(self, last_n_chapters: int = 20) -> Dict[str, int]: """获取最近N章的钩子类型使用统计""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( """ SELECT hook_type FROM chapter_reading_power WHERE hook_type IS NOT NULL AND hook_type != '' ORDER BY chapter DESC LIMIT ? """, (last_n_chapters,), ) stats = {} for row in cursor.fetchall(): hook = row["hook_type"] stats[hook] = stats.get(hook, 0) + 1 return stats # ==================== v5.4 审查指标 ==================== def save_review_metrics(self, metrics: ReviewMetrics) -> None: """保存审查指标记录""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO review_metrics (start_chapter, end_chapter, overall_score, dimension_scores, severity_counts, critical_issues, report_file, notes, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) ON CONFLICT(start_chapter, end_chapter) DO UPDATE SET overall_score = excluded.overall_score, dimension_scores = excluded.dimension_scores, severity_counts = excluded.severity_counts, critical_issues = excluded.critical_issues, report_file = excluded.report_file, notes = excluded.notes, updated_at = CURRENT_TIMESTAMP """, ( metrics.start_chapter, metrics.end_chapter, metrics.overall_score, json.dumps(metrics.dimension_scores, ensure_ascii=False), json.dumps(metrics.severity_counts, ensure_ascii=False), json.dumps(metrics.critical_issues, ensure_ascii=False), metrics.report_file, metrics.notes, ), ) conn.commit() def get_recent_review_metrics(self, limit: int = 5) -> List[Dict]: """获取最近审查记录""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( """ SELECT * FROM review_metrics ORDER BY end_chapter DESC, start_chapter DESC LIMIT ? """, (limit,), ) return [ self._row_to_dict( row, parse_json=["dimension_scores", "severity_counts", "critical_issues"], ) for row in cursor.fetchall() ] def get_review_trend_stats(self, last_n: int = 5) -> Dict[str, Any]: """获取审查趋势统计""" records = self.get_recent_review_metrics(last_n) if not records: return { "count": 0, "overall_avg": 0.0, "dimension_avg": {}, "severity_totals": {}, "recent_ranges": [], } overall_scores: List[float] = [] dimension_totals: Dict[str, float] = {} dimension_counts: Dict[str, int] = {} severity_totals: Dict[str, int] = {} for record in records: score = record.get("overall_score") if score is not None: try: overall_scores.append(float(score)) except (TypeError, ValueError): pass dimensions = record.get("dimension_scores") or {} if isinstance(dimensions, dict): for key, value in dimensions.items(): try: val = float(value) except (TypeError, ValueError): continue dimension_totals[key] = dimension_totals.get(key, 0.0) + val dimension_counts[key] = dimension_counts.get(key, 0) + 1 severities = record.get("severity_counts") or {} if isinstance(severities, dict): for key, value in severities.items(): try: count = int(value) except (TypeError, ValueError): continue severity_totals[key] = severity_totals.get(key, 0) + count overall_avg = round(sum(overall_scores) / len(overall_scores), 2) if overall_scores else 0.0 dimension_avg = { key: round(dimension_totals[key] / dimension_counts[key], 2) for key in dimension_totals if dimension_counts.get(key, 0) > 0 } recent_ranges = [ { "start_chapter": record.get("start_chapter"), "end_chapter": record.get("end_chapter"), "overall_score": record.get("overall_score", 0), } for record in records ] return { "count": len(records), "overall_avg": overall_avg, "dimension_avg": dimension_avg, "severity_totals": severity_totals, "recent_ranges": recent_ranges, } # ==================== 写作清单评分（Phase F） ==================== def save_writing_checklist_score(self, meta: WritingChecklistScoreMeta) -> None: """保存章节写作清单评分。""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( """ INSERT INTO writing_checklist_scores ( chapter, template, total_items, required_items, completed_items, completed_required, total_weight, completed_weight, completion_rate, score, score_breakdown, pending_items, source, notes ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(chapter) DO UPDATE SET template=excluded.template, total_items=excluded.total_items, required_items=excluded.required_items, completed_items=excluded.completed_items, completed_required=excluded.completed_required, total_weight=excluded.total_weight, completed_weight=excluded.completed_weight, completion_rate=excluded.completion_rate, score=excluded.score, score_breakdown=excluded.score_breakdown, pending_items=excluded.pending_items, source=excluded.source, notes=excluded.notes, updated_at=CURRENT_TIMESTAMP """, ( meta.chapter, meta.template, meta.total_items, meta.required_items, meta.completed_items, meta.completed_required, meta.total_weight, meta.completed_weight, meta.completion_rate, meta.score, json.dumps(meta.score_breakdown, ensure_ascii=False), json.dumps(meta.pending_items, ensure_ascii=False), meta.source, meta.notes, ), ) conn.commit() def get_writing_checklist_score(self, chapter: int) -> Optional[Dict[str, Any]]: """获取指定章节的写作清单评分。""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( "SELECT * FROM writing_checklist_scores WHERE chapter = ?", (chapter,), ) row = cursor.fetchone() if not row: return None return self._row_to_dict(row, parse_json=["score_breakdown", "pending_items"]) def get_recent_writing_checklist_scores(self, limit: int = 10) -> List[Dict[str, Any]]: """获取最近章节写作清单评分。""" with self._get_conn() as conn: cursor = conn.cursor() cursor.execute( """ SELECT * FROM writing_checklist_scores ORDER BY chapter DESC LIMIT ? """, (limit,), ) return [ self._row_to_dict(row, parse_json=["score_breakdown", "pending_items"]) for row in cursor.fetchall() ] def get_writing_checklist_score_trend(self, last_n: int = 10) -> Dict[str, Any]: """获取写作清单评分趋势统计。""" records = self.get_recent_writing_checklist_scores(limit=max(1, int(last_n))) if not records: return { "count": 0, "score_avg": 0.0, "completion_avg": 0.0, "required_completion_avg": 0.0, "recent": [], } scores: List[float] = [] completion_rates: List[float] = [] required_rates: List[float] = [] for row in records: try: scores.append(float(row.get("score", 0.0))) except (TypeError, ValueError): pass try: completion_rates.append(float(row.get("completion_rate", 0.0))) except (TypeError, ValueError): pass required_items = int(row.get("required_items") or 0) completed_required = int(row.get("completed_required") or 0) if required_items > 0: required_rates.append(completed_required / required_items) else: required_rates.append(1.0) return { "count": len(records), "score_avg": round(sum(scores) / len(scores), 2) if scores else 0.0, "completion_avg": round(sum(completion_rates) / len(completion_rates), 4) if completion_rates else 0.0, "required_completion_avg": round(sum(required_rates) / len(required_rates), 4) if required_rates else 0.0, "recent": [ { "chapter": row.get("chapter"), "score": row.get("score"), "completion_rate": row.get("completion_rate"), } for row in records ], }