oct_migrate.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/migrate/oct_migrate.py
 
"""
Purpose
-------
FS-539 — ``oct migrate option-c`` relocates a legacy Option C project
into the canonical ``.option_c/`` dotdir layout introduced in v0.19.0.
 
The command **copies** (never moves) each artefact so in-flight work on
the legacy layout is not broken mid-release. Legacy files stay in place
for one release; ``oct audit`` warns about the duplication and a future
release will offer removal.
 
Responsibilities
----------------
- Detect migratable artefacts: ``.octrc.json``, debug config (in either
  ``oc_diagnostics/`` or ``diagnostics/``), ``logs/``.
- In ``--dry-run`` mode, describe the plan without touching the disk.
- Copy each artefact to ``.option_c/`` with ``.bk`` backup when a
  destination already exists and differs.
- Write ``oc_status.json`` recording the migration stage and source
  paths.
 
Diagnostics
-----------
Domain: MIGRATE
Levels:
    L2 — lifecycle: start / end
    L3 — per-artefact decisions
 
Contracts
---------
- Copies (never moves) — legacy files remain in place for one release.
- ``--dry-run`` performs no filesystem writes.
- A pre-existing destination that differs is backed up to ``.bk`` before overwrite.
 
Dependencies
------------
- oct.core.diagnostics (_dbg structured logger)
- oct.core.option_c_dir (OcStatus, write_oc_status, option_c_dir)
- oct.core.project_root (project root detection)
- click (CLI framework)
"""
 
from __future__ import annotations
 
import json
import shutil
from dataclasses import dataclass, field
from pathlib import Path
 
import click
 
from oct.core.diagnostics import _dbg
from oct.core.option_c_dir import (
    OcStatus,
    option_c_dir,
    write_oc_status,
)
from oct.core.project_root import get_project_root
 
 
@dataclass
class MigrationResult:
    """Structured output from :func:`run_migrate_option_c`."""
 
    copied: list[tuple[str, Path, Path]] = field(default_factory=list)
    skipped: list[tuple[str, str]] = field(default_factory=list)
    backed_up: list[Path] = field(default_factory=list)
    messages: list[str] = field(default_factory=list)
    stage: str = "bootstrap"
 
 
