feat(drive-arch): corrupt-FS hardening + survey subcommand (DA-002/DA-003)

tools/drive-archaeologist/src/drive_archaeologist/classifier.py

@@ -2,10 +2,22 @@
 File classification based on extension profiles.
 """
 
+import re
 from pathlib import Path
 
 from .profiles import CLASSIFICATION_PROFILES
 
+# RINEX 2 short filename: ssssdddh.yyt (site, DOY, session, 2-digit year, type).
+# The profile extension list only enumerates years .15–.22; real archives span
+# far more (e.g. .05o, .23o) and include Hatanaka (.yyd) and met (.yym) files.
+# o=obs n=nav g=GLONASS-nav d=Hatanaka m=met
+_RINEX_SHORT_RE = re.compile(r"^[a-z0-9]{4}\d{3}[a-x0-9]\.\d{2}[ondgm]$", re.IGNORECASE)
+
+# Leica raw GNSS: .m00, .m01, ... — an extension family, so it cannot live in
+# the static profile map. 3,665 of these surfaced unclassified on the first
+# real GNSS-bearing drive (DOSTB20150918 $RECYCLE.BIN).
+_LEICA_RAW_RE = re.compile(r"\.m\d{2}$", re.IGNORECASE)
+
 
 class Classifier:
     """
@@ -33,7 +45,8 @@ def _build_extension_map(self) -> dict[str, str]:
 
     def classify(self, filepath: Path) -> str | None:
         """
-        Classify a file based on its extension.
+        Classify a file based on its extension, with a RINEX short-name
+        regex fallback for year-extensions absent from the profile list.
 
         Args:
             filepath: The path to the file.
@@ -42,7 +55,14 @@ def classify(self, filepath: Path) -> str | None:
             The classification category as a string, or None if no match is found.
         """
         extension = filepath.suffix.lower()
-        return self._extension_map.get(extension)
+        category = self._extension_map.get(extension)
+        if category is not None:
+            return category
+        if _RINEX_SHORT_RE.match(filepath.name):
+            return "GNSS Data"
+        if _LEICA_RAW_RE.search(filepath.name):
+            return "GNSS Raw (Leica)"
+        return None
 
     def classify_by_ext(self, ext: str) -> str | None:
         return self._extension_map.get(ext.lower())

tools/drive-archaeologist/src/drive_archaeologist/cli.py

@@ -1,12 +1,14 @@
 """
 CLI interface for drive-archaeologist using Click framework.
-Provides the main 'scan' command for Phase 0.
+Provides the 'scan' command (full JSONL catalog) and the 'survey'
+command (fast wipe/keep triage, DA-003).
 """
 
 from pathlib import Path
 
 import click
 from rich.console import Console
+from rich.table import Table
 
 from .scanner import DeepScanner
 
@@ -29,13 +31,43 @@ def main():
     help="Output file path (default: scan_<name>_<timestamp>.jsonl)",
 )
 @click.option("--resume", "-r", is_flag=True, help="Resume a previous interrupted scan")
+@click.option("--force", is_flag=True, help="Overwrite an existing output file (otherwise refused)")
+@click.option(
+    "--include-hidden",
+    is_flag=True,
+    help="Also scan hidden (dot-prefixed) and system entries ($RECYCLE.BIN, .Trash*, ...)",
+)
+@click.option(
+    "--exclude",
+    "-x",
+    "excludes",
+    multiple=True,
+    help="Glob (relative to scan root, or bare name) to skip; repeatable",
+)
+@click.option(
+    "--max-archive-depth",
+    type=int,
+    default=3,
+    show_default=True,
+    help="How many levels of nested archives to extract (0 = never extract)",
+)
 @click.option(
     "--ingest", is_flag=True, help="Dispatch classified GNSS files to the ingestion pipeline"
 )
 @click.option(
     "--dry-run", is_flag=True, help="Log what would be dispatched without sending to Celery"
 )
