migrate_sessions.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/tools/migrate_sessions.py
 
"""
One-time migration script — legacy session-file relocation.
 
Purpose
-------
Moves legacy files into the new ``oc_chat/sessions/`` structure so
older installs upgrade cleanly:
 
1. Root-level ``chat_log_*.json`` → ``oc_chat/sessions/Chat_logs/``
2. Root-level ``sessions/*.json`` → ``oc_chat/sessions/`` (as
   lightweight metadata files)
 
Run with: ``python -m oc_chat.tools.migrate_sessions``. Safe to
re-run — files already in the target structure are skipped.
 
Responsibilities
----------------
- :func:`_ensure_dirs` — create target directories idempotently.
- :func:`_migrate_root_chat_logs` — relocate legacy chat-log files.
- :func:`_migrate_old_sessions_dir` — relocate the legacy
  ``sessions/`` directory.
- :func:`main` — CLI entry point that prints a human-readable summary.
 
Diagnostics
-----------
Domain: OC_CHAT.TOOLS
Levels:
    L2 — lifecycle events
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Idempotent: running twice on an already-migrated tree is a no-op.
- Never overwrites a destination file silently — duplicate names are
  preserved by skipping with a warning.
- All ``print()`` calls are user-facing CLI output, not diagnostic.
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
 
import json
import shutil
from pathlib import Path
 
from oc_diagnostics import _dbg
 
PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent  # three levels up from tools/
NEW_SESSIONS_DIR = PROJECT_ROOT / "oc_chat" / "sessions"
NEW_CHAT_LOGS_DIR = NEW_SESSIONS_DIR / "Chat_logs"
OLD_SESSIONS_DIR = PROJECT_ROOT / "sessions"
 
 
def _ensure_dirs():
    """Create target directories if they don't exist."""
    NEW_SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
    NEW_CHAT_LOGS_DIR.mkdir(parents=True, exist_ok=True)
 
 
def _migrate_root_chat_logs() -> list[Path]:
    """Move ``chat_log_*.json`` from project root into ``Chat_logs/``."""
    migrated = []
    for f in sorted(PROJECT_ROOT.glob("chat_log_*.json")):
        dest = NEW_CHAT_LOGS_DIR / f.name
        if dest.exists():
            print(f"  [skip] {f.name} — already in Chat_logs/")
            continue
        shutil.move(str(f), str(dest))
        print(f"  [move] {f.name} → oc_chat/sessions/Chat_logs/{f.name}")
        migrated.append(dest)
    return migrated
 
 
def _convert_old_session_to_metadata(old_path: Path) -> dict | None:
    """Convert a legacy session JSON (full conversation) to lightweight metadata format."""
    try:
        with open(old_path, "r", encoding="utf-8") as f:
            data = json.load(f)
    except (json.JSONDecodeError, OSError) as e:
        print(f"  [warn] Could not read {old_path.name}: {e}")
        return None
 
    messages = data.get("messages", [])
    msg_count = sum(1 for m in messages if m.get("role") in ("user", "assistant"))
    tool_count = sum(1 for m in messages if m.get("role") == "tool")
 
    # Estimate token usage from messages if available
    total_prompt = 0
    total_completion = 0
    total_tokens = 0
    for m in messages:
        usage = m.get("usage")
        if usage:
            total_prompt += usage.get("prompt_tokens", 0)
            total_completion += usage.get("completion_tokens", 0)
            total_tokens += usage.get("total_tokens", 0)
 
    # Determine a chat log name based on the session
    session_name = data.get("session_name", old_path.stem)
 
    metadata = {
        "session_name": session_name,
        "created": data.get("timestamp", session_name.split("_", 1)[1] if "_" in session_name else "unknown"),
        "last_modified": data.get("timestamp", "unknown"),
        "model": data.get("model", "unknown"),
        "reasoning_effort": data.get("reasoning_effort", "high"),
        "system_prompt": messages[0]["content"] if messages and messages[0].get("role") == "system" else "",
        "stats": {
            "message_count": msg_count,
            "tool_call_count": tool_count,
            "total_prompt_tokens": total_prompt,
            "total_completion_tokens": total_completion,
            "total_tokens": total_tokens,
        },
        "chat_logs": [
            f"Chat_logs/{session_name}_log.json"
        ],
    }
    return metadata
 
 
