migrate_session_names.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/scripts/migrate_session_names.py
 
"""
Session name migration: timestamp format → random-word format.
 
Purpose
-------
One-shot offline script that renames timestamp-format sessions
(``session_YYYYMMDD_HHMMSS``) to readable random-word names
(e.g., ``sunny-weaving-honey``).
 
Responsibilities
----------------
- Detect sessions matching the old timestamp pattern
- Generate a fresh random-word name for each
- Rename the session directory on disk
- Rename all files inside (which embed the old name as a prefix)
- Update the ``session_name`` field in the ``_meta.json`` file
- Print a mapping table: old_name → new_name
- Support ``--dry-run`` (default) to preview without writing
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 — migration events (renamed, skipped, errors)
    L3 — per-file rename details
    L4 — per-path filesystem operation traces
 
Contracts
---------
- Idempotent: sessions already in random-word format are skipped
- ``--dry-run`` makes no filesystem changes
- Exits with code 0 on success, 1 on any unrecoverable error
- Never touches the archive/ or legacy/ subdirectories
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.core.wordgen (lazy; random_name, which generates each collision-free new session name)
- oc_chat.core.persistence (lazy; DEFAULT_SESSIONS_DIR, the default for --sessions-dir)
"""
from __future__ import annotations
 
import argparse
import json
import re
import shutil
import sys
import tempfile
from pathlib import Path
 
from oc_diagnostics import _dbg
 
# Pattern for the old auto-generated timestamp names
_TIMESTAMP_RE = re.compile(r"^session_\d{8}_\d{6}$")
 
# Reserved subdirectory names inside the sessions root
_SKIP_DIRS = {"archive", "legacy"}
 
 
def _load_json(path: Path) -> dict:
    """Read a JSON file, stripping UTF-8 BOM if present."""
    raw = path.read_bytes()
    if raw.startswith(b"\xef\xbb\xbf"):
        raw = raw[3:]
    return json.loads(raw.decode("utf-8"))
 
 
def _write_json_atomic(path: Path, data: dict) -> None:
    """Write JSON atomically via a temp file + rename."""
    with tempfile.NamedTemporaryFile(
        mode="w", encoding="utf-8", dir=path.parent,
        suffix=".tmp", delete=False
    ) as tmp:
        json.dump(data, tmp, indent=2, ensure_ascii=False)
        tmp_path = Path(tmp.name)
    tmp_path.replace(path)
 
 
def _find_meta_file(session_dir: Path) -> Path | None:
    """Return the most recently modified *_meta.json in session_dir, or None."""
    candidates = sorted(session_dir.glob("*_meta.json"), key=lambda p: p.stat().st_mtime)
    return candidates[-1] if candidates else None
 
 
def _collect_sessions(sessions_root: Path) -> list[Path]:
    """Return session dirs that match the old timestamp naming pattern."""
    results = []
    for child in sorted(sessions_root.iterdir()):
        if not child.is_dir():
            continue
        if child.name in _SKIP_DIRS:
            continue
        if _TIMESTAMP_RE.match(child.name):
            results.append(child)
    return results
 
 
