Add new markdown files for various documents and implement document conversion script
- Created markdown files for CDC Diagnostic Criteria for Creutzfeldt-Jakob Disease and multiple media documents. - Added a script to convert JSON documents containing HTML into markdown format. - Implemented functions to extract titles and HTML content from JSON files. - Enhanced markdown generation with support for various HTML elements including tables, lists, and images. - Included error handling and logging for file processing.
This commit is contained in:
@@ -0,0 +1,485 @@
|
||||
#!/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
|
||||
from typing import Iterable
|
||||
|
||||
from bs4 import BeautifulSoup, NavigableString, Tag
|
||||
|
||||
|
||||
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 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
|
||||
pattern = os.path.join(search_dir, f"*{docid}*.json")
|
||||
for path in sorted(glob.glob(pattern)):
|
||||
try:
|
||||
with open(path, "r", encoding="utf-8") as f:
|
||||
data = json.load(f)
|
||||
except Exception:
|
||||
continue
|
||||
# direct title
|
||||
title = recursive_search_for_key(data, "title")
|
||||
if title:
|
||||
# some files include empty titles; skip blanks
|
||||
if isinstance(title, str) and title.strip():
|
||||
return title.strip()
|
||||
# sometimes summary objects contain results with titles
|
||||
# try 'searchResults' -> 'results' list
|
||||
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 r.get("title")
|
||||
return None
|
||||
|
||||
|
||||
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
|
||||
|
||||
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)
|
||||
# first consult capture_index.jsonl if present
|
||||
title_from_index = find_title_in_capture_index(docid, os.path.dirname(path) or ".")
|
||||
if not title_from_index:
|
||||
title_from_index = find_title_for_docid(docid, os.path.dirname(path) or ".")
|
||||
if title_from_index:
|
||||
article_name = title_from_index
|
||||
|
||||
# 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)
|
||||
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 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:
|
||||
md = "# " + article_name.strip() + "\n\n" + md
|
||||
except Exception:
|
||||
# be conservative: if anything goes wrong, keep the original md
|
||||
pass
|
||||
|
||||
try:
|
||||
os.makedirs(target_dir, exist_ok=True)
|
||||
with open(out_path, "w", encoding="utf-8") as f:
|
||||
f.write(md)
|
||||
# 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
|
||||
for f in glob.glob(os.path.join(output_dir, "*.md")):
|
||||
try:
|
||||
os.remove(f)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
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())
|
||||
Reference in New Issue
Block a user