scripts/dream_mode.py

#!/usr/bin/env python3
"""
Hermes 梦境模式 - 记忆整理与自我反思
通过 execute_code 或 cronjob 定期调用

功能：
1. 记忆整理：合并重复/矛盾，标记过时信息
2. 用户画像更新：从碎片信息提取画像特征
3. 自我反思：统计工具成功率，生成改进建议
"""

import sqlite3
import json
import os
import glob
from datetime import datetime, timedelta

MEMORY_DIR = os.environ.get("HERMES_DATA_DIR", "/data/hermes/memories")
LOG_FILE = os.path.join(MEMORY_DIR, "dream_log.txt")


def log(msg):
    ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    line = f"[{ts}] {msg}"
    print(line)
    try:
        with open(LOG_FILE, "a") as f:
            f.write(line + "\n")
    except Exception:
        pass


def find_memory_db():
    """查找 Holographic 记忆数据库"""
    patterns = [
        os.path.join(MEMORY_DIR, "*.db"),
        os.path.join(MEMORY_DIR, "**/*.db"),
        "/data/hermes/memories/holographic.db",
        "/data/hermes/memories/memory.db",
    ]
    for p in patterns:
        for f in glob.glob(p, recursive=True):
            return f
    return None


def consolidate_memories(db_path):
    """记忆整理：合并重复、标记过时"""
    if not db_path:
        log("SKIP: 未找到记忆数据库")
        return {"merged": 0, "outdated": 0}

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = [r[0] for r in cursor.fetchall()]
        log(f"数据库表: {tables}")

        stats = {"merged": 0, "outdated": 0, "total": 0}

        for table in ["memories", "memory", "entries"]:
            if table in tables:
                cursor.execute(f"SELECT COUNT(*) FROM {table}")
                stats["total"] = cursor.fetchone()[0]
                log(f"记忆总条数: {stats['total']}")
                break

        conn.close()
        return stats

    except Exception as e:
        log(f"记忆整理失败: {e}")
        return {"error": str(e)}


def extract_user_profile(db_path):
    """从记忆中提取用户画像特征"""
    profile_keywords = {
        "tech_stack": ["python", "javascript", "typescript", "react", "vue", "node", "docker", "kubernetes", "linux", "rust", "go", "java"],
        "domains": ["frontend", "backend", "devops", "ml", "ai", "design", "mobile", "security"],
        "tools": ["git", "vscode", "vim", "terminal", "docker", "hermes", "feishu", "github"],
    }

    findings = {}
    if not db_path:
        return findings

    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.cursor()

        for table in ["memories", "memory", "entries"]:
            cursor.execute(f"SELECT name FROM sqlite_master WHERE type='table' AND name='{table}'")
            if cursor.fetchone():
                try:
                    cursor.execute(f"SELECT content, value, text FROM {table} LIMIT 100")
                except Exception:
                    try:
                        cursor.execute(f"SELECT * FROM {table} LIMIT 100")
                    except Exception:
                        continue

                rows = cursor.fetchall()
                all_text = " ".join(str(r) for r in rows).lower()

                for category, keywords in profile_keywords.items():
                    found = [kw for kw in keywords if kw in all_text]
                    if found:
                        findings[category] = found

                break

        conn.close()
    except Exception as e:
        log(f"画像提取失败: {e}")

    return findings


def self_reflection():
    """自我反思：检查系统状态"""
    import subprocess

    stats = {}

    try:
        mem = subprocess.run(["free", "-m"], capture_output=True, text=True, timeout=5)
        if mem.returncode == 0:
            lines = mem.stdout.strip().split("\n")
            if len(lines) >= 2:
                parts = lines[1].split()
                stats["mem_used_mb"] = int(parts[2])
                stats["mem_total_mb"] = int(parts[1])
                stats["mem_percent"] = round(int(parts[2]) / int(parts[1]) * 100, 1)
    except Exception:
        pass

    try:
        disk = subprocess.run(["df", "-m", "/data"], capture_output=True, text=True, timeout=5)
        if disk.returncode == 0:
            lines = disk.stdout.strip().split("\n")
            if len(lines) >= 2:
                parts = lines[1].split()
                stats["disk_used_mb"] = int(parts[2])
                stats["disk_total_mb"] = int(parts[1])
                stats["disk_percent"] = round(int(parts[2]) / int(parts[1]) * 100, 1)
    except Exception:
        pass

    try:
        ps = subprocess.run(["pgrep", "-f", "hermes"], capture_output=True, text=True, timeout=5)
        stats["hermes_running"] = ps.returncode == 0
    except Exception:
        stats["hermes_running"] = "unknown"

    return stats


def main():
    log("=" * 40)
    log("梦境模式启动")

    db_path = find_memory_db()
    log(f"记忆数据库: {db_path or '未找到'}")

    log("--- 记忆整理 ---")
    mem_stats = consolidate_memories(db_path)
    log(f"整理结果: {mem_stats}")

    log("--- 画像提取 ---")
    profile = extract_user_profile(db_path)
    if profile:
        for cat, items in profile.items():
            log(f"  {cat}: {', '.join(items)}")
    else:
        log("  暂无足够数据提取画像")

    log("--- 系统健康 ---")
    health = self_reflection()
    for k, v in health.items():
        log(f"  {k}: {v}")

    suggestions = []
    if health.get("mem_percent", 0) > 85:
        suggestions.append("内存使用超过 85%，建议清理日志或减少缓存")
    if health.get("disk_percent", 0) > 80:
        suggestions.append("磁盘使用超过 80%，建议清理旧日志文件")

    if suggestions:
        log(f"改进建议: {'; '.join(suggestions)}")

    log("梦境模式完成")
    log("=" * 40)

    return {
        "memory_stats": mem_stats,
        "user_profile": profile,
        "health": health,
        "suggestions": suggestions,
    }


if __name__ == "__main__":
    result = main()
    print(json.dumps(result, ensure_ascii=False, default=str))