Files
statdx/scrapers/document_to_markdown.py
T
Ross 2f8df5f820 Add comprehensive articles on Vascular Dementia and Wallerian Degeneration
- Created a detailed article for Vascular Dementia covering key facts, terminology, imaging findings, differential diagnoses, pathology, clinical issues, and diagnostic checklist.
- Developed an extensive article on Wallerian Degeneration including key facts, terminology, imaging features, differential diagnoses, pathology, clinical issues, and diagnostic checklist.
2025-10-17 22:11:27 +01:00

718 lines
27 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
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"![{alt}]({src})"
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 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
# 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", help="Clear output directory before processing")
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())