migrate_to_oc_diagnostics.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# tools/migrate_to_oc_diagnostics.py
 
"""
Migration Script: Migrate a project from embedded diagnostics to oc_diagnostics
================================================================================
 
Purpose
-------
Rewrite Python import paths and Markdown references in an existing project so
that all usage of the old embedded ``diagnostics/`` package is replaced by the
new standalone ``oc_diagnostics`` package.
 
Responsibilities
----------------
- Walk the project tree, skipping common non-project directories
  (``EXCLUDE_DIRS``).
- Rewrite Python import statements in ``.py`` files using a curated list of
  ``(pattern, replacement)`` pairs.
- Rewrite documentation references in ``.md`` files.
- Create a ``.bak`` file next to every modified file so changes are reversible.
- Handle the old ``diagnostics/`` directory in one of two ways:
    - **Default (backup):** rename ``diagnostics/`` to ``diagnostics.bak/``.
      The original content is preserved; the active project no longer uses it.
    - **``--delete-original``:** permanently delete ``diagnostics/`` with no
      backup copy.  Use only when a backup already exists.
- Support ``--dry-run`` to preview all actions without modifying the filesystem.
 
Diagnostics
-----------
Domain: MIGRATE
Levels:
    L2 — lifecycle
    L3 — semantic details
    L4 — deep tracing
 
Contracts
---------
- Must not modify files outside the target project root.
- Must skip directories in ``EXCLUDE_DIRS`` to avoid corrupting virtual
  environments, VCS metadata, caches, and third-party packages.
- In ``--dry-run`` mode no files must be written, renamed, or deleted.
- Every modified file must have a ``.bak`` copy written before the rewrite.
- The ``diagnostics/`` backup copy is created by default; ``--delete-original``
  skips the backup and deletes instead.
 
Run this script from the root of the project being migrated:
 
    python tools/migrate_to_oc_diagnostics.py [--dry-run] [--delete-original]
"""
 
import argparse
import os
import re
import shutil
 
 
# ---------------------------------------------------------------------------
# DIRECTORIES TO SKIP DURING TREE WALK
# ---------------------------------------------------------------------------
 
EXCLUDE_DIRS = {
    ".git",
    "__pycache__",
    "venv",
    "env",
    ".venv",
    ".env",
    "node_modules",
    "diagnostics.bak",  # skip our own backup folder
    ".tox",
    ".mypy_cache",
    ".pytest_cache",
    "dist",
    "build",
    "*.egg-info",
}
 
 
# ---------------------------------------------------------------------------
# PYTHON IMPORT REPLACEMENTS
# ---------------------------------------------------------------------------
 
PYTHON_REPLACEMENTS = [
    # unified_debug
    (r"from\s+diagnostics\.unified_debug\s+import\s+_dbg",
     r"from oc_diagnostics import _dbg"),
 
    (r"from\s+diagnostics\.unified_debug\s+import\s+DEBUG_LEVELS",
     r"from oc_diagnostics import DEBUG_LEVELS"),
 
    # relative unified_debug
    (r"from\s+\.\s*diagnostics\.unified_debug\s+import\s+_dbg",
     r"from oc_diagnostics import _dbg"),
 
    (r"from\s+\.\s*diagnostics\.unified_debug\s+import\s+DEBUG_LEVELS",
     r"from oc_diagnostics import DEBUG_LEVELS"),
 
    # instrumentation middleware
    (r"from\s+diagnostics\.instrumentation_middleware\s+import\s+apply_instrumentation",
     r"from oc_diagnostics.instrumentation_middleware import apply_instrumentation"),
 
    (r"from\s+diagnostics\.instrumentation_middleware\s+import\s+enable_instrumentation_middleware",
     r"from oc_diagnostics.instrumentation_middleware import enable_instrumentation_middleware"),
 
    # relative instrumentation middleware
    (r"from\s+\.\s*diagnostics\.instrumentation_middleware\s+import\s+apply_instrumentation",
     r"from oc_diagnostics.instrumentation_middleware import apply_instrumentation"),
 
    (r"from\s+\.\s*diagnostics\.instrumentation_middleware\s+import\s+enable_instrumentation_middleware",
     r"from oc_diagnostics.instrumentation_middleware import enable_instrumentation_middleware"),
 
    # id instrumentation
    (r"from\s+diagnostics\.id_instrumentation\s+import\s+instrument_ids",
     r"from oc_diagnostics.id_instrumentation import instrument_ids"),
 
    (r"from\s+\.\s*diagnostics\.id_instrumentation\s+import\s+instrument_ids",
     r"from oc_diagnostics.id_instrumentation import instrument_ids"),
 
    # UI event logger
    (r"from\s+diagnostics\.ui_event_logger\s+import\s+ui_event_logger",
     r"from oc_diagnostics.ui_event_logger import ui_event_logger"),
 
    (r"from\s+\.\s*diagnostics\.ui_event_logger\s+import\s+ui_event_logger",
     r"from oc_diagnostics.ui_event_logger import ui_event_logger"),
]
 
 
# ---------------------------------------------------------------------------
# MARKDOWN REPLACEMENTS
# ---------------------------------------------------------------------------
 
