from __future__ import annotations import json import re from dataclasses import dataclass from pathlib import Path from typing import Any PROFILE_IDS = { "emotion": { "kobe92-perspective", "niepanchongsheng-perspective", "chaojiyangjia-perspective", "tuixuechaogu-perspective", "chenxiaoqun-perspective", "zhiyechaoshou-perspective", }, "first_board": { "beijingchaojia-perspective", "chuangshiji-perspective", "xuxiang-perspective", "foshanwuyingjiao-perspective", }, "leader": { "zhaolaoge-perspective", "fangxinxia-perspective", "xiaoe-perspective", "sunge-perspective", "liuyizhonglu-perspective", }, "trend": { "zhangdetao-perspective", "zhangmengzhu-perspective", "zuoshouxinyi-perspective", }, "low_absorption": { "qiaobangzhu-perspective", "asking-perspective", "longfeihu-perspective", "ruihexian-perspective", }, "macro": {"shuipi-perspective"}, } @dataclass(frozen=True, slots=True) class MentorSkill: id: str name: str description: str tagline: str focus: tuple[str, ...] grade: str evidence_label: str evidence_note: str profile: str content: str private: bool def public(self) -> dict[str, Any]: return { "id": self.id, "name": self.name, "description": self.description, "tagline": self.tagline, "focus": list(self.focus), "grade": self.grade, "evidence_label": self.evidence_label, "evidence_note": self.evidence_note, "private": self.private, } class MentorSkillError(ValueError): pass class MentorSkillRegistry: def __init__(self, public_root: Path, private_root: Path) -> None: self._public_root = public_root self._private_root = private_root def list(self, include_private: bool) -> tuple[MentorSkill, ...]: skills = {item.id: item for item in self._load_root(self._public_root, False)} if include_private: skills.update({item.id: item for item in self._load_root(self._private_root, True)}) return tuple(sorted(skills.values(), key=lambda item: (item.name.casefold(), item.id))) def get(self, skill_id: str, include_private: bool) -> MentorSkill: match = next((item for item in self.list(include_private) if item.id == skill_id), None) if match is None: raise MentorSkillError("思维模型不存在或当前账号不可见。") return match def _load_root(self, root: Path, private: bool) -> tuple[MentorSkill, ...]: if not root.is_dir(): return () catalog = _catalog(root) result = [] for directory in sorted(root.iterdir(), key=lambda item: item.name): path = directory / "SKILL.md" if directory.is_dir() and path.is_file(): result.append(_read_skill(path, catalog, private)) return tuple(result) def _read_skill(path: Path, catalog: dict[str, Any], private: bool) -> MentorSkill: if path.stat().st_size > 200_000: raise MentorSkillError(f"Skill文件过大：{path.parent.name}") content = path.read_text(encoding="utf-8") metadata = _frontmatter(content) raw_id = metadata.get("name") or path.parent.name skill_id = re.sub(r"[^A-Za-z0-9_-]+", "-", raw_id).strip("-").lower() if not skill_id: raise MentorSkillError(f"Skill缺少有效ID：{path.parent.name}") heading = re.search(r"^#\s+(.+?)(?:\s*[·|]\s*.+)?$", content, re.MULTILINE) name = heading.group(1).strip() if heading else path.parent.name description_raw = metadata.get("description", "") purpose = re.search(r"用途[：:]\s*([^\n]+)", description_raw) description = purpose.group(1).strip() if purpose else _first_sentence(description_raw) tagline_match = re.search(r'^>\s*["“「](.+?)["”」]\s*$', content, re.MULTILINE) focus = tuple( item.strip() for item in re.findall(r"^###\s+模型\d+[：:]\s*(.+)$", content, re.MULTILINE)[:4] ) item = catalog.get(skill_id) if isinstance(catalog.get(skill_id), dict) else {} evidence = item.get("evidence") if isinstance(item.get("evidence"), dict) else {} grade = str(evidence.get("grade") or "C").upper() if grade not in {"A", "B", "C"}: grade = "C" return MentorSkill( id=skill_id, name=name.removesuffix("-perspective").strip(), description=description, tagline=tagline_match.group(1).strip() if tagline_match else "", focus=focus, grade=grade, evidence_label=str(evidence.get("label") or "公开资料"), evidence_note=str(evidence.get("note") or "素材等级待进一步核验"), profile=_profile(skill_id, f"{description} {' '.join(focus)}"), content=content, private=private, ) def _catalog(root: Path) -> dict[str, Any]: path = root / "mentor_catalog.json" if not path.is_file(): return {} try: payload = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as exc: raise MentorSkillError(f"思维模型目录无法读取：{root.name}") from exc mentors = payload.get("mentors", payload) if isinstance(payload, dict) else {} if not isinstance(mentors, dict): raise MentorSkillError(f"思维模型目录格式错误：{root.name}") return mentors def _profile(skill_id: str, text: str) -> str: for profile, identifiers in PROFILE_IDS.items(): if skill_id in identifiers: return profile keywords = ( ("macro", ("宏观", "政策", "指数", "ETF")), ("trend", ("趋势", "资金面", "动能")), ("first_board", ("首板", "打板")), ("leader", ("龙头", "连板", "空间板")), ("low_absorption", ("低吸", "反包", "承接")), ("emotion", ("情绪", "周期", "退潮")), ) return next( (profile for profile, words in keywords if any(word in text for word in words)), "emotion", ) def _frontmatter(content: str) -> dict[str, str]: if not content.startswith("---"): return {} end = content.find("\n---", 3) if end < 0: return {} lines = content[3:end].strip().splitlines() result: dict[str, str] = {} index = 0 while index < len(lines): line = lines[index] if ":" not in line: index += 1 continue key, value = line.split(":", 1) value = value.strip() if value == "|": block = [] index += 1 while index < len(lines) and (lines[index].startswith(" ") or not lines[index].strip()): block.append(lines[index].strip()) index += 1 result[key.strip()] = "\n".join(block).strip() continue result[key.strip()] = value.strip('"\'') index += 1 return result def _first_sentence(value: str) -> str: compact = " ".join(line.strip() for line in value.splitlines() if line.strip()) return re.split(r"[。；]", compact, maxsplit=1)[0].strip()