ross/statdx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Cache and manage image group metadata for enhanced document processing

Browse Source
This commit is contained in:
Ross
2025-10-18 09:20:57 +01:00
parent 2585d69fbe
commit 90ac197cbe
1 changed files with 48 additions and 256 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scrapers/document_to_markdown.py
+49 -257
View File
@@ -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 ---
image_data_to_add = {}
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")
try:
input_dir = os.path.dirname(path) or "."
image_data = IMAGE_GROUPS[group_id]
image_data_to_add[image_data["name"]] = image_data["images"]
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.
except KeyError:
logger.warning(f"Image group {group_id} not found in IMAGE_GROUPS.")
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 <name>.meta.json or <name>.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
# 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
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