diff --git a/scrapers/capture_passive_playwright_async.py b/scrapers/capture_passive_playwright_async.py index f4fd76e..32f7db4 100644 --- a/scrapers/capture_passive_playwright_async.py +++ b/scrapers/capture_passive_playwright_async.py @@ -249,23 +249,28 @@ async def run(args): try: parsed = json.loads(txt) pretty = json.dumps(parsed, ensure_ascii=False, indent=2) - fname = f"{safe}_{h}_{ts}.json" - body_path = str(output / fname) - save_text(output / fname, pretty) - try: - print(f"{now_ts()}\tSAVED\t{body_path}") - except Exception: - pass + content_bytes = pretty.encode('utf-8') excerpt = pretty[:800] + use_text = pretty + ext = 'json' except Exception: - fname = f"{safe}_{h}_{ts}.txt" - body_path = str(output / fname) - save_text(output / fname, txt) + content_bytes = txt.encode('utf-8') + excerpt = txt[:800] + use_text = txt + ext = 'txt' + + # deterministic filename based on content hash (sha1) + content_sha1 = hashlib.sha1(content_bytes).hexdigest() + fname = f"{safe}_{content_sha1}.{ext}" + body_path = str(output / fname) + + # if file already exists, don't rewrite body; just update meta/index + if not os.path.exists(body_path): + save_text(output / fname, use_text) try: print(f"{now_ts()}\tSAVED\t{body_path}") except Exception: pass - excerpt = txt[:800] else: try: data = await resp.body() @@ -278,26 +283,29 @@ async def run(args): ext = guess_ext(ctype) # Put images into an images/ subdirectory, other binaries into assets/ if ctype.startswith('image/'): - # Images don't need per-capture timestamps; use stable names with hash subdir = output / 'images' subdir.mkdir(parents=True, exist_ok=True) - fname = f"{safe}_{h}.{ext}" + content_sha1 = hashlib.sha1(data).hexdigest() + fname = f"{safe}_{content_sha1}.{ext}" body_path = str(subdir / fname) - save_binary(subdir / fname, data) - try: - print(f"{now_ts()}\tSAVED\t{body_path}") - except Exception: - pass + if not os.path.exists(body_path): + save_binary(subdir / fname, data) + try: + print(f"{now_ts()}\tSAVED\t{body_path}") + except Exception: + pass else: subdir = output / 'assets' subdir.mkdir(parents=True, exist_ok=True) - fname = f"{safe}_{h}_{ts}.{ext}" + content_sha1 = hashlib.sha1(data).hexdigest() + fname = f"{safe}_{content_sha1}.{ext}" body_path = str(subdir / fname) - save_binary(subdir / fname, data) - try: - print(f"{now_ts()}\tSAVED\t{body_path}") - except Exception: - pass + if not os.path.exists(body_path): + save_binary(subdir / fname, data) + try: + print(f"{now_ts()}\tSAVED\t{body_path}") + except Exception: + pass except Exception: # final fallback: try binary try: @@ -340,11 +348,12 @@ async def run(args): except Exception: content_hash = None - # prepare the meta filename we'll write for this response - new_meta_name = f"{safe}_{h}_{ts}.meta.json" + # prepare the meta filename we'll write for this response (canonical by content when available) + meta_base = content_hash or h + new_meta_name = f"{safe}_{meta_base}.meta.json" new_meta_abs = os.path.abspath(str(output / new_meta_name)) - # If older responses exist with the same content, delete them (fast in-memory lookup) + # If older responses exist with the same content, handle them according to dedupe policy try: matches = in_memory_index.get(content_hash, []) if content_hash else [] for j in matches: @@ -352,20 +361,55 @@ async def run(args): metaf = j.get('meta_file') try: if body and os.path.exists(body) and os.path.abspath(body) != os.path.abspath(body_path): - os.remove(body) + if args.dry_run: + print(f"DRYRUN: would remove old body {body}") + else: + if args.dedupe_policy == 'delete': + os.remove(body) + elif args.dedupe_policy == 'move': + dupdir = output / 'duplicates' + dupdir.mkdir(parents=True, exist_ok=True) + try: + os.replace(body, dupdir / os.path.basename(body)) + except Exception: + os.remove(body) + elif args.dedupe_policy == 'symlink': + # replace old body with symlink to canonical + try: + os.remove(body) + os.symlink(os.path.abspath(body_path), body) + except Exception: + pass try: with open(output / '_debug_events.log', 'a', encoding='utf-8') as dbg: - dbg.write(f"{now_ts()}\tDELETED_OLD_BODY\t{body}\n") + dbg.write(f"{now_ts()}\tDEDUP_BODY\t{body}\n") except Exception: pass except Exception: pass try: if metaf and os.path.exists(metaf) and os.path.abspath(metaf) != new_meta_abs: - os.remove(metaf) + if args.dry_run: + print(f"DRYRUN: would remove old meta {metaf}") + else: + if args.dedupe_policy == 'delete': + os.remove(metaf) + elif args.dedupe_policy == 'move': + dupdir = output / 'duplicates' + dupdir.mkdir(parents=True, exist_ok=True) + try: + os.replace(metaf, dupdir / os.path.basename(metaf)) + except Exception: + os.remove(metaf) + elif args.dedupe_policy == 'symlink': + try: + os.remove(metaf) + os.symlink(new_meta_abs, metaf) + except Exception: + pass try: with open(output / '_debug_events.log', 'a', encoding='utf-8') as dbg: - dbg.write(f"{now_ts()}\tDELETED_OLD_META\t{metaf}\n") + dbg.write(f"{now_ts()}\tDEDUP_META\t{metaf}\n") except Exception: pass except Exception: @@ -383,10 +427,10 @@ async def run(args): 'response_excerpt': excerpt, 'content_hash': content_hash, } - meta_name = f"{safe}_{h}_{ts}.meta.json" - save_text(output / meta_name, json.dumps(meta, ensure_ascii=False, indent=2)) + # write canonical meta (update if exists) + save_text(output / new_meta_name, json.dumps(meta, ensure_ascii=False, indent=2)) try: - entry = {'url': url, 'resource_type': rtype, 'timestamp': ts, 'body_file': body_path, 'meta_file': str(output / meta_name), 'excerpt': excerpt, 'content_hash': content_hash} + entry = {'url': url, 'resource_type': rtype, 'timestamp': ts, 'body_file': body_path, 'meta_file': str(output / new_meta_name), 'excerpt': excerpt, 'content_hash': content_hash} with open(index_path, 'a', encoding='utf-8') as idx: idx.write(json.dumps(entry, ensure_ascii=False) + '\n') # update in-memory index @@ -485,6 +529,8 @@ def parse_args(): parser.add_argument('--post-login-selector', default='#ds-app', help='Selector that indicates a successful login') parser.add_argument('--channel', default=os.getenv('PLAYWRIGHT_CHROME_CHANNEL', 'chrome')) parser.add_argument('--capture-types', default='xhr,fetch,document,other') + parser.add_argument('--dedupe-policy', default='delete', choices=['delete', 'move', 'keep', 'symlink'], help="What to do with older duplicate files: delete/move/keep/symlink") + parser.add_argument('--dry-run', action='store_true', help='If set, show dedupe actions without performing them') return parser.parse_args() diff --git a/scrapers/migrate_captured_files.py b/scrapers/migrate_captured_files.py new file mode 100644 index 0000000..01a9bc2 --- /dev/null +++ b/scrapers/migrate_captured_files.py @@ -0,0 +1,213 @@ +#!/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 '_' 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)