210 lines
7.2 KiB
Python
210 lines
7.2 KiB
Python
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()
|