-def scan(path: Path, output: Path | None, resume: bool, ingest: bool, dry_run: bool):
+def scan(
+    path: Path,
+    output: Path | None,
+    resume: bool,
+    force: bool,
+    include_hidden: bool,
+    excludes: tuple[str, ...],
+    max_archive_depth: int,
+    ingest: bool,
+    dry_run: bool,
+):
     """Scan a drive or directory and produce a JSONL file with metadata"""
     on_classified = None
     if ingest or dry_run:
@@ -44,7 +76,16 @@ def scan(path: Path, output: Path | None, resume: bool, ingest: bool, dry_run: b
         on_classified = make_dispatch_callback(dry_run=dry_run)
 
     try:
-        scanner = DeepScanner(path, output_file=output, resume=resume, on_classified=on_classified)
+        scanner = DeepScanner(
+            path,
+            output_file=output,
+            resume=resume,
+            on_classified=on_classified,
+            include_hidden=include_hidden,
+            excludes=list(excludes),
+            max_archive_depth=max_archive_depth,
+            force=force,
+        )
         scanner.scan()
     except KeyboardInterrupt:
         console.print("\n[yellow]Warning: Scan interrupted by user[/yellow]")
@@ -55,5 +96,83 @@ def scan(path: Path, output: Path | None, resume: bool, ingest: bool, dry_run: b
         raise click.Abort() from None
 
 
+@main.command()
+@click.argument("path", type=click.Path(exists=True, path_type=Path))
+@click.option(
+    "--include-hidden/--no-include-hidden",
+    default=True,
+    show_default=True,
+    help="Survey hidden/system entries too (a stick whose content sits in .Trash-1000 "
+    "would otherwise read as empty)",
+)
+@click.option(
+    "--exclude",
+    "-x",
+    "excludes",
+    multiple=True,
+    help="Glob (relative to survey root, or bare name) to skip; repeatable",
+)
+@click.option(
+    "--extract-archives",
+    is_flag=True,
+    help="Also look inside archives (slower; extracts to $TMPDIR)",
+)
+@click.option("--top", type=int, default=12, show_default=True, help="Rows in the category table")
+def survey(
+    path: Path,
+    include_hidden: bool,
+    excludes: tuple[str, ...],
+    extract_archives: bool,
+    top: int,
+):
+    """Fast triage: what's on this drive, and is it safe to wipe?
+
+    Walks once with the same classifier as `scan` but writes NO catalog,
+    computes NO hashes, and (by default) opens NO archives. Prints a
+    category/extension breakdown and a wipe/keep verdict with explicit
+    disclosure of anything the walk did not cover.
+    """
+    try:
+        scanner = DeepScanner(
+            path,
+            stats_only=True,
+            include_hidden=include_hidden,
+            excludes=list(excludes),
+            max_archive_depth=1 if extract_archives else 0,
+        )
+        scanner.scan()
+    except KeyboardInterrupt:
+        console.print("\n[yellow]Survey interrupted[/yellow]")
+        raise click.Abort() from None
+    except Exception as e:
+        console.print(f"[red]Error: {e}[/red]")
+        raise click.Abort() from None
+
+    stats = scanner.stats
+    table = Table(title=f"Survey: {path}")
+    table.add_column("Category", style="cyan")
+    table.add_column("Files", justify="right")
+    table.add_column("Top extensions", style="dim")
+    ext_by_cat: dict[str, list[str]] = {}
+    for category, count in stats.categories.most_common(top):
+        exts = [
+            e
+            for e, _ in stats.extensions.most_common()
+            if scanner.classifier.classify_by_ext(e) == category
+        ][:4]
+        ext_by_cat[category] = exts
+        table.add_row(category, f"{count:,}", " ".join(exts))
+    console.print(table)
+    console.print(
+        f"[bold]Total:[/bold] {scanner.file_count:,} files, {stats.total_bytes / (1024**3):.2f} GiB"
+    )
+
+    verdict, warnings = scanner.survey_verdict()
+    for w in warnings:
+        console.print(f"[yellow]⚠ {w}[/yellow]")
+    color = "red" if stats.gnss_files or stats.metadata_inconsistent else "green"
+    console.print(f"[bold {color}]Verdict: {verdict}[/bold {color}]")
+
+
 if __name__ == "__main__":
     main()