zhiyi/scripts/migrate_hermes_memory.py

#!/usr/bin/env python3
"""
批量迁移 Hermes 记忆到织忆。
- 读取 ~/.hermes/memory_db/memory.fts.db
- 直接写入 Redis + JSONL（绕过 commit 的索引重建）
- 最后一次性全量重建索引
"""
import sqlite3, json, os, time, uuid
from pathlib import Path
import sys

# Add zhiyi src to path
ZHIYI_ROOT = Path("/home/muc/projects/zhiyi")
sys.path.insert(0, str(ZHIYI_ROOT / "src"))

MEMORY_DB = Path.home() / ".hermes/memory_db/memory.fts.db"
ZHIYI_DATA = ZHIYI_ROOT / "data"
EPISODES_DIR = Path.home() / ".memory-fabric/zhiyi/episodes"
DISTILLED_DIR = Path.home() / ".memory-fabric/zhiyi/distilled"

# Redis config
REDIS_HOST = "localhost"
REDIS_PORT = 6379

SKIP_PATTERNS = [
    "Context Compaction", "--- END OF", "## Active Task", "## Relevant Memory",
    "Review the conversation above", "[System note:", "造成这种情况的原因",
]

def is_valid(content: str) -> bool:
    if not content or len(content.strip()) < 10:
        return False
    for pat in SKIP_PATTERNS:
        if pat in content:
            return False
    return True

def get_tag_category(tag: str) -> str:
    """Map Hermes tag to ZhiYi category."""
    if tag in ("偏好", "配置", "环境", "认知突破"):
        return "distilled"
    return "distilled"

def load_hermes_memories():
    """Read all memories from Hermes FTS db."""
    conn = sqlite3.connect(str(MEMORY_DB))
    conn.row_factory = sqlite3.Row
    cur = conn.cursor()
    cur.execute("SELECT content, tag, at FROM memory ORDER BY at DESC")
    rows = cur.fetchall()
    conn.close()
    return [dict(r) for r in rows]

def write_jsonl(path: Path, doc: dict):
    """Append one doc to JSONL file."""
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "a", encoding="utf-8") as f:
        f.write(json.dumps(doc, ensure_ascii=False) + "\n")

def write_redis(episode_id: str, doc: dict):
    """Write to Redis hash."""
    import redis
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
    key = f"commit:{episode_id}"
    r.hset(key, mapping={
        "id": episode_id,
        "content": doc["content"],
        "category": doc.get("category", "distilled"),
        "tags": doc.get("tags", ""),
        "created_at": doc.get("created_at", ""),
    })
    r.expire(key, 86400 * 30)  # 30 day TTL

def publish_event(episode_id: str, category: str):
    """Publish event to Redis Streams."""
    import redis
    r = redis.Redis(host=REDIS_HOST, port=REDIS_PORT, decode_responses=True)
    r.xadd("zhiyi:events", {
        "type": "commit",
        "episode_id": episode_id,
        "category": category,
    })

def migrate():
    rows = load_hermes_memories()
    print(f"Loaded {len(rows)} entries from Hermes memory")

    # Filter valid entries
    valid = [r for r in rows if is_valid(r["content"])]
    print(f"Valid entries: {len(valid)} (skipped {len(rows) - len(valid)})")

    # Get existing episode IDs to avoid duplicates (check JSONL)
    existing_ids = set()
    for f in DISTILLED_DIR.glob("*.jsonl"):
        for line in open(f, encoding="utf-8"):
            try:
                d = json.loads(line)
                if "episode_id" in d:
                    existing_ids.add(d["episode_id"])
            except:
                pass

    print(f"Existing distilled entries: {len(existing_ids)}")

    imported = skipped_dup = 0
    for i, row in enumerate(valid):
        content = row["content"].strip()
        tag = row["tag"] or "general"
        created_at = row["at"]

        # Generate stable ID from content hash
        import hashlib
        h = hashlib.sha256(content.encode()).hexdigest()[:16]
        episode_id = f"hermes-mig-{h}"

        if episode_id in existing_ids:
            skipped_dup += 1
            continue

        category = get_tag_category(tag)
        doc = {
            "episode_id": episode_id,
            "content": content,
            "category": category,
            "tags": tag,
            "created_at": created_at,
        }

        # Write to JSONL
        if category == "distilled":
            jsonl_path = DISTILLED_DIR / f"hermes-migrated-{created_at[:7]}.jsonl"
        else:
            jsonl_path = EPISODES_DIR / f"hermes-migrated-{created_at[:7]}.jsonl"

        write_jsonl(jsonl_path, doc)

        # Write to Redis
        try:
            write_redis(episode_id, doc)
            publish_event(episode_id, category)
        except Exception as e:
            print(f"  Redis error: {e}")

        imported += 1
        if (i + 1) % 50 == 0:
            print(f"  {i+1}/{len(valid)} processed, imported={imported}")

    print(f"\nMigration done: {imported} imported, {skipped_dup} duplicate skipped")

    # Save migration metadata
    meta = {
        "migrated_at": time.strftime("%Y-%m-%dT%H:%M:%S"),
        "total_loaded": len(rows),
        "valid": len(valid),
        "imported": imported,
        "skipped_dup": skipped_dup,
        "source": "hermes-memory-db",
    }
    print(f"Metadata: {json.dumps(meta, ensure_ascii=False)}")

    # Now trigger one final index rebuild
    print("\nTriggering index rebuild...")
    import requests
    try:
        r = requests.post("http://localhost:7821/api/v1/rebuild-index", timeout=60)
        print(f"Rebuild: {r.status_code} {r.text[:200]}")
    except Exception as e:
        print(f"Rebuild trigger failed: {e}")
        print("NOTE: You may need to restart ZhiYi to see migrated memories in recall")

    print("\nMigration complete!")
    return meta

if __name__ == "__main__":
    migrate()