MD_REPLACEMENTS = [
    (r"\bdiagnostics/", r"oc_diagnostics/"),
    (r"DIAGNOSTICS_README\.md", r"REFERENCE.md"),
    (r"DIAGNOSTICS_INTEGRATION_GUIDE_README\.md", r"INTEGRATION_GUIDE.md"),
    (r"from diagnostics\.", r"from oc_diagnostics."),
    (r"`diagnostics/`", r"`oc_diagnostics/`"),
]
 
 
# ---------------------------------------------------------------------------
# FILE PROCESSING
# ---------------------------------------------------------------------------
 
def _should_exclude(dirpath: str) -> bool:
    """Return True if this directory should be skipped during the tree walk."""
    parts = set(os.path.normpath(dirpath).split(os.sep))
    return bool(parts & EXCLUDE_DIRS)
 
 
def rewrite_file(path: str, replacements: list, dry_run: bool) -> bool:
    """Rewrite a file using a list of (pattern, replacement) pairs.
 
    Returns True if the file would be (or was) modified.
    Creates a ``.bak`` backup before writing when not in dry-run mode.
    """
    try:
        with open(path, "r", encoding="utf-8") as f:
            original = f.read()
    except (OSError, UnicodeDecodeError):
        return False
 
    updated = original
    for pattern, repl in replacements:
        updated = re.sub(pattern, repl, updated)
 
    if updated == original:
        return False
 
    if not dry_run:
        backup = path + ".bak"
        with open(backup, "w", encoding="utf-8") as f:
            f.write(original)
        with open(path, "w", encoding="utf-8") as f:
            f.write(updated)
 
    return True
 
 
def migrate_project(root: str = ".", dry_run: bool = False) -> list[str]:
    """Walk *root* and rewrite all Python and Markdown files.
 
    Returns the list of files that were (or would be) modified.
    """
    modified = []
 
    for dirpath, dirnames, filenames in os.walk(root):
        # Prune excluded directories in-place so os.walk skips their subtrees.
        dirnames[:] = [
            d for d in dirnames
            if d not in EXCLUDE_DIRS and not d.endswith(".egg-info")
        ]
 
        if _should_exclude(dirpath):
            continue
 
        for filename in filenames:
            full = os.path.join(dirpath, filename)
 
            if filename.endswith(".py"):
                if rewrite_file(full, PYTHON_REPLACEMENTS, dry_run):
                    modified.append(full)
 
            elif filename.endswith(".md"):
                if rewrite_file(full, MD_REPLACEMENTS, dry_run):
                    modified.append(full)
 
    return modified
 
 
# ---------------------------------------------------------------------------
# HANDLE OLD diagnostics/ FOLDER
# ---------------------------------------------------------------------------
 
def handle_old_diagnostics(delete_original: bool, dry_run: bool) -> str:
    """Handle the old ``diagnostics/`` directory.
 
    Default (backup): rename ``diagnostics/`` → ``diagnostics.bak/``.
    With ``--delete-original``: permanently delete ``diagnostics/``.
 
    Returns a human-readable status string.
    """
    old_path = os.path.join(".", "diagnostics")
    backup_path = os.path.join(".", "diagnostics.bak")
 
    if not os.path.exists(old_path):
        return "No diagnostics/ folder found (already removed or never present)."
 
    if delete_original:
        action = f"Deleted diagnostics/ (no backup created)."
        if not dry_run:
            shutil.rmtree(old_path)
    else:
        action = f"Renamed diagnostics/ → diagnostics.bak/ (backup preserved)."
        if not dry_run:
            if os.path.exists(backup_path):
                shutil.rmtree(backup_path)
            shutil.move(old_path, backup_path)
 
    return action
 
 
# ---------------------------------------------------------------------------
# MAIN
# ---------------------------------------------------------------------------
 
if __name__ == "__main__":
    parser = argparse.ArgumentParser(
        description="Migrate a project from embedded diagnostics/ to oc_diagnostics.",
    )
    parser.add_argument(
        "--dry-run",
        action="store_true",
        help="Preview all changes without modifying the filesystem.",
    )
    parser.add_argument(
        "--delete-original",
        action="store_true",
        help=(
            "Delete diagnostics/ without creating a backup. "
            "Default behavior renames it to diagnostics.bak/ instead."
        ),
    )
    args = parser.parse_args()
 
    dry_prefix = "[DRY RUN] " if args.dry_run else ""
 
    print(f"\n{dry_prefix}Migrating project to oc_diagnostics...\n")
    if args.dry_run:
        print("  (No files will be modified.)\n")
 
    modified_files = migrate_project(".", dry_run=args.dry_run)
 
    if modified_files:
        print(f"{dry_prefix}Would update:" if args.dry_run else "Updated the following files:\n")
        for f in modified_files:
            print("  -", f)
        if not args.dry_run:
            print("\n  (.bak backups created alongside each modified file)\n")
    else:
        print("No Python or Markdown files required changes.\n")
 
    # Handle old diagnostics/ directory
    result = handle_old_diagnostics(
        delete_original=args.delete_original,
        dry_run=args.dry_run,
    )
    print(f"{dry_prefix}{result}\n")
 
    print(f"{dry_prefix}Migration {'preview' if args.dry_run else 'complete'}.\n")