def _migrate_old_sessions_dir() -> list[Path]:
    """Convert old ``sessions/*.json`` → new lightweight metadata in ``oc_chat/sessions/`` and move messages as chat logs."""
    func = "_migrate_old_sessions_dir"
    if not OLD_SESSIONS_DIR.is_dir():
        print("  [info] No old sessions/ directory found — nothing to migrate.")
        return []
 
    migrated = []
    for f in sorted(OLD_SESSIONS_DIR.glob("*.json")):
        metadata = _convert_old_session_to_metadata(f)
        if metadata is None:
            continue
 
        # Write the lightweight session metadata file
        meta_path = NEW_SESSIONS_DIR / f.name
        with open(meta_path, "w", encoding="utf-8") as mf:
            json.dump(metadata, mf, indent=2, ensure_ascii=False)
        print(f"  [meta] {f.name} → oc_chat/sessions/{f.name} (lightweight metadata)")
 
        # Also save the full conversation as a chat log
        try:
            with open(f, "r", encoding="utf-8") as of:
                old_data = json.load(of)
        except (json.JSONDecodeError, OSError) as e:
            # VALIDATION_ERROR — legacy file is malformed/unreadable; keep migrating with empty body
            _dbg("OC_CHAT.TOOLS", func, f"VALIDATION_ERROR: {f.name}: {type(e).__name__}: {e}", 3)
            old_data = {}
 
        log_name = f"{metadata['session_name']}_log.json"
        log_path = NEW_CHAT_LOGS_DIR / log_name
 
        # Add a message_stats entry to the chat log
        log_data = {
            "session_name": metadata["session_name"],
            "log_index": 1,
            "timestamp": metadata["created"],
            "model": metadata["model"],
            "reasoning_effort": metadata["reasoning_effort"],
            "messages": old_data.get("messages", []),
            "usage": {
                "prompt_tokens": metadata["stats"]["total_prompt_tokens"],
                "completion_tokens": metadata["stats"]["total_completion_tokens"],
                "total_tokens": metadata["stats"]["total_tokens"],
            },
        }
 
        with open(log_path, "w", encoding="utf-8") as lf:
            json.dump(log_data, lf, indent=2, ensure_ascii=False)
        print(f"  [log]  {f.name} → oc_chat/sessions/Chat_logs/{log_name}")
 
        migrated.append(meta_path)
 
    return migrated
 
 
def main() -> None:
    """Run the one-time session migration."""
    func = "main"
    _dbg("OC_CHAT.TOOLS", func, f"source={OLD_SESSIONS_DIR} target={NEW_SESSIONS_DIR}", 2)
    print("=" * 60)
    print("Session Migration Tool")
    print(f"  Source sessions/: {OLD_SESSIONS_DIR}")
    print(f"  Source root chat_logs: {PROJECT_ROOT}")
    print(f"  Target: {NEW_SESSIONS_DIR}")
    print("=" * 60)
 
    _ensure_dirs()
 
    print("\n--- Migrating root chat_log_*.json files ---")
    chat_logs = _migrate_root_chat_logs()
    print(f"  Migrated {len(chat_logs)} chat log(s).")
 
    print("\n--- Converting old sessions/ to new format ---")
    sessions = _migrate_old_sessions_dir()
    print(f"  Migrated {len(sessions)} session(s).")
 
    # Cleanup: remove old sessions/ directory if empty
    if OLD_SESSIONS_DIR.is_dir() and not list(OLD_SESSIONS_DIR.iterdir()):
        OLD_SESSIONS_DIR.rmdir()
        print(f"\n  [cleanup] Removed empty old sessions/ directory.")
 
    print("\n✅ Migration complete!")
    print(f"   Sessions:    {NEW_SESSIONS_DIR}")
    print(f"   Chat logs:   {NEW_CHAT_LOGS_DIR}")
 
 
if __name__ == "__main__":
    main()