statdx/scrapers/migrate_captured_files.py

#!/usr/bin/env python3
"""Migrate timestamped captured files to content-hash deterministic names.

This script scans an output capture directory (default: xhr_captured_async),
finds body files and their meta files that include timestamps in filenames,
computes content SHA1, renames/moves them to canonical names like
`{safe}_{sha1}.json` or `{safe}_{sha1}.{ext}` under the same directory structure,
updates meta content (response_body_file) and meta filename to `{safe}_{sha1}.meta.json`,
and rewrites `capture_index.jsonl` so entries point to the canonical files.

It supports --dry-run and --dedupe-policy (delete/move/keep/symlink) for handling
old duplicates that would collide on the canonical name.

Usage:
  python scrapers/migrate_captured_files.py --input-dir xhr_captured_async --dry-run

"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
import shutil
from pathlib import Path
from typing import Dict, List


def compute_sha1(path: Path) -> str:
    h = hashlib.sha1()
    with open(path, "rb") as f:
        while True:
            b = f.read(8192)
            if not b:
                break
            h.update(b)
    return h.hexdigest()


def canonical_name_for(path: Path, safe: str, sha1: str) -> Path:
    # preserve extension
    ext = path.suffix.lstrip('.')
    if not ext:
        ext = 'bin'
    return path.with_name(f"{safe}_{sha1}.{ext}")


def canonical_meta_name(safe: str, sha1: str, directory: Path) -> Path:
    return directory / f"{safe}_{sha1}.meta.json"


def find_safe_from_filename(fname: str) -> str:
    # safe prefix is everything before the first '_<sha_or_timestamp>' pattern
    # but we can be conservative: find the substring up to the first '_[0-9a-f]{8,}' or a timestamp pattern
    import re
    m = re.match(r"^(.+?)_(?:[0-9a-f]{8,32}|\d{8}T\d{6}Z|\d{4}-\d{2}-\d{2}T)", fname)
    if m:
        return m.group(1)
    # fallback: strip final numeric groups
    parts = fname.split('_')
    if len(parts) > 1:
        return '_'.join(parts[:-1])
    return fname


def load_index(index_path: Path) -> List[Dict]:
    if not index_path.exists():
        return []
    out = []
    with open(index_path, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if not line:
                continue
            try:
                out.append(json.loads(line))
            except Exception:
                continue
    return out


def write_index(index_path: Path, entries: List[Dict]):
    with open(index_path, 'w', encoding='utf-8') as f:
        for e in entries:
            f.write(json.dumps(e, ensure_ascii=False) + '\n')


def migrate(input_dir: Path, dedupe_policy: str = 'delete', dry_run: bool = True):
    # gather candidate files: top-level json/text files and files in images/ and assets/
    files = list(input_dir.glob('*.json')) + list(input_dir.glob('*.txt')) + list(input_dir.glob('*.bin'))
    files += list(input_dir.joinpath('images').glob('*')) if (input_dir / 'images').exists() else []
    files += list(input_dir.joinpath('assets').glob('*')) if (input_dir / 'assets').exists() else []

    # map sha1 -> canonical path (to detect collisions)
    canonical_map: Dict[str, Path] = {}
    # entries to update in index: old_path -> new_path
    path_updates: Dict[str, str] = {}

    for f in files:
        if f.is_dir():
            continue
        # skip meta files
        if f.name.endswith('.meta.json'):
            continue
        try:
            sha1 = compute_sha1(f)
        except Exception as e:
            print(f"Failed to hash {f}: {e}")
            continue
        safe = find_safe_from_filename(f.name)
        # canonical target
        target = f.with_name(f"{safe}_{sha1}{f.suffix}")
        # if already canonical, skip
        if f.resolve() == target.resolve():
            canonical_map[sha1] = target
            continue
        # if canonical target exists and is different file -> dedupe action
        if target.exists() and target.resolve() != f.resolve():
            print(f"Collision: {f} -> {target} (already exists)")
            if dry_run:
                print(f"DRYRUN would {dedupe_policy} {f}")
            else:
                if dedupe_policy == 'delete':
                    f.unlink()
                elif dedupe_policy == 'move':
                    dupdir = input_dir / 'duplicates'
                    dupdir.mkdir(parents=True, exist_ok=True)
                    shutil.move(str(f), str(dupdir / f.name))
                elif dedupe_policy == 'symlink':
                    try:
                        f.unlink()
                    except Exception:
                        pass
                    try:
                        os.symlink(str(target.resolve()), str(f))
                    except Exception:
                        pass
                elif dedupe_policy == 'keep':
                    pass
            path_updates[str(f)] = str(target)
            canonical_map[sha1] = target
            continue
        # otherwise move/rename to canonical target
        print(f"Renaming: {f} -> {target}")
        if not dry_run:
            try:
                os.replace(str(f), str(target))
            except Exception as e:
                print(f"Failed to move {f} -> {target}: {e}")
                continue
        path_updates[str(f)] = str(target)
        canonical_map[sha1] = target

    # Update meta files: for any meta files that point to old body paths, rewrite them
    metas = list(input_dir.glob('*.meta.json')) + list(input_dir.joinpath('images').glob('*.meta.json')) if (input_dir / 'images').exists() else []
    metas += list(input_dir.joinpath('assets').glob('*.meta.json')) if (input_dir / 'assets').exists() else []
    for m in metas:
        try:
            with open(m, 'r', encoding='utf-8') as f:
                jd = json.load(f)
        except Exception:
            continue
        bf = jd.get('response_body_file')
        if bf and bf in path_updates:
            newbf = path_updates[bf]
            jd['response_body_file'] = newbf
            # write canonical meta name
            safe = find_safe_from_filename(os.path.basename(newbf))
            sha1 = compute_sha1(Path(newbf)) if Path(newbf).exists() else None
            if sha1:
                new_meta = m.with_name(f"{safe}_{sha1}.meta.json")
            else:
                new_meta = m
            print(f"Updating meta {m} -> {new_meta}")
            if not dry_run:
                with open(new_meta, 'w', encoding='utf-8') as out:
                    json.dump(jd, out, ensure_ascii=False, indent=2)
                if new_meta.resolve() != m.resolve():
                    try:
                        m.unlink()
                    except Exception:
                        pass
            path_updates[str(m)] = str(new_meta)

    # Rewrite capture_index.jsonl
    idx_path = input_dir / 'capture_index.jsonl'
    index_entries = load_index(idx_path)
    changed = False
    for e in index_entries:
        bf = e.get('body_file')
        mf = e.get('meta_file')
        if bf and bf in path_updates:
            e['body_file'] = path_updates[bf]
            changed = True
        if mf and mf in path_updates:
            e['meta_file'] = path_updates[mf]
            changed = True
    if changed:
        print(f"Updating index {idx_path}")
        if not dry_run:
            write_index(idx_path, index_entries)

    print('Migration complete')


if __name__ == '__main__':
    p = argparse.ArgumentParser()
    p.add_argument('--input-dir', default='xhr_captured_async')
    p.add_argument('--dedupe-policy', default='delete', choices=['delete', 'move', 'keep', 'symlink'])
    p.add_argument('--dry-run', action='store_true')
    args = p.parse_args()

    migrate(Path(args.input_dir), dedupe_policy=args.dedupe_policy, dry_run=args.dry_run)