beamuswayne--simp-skill
wehub-resource-sync
349 行
13 KiB
Python
349 行
13 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
simp-skill · Skill Writer
|
||
管理心上人档案的创建、列表和版本控制
|
||
|
||
用法：
|
||
python3 skill_writer.py --action list
|
||
python3 skill_writer.py --action init --slug xiaomei
|
||
python3 skill_writer.py --action backup --slug xiaomei
|
||
python3 skill_writer.py --action rollback --slug xiaomei --version v1
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import shutil
|
||
import argparse
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
logging.basicConfig(level=logging.INFO, format="%(message)s")
|
||
logger = logging.getLogger(__name__)
|
||
|
||
DEFAULT_BASE_DIR = Path("crushes")
|
||
SIGNAL_SCORE_MIN = -15
|
||
SIGNAL_SCORE_MAX = 25
|
||
|
||
|
||
def init_crush(slug: str, base_dir: Path = DEFAULT_BASE_DIR) -> None:
|
||
"""初始化心上人档案目录结构"""
|
||
crush_dir = base_dir / slug
|
||
|
||
dirs = [
|
||
crush_dir,
|
||
crush_dir / "memories" / "chats",
|
||
crush_dir / "memories" / "social",
|
||
crush_dir / "memories" / "photos",
|
||
crush_dir / "versions",
|
||
crush_dir / "snapshots",
|
||
]
|
||
for d in dirs:
|
||
d.mkdir(parents=True, exist_ok=True)
|
||
|
||
now = datetime.now()
|
||
|
||
profile_path = crush_dir / "profile.md"
|
||
if not profile_path.exists():
|
||
profile_path.write_text(
|
||
f"---\n"
|
||
f"nickname: \"[待填写]\"\n"
|
||
f"slug: {slug}\n"
|
||
f"gender: \"[待填写]\"\n"
|
||
f"age: \"[待填写]\"\n"
|
||
f"occupation: \"[待填写]\"\n"
|
||
f"city: \"[待填写]\"\n"
|
||
f"mbti: \"[待填写]\"\n"
|
||
f"zodiac: \"[待填写]\"\n"
|
||
f"personality_type: \"[感性型/理性型/傲娇型/温柔型]\"\n"
|
||
f"how_met: \"[待填写]\"\n"
|
||
f"created_at: \"{now.strftime('%Y-%m-%d')}\"\n"
|
||
f"---\n\n"
|
||
f"## 性格画像\n\n"
|
||
f"[在这里描述ta的性格特征]\n\n"
|
||
f"## 最打动ta的方式\n\n"
|
||
f"[基于性格分析，什么样的话/行为最有效]\n\n"
|
||
f"## 用户自身风格\n\n"
|
||
f"[用户的说话风格、消息习惯、偏好模式]\n\n"
|
||
f"## 注意事项\n\n"
|
||
f"[ta特别在意或反感的事]\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
state_path = crush_dir / "state.md"
|
||
if not state_path.exists():
|
||
state_path.write_text(
|
||
f"---\n"
|
||
f"current_stage: 未知\n"
|
||
f"signal_score: null\n"
|
||
f"last_signal_score: null\n"
|
||
f"score_trend: stable\n"
|
||
f"recommended_mode: hybrid\n"
|
||
f"last_updated: \"{now.strftime('%Y-%m-%dT%H:%M:%S')}\"\n"
|
||
f"milestones_done: 0\n"
|
||
f"---\n\n"
|
||
f"## 当前状态（一句话）\n\n"
|
||
f"[运行 /simp analyze 后自动生成]\n\n"
|
||
f"## 最近信号（最新3条）\n\n"
|
||
f"[暂无信号记录]\n\n"
|
||
f"## 当前策略方向\n\n"
|
||
f"[运行 /simp analyze 后生成]\n\n"
|
||
f"## 下一步建议\n\n"
|
||
f"[运行 /simp analyze 后生成]\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
events_path = crush_dir / "events.jsonl"
|
||
if not events_path.exists():
|
||
events_path.touch()
|
||
|
||
interactions_path = crush_dir / "interactions.jsonl"
|
||
if not interactions_path.exists():
|
||
interactions_path.touch()
|
||
|
||
strategy_path = crush_dir / "strategy.md"
|
||
if not strategy_path.exists():
|
||
strategy_path.write_text(
|
||
f"# 追求策略\n\n"
|
||
f"> 由 simp-skill 生成 | 最后更新：{now.strftime('%Y-%m-%d')}\n\n"
|
||
f"## 当前阶段\n\n"
|
||
f"[待评估]\n\n"
|
||
f"## 推荐模式\n\n"
|
||
f"[纯情模式/策略模式/混合模式]\n\n"
|
||
f"## 本阶段重点\n\n"
|
||
f"[待生成]\n\n"
|
||
f"## 近期行动计划\n\n"
|
||
f"[待生成]\n",
|
||
encoding="utf-8",
|
||
)
|
||
|
||
meta_path = crush_dir / "meta.json"
|
||
if not meta_path.exists():
|
||
meta = {
|
||
"slug": slug,
|
||
"nickname": "[待填写]",
|
||
"created_at": now.isoformat(),
|
||
"updated_at": now.isoformat(),
|
||
"version": "v1",
|
||
"current_stage": "未知",
|
||
"signal_score": None,
|
||
"mode": "hybrid",
|
||
"event_count": 0,
|
||
"last_snapshot": None,
|
||
"interaction_count": 0,
|
||
"last_interaction": None,
|
||
"consecutive_days": 0,
|
||
}
|
||
meta_path.write_text(json.dumps(meta, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
logger.info("✅ 档案目录创建成功：%s/", crush_dir)
|
||
logger.info(" ├── profile.md （心上人基本信息）")
|
||
logger.info(" ├── state.md （当前状态快照）")
|
||
logger.info(" ├── events.jsonl （事件日志）")
|
||
logger.info(" ├── strategy.md （追求策略）")
|
||
logger.info(" ├── meta.json （元数据）")
|
||
logger.info(" ├── snapshots/ （定期快照）")
|
||
logger.info(" └── memories/")
|
||
logger.info(" ├── chats/ （放聊天记录）")
|
||
logger.info(" ├── social/ （放社交媒体截图）")
|
||
logger.info(" └── photos/ （放照片）")
|
||
logger.info("")
|
||
logger.info("下一步：")
|
||
logger.info(" 1. 编辑 %s/profile.md 填写心上人信息", crush_dir)
|
||
logger.info(" 2. 运行 /simp analyze 开始分析信号")
|
||
logger.info(" 3. 把聊天记录放到 %s/memories/chats/ 并运行 chat_parser.py", crush_dir)
|
||
|
||
|
||
def list_crushes(base_dir: Path = DEFAULT_BASE_DIR) -> None:
|
||
"""列出所有心上人档案"""
|
||
if not base_dir.exists():
|
||
logger.info("还没有任何心上人档案。运行 /simp create <名字> 开始吧！")
|
||
return
|
||
|
||
crushes = [d for d in base_dir.iterdir() if d.is_dir()]
|
||
if not crushes:
|
||
logger.info("还没有任何心上人档案。运行 /simp create <名字> 开始吧！")
|
||
return
|
||
|
||
logger.info("💝 心上人档案列表（共 %d 个）", len(crushes))
|
||
logger.info("")
|
||
for crush_dir in sorted(crushes):
|
||
meta_path = crush_dir / "meta.json"
|
||
if meta_path.exists():
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8"))
|
||
score = meta.get("signal_score")
|
||
score_str = f"{score}/25" if score is not None else "未评估"
|
||
stage = meta.get("current_stage", "未知")
|
||
updated = meta.get("updated_at", "")[:10]
|
||
logger.info(" 📁 %s", crush_dir.name)
|
||
logger.info(" 阶段：%s | 信号评分：%s | 最后更新：%s", stage, score_str, updated)
|
||
else:
|
||
logger.info(" 📁 %s", crush_dir.name)
|
||
logger.info("")
|
||
|
||
|
||
def backup_crush(slug: str, base_dir: Path = DEFAULT_BASE_DIR) -> str:
|
||
"""备份当前版本，返回版本名称；档案不存在时返回空字符串"""
|
||
crush_dir = base_dir / slug
|
||
if not crush_dir.exists():
|
||
logger.error("❌ 档案不存在：%s", slug)
|
||
return ""
|
||
|
||
meta_path = crush_dir / "meta.json"
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8")) if meta_path.exists() else {}
|
||
current_version = meta.get("version", "v1")
|
||
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
|
||
version_name = f"{current_version}_{timestamp}"
|
||
|
||
version_dir = crush_dir / "versions" / version_name
|
||
version_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
for filename in ["profile.md", "state.md", "strategy.md", "meta.json"]:
|
||
src = crush_dir / filename
|
||
if src.exists():
|
||
shutil.copy2(src, version_dir / filename)
|
||
|
||
v_num = int(current_version[1:]) + 1
|
||
new_version = f"v{v_num}"
|
||
updated_meta = {**meta, "version": new_version, "updated_at": datetime.now().isoformat()}
|
||
meta_path.write_text(json.dumps(updated_meta, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
|
||
logger.info("✅ 已备份版本 %s → %s", current_version, version_name)
|
||
logger.info(" 当前版本升级为 %s", new_version)
|
||
return version_name
|
||
|
||
|
||
def rollback_crush(slug: str, version: str, base_dir: Path = DEFAULT_BASE_DIR) -> None:
|
||
"""回滚到指定版本"""
|
||
crush_dir = base_dir / slug
|
||
versions_dir = crush_dir / "versions"
|
||
|
||
if not versions_dir.exists() or not any(versions_dir.iterdir()):
|
||
logger.error("❌ 没有找到版本历史")
|
||
return
|
||
|
||
matching = [d for d in versions_dir.iterdir() if d.name.startswith(version)]
|
||
if not matching:
|
||
available = [d.name for d in versions_dir.iterdir()]
|
||
logger.error("❌ 版本 %s 不存在", version)
|
||
logger.error(" 可用版本：%s", ", ".join(available))
|
||
return
|
||
|
||
backup_crush(slug, base_dir)
|
||
|
||
target_dir = sorted(matching)[-1]
|
||
# events.jsonl 是不可变历史，不参与回滚
|
||
for filename in ["profile.md", "state.md", "strategy.md", "meta.json"]:
|
||
src = target_dir / filename
|
||
if src.exists():
|
||
shutil.copy2(src, crush_dir / filename)
|
||
|
||
logger.info("✅ 已回滚到版本 %s", target_dir.name)
|
||
|
||
|
||
def list_versions(slug: str, base_dir: Path = DEFAULT_BASE_DIR) -> None:
|
||
"""列出版本历史"""
|
||
crush_dir = base_dir / slug
|
||
versions_dir = crush_dir / "versions"
|
||
|
||
if not versions_dir.exists() or not list(versions_dir.iterdir()):
|
||
logger.info("档案 %s 没有版本历史", slug)
|
||
return
|
||
|
||
versions = sorted(versions_dir.iterdir())
|
||
logger.info("📚 %s 的版本历史（共 %d 个版本）", slug, len(versions))
|
||
for v in versions:
|
||
meta_path = v / "meta.json"
|
||
if meta_path.exists():
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8"))
|
||
stage = meta.get("current_stage", "")
|
||
score = meta.get("signal_score", "")
|
||
logger.info(" - %s 阶段：%s 评分：%s", v.name, stage, score)
|
||
else:
|
||
logger.info(" - %s", v.name)
|
||
|
||
|
||
def update_meta(slug: str, base_dir: Path = DEFAULT_BASE_DIR, **kwargs: object) -> None:
|
||
"""更新档案元数据"""
|
||
crush_dir = base_dir / slug
|
||
meta_path = crush_dir / "meta.json"
|
||
if not meta_path.exists():
|
||
logger.error("❌ 档案不存在：%s", slug)
|
||
return
|
||
|
||
if "signal_score" in kwargs:
|
||
score = kwargs["signal_score"]
|
||
if score is not None and not (SIGNAL_SCORE_MIN <= int(score) <= SIGNAL_SCORE_MAX):
|
||
logger.error(
|
||
"❌ 信号评分必须在 %d-%d 之间，当前值：%s",
|
||
SIGNAL_SCORE_MIN,
|
||
SIGNAL_SCORE_MAX,
|
||
score,
|
||
)
|
||
return
|
||
|
||
meta = json.loads(meta_path.read_text(encoding="utf-8"))
|
||
updated_meta = {**meta, **kwargs, "updated_at": datetime.now().isoformat()}
|
||
meta_path.write_text(json.dumps(updated_meta, ensure_ascii=False, indent=2), encoding="utf-8")
|
||
logger.info("✅ 档案元数据已更新")
|
||
|
||
|
||
def main() -> None:
|
||
parser = argparse.ArgumentParser(
|
||
description="simp-skill · 档案管理器",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
parser.add_argument(
|
||
"--action",
|
||
required=True,
|
||
choices=["list", "init", "backup", "rollback", "versions", "update-meta"],
|
||
help="操作类型",
|
||
)
|
||
parser.add_argument("--slug", help="心上人档案名（拼音或英文）")
|
||
parser.add_argument("--version", help="版本号（rollback 时使用）")
|
||
parser.add_argument("--stage", help="更新当前阶段")
|
||
parser.add_argument("--score", type=int, help=f"更新信号评分（{SIGNAL_SCORE_MIN}~{SIGNAL_SCORE_MAX}）")
|
||
parser.add_argument("--mode", choices=["sweet", "strategic", "hybrid"], help="更新追求模式")
|
||
parser.add_argument("--base-dir", default="crushes", help="档案根目录（默认：crushes/）")
|
||
|
||
args = parser.parse_args()
|
||
base_dir = Path(args.base_dir)
|
||
|
||
if args.action == "list":
|
||
list_crushes(base_dir)
|
||
elif args.action == "init":
|
||
if not args.slug:
|
||
logger.error("❌ 请提供 --slug 参数")
|
||
return
|
||
init_crush(args.slug, base_dir)
|
||
elif args.action == "backup":
|
||
if not args.slug:
|
||
logger.error("❌ 请提供 --slug 参数")
|
||
return
|
||
backup_crush(args.slug, base_dir)
|
||
elif args.action == "rollback":
|
||
if not args.slug or not args.version:
|
||
logger.error("❌ 请提供 --slug 和 --version 参数")
|
||
return
|
||
rollback_crush(args.slug, args.version, base_dir)
|
||
elif args.action == "versions":
|
||
if not args.slug:
|
||
logger.error("❌ 请提供 --slug 参数")
|
||
return
|
||
list_versions(args.slug, base_dir)
|
||
elif args.action == "update-meta":
|
||
if not args.slug:
|
||
logger.error("❌ 请提供 --slug 参数")
|
||
return
|
||
kwargs: dict[str, object] = {}
|
||
if args.stage:
|
||
kwargs["current_stage"] = args.stage
|
||
if args.score is not None:
|
||
kwargs["signal_score"] = args.score
|
||
if args.mode:
|
||
kwargs["mode"] = args.mode
|
||
update_meta(args.slug, base_dir, **kwargs)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|