Ross
999 lines
41 KiB
Python
999 lines
41 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Convert captured STATdx document JSON files (containing HTML) into Markdown files.
|
|
|
|
Usage:
|
|
python scrapers/document_to_markdown.py --input-dir xhr_captured_async --output-dir docs_md
|
|
|
|
The script looks for .json files in the input directory. For each file it tries
|
|
to extract HTML from common keys like `documentHtml`, `html`, or `content` and
|
|
converts that HTML to Markdown using BeautifulSoup-based rules.
|
|
|
|
Output: a .md file next to the JSON (or in --output-dir) with the same base name.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import glob
|
|
import json
|
|
import os
|
|
import re
|
|
from datetime import datetime
|
|
import shutil
|
|
from typing import Iterable
|
|
|
|
from bs4 import BeautifulSoup, NavigableString, Tag
|
|
import html
|
|
import hashlib
|
|
import urllib.parse
|
|
|
|
|
|
def text_of(node) -> str:
|
|
"""Return the text content of a node, stripping extra whitespace."""
|
|
if node is None:
|
|
return ""
|
|
s = node.get_text(separator=" ", strip=True) if hasattr(node, "get_text") else str(node).strip()
|
|
# collapse multiple spaces
|
|
return re.sub(r"\s+", " ", s)
|
|
|
|
|
|
def node_to_md(node, indent=0) -> str:
|
|
"""Recursively convert a BeautifulSoup node to Markdown."""
|
|
if node is None:
|
|
return ""
|
|
|
|
if isinstance(node, NavigableString):
|
|
return str(node)
|
|
|
|
if isinstance(node, Tag):
|
|
name = node.name.lower()
|
|
# headings
|
|
if name in ("h1", "h2", "h3", "h4", "h5", "h6"):
|
|
level = int(name[1])
|
|
return "\n" + ("#" * level) + " " + text_of(node) + "\n\n"
|
|
|
|
if name == "p":
|
|
return "\n" + text_of(node) + "\n\n"
|
|
|
|
if name in ("strong", "b"):
|
|
return "**" + text_of(node) + "**"
|
|
|
|
if name in ("em", "i"):
|
|
return "*" + text_of(node) + "*"
|
|
|
|
if name == "a":
|
|
href = node.get("href") or ""
|
|
text = text_of(node) or href
|
|
return f"[{text}]({href})"
|
|
|
|
if name == "img":
|
|
src = node.get("src") or node.get("data-src") or ""
|
|
alt = node.get("alt") or ""
|
|
return f""
|
|
|
|
if name in ("code",) and node.parent and node.parent.name == "pre":
|
|
# handled in pre
|
|
return str(node)
|
|
|
|
if name == "pre":
|
|
# code block
|
|
code_text = node.get_text() or ""
|
|
# try to detect language from class like language-python
|
|
classes = " ".join(node.get("class") or [])
|
|
lang_match = re.search(r"language-([a-zA-Z0-9_+-]+)", classes)
|
|
lang = lang_match.group(1) if lang_match else ""
|
|
fence = "```" + (lang or "")
|
|
return "\n" + fence + "\n" + code_text.rstrip() + "\n" + "```\n\n"
|
|
|
|
if name in ("ul", "ol"):
|
|
out = "\n"
|
|
for li in node.find_all("li", recursive=False):
|
|
prefix = "- " if name == "ul" else "1. "
|
|
# indent nested lists
|
|
content = node_to_md(li, indent=indent + 2).strip()
|
|
content = content.replace("\n", "\n" + " " * (indent + 2))
|
|
out += " " * indent + prefix + content + "\n"
|
|
out += "\n"
|
|
return out
|
|
|
|
if name == "li":
|
|
parts = []
|
|
for child in node.children:
|
|
parts.append(node_to_md(child, indent=indent))
|
|
return "".join(parts)
|
|
|
|
if name == "blockquote":
|
|
content = text_of(node)
|
|
lines = content.splitlines()
|
|
return "\n" + "\n".join(
|
|
"> " + line for line in lines if line.strip()
|
|
) + "\n\n"
|
|
|
|
if name == "table":
|
|
# simple table conversion: try header row then body rows
|
|
rows = []
|
|
for tr in node.find_all("tr"):
|
|
cells = [text_of(td) for td in tr.find_all(["th", "td"])]
|
|
rows.append(cells)
|
|
if not rows:
|
|
return ""
|
|
# header is first row
|
|
header = rows[0]
|
|
sep = ["---"] * len(header)
|
|
out = "| " + " | ".join(header) + " |\n"
|
|
out += "| " + " | ".join(sep) + " |\n"
|
|
for r in rows[1:]:
|
|
out += "| " + " | ".join(r) + " |\n"
|
|
out += "\n"
|
|
return out
|
|
|
|
# block-level elements we want to preserve newlines for
|
|
if name in ("div", "section", "article", "main", "header", "footer"):
|
|
parts = [node_to_md(child, indent=indent) for child in node.children]
|
|
return "".join(parts)
|
|
|
|
# fallback: inline rendering of children
|
|
parts = []
|
|
for child in node.children:
|
|
parts.append(node_to_md(child, indent=indent))
|
|
return "".join(parts)
|
|
|
|
# should never get here, but ensure a string is returned
|
|
return ""
|
|
|
|
|
|
def html_to_markdown(html: str) -> str:
|
|
"""Convert HTML fragment/string to markdown text."""
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
# If there's an <article> use that, otherwise body, otherwise whole doc
|
|
candidate = soup.find("article") or soup.find("body") or soup
|
|
md = node_to_md(candidate)
|
|
# cleanup: collapse 3+ newlines to 2
|
|
md = re.sub(r"\n{3,}", "\n\n", md)
|
|
# strip leading/trailing whitespace
|
|
return md.strip() + "\n"
|
|
|
|
|
|
def find_html_in_json(obj) -> str | None:
|
|
"""Heuristics to find an HTML payload in a JSON object."""
|
|
if not isinstance(obj, dict):
|
|
return None
|
|
# common keys
|
|
for key in ("documentHtml", "html", "content", "document_html", "bodyHtml"):
|
|
if key in obj and isinstance(obj[key], str) and ("<" in obj[key] and ">" in obj[key]):
|
|
return obj[key]
|
|
|
|
# search nested
|
|
for v in obj.values():
|
|
if isinstance(v, str) and "<" in v and ">" in v:
|
|
return v
|
|
return None
|
|
|
|
|
|
def slugify(s: str, max_len: int = 80) -> str:
|
|
"""Create a filesystem-friendly slug from a string."""
|
|
if not s:
|
|
return ""
|
|
s = s.lower()
|
|
# replace spaces and slashes with hyphens
|
|
s = re.sub(r"[\s/]+", "-", s)
|
|
# remove characters that are not alnum, dash, or underscore
|
|
s = re.sub(r"[^a-z0-9\-_]+", "", s)
|
|
# collapse multiple hyphens
|
|
s = re.sub(r"-+", "-", s)
|
|
s = s.strip("-_")
|
|
if max_len and len(s) > max_len:
|
|
s = s[:max_len].rstrip("-_")
|
|
return s
|
|
|
|
|
|
def find_article_name_in_json(obj) -> str | None:
|
|
"""Look for an article name in known keys or nested values.
|
|
|
|
Common keys include: aname, articleName, name, title, documentTitle
|
|
"""
|
|
if not isinstance(obj, dict):
|
|
return None
|
|
|
|
candidates = [
|
|
"aname",
|
|
"articleName",
|
|
"article_name",
|
|
"name",
|
|
"title",
|
|
"documentTitle",
|
|
"document_title",
|
|
]
|
|
for key in candidates:
|
|
if key in obj:
|
|
v = obj.get(key)
|
|
if isinstance(v, str) and v.strip():
|
|
return v.strip()
|
|
|
|
# try shallow nested search for non-empty strings
|
|
for v in obj.values():
|
|
if isinstance(v, str) and v.strip():
|
|
# skip values that look like html (we prefer explicit name keys)
|
|
if "<" in v or ">" in v:
|
|
continue
|
|
# short string looks promising
|
|
if len(v.strip()) <= 200:
|
|
return v.strip()
|
|
|
|
return None
|
|
|
|
|
|
def extract_title_from_html(html: str) -> str | None:
|
|
"""Try to extract a human-friendly title from HTML content.
|
|
|
|
Checks in order: <h1>, <meta property="og:title">, <meta name="title">, <title>, then first <h2>.
|
|
"""
|
|
if not html:
|
|
return None
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
# h1 first
|
|
h1 = soup.find("h1")
|
|
if h1 and text_of(h1).strip():
|
|
return text_of(h1).strip()
|
|
|
|
# OpenGraph title
|
|
og = soup.find("meta", property="og:title")
|
|
if og and og.get("content"):
|
|
return og.get("content").strip()
|
|
|
|
# meta name=title
|
|
m = soup.find("meta", attrs={"name": "title"})
|
|
if m and m.get("content"):
|
|
return m.get("content").strip()
|
|
|
|
# document <title>
|
|
t = soup.find("title")
|
|
if t and t.string:
|
|
return t.string.strip()
|
|
|
|
# fallback to first h2
|
|
h2 = soup.find("h2")
|
|
if h2 and text_of(h2).strip():
|
|
return text_of(h2).strip()
|
|
|
|
return None
|
|
|
|
|
|
def recursive_search_for_key(obj, key: str):
|
|
"""Recursively search for the first occurrence of key in a nested JSON-like object."""
|
|
if isinstance(obj, dict):
|
|
if key in obj and isinstance(obj[key], str) and obj[key].strip():
|
|
return obj[key].strip()
|
|
for v in obj.values():
|
|
res = recursive_search_for_key(v, key)
|
|
if res:
|
|
return res
|
|
elif isinstance(obj, list):
|
|
for item in obj:
|
|
res = recursive_search_for_key(item, key)
|
|
if res:
|
|
return res
|
|
return None
|
|
|
|
|
|
def recursive_search_for_value(obj, value: str) -> bool:
|
|
"""Return True if `value` appears as a substring in any string value inside obj."""
|
|
if value is None:
|
|
return False
|
|
if isinstance(obj, str):
|
|
return value in obj
|
|
if isinstance(obj, dict):
|
|
for v in obj.values():
|
|
if recursive_search_for_value(v, value):
|
|
return True
|
|
elif isinstance(obj, list):
|
|
for it in obj:
|
|
if recursive_search_for_value(it, value):
|
|
return True
|
|
return False
|
|
|
|
|
|
def find_title_for_docid(docid: str, search_dir: str) -> str | None:
|
|
"""Look for files in search_dir that contain the docid and try to extract a title.
|
|
|
|
This checks nearby summary/media JSONs which often contain a `title` field.
|
|
"""
|
|
if not docid:
|
|
return None
|
|
|
|
# Phase 1: scan all JSON files for breadcrumb/list-style captures and prefer
|
|
# their enhancedDocumentName/name when present. Breadcrumb captures often
|
|
# don't include the docid in their filename, so scan every JSON once.
|
|
all_paths = sorted(glob.glob(os.path.join(search_dir, "*.json")))
|
|
for path in all_paths:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except Exception:
|
|
continue
|
|
if isinstance(data, list):
|
|
for item in data:
|
|
try:
|
|
if isinstance(item, dict) and item.get("documentId") == docid:
|
|
ed = item.get("enhancedDocumentName") or item.get("name")
|
|
if isinstance(ed, str) and ed.strip():
|
|
return html.unescape(ed.strip())
|
|
except Exception:
|
|
continue
|
|
|
|
# Phase 2: look only at files whose filename contains the docid for other
|
|
# title signals (top-level names, recursive title keys, searchResults).
|
|
pattern = os.path.join(search_dir, f"*{docid}*.json")
|
|
paths = sorted(glob.glob(pattern))
|
|
for path in paths:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except Exception:
|
|
continue
|
|
|
|
# top-level keys commonly used
|
|
if isinstance(data, dict):
|
|
for key in ("enhancedDocumentName", "documentName", "name"):
|
|
if key in data:
|
|
v = data.get(key)
|
|
if isinstance(v, str) and v.strip():
|
|
return html.unescape(v.strip())
|
|
|
|
# fallback: direct title anywhere in the JSON
|
|
title = recursive_search_for_key(data, "title")
|
|
if title and isinstance(title, str) and title.strip():
|
|
return html.unescape(title.strip())
|
|
|
|
# sometimes summary objects contain results with titles
|
|
sr = data.get("searchResults") if isinstance(data, dict) else None
|
|
if isinstance(sr, dict):
|
|
results = sr.get("results")
|
|
if isinstance(results, list):
|
|
for r in results:
|
|
if isinstance(r, dict) and r.get("id") == docid and r.get("title"):
|
|
return html.unescape(r.get("title"))
|
|
|
|
return None
|
|
|
|
|
|
def find_breadcrumbs_for_docid(docid: str, search_dir: str) -> list:
|
|
"""Return a list of breadcrumb strings associated with docid by scanning
|
|
captured JSONs. Preserves discovery order and HTML-unescapes values.
|
|
"""
|
|
if not docid:
|
|
return []
|
|
|
|
def collect_names_from_breadcrumbs(bc_list):
|
|
names = []
|
|
if isinstance(bc_list, list):
|
|
for item in bc_list:
|
|
if isinstance(item, dict) and item.get("name"):
|
|
names.append(html.unescape(item.get("name")))
|
|
return names
|
|
|
|
# First, prefer files whose filename contains the docid. These often
|
|
# include a structured `breadcrumbs` list (ordered parents) and may
|
|
# include a leaf node entry for the document.
|
|
candidates = []
|
|
doc_paths = sorted(glob.glob(os.path.join(search_dir, f"*{docid}*.json")))
|
|
for path in doc_paths:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except Exception:
|
|
continue
|
|
|
|
# look for a breadcrumbs list anywhere in this JSON
|
|
# and prefer the first full list we find
|
|
def recurse_find_breadcrumb_lists(obj):
|
|
out = []
|
|
if isinstance(obj, dict):
|
|
for k, v in obj.items():
|
|
if k == "breadcrumbs" and isinstance(v, list):
|
|
names = collect_names_from_breadcrumbs(v)
|
|
if names:
|
|
out.append(names)
|
|
out.extend(recurse_find_breadcrumb_lists(v))
|
|
elif isinstance(obj, list):
|
|
for item in obj:
|
|
out.extend(recurse_find_breadcrumb_lists(item))
|
|
return out
|
|
|
|
bc_lists = recurse_find_breadcrumb_lists(data)
|
|
if bc_lists:
|
|
# take the first breadcrumbs list as base path
|
|
base = bc_lists[0].copy()
|
|
# try to find a leaf name for this doc within the JSON
|
|
leaf = None
|
|
def recurse_find_leaf(obj):
|
|
nonlocal leaf
|
|
if leaf:
|
|
return
|
|
if isinstance(obj, dict):
|
|
# node that references the document directly
|
|
if obj.get("documentId") == docid and obj.get("name"):
|
|
leaf = html.unescape(obj.get("name"))
|
|
return
|
|
# some files use enhancedDocumentName/documentName
|
|
for k in ("enhancedDocumentName", "documentName", "name", "title"):
|
|
if k in obj and isinstance(obj[k], str) and obj.get("documentId") == docid:
|
|
leaf = html.unescape(obj.get(k))
|
|
return
|
|
for v in obj.values():
|
|
recurse_find_leaf(v)
|
|
elif isinstance(obj, list):
|
|
for item in obj:
|
|
recurse_find_leaf(item)
|
|
|
|
recurse_find_leaf(data)
|
|
if leaf and (not base or base[-1] != leaf):
|
|
base.append(leaf)
|
|
if base and base not in candidates:
|
|
candidates.append(base)
|
|
|
|
# If we found candidate breadcrumb paths from docid-matching files, return the first
|
|
if candidates:
|
|
return candidates[0]
|
|
|
|
# Otherwise, scan all JSONs for entries that explicitly list the docid and
|
|
# attempt to reconstruct a breadcrumb path by combining any nearby breadcrumbs
|
|
all_paths = sorted(glob.glob(os.path.join(search_dir, "*.json")))
|
|
for path in all_paths:
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except Exception:
|
|
continue
|
|
|
|
# list-of-dicts captures where an item references the documentId
|
|
if isinstance(data, list):
|
|
for item in data:
|
|
if isinstance(item, dict) and item.get("documentId") == docid:
|
|
# if this file also contains a breadcrumbs list elsewhere, use it
|
|
bc_lists = []
|
|
if isinstance(data, dict):
|
|
bc_lists = []
|
|
# try to locate breadcrumbs within same file by re-loading
|
|
def find_bcs(obj):
|
|
if isinstance(obj, dict):
|
|
if obj.get("breadcrumbs") and isinstance(obj.get("breadcrumbs"), list):
|
|
return collect_names_from_breadcrumbs(obj.get("breadcrumbs"))
|
|
for v in obj.values():
|
|
res = find_bcs(v)
|
|
if res:
|
|
return res
|
|
elif isinstance(obj, list):
|
|
for it in obj:
|
|
res = find_bcs(it)
|
|
if res:
|
|
return res
|
|
return None
|
|
|
|
bcs = find_bcs(data)
|
|
path_names = bcs or []
|
|
# append the item's own enhancedDocumentName/name if present
|
|
leaf = item.get("enhancedDocumentName") or item.get("name")
|
|
if isinstance(leaf, str) and leaf.strip():
|
|
leaf = html.unescape(leaf.strip())
|
|
if not path_names or path_names[-1] != leaf:
|
|
path_names = path_names + [leaf]
|
|
if path_names:
|
|
return path_names
|
|
|
|
return []
|
|
|
|
|
|
def find_title_in_capture_index(docid: str, base_dir: str) -> str | None:
|
|
"""Look for a title for docid inside a capture_index.jsonl file in base_dir.
|
|
|
|
capture_index.jsonl contains many JSON lines with search results; prefer titles found there.
|
|
"""
|
|
idx_path = os.path.join(base_dir, "capture_index.jsonl")
|
|
if not os.path.exists(idx_path):
|
|
# try parent
|
|
idx_path = os.path.join(os.path.dirname(base_dir), "capture_index.jsonl")
|
|
if not os.path.exists(idx_path):
|
|
return None
|
|
|
|
try:
|
|
with open(idx_path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
j = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
# top-level id
|
|
if j.get("id") == docid and j.get("title"):
|
|
return j.get("title")
|
|
# inside searchResults.results
|
|
sr = j.get("searchResults")
|
|
if isinstance(sr, dict):
|
|
results = sr.get("results")
|
|
if isinstance(results, list):
|
|
for r in results:
|
|
if isinstance(r, dict) and r.get("id") == docid and r.get("title"):
|
|
return r.get("title")
|
|
except Exception:
|
|
return None
|
|
return None
|
|
|
|
|
|
def process_file(path: str, out_dir: str, overwrite: bool = False) -> tuple[bool, str]:
|
|
"""Process one JSON file. Returns (success, output_path_or_error)."""
|
|
base = os.path.basename(path)
|
|
name, _ = os.path.splitext(base)
|
|
|
|
# attempt to extract article name for nicer filenames
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data_peek = json.load(f)
|
|
except Exception:
|
|
data_peek = None
|
|
|
|
# deterministic doc id (if present in filename)
|
|
docid = None
|
|
|
|
article_name = None
|
|
if isinstance(data_peek, dict):
|
|
article_name = find_article_name_in_json(data_peek)
|
|
|
|
# if no explicit article name, try to infer from docid present in filename
|
|
if not article_name:
|
|
# try to parse a uuid-like docid from filename
|
|
m = re.search(r"([0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})", base)
|
|
if m:
|
|
docid = m.group(1)
|
|
# Prefer nearby breadcrumb/document captures (which include
|
|
# enhancedDocumentName/name) over titles that appear in
|
|
# capture_index.jsonl. This makes section-level names
|
|
# like "Brain Tumor in Newborn/Infant" the canonical title.
|
|
title_from_nearby = find_title_for_docid(docid, os.path.dirname(path) or ".")
|
|
if not title_from_nearby:
|
|
title_from_nearby = find_title_in_capture_index(docid, os.path.dirname(path) or ".")
|
|
if title_from_nearby:
|
|
article_name = title_from_nearby
|
|
|
|
# if still not found, try to extract from inline HTML
|
|
if not article_name and isinstance(data_peek, dict):
|
|
html_candidate = find_html_in_json(data_peek)
|
|
if html_candidate:
|
|
title_from_html = extract_title_from_html(html_candidate)
|
|
if title_from_html:
|
|
article_name = title_from_html
|
|
|
|
if article_name:
|
|
slug = slugify(article_name)
|
|
# prefer deterministic filename based on docid when available to avoid duplicates
|
|
if docid:
|
|
out_base = f"{slug}_{docid}" if slug else docid
|
|
else:
|
|
out_base = f"{slug}_{name}" if slug else name
|
|
# place titled articles into an articles/ subfolder
|
|
target_dir = os.path.join(out_dir, "articles")
|
|
|
|
if isinstance(article_name, str) and article_name.startswith("https"):
|
|
target_dir = os.path.join(out_dir, "external")
|
|
else:
|
|
# per request: only save files that have titles
|
|
return False, "no-title"
|
|
|
|
out_path = os.path.join(target_dir, out_base + ".md")
|
|
|
|
if os.path.exists(out_path) and not overwrite:
|
|
return False, f"exists: {out_path}"
|
|
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except Exception as e:
|
|
return False, f"json load error: {e}"
|
|
|
|
html = find_html_in_json(data)
|
|
if not html:
|
|
return False, "no-html-found"
|
|
|
|
md = html_to_markdown(html)
|
|
|
|
# If we have an article_name (extracted earlier for filename), prepend it as H1
|
|
try:
|
|
if article_name:
|
|
# avoid duplicating if the markdown already starts with the same header
|
|
first_line = md.lstrip().splitlines()[0] if md.strip() else ""
|
|
normalized_first = re.sub(r"[#\s]+", "", first_line).strip().lower()
|
|
normalized_title = re.sub(r"[^a-z0-9]+", "", article_name.lower())
|
|
if not normalized_first or normalized_title not in normalized_first:
|
|
# Build YAML frontmatter with title and breadcrumbs (and docid)
|
|
front_lines = ["---"]
|
|
# use JSON quoting to safely escape strings in YAML
|
|
front_lines.append(f"title: {json.dumps(article_name.strip())}")
|
|
if docid:
|
|
front_lines.append(f"docid: {json.dumps(docid)}")
|
|
# collect breadcrumbs and render as YAML list if present
|
|
crumbs = []
|
|
if docid:
|
|
crumbs = find_breadcrumbs_for_docid(docid, os.path.dirname(path) or ".")
|
|
if crumbs:
|
|
front_lines.append("breadcrumbs:")
|
|
for c in crumbs:
|
|
front_lines.append(f" - {json.dumps(c)}")
|
|
front_lines.append("---\n")
|
|
front = "\n".join(front_lines)
|
|
|
|
# Avoid duplicating an H1 that matches the title in the body
|
|
md_lines = md.lstrip().splitlines()
|
|
if md_lines:
|
|
first = md_lines[0].strip()
|
|
normalized_first = re.sub(r"[#\s]+", "", first).strip().lower()
|
|
normalized_title = re.sub(r"[^a-z0-9]+", "", article_name.lower())
|
|
if first.startswith("#") and normalized_title in normalized_first:
|
|
# drop the first line (existing H1)
|
|
md_lines = md_lines[1:]
|
|
|
|
md = front + "\n".join(md_lines).lstrip()
|
|
except Exception:
|
|
# be conservative: if anything goes wrong, keep the original md
|
|
pass
|
|
|
|
# (channel/frontmatter injection removed - handled elsewhere)
|
|
|
|
# --- find and save related image/response files ---
|
|
try:
|
|
input_dir = os.path.dirname(path) or "."
|
|
|
|
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.
|
|
|
|
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"")
|
|
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
|
|
|
|
# Before writing, compute a normalized content hash to detect duplicates
|
|
def normalize_for_hash(s: str) -> str:
|
|
# remove ISO8601-like timestamps and common date patterns, collapse whitespace
|
|
s2 = re.sub(r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:Z|[+-]\d{2}:?\d{2})?", "", s)
|
|
s2 = re.sub(r"\b\d{1,2}/\d{1,2}/\d{2,4}\b", "", s2)
|
|
# remove lines that are just timestamps or contain 'Last updated'
|
|
s2 = "\n".join([ln for ln in s2.splitlines() if not re.match(r"^\s*(Last updated|LastVisited|LastVisitedAsDate|Updated).*$", ln, re.I)])
|
|
s2 = re.sub(r"\s+", " ", s2).strip()
|
|
return s2
|
|
|
|
def content_hash(s: str) -> str:
|
|
nh = normalize_for_hash(s)
|
|
return hashlib.sha256(nh.encode("utf-8")).hexdigest()
|
|
|
|
# index stored at top-level out_dir to dedupe across articles/external
|
|
index_path = os.path.join(out_dir, "_content_index.json")
|
|
index = {}
|
|
try:
|
|
if os.path.exists(index_path):
|
|
with open(index_path, "r", encoding="utf-8") as ix:
|
|
index = json.load(ix)
|
|
except Exception:
|
|
index = {}
|
|
|
|
ch = content_hash(md)
|
|
existing = index.get(ch)
|
|
if existing and os.path.exists(existing) and not overwrite:
|
|
# content already saved elsewhere — skip writing
|
|
return False, f"duplicate: {existing}"
|
|
|
|
try:
|
|
os.makedirs(target_dir, exist_ok=True)
|
|
with open(out_path, "w", encoding="utf-8") as f:
|
|
f.write(md)
|
|
# update index
|
|
try:
|
|
index[ch] = out_path
|
|
with open(index_path, "w", encoding="utf-8") as ix:
|
|
json.dump(index, ix, indent=2, ensure_ascii=False)
|
|
except Exception:
|
|
pass
|
|
# log saved file path with timestamp
|
|
try:
|
|
log_path = os.path.join(target_dir, "_saved_files.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}\n")
|
|
except Exception:
|
|
# non-fatal if logging fails
|
|
pass
|
|
except Exception as e:
|
|
return False, f"write error: {e}"
|
|
|
|
return True, out_path
|
|
|
|
|
|
def main(argv: Iterable[str] | None = None) -> int:
|
|
p = argparse.ArgumentParser(description="Convert captured JSON HTML to Markdown")
|
|
p.add_argument("--input-dir", default="xhr_captured_async", help="Directory with captured .json files")
|
|
p.add_argument("--output-dir", default="docs_md", help="Where to write .md files (defaults to input-dir) ")
|
|
p.add_argument("--pattern", default="*.json", help="Glob pattern to find files in input dir")
|
|
p.add_argument("--overwrite", action="store_true", help="Overwrite existing .md files")
|
|
p.add_argument("--verbose", "-v", action="store_true", help="Print processing details")
|
|
p.add_argument("--clear", "-c", action="store_true", default=False, help="Clear output directory before processing (default: false)")
|
|
args = p.parse_args(list(argv) if argv is not None else None)
|
|
|
|
input_dir = args.input_dir
|
|
output_dir = args.output_dir or input_dir
|
|
|
|
if args.clear and os.path.exists(output_dir):
|
|
# remove all .md files in output_dir
|
|
shutil.rmtree(output_dir, ignore_errors=True)
|
|
|
|
files = sorted(glob.glob(os.path.join(input_dir, args.pattern)))
|
|
if not files:
|
|
print(f"No files found in {input_dir} matching {args.pattern}")
|
|
return 1
|
|
|
|
ok = 0
|
|
for path in files:
|
|
success, info = process_file(path, output_dir, overwrite=args.overwrite)
|
|
if success:
|
|
ok += 1
|
|
if args.verbose:
|
|
print(f"WROTE: {info}")
|
|
else:
|
|
if args.verbose:
|
|
print(f"SKIP: {path} -> {info}")
|
|
|
|
print(f"Converted {ok}/{len(files)} files to Markdown in {output_dir}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|