From 90ac197cbec508a3eba8f71d70e1cf609819a41c Mon Sep 17 00:00:00 2001 From: Ross Date: Sat, 18 Oct 2025 09:20:57 +0100 Subject: [PATCH] feat: Cache and manage image group metadata for enhanced document processing --- scrapers/document_to_markdown.py | 304 +++++-------------------------- 1 file changed, 48 insertions(+), 256 deletions(-) diff --git a/scrapers/document_to_markdown.py b/scrapers/document_to_markdown.py index 10b7128..a974c0c 100644 --- a/scrapers/document_to_markdown.py +++ b/scrapers/document_to_markdown.py @@ -26,7 +26,11 @@ from bs4 import BeautifulSoup, NavigableString, Tag import html import hashlib import urllib.parse +from pprint import pformat +from loguru import logger +import sys +IMAGE_GROUPS = {} def text_of(node) -> str: """Return the text content of a node, stripping extra whitespace.""" @@ -576,6 +580,7 @@ def process_file(path: str, out_dir: str, overwrite: bool = False) -> tuple[bool target_dir = os.path.join(out_dir, "articles") if isinstance(article_name, str) and article_name.startswith("https"): + return False, "external-url" target_dir = os.path.join(out_dir, "external") else: # per request: only save files that have titles @@ -640,266 +645,21 @@ def process_file(path: str, out_dir: str, overwrite: bool = False) -> tuple[bool # (channel/frontmatter injection removed - handled elsewhere) - # --- find and save related image/response files --- - try: - input_dir = os.path.dirname(path) or "." + image_data_to_add = {} - def find_image_candidates(search_dir: str, docid: str | None, article_name: str | None, image_group_ids: list | None = None) -> list: - """Find candidate media/image files in search_dir related to the document. + if "imageGroups" in data_peek: + for img_group in data_peek.get("imageGroups", []): + logger.debug(f"Image group: {pformat(img_group)}") + group_id = img_group.get("imageGroupId") - Preference order: - - files whose filename contains the docid - - files that contain any image_group_id - - files whose content contains the docid or article_name - - filenames that include 'image_' or 'thumbnail' heuristically - """ - candidates = [] - patterns = ["*image*.*", "*media*.*", "*resource*.*", "*thumbnail*.*"] - seen = set() - img_ids = set(image_group_ids or []) - - for pat in patterns: - for p in glob.glob(os.path.join(search_dir, pat)): - if p in seen: - continue - seen.add(p) - bn = os.path.basename(p) - - # prefer files that reference docid in filename - if docid and docid in bn: - candidates.append(p) - continue - - # if filename contains 'media' and we have image group ids, inspect JSON quickly - if img_ids and 'media' in bn.lower() and bn.lower().endswith('.json'): - try: - with open(p, 'r', encoding='utf-8') as pf: - j = json.load(pf) - except Exception: - j = None - if j is not None: - # search for imageGroupId tokens or referenced imageGroupId in the JSON - jtxt = json.dumps(j) - if any(igid in jtxt for igid in img_ids): - candidates.append(p) - continue - - # fallback to searching the file content for docid or article_name - try: - with open(p, 'r', encoding='utf-8', errors='ignore') as pf: - txt = pf.read() - except Exception: - txt = "" - if docid and docid in txt: - candidates.append(p) - continue - if article_name and article_name in txt: - candidates.append(p) - continue - - # heuristic filename checks - if 'image_' in bn.lower() or 'thumbnail' in bn.lower() or bn.lower().startswith('img_'): - candidates.append(p) - return sorted(set(candidates)) - - # gather imageGroup ids from the capture JSON (if present) - image_group_ids = [] - try: - # common structure: data.get('imageGroups') -> list of dicts with imageGroupId - igs = None - if isinstance(data, dict): - igs = data.get('imageGroups') or data.get('image_groups') - if isinstance(igs, list): - for it in igs: - if isinstance(it, dict) and it.get('imageGroupId'): - image_group_ids.append(it.get('imageGroupId')) - except Exception: - image_group_ids = [] - - img_candidates = find_image_candidates(input_dir, docid, article_name, image_group_ids) - copied_map = {} - if img_candidates: - images_dir = os.path.join(target_dir, "images") - os.makedirs(images_dir, exist_ok=True) - - # Expand media JSON candidates: parse them for referenced image filenames - expanded = list(img_candidates) - # mapping of imageId -> caption discovered in media JSONs - media_captions = {} - for candidate in list(img_candidates): - bn = os.path.basename(candidate).lower() - if bn.endswith('.json') and ('media' in bn or 'image' in bn or 'resource' in bn): - try: - with open(candidate, 'r', encoding='utf-8') as mf: - mj = json.load(mf) - except Exception: - mj = None - if mj is None: - continue - - # media JSONs may be lists of group objects or a dict; normalize to list - entries = mj if isinstance(mj, list) else ([mj] if isinstance(mj, dict) else []) - found_image_ids = set() - for entry in entries: - if not isinstance(entry, dict): - continue - gid = entry.get('groupId') or entry.get('imageGroupId') or entry.get('groupID') - # if group id matches any of the capture's image group ids, collect imageIds - if (not image_group_ids) or (gid and gid in image_group_ids): - images = entry.get('images') or entry.get('imageList') or [] - if isinstance(images, list): - for img_obj in images: - if isinstance(img_obj, dict): - iid = img_obj.get('imageId') or img_obj.get('id') - if iid: - found_image_ids.add(iid) - # also consider thumbnailUrl fields - thumb = img_obj.get('thumbnailUrl') or img_obj.get('url') or img_obj.get('src') - if isinstance(thumb, str): - m = re.search(r"/thumbnail/([0-9a-fA-F-]+)", thumb) - if m: - found_image_ids.add(m.group(1)) - - # for every found image id, record caption and glob files in input_dir that contain it - for iid in found_image_ids: - # capture caption from media JSON (if available) - # find the image object in entries to extract caption - for entry in entries: - if not isinstance(entry, dict): - continue - images = entry.get('images') or entry.get('imageList') or [] - if isinstance(images, list): - for img_obj in images: - if isinstance(img_obj, dict): - iid2 = img_obj.get('imageId') or img_obj.get('id') - if iid2 == iid: - cap = img_obj.get('caption') or img_obj.get('title') or img_obj.get('imageTitle') - if isinstance(cap, str) and cap.strip(): - media_captions[iid] = cap.strip() - for match in glob.glob(os.path.join(input_dir, f"*{iid}*")): - if match not in expanded: - expanded.append(match) - - # now copy files for all expanded candidates, grouping by stem - for src in expanded: - base = os.path.basename(src) - stem = base.split(".")[0] - for match in glob.glob(os.path.join(input_dir, stem + "*")): - try: - dst = os.path.join(images_dir, os.path.basename(match)) - if os.path.abspath(match) == os.path.abspath(dst): - continue - shutil.copy2(match, dst) - copied_map[os.path.basename(match)] = os.path.join("images", os.path.basename(match)) - # if match references a known media imageId, record caption association so we can render it in the article - for iid, cap in media_captions.items(): - if iid in os.path.basename(match): - # store caption mapping keyed by the relative path - try: - # use a separate mapping to avoid clobbering copied_map - if '_captions_map' not in locals(): - _captions_map = {} - except Exception: - _captions_map = {} - _captions_map[copied_map[os.path.basename(match)]] = cap - except Exception: - continue - - # rewrite markdown image links that reference captured filenames or urls - image_list = [] - if copied_map: - # for each original filename, replace occurrences in md - for orig, rel in copied_map.items(): - # plain filename - md = md.replace(orig, rel) - # URL-encoded variants (common in src attributes) - try: - md = md.replace(urllib.parse.quote(orig), rel) - except Exception: - pass - # if the original was a full URL, replace filename part - md = re.sub(r"https?://[^)\s]*" + re.escape(orig), rel, md) - # track image files (only common image extensions) - lower = orig.lower() - if any(lower.endswith(ext) for ext in ('.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg')): - image_list.append(rel) - else: - # also consider the rel if it points to an actual image file in copied_map - if any(rel.lower().endswith(ext) for ext in ('.png', '.jpg', '.jpeg', '.gif', '.webp', '.svg')): - image_list.append(rel) - - # If we copied images, append an Images section to the markdown and log the association - if image_list: - # try to extract captions from any copied .meta.json files next to the images - captions = {} - for img_rel in list(image_list): - img_path = os.path.join(target_dir, img_rel) - meta_path = None - # look for .meta.json or .json in the images dir - base = os.path.basename(img_path) - stem = os.path.splitext(base)[0] - for cand in (stem + '.meta.json', stem + '.json'): - candp = os.path.join(os.path.dirname(img_path), cand) - if os.path.exists(candp): - meta_path = candp - break - if meta_path: - try: - with open(meta_path, 'r', encoding='utf-8') as mf: - mj = json.load(mf) - # common keys for captions/titles - for k in ('caption', 'alt', 'title', 'name'): - if k in mj and isinstance(mj[k], str) and mj[k].strip(): - captions[img_rel] = mj[k].strip() - break - except Exception: - pass - # if a caption was discovered earlier during media JSON parsing, use it (keyed by relative path) - try: - if '_captions_map' in locals() and img_rel in _captions_map and img_rel not in captions: - captions[img_rel] = _captions_map[img_rel] - except Exception: - pass - - # build Images markdown block - imgs_md_lines = ["\n\n## Images\n"] - for img in image_list: - cap = captions.get(img) or os.path.basename(img) - imgs_md_lines.append(f"![{cap}]({img})") - imgs_md_lines.append('') - - md = md.rstrip() + "\n" + "\n".join(imgs_md_lines) + "\n" - - # update a top-level images index so we can see associations across run try: - images_index_path = os.path.join(out_dir, "_images_index.json") - images_index = {} - if os.path.exists(images_index_path): - with open(images_index_path, 'r', encoding='utf-8') as ixf: - try: - images_index = json.load(ixf) - except Exception: - images_index = {} - # store relative image paths for this article - rels = image_list - images_index[out_path] = rels - with open(images_index_path, 'w', encoding='utf-8') as ixf: - json.dump(images_index, ixf, indent=2, ensure_ascii=False) - except Exception: - pass + image_data = IMAGE_GROUPS[group_id] + image_data_to_add[image_data["name"]] = image_data["images"] - # append per-target saved images log - try: - log_path = os.path.join(target_dir, "_saved_images.log") - from datetime import timezone - with open(log_path, 'a', encoding='utf-8') as lf: - lf.write(f"{datetime.now(timezone.utc).isoformat()}\t{out_path}\t{','.join(image_list)}\n") - except Exception: - pass - - except Exception: - # non-fatal - pass + except KeyError: + logger.warning(f"Image group {group_id} not found in IMAGE_GROUPS.") + + logger.debug(f"Extracted image data to add: {pformat(image_data_to_add)}") # Before writing, compute a normalized content hash to detect duplicates def normalize_for_hash(s: str) -> str: @@ -979,8 +739,39 @@ def main(argv: Iterable[str] | None = None) -> int: print(f"No files found in {input_dir} matching {args.pattern}") return 1 + # Start by caching image group metadata + for path in files: + base = os.path.basename(path) + # only process files that match the desired prefix + if not base.startswith("app.statdx.com_document_") or base.endswith("meta.json") or not "_media_" in base: + continue + + logger.debug(f"Caching image groups from media file: {path}") + with open(path, "r", encoding="utf-8") as f: + image_group_data = json.load(f) + + for group in image_group_data: + + group_id = group.get("groupId") + group_name = group.get("name") + images = group.get("images", []) + IMAGE_GROUPS[group_id] = { + "name": group_name, + "images": images, + } + #logger.debug(f"Image group {n}: {pformat(group)}") + + logger.debug(f"Cached {len(IMAGE_GROUPS)} image groups from media files.") + ok = 0 for path in files: + base = os.path.basename(path) + # only process files that match the desired prefix + if not base.startswith("app.statdx.com_document_") or base.endswith("meta.json"): + #if args.verbose: + # print(f"SKIP (not matching prefix): {path}") + continue + success, info = process_file(path, output_dir, overwrite=args.overwrite) if success: ok += 1 @@ -989,6 +780,7 @@ def main(argv: Iterable[str] | None = None) -> int: else: if args.verbose: print(f"SKIP: {path} -> {info}") + continue print(f"Converted {ok}/{len(files)} files to Markdown in {output_dir}") return 0