feat: Implement migration script for renaming timestamped files to content-hash based names with deduplication options
This commit is contained in:
Ross
@@ -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()
|
||||
|
||||
|
||||
|
||||
@@ -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 '_<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)
|
||||
Reference in New Issue
Block a user