#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ContextManager - assemble context packs with weighted priorities. """ from __future__ import annotations import json from pathlib import Path from typing import Any, Dict, List, Optional from .config import get_config from .index_manager import IndexManager from .snapshot_manager import SnapshotManager, SnapshotVersionMismatch class ContextManager: DEFAULT_TEMPLATE = "plot" TEMPLATE_WEIGHTS = { "plot": {"core": 0.40, "scene": 0.35, "global": 0.25}, "battle": {"core": 0.35, "scene": 0.45, "global": 0.20}, "emotion": {"core": 0.45, "scene": 0.35, "global": 0.20}, "transition": {"core": 0.50, "scene": 0.25, "global": 0.25}, } def __init__(self, config=None, snapshot_manager: Optional[SnapshotManager] = None): self.config = config or get_config() self.snapshot_manager = snapshot_manager or SnapshotManager(self.config) self.index_manager = IndexManager(self.config) def build_context( self, chapter: int, template: str | None = None, use_snapshot: bool = True, save_snapshot: bool = True, max_chars: Optional[int] = None, ) -> Dict[str, Any]: template = template or self.DEFAULT_TEMPLATE if template not in self.TEMPLATE_WEIGHTS: template = self.DEFAULT_TEMPLATE if use_snapshot: try: cached = self.snapshot_manager.load_snapshot(chapter) if cached: return cached.get("payload", cached) except SnapshotVersionMismatch: # Snapshot incompatible; rebuild below. pass pack = self._build_pack(chapter) assembled = self.assemble_context(pack, template=template, max_chars=max_chars) if save_snapshot: meta = {"template": template} self.snapshot_manager.save_snapshot(chapter, assembled, meta=meta) return assembled def assemble_context( self, pack: Dict[str, Any], template: str = DEFAULT_TEMPLATE, max_chars: Optional[int] = None, ) -> Dict[str, Any]: weights = self.TEMPLATE_WEIGHTS.get(template, self.TEMPLATE_WEIGHTS[self.DEFAULT_TEMPLATE]) max_chars = max_chars or 8000 sections = {} for section_name in ["core", "scene", "global", "memory", "preferences", "alerts"]: if section_name in pack: sections[section_name] = pack[section_name] assembled: Dict[str, Any] = {"meta": pack.get("meta", {}), "sections": {}} for name, content in sections.items(): weight = weights.get(name, 0.0) budget = int(max_chars * weight) if weight > 0 else None text = json.dumps(content, ensure_ascii=False) if budget is not None and len(text) > budget: text = text[:budget] assembled["sections"][name] = {"content": content, "text": text, "budget": budget} assembled["template"] = template assembled["weights"] = weights return assembled def filter_invalid_items(self, items: List[Dict[str, Any]], source_type: str, id_key: str) -> List[Dict[str, Any]]: confirmed = self.index_manager.get_invalid_ids(source_type, status="confirmed") pending = self.index_manager.get_invalid_ids(source_type, status="pending") result = [] for item in items: item_id = str(item.get(id_key, "")) if item_id in confirmed: continue if item_id in pending: item = dict(item) item["warning"] = "pending_invalid" result.append(item) return result def apply_confidence_filter(self, items: List[Dict[str, Any]], min_confidence: float) -> List[Dict[str, Any]]: filtered: List[Dict[str, Any]] = [] for item in items: conf = item.get("confidence") if conf is None or conf >= min_confidence: filtered.append(item) return filtered def _build_pack(self, chapter: int) -> Dict[str, Any]: state = self._load_state() core = { "chapter_outline": self._load_outline(chapter), "protagonist_snapshot": state.get("protagonist_state", {}), "recent_summaries": self._load_recent_summaries(chapter, window=3), "recent_meta": self._load_recent_meta(state, chapter, window=3), } scene = { "location_context": state.get("protagonist_state", {}).get("location", {}), "appearing_characters": self._load_recent_appearances(), } scene["appearing_characters"] = self.filter_invalid_items( scene["appearing_characters"], source_type="entity", id_key="entity_id" ) global_ctx = { "worldview_skeleton": self._load_setting("世界观"), "power_system_skeleton": self._load_setting("力量体系"), "style_contract_ref": self._load_setting("风格契约"), } preferences = self._load_json_optional(self.config.webnovel_dir / "preferences.json") memory = self._load_json_optional(self.config.webnovel_dir / "project_memory.json") return { "meta": {"chapter": chapter}, "core": core, "scene": scene, "global": global_ctx, "preferences": preferences, "memory": memory, "alerts": { "disambiguation_warnings": state.get("disambiguation_warnings", [])[-10:], "disambiguation_pending": state.get("disambiguation_pending", [])[-10:], }, } def _load_state(self) -> Dict[str, Any]: path = self.config.state_file if not path.exists(): return {} return json.loads(path.read_text(encoding="utf-8")) def _load_outline(self, chapter: int) -> str: outline_dir = self.config.outline_dir patterns = [ f"第{chapter}章*.md", f"第{chapter:02d}章*.md", f"第{chapter:03d}章*.md", f"第{chapter:04d}章*.md", ] for pattern in patterns: matches = list(outline_dir.glob(pattern)) if matches: return matches[0].read_text(encoding="utf-8") return f"[大纲未找到: 第{chapter}章]" def _load_recent_summaries(self, chapter: int, window: int = 3) -> List[Dict[str, Any]]: summaries = [] summaries_dir = self.config.webnovel_dir / "summaries" for ch in range(max(1, chapter - window), chapter): path = summaries_dir / f"ch{ch:04d}.md" if path.exists(): summaries.append({"chapter": ch, "summary": path.read_text(encoding="utf-8")}) return summaries def _load_recent_meta(self, state: Dict[str, Any], chapter: int, window: int = 3) -> List[Dict[str, Any]]: meta = state.get("chapter_meta", {}) or {} results = [] for ch in range(max(1, chapter - window), chapter): for key in (f"{ch:04d}", str(ch)): if key in meta: results.append({"chapter": ch, **meta.get(key, {})}) break return results def _load_recent_appearances(self) -> List[Dict[str, Any]]: appearances = self.index_manager.get_recent_appearances() return appearances or [] def _load_setting(self, keyword: str) -> str: settings_dir = self.config.settings_dir candidates = [ settings_dir / f"{keyword}.md", ] for path in candidates: if path.exists(): return path.read_text(encoding="utf-8") # fallback: any file containing keyword matches = list(settings_dir.glob(f"*{keyword}*.md")) if matches: return matches[0].read_text(encoding="utf-8") return f"[{keyword}设定未找到]" def _load_json_optional(self, path: Path) -> Dict[str, Any]: if not path.exists(): return {} try: return json.loads(path.read_text(encoding="utf-8")) except json.JSONDecodeError: return {} def main(): import argparse from .cli_output import print_success, print_error parser = argparse.ArgumentParser(description="Context Manager CLI") parser.add_argument("--project-root", type=str, help="项目根目录") parser.add_argument("--chapter", type=int, required=True) parser.add_argument("--template", type=str, default=ContextManager.DEFAULT_TEMPLATE) parser.add_argument("--no-snapshot", action="store_true") parser.add_argument("--max-chars", type=int, default=8000) args = parser.parse_args() config = None if args.project_root: from .config import DataModulesConfig config = DataModulesConfig.from_project_root(args.project_root) manager = ContextManager(config) try: payload = manager.build_context( chapter=args.chapter, template=args.template, use_snapshot=not args.no_snapshot, save_snapshot=True, max_chars=args.max_chars, ) print_success(payload, message="context_built") try: manager.index_manager.log_tool_call("context_manager:build", True, chapter=args.chapter) except Exception: pass except Exception as exc: print_error("CONTEXT_BUILD_FAILED", str(exc), suggestion="请检查项目结构与依赖文件") try: manager.index_manager.log_tool_call( "context_manager:build", False, error_code="CONTEXT_BUILD_FAILED", error_message=str(exc), chapter=args.chapter ) except Exception: pass if __name__ == "__main__": main()