"""CLI subcommand: ``dataforge repair [--dry-run | --apply]``.""" from __future__ import annotations import hashlib from datetime import UTC, datetime from pathlib import Path from typing import Annotated import pandas as pd import typer from rich.console import Console from rich.panel import Panel from dataforge.cli.common import load_schema, read_csv from dataforge.detectors import run_all_detectors from dataforge.detectors.base import Issue, Schema from dataforge.repairers import build_repairers from dataforge.repairers.base import ProposedFix, RepairAttempt, RetryContext from dataforge.safety import SafetyContext, SafetyFilter, SafetyResult, SafetyVerdict from dataforge.transactions.log import ( append_applied_event, append_created_transaction, cache_dir_for, sha256_bytes, snapshot_path_for, ) from dataforge.transactions.txn import CellFix, RepairTransaction, generate_txn_id from dataforge.ui.repair_diff import render_repair_diff from dataforge.verifier import SMTVerifier, VerificationVerdict _console = Console(stderr=True) def apply_fixes_to_csv(path: Path, fixes: list[CellFix]) -> str: """Apply ordered cell fixes to a CSV and return the post-state SHA-256. Args: path: Source CSV path. fixes: Ordered list of cell fixes to apply. Returns: SHA-256 of the written file bytes. Raises: ValueError: If a fix references a missing row/column or stale old value. """ df = read_csv(path) for fix in fixes: if fix.operation != "update": raise ValueError(f"Unsupported repair operation '{fix.operation}' for row {fix.row}.") if fix.column not in df.columns: raise ValueError(f"Column '{fix.column}' not found in '{path}'.") if fix.row < 0 or fix.row >= len(df.index): raise ValueError(f"Row {fix.row} is out of bounds for '{path}'.") current_value = str(df.at[fix.row, fix.column]) if current_value != fix.old_value: raise ValueError( f"Refusing to apply stale fix for row {fix.row}, column '{fix.column}': " f"expected '{fix.old_value}', found '{current_value}'." ) df.at[fix.row, fix.column] = fix.new_value df.to_csv(path, index=False, lineterminator="\n") return hashlib.sha256(path.read_bytes()).hexdigest() def _resolve_schema(schema_path: Path | None) -> Schema | None: """Resolve an optional schema path into a parsed Schema.""" if schema_path is None: return None return load_schema(schema_path) def _print_error(message: str, *, hint: str | None = None) -> None: """Render a rich-formatted CLI error.""" body = f"[bold red]{message}[/bold red]" if hint: body = f"{body}\n\n[dim]{hint}[/dim]" _console.print(Panel(body, title="Repair Error", style="red")) def _propose_repairs( issues: list[Issue], path: Path, working_df: pd.DataFrame, schema: Schema | None, *, allow_llm: bool, model: str, allow_pii: bool, confirm_pii: bool, confirm_escalations: bool, interactive: bool, ) -> tuple[list[ProposedFix], list[list[RepairAttempt]]]: """Run repairers and gates issue-by-issue against the working dataframe.""" repairers = build_repairers( cache_dir=cache_dir_for(path), allow_llm=allow_llm, model=model, ) safety_filter = SafetyFilter() verifier = SMTVerifier() safety_context = SafetyContext( allow_pii=allow_pii, confirm_pii=confirm_pii, confirm_escalations=confirm_escalations, ) accepted_fixes: list[ProposedFix] = [] attempt_groups: list[list[RepairAttempt]] = [] for issue in issues: attempts: list[RepairAttempt] = [] repairer = repairers.get(issue.issue_type) if repairer is None: attempts.append( RepairAttempt( issue=issue, attempt_number=1, status="attempted_not_fixed", reason="No repairer is registered for this issue type.", ) ) attempt_groups.append(attempts) continue accepted = False retry_context = RetryContext(issue=issue) for attempt_number in range(1, 4): candidate = repairer.propose(issue, working_df, schema, retry_context=retry_context) if candidate is None: attempts.append( RepairAttempt( issue=issue, attempt_number=attempt_number, status="attempted_not_fixed", reason="No repair proposal was available for this issue.", ) ) break preferred = safety_filter.choose_preferred([candidate], schema, safety_context) safety_result = safety_filter.evaluate(preferred, schema, safety_context) if safety_result.verdict == SafetyVerdict.ESCALATE and interactive: safety_context, safety_result = _resolve_escalation( preferred, schema, safety_context, safety_filter, safety_result, ) if safety_result.verdict == SafetyVerdict.DENY: attempts.append( RepairAttempt( issue=issue, attempt_number=attempt_number, fix=preferred, status="denied", reason=safety_result.reason, ) ) retry_context = _build_retry_context(issue, attempts) continue if safety_result.verdict == SafetyVerdict.ESCALATE: attempts.append( RepairAttempt( issue=issue, attempt_number=attempt_number, fix=preferred, status="escalated", reason=safety_result.reason, ) ) break verifier_result = verifier.verify(working_df, [preferred], schema) if verifier_result.verdict == VerificationVerdict.ACCEPT: accepted_fixes.append(preferred) working_df.at[preferred.fix.row, preferred.fix.column] = preferred.fix.new_value attempts.append( RepairAttempt( issue=issue, attempt_number=attempt_number, fix=preferred, status="accepted", reason=verifier_result.reason, ) ) accepted = True break attempts.append( RepairAttempt( issue=issue, attempt_number=attempt_number, fix=preferred, status=( "rejected" if verifier_result.verdict == VerificationVerdict.REJECT else "unknown" ), reason=verifier_result.reason, unsat_core=verifier_result.unsat_core, ) ) retry_context = _build_retry_context(issue, attempts) if ( not accepted and attempts and attempts[-1].status not in {"attempted_not_fixed", "escalated"} ): last_reason = attempts[-1].reason attempts[-1] = attempts[-1].model_copy( update={ "status": "attempted_not_fixed", "reason": ( f"Issue was attempted but not fixed after {len(attempts)} attempt(s). " f"Last failure: {last_reason}" ), } ) attempt_groups.append(attempts) return accepted_fixes, attempt_groups def _build_retry_context(issue: Issue, attempts: list[RepairAttempt]) -> RetryContext: """Build retry hints from previous failed attempts.""" rejected_values = frozenset( attempt.fix.fix.new_value for attempt in attempts if attempt.fix is not None and attempt.status in {"denied", "rejected", "unknown"} ) hints: list[str] = [] for attempt in attempts: hints.append(attempt.reason) hints.extend(attempt.unsat_core) return RetryContext( issue=issue, previous_attempts=tuple(attempts), rejected_values=rejected_values, hints=tuple(hints), ) def _resolve_escalation( candidate: ProposedFix, schema: Schema | None, context: SafetyContext, safety_filter: SafetyFilter, safety_result: SafetyResult, ) -> tuple[SafetyContext, SafetyResult]: """Prompt for safety escalations and re-evaluate if the user confirms.""" if "NO_PII_OVERWRITE" in safety_result.rule_ids: confirmed = typer.confirm( f"Candidate fix for row {candidate.fix.row}, column '{candidate.fix.column}' " "touches PII. Confirm this edit?", default=False, ) if confirmed: updated = context.model_copy(update={"confirm_pii": True}) return updated, safety_filter.evaluate(candidate, schema, updated) return context, safety_result confirmed = typer.confirm( f"Candidate fix for row {candidate.fix.row}, column '{candidate.fix.column}' " "touches an aggregate-sensitive column. Confirm this edit?", default=False, ) if confirmed: updated = context.model_copy(update={"confirm_escalations": True}) return updated, safety_filter.evaluate(candidate, schema, updated) return context, safety_result def _render_attempt_summary( attempt_groups: list[list[RepairAttempt]], console: Console, ) -> int: """Render a summary for issues that were not accepted.""" failed_groups = [ attempts for attempts in attempt_groups if attempts and attempts[-1].status != "accepted" ] if not failed_groups: return 0 lines: list[str] = [] for attempts in failed_groups: final_attempt = attempts[-1] issue = final_attempt.issue prefix = "" if any(label.startswith("fd::") for label in final_attempt.unsat_core): prefix = "functional dependency rejection - " elif any(label.startswith("domain::") for label in final_attempt.unsat_core): prefix = "domain bound rejection - " lines.append( f"{issue.issue_type} at {issue.row}:{issue.column} " f"after {len(attempts)} attempt(s): {prefix}{final_attempt.reason}" ) console.print("[bold yellow]Attempted But Not Fixed[/bold yellow]") for line in lines: console.print(line, overflow="fold") return len(failed_groups) def _apply_transaction( path: Path, fixes: list[ProposedFix], source_bytes: bytes, ) -> str: """Write a transaction record, apply fixes, and append the applied event.""" resolved_path = path.resolve() txn_id = generate_txn_id() snapshot_path = snapshot_path_for(resolved_path, txn_id) snapshot_path.parent.mkdir(parents=True, exist_ok=True) snapshot_path.write_bytes(source_bytes) transaction = RepairTransaction( txn_id=txn_id, created_at=datetime.now(UTC), source_path=str(resolved_path), source_sha256=sha256_bytes(source_bytes), source_snapshot_path=str(snapshot_path.resolve()), fixes=[proposal.fix for proposal in fixes], applied=False, ) log_path = append_created_transaction(transaction) try: post_sha256 = apply_fixes_to_csv(path, [proposal.fix for proposal in fixes]) append_applied_event(log_path, txn_id, post_sha256=post_sha256) except Exception: path.write_bytes(source_bytes) raise return txn_id def repair( path: Annotated[ Path, typer.Argument( exists=True, readable=True, help="Path to the CSV file to repair.", ), ], schema: Annotated[ Path | None, typer.Option( "--schema", exists=True, readable=True, help="Path to a YAML schema file with column types and FDs.", ), ] = None, dry_run: Annotated[ bool, typer.Option("--dry-run", help="Show proposed fixes without changing the file."), ] = False, apply: Annotated[ bool, typer.Option("--apply", help="Apply fixes and record a reversible transaction."), ] = False, allow_llm: Annotated[ bool, typer.Option( "--allow-llm", help="Allow fd_violation repair to call the configured LLM provider if needed.", ), ] = False, allow_pii: Annotated[ bool, typer.Option( "--allow-pii", help="Allow PII-targeting fixes to be considered by the safety layer.", ), ] = False, confirm_pii: Annotated[ bool, typer.Option( "--confirm-pii", help="Non-interactively confirm any PII-targeting fixes allowed via --allow-pii.", ), ] = False, confirm_escalations: Annotated[ bool, typer.Option( "--confirm-escalations", help="Non-interactively confirm soft safety escalations such as aggregate-sensitive edits.", ), ] = False, llm_model: Annotated[ str, typer.Option("--llm-model", help="Model name for fd_violation LLM fallback."), ] = "gemini-2.0-flash", ) -> None: """Detect, propose, and optionally apply reversible repairs to a CSV.""" if dry_run == apply: _print_error( "Choose exactly one of --dry-run or --apply.", hint="Example: dataforge repair data.csv --dry-run", ) raise typer.Exit(code=2) try: parsed_schema = _resolve_schema(schema) df = read_csv(path) except Exception as exc: _print_error(str(exc)) raise typer.Exit(code=2) from exc issues = run_all_detectors(df, parsed_schema) accepted_fixes, attempt_groups = _propose_repairs( issues, path, df.copy(deep=True), parsed_schema, allow_llm=allow_llm, model=llm_model, allow_pii=allow_pii, confirm_pii=confirm_pii, confirm_escalations=confirm_escalations, interactive=apply, ) output_console = Console() render_repair_diff(accepted_fixes, output_console, file_path=str(path)) failed_issue_count = _render_attempt_summary(attempt_groups, output_console) if not accepted_fixes and failed_issue_count == 0: raise typer.Exit(code=1) if dry_run: raise typer.Exit(code=0 if accepted_fixes else 1) if not accepted_fixes: raise typer.Exit(code=1) batch_safety = SafetyFilter().evaluate_batch(accepted_fixes) if batch_safety.verdict != SafetyVerdict.ALLOW: _print_error(batch_safety.reason) raise typer.Exit(code=1) source_bytes = path.read_bytes() try: txn_id = _apply_transaction(path, accepted_fixes, source_bytes) except Exception as exc: _print_error( f"Failed to apply repairs: {exc}", hint="The source file was restored to its pre-apply bytes.", ) raise typer.Exit(code=1) from exc output_console.print( Panel( f"[green]Applied {len(accepted_fixes)} fix(es).[/green]\n" f"Transaction ID: [bold]{txn_id}[/bold]", title="Repair Applied", style="green", ) ) if failed_issue_count: output_console.print( Panel( f"[yellow]{failed_issue_count} issue(s) were attempted but not fixed.[/yellow]", title="Week 3 Summary", style="yellow", ) )