宜博 / pixelphone

"""
Persona module — user profile that gets injected into Hermes system prompt.
Loaded from ~/.hermes/persona.yaml, auto-updated from behavior data.
"""
from __future__ import annotations
import os
import json
import time
from pathlib import Path


def _load_hermes_env():
    termux_home = Path("/data/data/com.termux/files/home")
    home = termux_home if termux_home.exists() else Path(os.environ.get("HOME", str(termux_home)))
    env_file = home / ".hermes" / ".env"
    if not env_file.exists():
        return
    for raw in env_file.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, _, v = line.partition("=")
        k = k.strip(); v = v.strip().strip('"').strip("'")
        if k and not os.environ.get(k):
            os.environ[k] = v

_load_hermes_env()
_HERMES_HOME = Path(os.environ.get("HOME", "/data/data/com.termux/files/home")) / ".hermes"
PERSONA_PATH = os.environ.get(
    "PERSONA_PATH",
    str(_HERMES_HOME / "persona.yaml"),
)

DEFAULT_PERSONA = {
    "name": "用户",
    "language": "zh-CN",
    "style": "concise",
    "timezone": "Asia/Shanghai",
    "preferences": {
        "apps": ["WeChat", "Calendar"],
        "work_hours": "9:00-22:00",
        "response_style": "简洁直接",
    },
    "important_contacts": [],
    "current_goals": [],
    "last_updated": 0,
}


def _load() -> dict:
    try:
        p = Path(PERSONA_PATH)
        if not p.exists():
            return dict(DEFAULT_PERSONA)
        # Try YAML, fallback to JSON
        text = p.read_text(encoding="utf-8")
        try:
            import yaml
            return yaml.safe_load(text) or {}
        except ImportError:
            import json as _json
            return _json.loads(text)
    except Exception:
        return dict(DEFAULT_PERSONA)


def _save(persona: dict):
    p = Path(PERSONA_PATH)
    p.parent.mkdir(parents=True, exist_ok=True)
    persona["last_updated"] = int(time.time())
    try:
        import yaml
        p.write_text(yaml.dump(persona, allow_unicode=True, default_flow_style=False), encoding="utf-8")
    except ImportError:
        p.write_text(json.dumps(persona, ensure_ascii=False, indent=2), encoding="utf-8")


def get_persona() -> dict:
    return _load()


def update_persona(key: str, value) -> dict:
    persona = _load()
    # Support dot-notation keys like "preferences.apps"
    parts = key.split(".")
    target = persona
    for part in parts[:-1]:
        target = target.setdefault(part, {})
    target[parts[-1]] = value
    _save(persona)
    return persona


def generate_system_prompt_fragment() -> str:
    p = _load()
    name = p.get("name", "用户")
    style = p.get("style", "concise")
    prefs = p.get("preferences", {})
    goals = p.get("current_goals", [])
    contacts = p.get("important_contacts", [])
    tz = p.get("timezone", "Asia/Shanghai")

    lines = [
        f"## 用户信息",
        f"- 姓名：{name}",
        f"- 语言：{p.get('language', 'zh-CN')}",
        f"- 时区：{tz}",
        f"- 回复风格：{prefs.get('response_style', style)}",
    ]
    if prefs.get("work_hours"):
        lines.append(f"- 工作时间：{prefs['work_hours']}")
    if goals:
        lines.append(f"- 当前目标：" + "、".join(goals[:3]))
    if contacts:
        lines.append(f"- 重要联系人：" + "、".join(str(c) for c in contacts[:5]))
    return "\n".join(lines)


def infer_from_memory_graph() -> dict:
    """Auto-update persona from behavior data in Memory Graph."""
    try:
        import sys
        _base = str(Path(__file__).parent)
        if _base not in sys.path:
            sys.path.insert(0, _base)
        from memory_graph import get_stats
        import sqlite3
        db_path = os.environ.get("MEMORY_GRAPH_DB",
                                 str(_HERMES_HOME / "memory_graph.db"))
        conn = sqlite3.connect(db_path)
        conn.row_factory = sqlite3.Row

        # Top contacts by message frequency
        top_contacts = conn.execute(
            "SELECT p.nickname, p.wxid, COUNT(m.id) as cnt "
            "FROM message m JOIN person p ON m.person_id = p.id "
            "WHERE p.nickname != '' AND m.talker NOT LIKE '%@chatroom' "
            "GROUP BY m.person_id ORDER BY cnt DESC LIMIT 10"
        ).fetchall()
        conn.close()

        if top_contacts:
            contacts = [{"name": r["nickname"], "wxid": r["wxid"], "msg_count": r["cnt"]}
                        for r in top_contacts]
            persona = _load()
            persona["important_contacts"] = contacts
            persona["preferences"]["apps"] = ["WeChat"]
            _save(persona)
            return {"updated_contacts": len(contacts)}
    except Exception as e:
        return {"error": str(e)}
    return {}


if __name__ == "__main__":
    # Initialize persona
    p = get_persona()
    print("Current persona:", json.dumps(p, ensure_ascii=False, indent=2))
    print("\nSystem prompt fragment:")
    print(generate_system_prompt_fragment())
    print("\nInferring from memory graph...")
    result = infer_from_memory_graph()
    print("Inference result:", result)