_ARTEFACT_SPEC: list[tuple[str, str, list[str], str]] = [
    # (label, dest_relative_to_.option_c, [source_candidates], kind)
    ("octrc", "octrc.json", [".octrc.json"], "file"),
    (
        "debug_config",
        "debug_config.json",
        ["oc_diagnostics/debug_config.json", "diagnostics/debug_config.json"],
        "file",
    ),
    ("logs", "logs", ["logs"], "dir"),
]
 
 
def run_migrate_option_c(
    project_root: Path,
    *,
    dry_run: bool = False,
    no_backup: bool = False,
) -> MigrationResult:
    """Migrate a legacy Option C project into the ``.option_c/`` layout.
 
    Returns a :class:`MigrationResult` describing what was copied,
    skipped, or backed up. The caller renders the result.
    """
    func = "run_migrate_option_c"
    _dbg("MIGRATE", func, f"start root={project_root} dry_run={dry_run}", 2)
 
    result = MigrationResult()
    oc_dir = option_c_dir(project_root)
 
    if dry_run:
        result.messages.append(f"[DRY RUN] Migration target: {oc_dir}")
        for label, dest_rel, source_candidates, kind in _ARTEFACT_SPEC:
            src = _find_source(project_root, source_candidates)
            if src is None:
                result.messages.append(f"[DRY RUN] {label}: not found — skip")
                result.skipped.append((label, "not found"))
                continue
            dest = oc_dir / dest_rel
            if dest.exists():
                result.messages.append(
                    f"[DRY RUN] {label}: {src} -> {dest} (destination exists, "
                    f"{'overwrite' if no_backup else 'backup + overwrite'})"
                )
            else:
                result.messages.append(
                    f"[DRY RUN] {label}: {src} -> {dest}"
                )
        result.messages.append(
            f"[DRY RUN] Would write {oc_dir / 'oc_status.json'}"
        )
        result.messages.append("[DRY RUN] Migration plan complete (no files written)")
        _dbg("MIGRATE", func, "dry-run end", 2)
        return result
 
    oc_dir.mkdir(parents=True, exist_ok=True)
    (oc_dir / "cache").mkdir(exist_ok=True)
 
    migrated_from: dict[str, str] = {}
    pillars_found: set[str] = set()
 
    for label, dest_rel, source_candidates, kind in _ARTEFACT_SPEC:
        src = _find_source(project_root, source_candidates)
        if src is None:
            result.skipped.append((label, "not found"))
            _dbg("MIGRATE", func, f"{label}: no source found, skipping", 3)
            continue
 
        dest = oc_dir / dest_rel
 
        if dest.exists():
            if _content_matches(src, dest, kind):
                result.skipped.append((label, "already migrated"))
                result.messages.append(
                    f"  {label}: already at {dest} (identical) — skip"
                )
                _dbg("MIGRATE", func, f"{label}: dest identical, skip", 3)
                pillars_found.add(label)
                migrated_from[label] = str(
                    src.relative_to(project_root)
                )
                continue
            if not no_backup:
                backup = Path(str(dest) + ".bk")
                if kind == "dir":
                    if backup.exists():
                        shutil.rmtree(backup)
                    shutil.copytree(dest, backup)
                else:
                    shutil.copy2(dest, backup)
                result.backed_up.append(backup)
                result.messages.append(f"  {label}: backed up {dest} -> {backup}")
 
        if kind == "dir":
            if dest.exists():
                shutil.rmtree(dest)
            shutil.copytree(src, dest)
        else:
            shutil.copy2(src, dest)
 
        result.copied.append((label, src, dest))
        result.messages.append(f"  {label}: {src} -> {dest}")
        pillars_found.add(label)
        migrated_from[label] = src.relative_to(project_root).as_posix()
        _dbg("MIGRATE", func, f"{label}: copied {src} -> {dest}", 3)
 
    _pillar_label_to_name = {
        "octrc": "linter",
        "debug_config": "diagnostics",
        "logs": "linter",
    }
    pillar_names_found = {
        _pillar_label_to_name.get(p, p) for p in pillars_found
    }
    all_pillars = {"linter", "diagnostics", "git", "docs", "tests", "scaffold"}
    stage = "compliant" if pillar_names_found else "bootstrap"
    if pillar_names_found and pillar_names_found != all_pillars:
        stage = "migrating"
    result.stage = stage
 
    try:
        from importlib.metadata import PackageNotFoundError, version
        try:
            oct_version = version("oct")
        except PackageNotFoundError:
            oct_version = "unknown"
    except ImportError:
        oct_version = "unknown"
 
    import datetime
    today = datetime.date.today().isoformat()
    pillars_dict = {}
    for name in ("linter", "diagnostics", "git", "docs", "tests", "scaffold"):
        pillars_dict[name] = {
            "enabled": name in pillar_names_found,
            "since": today if name in pillar_names_found else None,
        }
 
    status = OcStatus(
        stage=stage,
        pillars=pillars_dict,
        migrated_from=migrated_from,
        oct_version_at_migration=oct_version,
    )
    write_oc_status(project_root, status)
    result.messages.append(f"  oc_status.json: stage={stage}")
 
    _dbg(
        "MIGRATE", func,
        f"end copied={len(result.copied)} skipped={len(result.skipped)} "
        f"stage={stage}",
        2,
    )
    return result
 
 
def _find_source(project_root: Path, candidates: list[str]) -> Path | None:
    for candidate in candidates:
        path = project_root / candidate
        if path.exists():
            return path
    return None
 
 
def _content_matches(src: Path, dest: Path, kind: str) -> bool:
    if kind == "file":
        try:
            return src.read_bytes() == dest.read_bytes()
        except OSError:
            return False
    if kind == "dir":
        return dest.is_dir()
    return False
 
 
# -- Click surface -------------------------------------------------------
 
@click.group()
@click.pass_context
def migrate(ctx):  # noqa: ARG001
    """Migrate project layout to a newer Option C structure."""
    pass
 
 
@migrate.command(name="option-c")
@click.option("--dry-run", is_flag=True, help="Show plan without writing anything.")
@click.option(
    "--no-backup", is_flag=True, default=False,
    help="Skip .bk backup when overwriting an existing destination.",
)
@click.pass_context
def migrate_option_c_cmd(ctx, dry_run: bool, no_backup: bool):
    """Migrate to the canonical .option_c/ dotdir layout (FS-539)."""
    project_root = get_project_root(ctx)
    result = run_migrate_option_c(
        project_root, dry_run=dry_run, no_backup=no_backup,
    )
    for msg in result.messages:
        click.echo(msg)
    if not dry_run and not result.copied and not any(
        reason == "already migrated" for _, reason in result.skipped
    ):
        click.echo("Nothing to migrate — no legacy artefacts found.")