Ross
132 lines
4.0 KiB
Python
132 lines
4.0 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Map titles from a content-set JSON to document IDs by scanning captured search-ajax responses.
|
|
|
|
Usage:
|
|
python map_titles_to_ids.py \
|
|
--titles /home/ross/statdx/xhr_captured_async/app.statdx.com_search_content-set_productVersionId_bf09eabd-282b-4f83-b937-88d54b49db83_7e4593f2e47ed5942af45c77b9a7ff08e635ecc0.json \
|
|
--captures-dir /home/ross/statdx/xhr_captured_async \
|
|
--out mapping.json
|
|
|
|
The script produces a JSON mapping of the form:
|
|
[{
|
|
"title": "Original Title",
|
|
"normalized": "original title",
|
|
"matches": [ {"id": "...", "slug": "...", "source_file": "..."}, ... ]
|
|
}, ...]
|
|
|
|
It uses exact normalized-title matching (lowercased, whitespace collapsed). If needed we can add fuzzy matching.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from collections import defaultdict
|
|
|
|
|
|
def normalize_title(s: str) -> str:
|
|
if not isinstance(s, str):
|
|
return ""
|
|
s = s.strip().lower()
|
|
# collapse whitespace
|
|
s = re.sub(r"\s+", " ", s)
|
|
return s
|
|
|
|
|
|
def scan_search_ajax_files(captures_dir: Path):
|
|
mapping = defaultdict(list)
|
|
for p in captures_dir.iterdir():
|
|
name = p.name
|
|
# look for files that appear to be search-ajax JSON (not meta.json)
|
|
if "search-ajax" in name and name.endswith('.json') and not name.endswith('.meta.json'):
|
|
try:
|
|
with p.open('r') as fh:
|
|
doc = json.load(fh)
|
|
except Exception:
|
|
# skip unreadable files
|
|
continue
|
|
|
|
# pattern: doc['searchResults']['results'] is a list of result objects
|
|
results = None
|
|
if isinstance(doc, dict):
|
|
results = doc.get('searchResults', {}).get('results')
|
|
if not results:
|
|
continue
|
|
|
|
for r in results:
|
|
title = r.get('title') or (r.get('metadata') or {}).get('title')
|
|
_id = r.get('id') or (r.get('metadata') or {}).get('search_id')
|
|
# metadata.slug can be useful
|
|
slug = None
|
|
if isinstance(r.get('metadata'), dict):
|
|
sl = r['metadata'].get('slug')
|
|
# sometimes slug is a list inside metadata
|
|
if isinstance(sl, list) and sl:
|
|
slug = sl[0]
|
|
else:
|
|
slug = sl
|
|
|
|
if title and _id:
|
|
norm = normalize_title(title)
|
|
mapping[norm].append({
|
|
'id': _id,
|
|
'title': title,
|
|
'slug': slug,
|
|
'source_file': name,
|
|
})
|
|
|
|
return mapping
|
|
|
|
|
|
def load_titles_file(titles_file: Path):
|
|
with titles_file.open('r') as fh:
|
|
data = json.load(fh)
|
|
# expecting a list of strings
|
|
titles = []
|
|
if isinstance(data, list):
|
|
for item in data:
|
|
if isinstance(item, str):
|
|
titles.append(item)
|
|
else:
|
|
titles.append(str(item))
|
|
else:
|
|
raise SystemExit('titles file does not contain a JSON array')
|
|
return titles
|
|
|
|
|
|
def build_output(titles, mapping):
|
|
out = []
|
|
for t in titles:
|
|
norm = normalize_title(t)
|
|
matches = mapping.get(norm, [])
|
|
out.append({
|
|
'title': t,
|
|
'normalized': norm,
|
|
'matches': matches,
|
|
})
|
|
return out
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument('--titles', required=True, type=Path)
|
|
p.add_argument('--captures-dir', required=True, type=Path)
|
|
p.add_argument('--out', required=True, type=Path)
|
|
args = p.parse_args()
|
|
|
|
titles = load_titles_file(args.titles)
|
|
mapping = scan_search_ajax_files(args.captures_dir)
|
|
out = build_output(titles, mapping)
|
|
|
|
# write output
|
|
with args.out.open('w') as fh:
|
|
json.dump(out, fh, indent=2)
|
|
|
|
print(f'Wrote {len(out)} title mappings to {args.out}')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|