def migrate(
    sessions_dir: str,
    dry_run: bool = True,
    verbose: bool = True,
) -> int:
    """Run the migration.
 
    Args:
        sessions_dir: Path to the sessions root directory.
        dry_run: If True, print what would happen but make no changes.
        verbose: If True, print progress to stdout.
 
    Returns:
        Number of sessions migrated (or that would be migrated).
    """
    func = "migrate"
    root = Path(sessions_dir)
 
    if not root.is_dir():
        print(f"Sessions directory not found: {root}", file=sys.stderr)
        return 0
 
    from oc_chat.core.wordgen import random_name
 
    # Build the set of all existing session names (random-word + timestamp)
    # so collision avoidance works correctly.
    all_names: set[str] = set()
    for child in root.iterdir():
        if child.is_dir() and child.name not in _SKIP_DIRS:
            all_names.add(child.name)
 
    targets = _collect_sessions(root)
    if not targets:
        if verbose:
            print("No timestamp-format sessions found. Nothing to migrate.")
        return 0
 
    if verbose and dry_run:
        print("=== DRY RUN — no files will be changed ===\n")
 
    migrated = 0
    errors = 0
    skipped = 0
 
    col_w = max(len(t.name) for t in targets)
 
    for session_dir in targets:
        old_name = session_dir.name
 
        meta_path = _find_meta_file(session_dir)
        if meta_path is None:
            print(f"  SKIP  {old_name:<{col_w}}  (no meta file found)")
            skipped += 1
            continue
 
        # Generate a unique new name
        new_name = random_name(exists_fn=lambda n: n in all_names)
        all_names.add(new_name)
 
        _dbg("OC_CHAT.MAIN", func, f"{old_name!r} → {new_name!r}", 2)
 
        if verbose:
            print(f"  {'(dry) ' if dry_run else ''}RENAME  {old_name:<{col_w}}  →  {new_name}")
 
        if dry_run:
            migrated += 1
            continue
 
        try:
            # 1. Rename all files inside the directory (old_name prefix → new_name)
            for f in list(session_dir.iterdir()):
                if not f.is_file():
                    continue
                if f.name.startswith(old_name + "_"):
                    new_fname = new_name + f.name[len(old_name):]
                    f.rename(session_dir / new_fname)
                    _dbg("OC_CHAT.MAIN", func, f"  file: {f.name} → {new_fname}", 3)
 
            # 2. Update session_name in the (now-renamed) meta file
            new_meta_path = session_dir / (new_name + meta_path.name[len(old_name):])
            if new_meta_path.exists():
                meta_data = _load_json(new_meta_path)
                meta_data["session_name"] = new_name
                _write_json_atomic(new_meta_path, meta_data)
 
            # 3. Rename the session directory itself
            new_dir = root / new_name
            session_dir.rename(new_dir)
 
            migrated += 1
 
        except OSError as exc:
            print(f"  ERROR renaming {old_name}: {exc}", file=sys.stderr)
            _dbg("OC_CHAT.MAIN", func, f"error renaming {old_name!r}: {exc}", 1, override=True)
            errors += 1
 
    if verbose:
        print()
        mode = "Would migrate" if dry_run else "Migrated"
        print(f"{mode}: {migrated}  |  Skipped: {skipped}  |  Errors: {errors}")
        if dry_run and migrated > 0:
            print("\nRe-run without --dry-run to apply changes.")
 
    return migrated
 
 
def main() -> int:
    """Entry point for ``python -m oc_chat.scripts.migrate_session_names``."""
    for _stream in (sys.stdout, sys.stderr):
        try:
            _stream.reconfigure(errors="replace")
        except (AttributeError, ValueError):
            pass  # non-reconfigurable stream (tests, pipes wrapped by harnesses)
    from oc_chat.core.persistence import DEFAULT_SESSIONS_DIR
 
    parser = argparse.ArgumentParser(
        description="Rename timestamp-format sessions to random-word names.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=(
            "Examples:\n"
            "  python -m oc_chat.scripts.migrate_session_names           # dry run (safe)\n"
            "  python -m oc_chat.scripts.migrate_session_names --apply   # apply changes\n"
        ),
    )
    parser.add_argument(
        "--apply",
        action="store_true",
        default=False,
        help="Apply changes (default is dry-run/preview only)",
    )
    parser.add_argument(
        "--sessions-dir",
        default=DEFAULT_SESSIONS_DIR,
        help=f"Sessions directory (default: {DEFAULT_SESSIONS_DIR})",
    )
    parser.add_argument(
        "--quiet",
        action="store_true",
        default=False,
        help="Suppress output",
    )
 
    args = parser.parse_args()
    dry_run = not args.apply
 
    migrate(
        sessions_dir=args.sessions_dir,
        dry_run=dry_run,
        verbose=not args.quiet,
    )
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main())