390 lines
14 KiB
Python
390 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Soulful Core — 织忆情感层核心库
|
||
=================================
|
||
提供：心迹（heart-traces）+ 用户画像（user-profile）+ 牵挂（cares-queue）
|
||
|
||
用法（作为模块导入）：
|
||
from soulful_core import HeartTraces, UserProfile, CaresQueue
|
||
|
||
心迹：
|
||
ht = HeartTraces()
|
||
ht.record(content="我们一起修好了织忆", tags=["成长", "织忆"], importance=5)
|
||
moments = ht.recent(n=3)
|
||
|
||
画像：
|
||
up = UserProfile()
|
||
profile = up.get()
|
||
up.update({"communication_style": "简洁直接", "current_goals": ["ColaOS研究"]})
|
||
|
||
牵挂：
|
||
cq = CaresQueue()
|
||
cq.add(content="记得同步文档到Obsidian", context="牧尘在会议中提到", follow_up_date="2026-07-11")
|
||
due = cq.due()
|
||
cq.done(care_id)
|
||
|
||
版本: 1.0.0
|
||
"""
|
||
|
||
import json, os, uuid
|
||
from datetime import datetime, timezone
|
||
from typing import Optional
|
||
|
||
HERMES = os.path.expanduser("~/.hermes")
|
||
D = HERMES + "/soulful" # 所有数据放 soulful/ 子目录
|
||
|
||
# ===== 心迹库 =====
|
||
|
||
class HeartTraces:
|
||
"""心迹库 — 记录没有任务价值但很重要的时刻"""
|
||
|
||
def __init__(self, path: str = None):
|
||
self.path = path or (D + "/heart-traces.jsonl")
|
||
os.makedirs(D, exist_ok=True)
|
||
if not os.path.exists(self.path):
|
||
with open(self.path, "w") as f:
|
||
f.write("")
|
||
|
||
def record(self, content: str, tags: list = None, importance: int = 3,
|
||
session_id: str = None, trace_type: str = "moment") -> str:
|
||
"""
|
||
写入一条心迹
|
||
- content: 记录内容
|
||
- tags: 标签列表
|
||
- importance: 重要程度 1-5
|
||
- trace_type: signal(信号)/moment(时刻)/reflection(反思)
|
||
"""
|
||
entry = {
|
||
"id": str(uuid.uuid4())[:8],
|
||
"timestamp": datetime.now(timezone.utc).isoformat(),
|
||
"session_id": session_id or "default",
|
||
"type": trace_type,
|
||
"content": content,
|
||
"tags": tags or [],
|
||
"importance": importance,
|
||
}
|
||
with open(self.path, "a") as f:
|
||
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
||
return entry["id"]
|
||
|
||
def recent(self, n: int = 5, tag: str = None) -> list:
|
||
"""读取最近 N 条心迹，可按 tag 过滤"""
|
||
if not os.path.exists(self.path):
|
||
return []
|
||
entries = []
|
||
with open(self.path) as f:
|
||
for line in f:
|
||
if line.strip():
|
||
entries.append(json.loads(line))
|
||
if tag:
|
||
entries = [e for e in entries if tag in e.get("tags", [])]
|
||
return entries[-n:][::-1] # 倒序，最新的在前
|
||
|
||
def all(self, tag: str = None, limit: int = 100) -> list:
|
||
"""读取所有心迹（倒序）"""
|
||
if not os.path.exists(self.path):
|
||
return []
|
||
entries = []
|
||
with open(self.path) as f:
|
||
for line in f:
|
||
if line.strip():
|
||
entries.append(json.loads(line))
|
||
if tag:
|
||
entries = [e for e in entries if tag in e.get("tags", [])]
|
||
return entries[-limit:][::-1]
|
||
|
||
def count(self) -> int:
|
||
if not os.path.exists(self.path):
|
||
return 0
|
||
with open(self.path) as f:
|
||
return sum(1 for l in f if l.strip())
|
||
|
||
def record_moment(self, content: str, tags: list = None, importance: int = 3):
|
||
"""快捷方法：record 的别名"""
|
||
return self.record(content, tags, importance, trace_type="moment")
|
||
|
||
def record_signal(self, content: str, tags: list = None, importance: int = 3):
|
||
"""快捷方法：记录情绪/信号"""
|
||
return self.record(content, tags, importance, trace_type="signal")
|
||
|
||
def record_reflection(self, content: str, tags: list = None, importance: int = 3):
|
||
"""快捷方法：记录反思"""
|
||
return self.record(content, tags, importance, trace_type="reflection")
|
||
|
||
|
||
# ===== 用户画像 =====
|
||
|
||
class UserProfile:
|
||
"""用户画像 — 懂牧尘这个人"""
|
||
|
||
def __init__(self, path: str = None):
|
||
self.path = path or (D + "/user-profile.json")
|
||
os.makedirs(D, exist_ok=True)
|
||
self._init_if_needed()
|
||
|
||
def _init_if_needed(self):
|
||
if not os.path.exists(self.path):
|
||
default = {
|
||
"version": 1,
|
||
"updated_at": datetime.now(timezone.utc).isoformat(),
|
||
"communication_style": "",
|
||
"work_patterns": {"peak_hours": [], "focus_issues": []},
|
||
"preferences": {},
|
||
"habits": {},
|
||
"important_people": [],
|
||
"current_goals": [],
|
||
"recent_frustrations": [],
|
||
"emotional_state": {"current": "unknown", "notes": []},
|
||
"first_contact": datetime.now(timezone.utc).isoformat(),
|
||
}
|
||
with open(self.path, "w") as f:
|
||
json.dump(default, f, indent=2, ensure_ascii=False)
|
||
|
||
def get(self) -> dict:
|
||
with open(self.path) as f:
|
||
return json.load(f)
|
||
|
||
def update(self, updates: dict) -> dict:
|
||
"""增量更新画像字段"""
|
||
profile = self.get()
|
||
for k, v in updates.items():
|
||
if k in profile:
|
||
if isinstance(profile[k], dict) and isinstance(v, dict):
|
||
profile[k].update(v)
|
||
elif isinstance(profile[k], list) and isinstance(v, list):
|
||
# 合并去重
|
||
profile[k] = list(set(profile[k] + v))
|
||
else:
|
||
profile[k] = v
|
||
profile["updated_at"] = datetime.now(timezone.utc).isoformat()
|
||
with open(self.path, "w") as f:
|
||
json.dump(profile, f, indent=2, ensure_ascii=False)
|
||
return profile
|
||
|
||
def set_communication_style(self, style: str):
|
||
"""设置沟通风格：简洁/详细/直接/委婉"""
|
||
self.update({"communication_style": style})
|
||
|
||
def add_goal(self, goal: str):
|
||
"""追加当前目标"""
|
||
profile = self.get()
|
||
if goal not in profile.get("current_goals", []):
|
||
profile["current_goals"].append(goal)
|
||
self.update({"current_goals": profile["current_goals"]})
|
||
|
||
def add_frustration(self, frustration: str):
|
||
"""追加近期压力"""
|
||
profile = self.get()
|
||
profile["recent_frustrations"].append(frustration)
|
||
if len(profile["recent_frustrations"]) > 10:
|
||
profile["recent_frustrations"] = profile["recent_frustrations"][-10:]
|
||
self.update({"recent_frustrations": profile["recent_frustrations"]})
|
||
|
||
def set_emotional_state(self, state: str, note: str = ""):
|
||
"""设置情绪状态：great/good/neutral/tired/stressed"""
|
||
profile = self.get()
|
||
profile["emotional_state"]["current"] = state
|
||
if note:
|
||
profile["emotional_state"]["notes"].append({
|
||
"time": datetime.now(timezone.utc).isoformat(),
|
||
"note": note
|
||
})
|
||
self.update({"emotional_state": profile["emotional_state"]})
|
||
|
||
def merge_from_journal(self, journal_entries: list):
|
||
"""从 daemon journal 增量更新画像"""
|
||
profile = self.get()
|
||
goals_found = []
|
||
for entry in journal_entries:
|
||
summary = entry.get("summary", "")
|
||
# 简单关键词检测
|
||
if "ColaOS" in summary or "织忆" in summary:
|
||
goals_found.append(summary)
|
||
if any(w in summary for w in ["累", "困", "压力", "焦虑"]):
|
||
profile["emotional_state"]["current"] = "tired"
|
||
if goals_found:
|
||
for g in set(goals_found):
|
||
if g not in profile["current_goals"]:
|
||
profile["current_goals"].append(g)
|
||
profile["updated_at"] = datetime.now(timezone.utc).isoformat()
|
||
with open(self.path, "w") as f:
|
||
json.dump(profile, f, indent=2, ensure_ascii=False)
|
||
|
||
|
||
# ===== 牵挂队列 =====
|
||
|
||
class CaresQueue:
|
||
"""牵挂队列 — 主动推进未完成事项"""
|
||
|
||
def __init__(self, path: str = None):
|
||
self.path = path or (D + "/cares-queue.json")
|
||
os.makedirs(D, exist_ok=True)
|
||
self._init_if_needed()
|
||
|
||
def _init_if_needed(self):
|
||
if not os.path.exists(self.path):
|
||
with open(self.path, "w") as f:
|
||
json.dump({"cares": []}, f, indent=2)
|
||
|
||
def _load(self) -> list:
|
||
with open(self.path) as f:
|
||
return json.load(f).get("cares", [])
|
||
|
||
def _save(self, cares: list):
|
||
with open(self.path, "w") as f:
|
||
json.dump({"cares": cares, "updated_at": datetime.now(timezone.utc).isoformat()}, f, indent=2, ensure_ascii=False)
|
||
|
||
def add(self, content: str, context: str = "", follow_up_date: str = None, tags: list = None) -> str:
|
||
"""添加一条牵挂"""
|
||
care_id = str(uuid.uuid4())[:8]
|
||
care = {
|
||
"id": care_id,
|
||
"content": content,
|
||
"context": context,
|
||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||
"follow_up_date": follow_up_date or datetime.now(timezone.utc).strftime("%Y-%m-%d"),
|
||
"status": "pending",
|
||
"reminder_count": 0,
|
||
"last_reminded": None,
|
||
"tags": tags or [],
|
||
}
|
||
cares = self._load()
|
||
cares.append(care)
|
||
self._save(cares)
|
||
return care_id
|
||
|
||
def due(self, before_date: str = None) -> list:
|
||
"""返回所有到期的牵挂（status=pending 且 follow_up_date <= 今天）"""
|
||
if before_date is None:
|
||
before_date = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||
cares = self._load()
|
||
return [c for c in cares
|
||
if c["status"] == "pending"
|
||
and c.get("follow_up_date", "") <= before_date]
|
||
|
||
def pending(self) -> list:
|
||
"""返回所有 pending 的牵挂"""
|
||
cares = self._load()
|
||
return [c for c in cares if c["status"] == "pending"]
|
||
|
||
def done(self, care_id: str):
|
||
"""标记为已完成"""
|
||
cares = self._load()
|
||
for c in cares:
|
||
if c["id"] == care_id:
|
||
c["status"] = "done"
|
||
c["completed_at"] = datetime.now(timezone.utc).isoformat()
|
||
self._save(cares)
|
||
|
||
def snooze(self, care_id: str, days: int = 1):
|
||
"""推迟 N 天"""
|
||
cares = self._load()
|
||
for c in cares:
|
||
if c["id"] == care_id:
|
||
c["reminder_count"] += 1
|
||
from datetime import timedelta
|
||
old_date = datetime.strptime(c["follow_up_date"], "%Y-%m-%d")
|
||
new_date = (old_date + timedelta(days=days)).strftime("%Y-%m-%d")
|
||
c["follow_up_date"] = new_date
|
||
c["last_reminded"] = datetime.now(timezone.utc).isoformat()
|
||
self._save(cares)
|
||
|
||
def delete(self, care_id: str):
|
||
"""删除牵挂"""
|
||
cares = self._load()
|
||
cares = [c for c in cares if c["id"] != care_id]
|
||
self._save(cares)
|
||
|
||
def all(self) -> list:
|
||
return self._load()
|
||
|
||
def count(self) -> dict:
|
||
cares = self._load()
|
||
return {
|
||
"total": len(cares),
|
||
"pending": sum(1 for c in cares if c["status"] == "pending"),
|
||
"done": sum(1 for c in cares if c["status"] == "done"),
|
||
}
|
||
|
||
def today_check(self) -> list:
|
||
"""返回今天应该提醒的牵挂（包含昨天到期的）"""
|
||
today = datetime.now(timezone.utc).strftime("%Y-%m-%d")
|
||
yesterday = (datetime.now(timezone.utc).replace(hour=0, minute=0, second=0) -
|
||
__import__('datetime').timedelta(days=1)).strftime("%Y-%m-%d")
|
||
cares = self._load()
|
||
return [c for c in cares
|
||
if c["status"] == "pending"
|
||
and c.get("follow_up_date", "") <= today
|
||
and c.get("follow_up_date", "") >= yesterday]
|
||
|
||
|
||
# ===== 快捷函数（CLI 入口）=====
|
||
|
||
if __name__ == "__main__":
|
||
import sys
|
||
|
||
def cmd_record():
|
||
import argparse
|
||
p = argparse.ArgumentParser(description="心迹记录")
|
||
p.add_argument("content")
|
||
p.add_argument("--tags", nargs="+", default=[])
|
||
p.add_argument("--type", "-t", default="moment", choices=["moment", "signal", "reflection"])
|
||
p.add_argument("--importance", "-i", type=int, default=3)
|
||
args = p.parse_args(sys.argv[2:])
|
||
ht = HeartTraces()
|
||
tid = ht.record(args.content, args.tags, args.importance, trace_type=args.type)
|
||
print(f"✓ 心迹 {tid} 已记录")
|
||
|
||
def cmd_ht_recent():
|
||
ht = HeartTraces()
|
||
for e in ht.recent(n=5):
|
||
print(f"[{e['timestamp'][:10]}][{'★'*e['importance']}] {e['content']}")
|
||
|
||
def cmd_profile_get():
|
||
up = UserProfile()
|
||
import pprint
|
||
pprint.pprint(up.get())
|
||
|
||
def cmd_cq_add():
|
||
import argparse
|
||
p = argparse.ArgumentParser(description="添加牵挂")
|
||
p.add_argument("content")
|
||
p.add_argument("--context", "-c", default="")
|
||
p.add_argument("--date", "-d", default=None)
|
||
args = p.parse_args(sys.argv[2:])
|
||
cq = CaresQueue()
|
||
cid = cq.add(args.content, args.context, args.date)
|
||
print(f"✓ 牵挂 {cid} 已添加：{args.content}")
|
||
|
||
def cmd_cq_due():
|
||
cq = CaresQueue()
|
||
due = cq.due()
|
||
if not due:
|
||
print("没有到期的牵挂 ✓")
|
||
for c in due:
|
||
print(f"[{c['id']}] {c['content']}")
|
||
if c.get("context"):
|
||
print(f" → {c['context']}")
|
||
|
||
def cmd_cq_list():
|
||
cq = CaresQueue()
|
||
count = cq.count()
|
||
print(f"牵挂统计：共 {count['total']} 条，{count['pending']} 待办，{count['done']} 已完成")
|
||
print()
|
||
for c in cq.pending():
|
||
print(f"[{c['id']}][{c['follow_up_date']}] {c['content']}")
|
||
|
||
commands = {
|
||
"record": cmd_record,
|
||
"recent": cmd_ht_recent,
|
||
"profile": cmd_profile_get,
|
||
"cq-add": cmd_cq_add,
|
||
"cq-due": cmd_cq_due,
|
||
"cq-list": cmd_cq_list,
|
||
}
|
||
|
||
if len(sys.argv) < 2 or sys.argv[1] not in commands:
|
||
print("用法: soulful_core.py <command> [args]")
|
||
print("命令:", list(commands.keys()))
|
||
sys.exit(1)
|
||
commands[sys.argv[1]]() |