Ross
943 lines
38 KiB
Python
943 lines
38 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
|
|
from pprint import pformat
|
|
from loguru import logger
|
|
import sys
|
|
|
|
IMAGE_GROUPS = {}
|
|
CAPTURE_INPUT_DIR = None
|
|
DOCUMENT_SUMMARYS = {}
|
|
|
|
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_find_key_any(obj, key: str):
|
|
"""Recursively search for key in nested JSON-like object and return its value (any type)."""
|
|
if isinstance(obj, dict):
|
|
if key in obj:
|
|
return obj[key]
|
|
for v in obj.values():
|
|
res = recursive_find_key_any(v, key)
|
|
if res is not None:
|
|
return res
|
|
elif isinstance(obj, list):
|
|
for item in obj:
|
|
res = recursive_find_key_any(item, key)
|
|
if res is not None:
|
|
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_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)
|
|
|
|
logger.debug(f"Processing file: {path}")
|
|
|
|
# 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
|
|
|
|
docid = base.split("_document_content_")[1].split("_")[0]
|
|
|
|
logger.debug(f"Extracted docid: {docid}")
|
|
#logger.debug(f"DOCUMENT_SUMMARYS keys: {DOCUMENT_SUMMARYS.get(docid)}")
|
|
|
|
article_name = DOCUMENT_SUMMARYS.get(docid).get("title")
|
|
|
|
# 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
|
|
|
|
logger.debug(f"Determined article name: {article_name}")
|
|
|
|
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"):
|
|
return False, "external-url"
|
|
target_dir = os.path.join(out_dir, "external")
|
|
else:
|
|
# per request: only save files that have titles
|
|
logger.error(f"No article name found for {path}; skipping.")
|
|
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)}")
|
|
|
|
|
|
summary_data = DOCUMENT_SUMMARYS.get(docid, {})
|
|
|
|
# Try to extract summary fields from the capture JSON (data_peek) or the loaded data
|
|
def safe_get(k):
|
|
v = None
|
|
try:
|
|
v = recursive_find_key_any(summary_data, k)
|
|
except Exception:
|
|
v = None
|
|
return v
|
|
|
|
# Authors (list of {key,value})
|
|
authors = safe_get('authors')
|
|
if isinstance(authors, list) and authors:
|
|
front_lines.append('authors:')
|
|
for a in authors:
|
|
try:
|
|
key = a.get('key')
|
|
val = a.get('value')
|
|
front_lines.append(f" - key: {json.dumps(key)}")
|
|
front_lines.append(f" value: {json.dumps(val)}")
|
|
except Exception:
|
|
continue
|
|
|
|
# Breadcrumbs: include name/slug/treeNodeId chain if available
|
|
bcs = safe_get('breadcrumbs')
|
|
if isinstance(bcs, list) and bcs:
|
|
front_lines.append('breadcrumbs:')
|
|
for bc in bcs:
|
|
if not isinstance(bc, dict):
|
|
continue
|
|
name = bc.get('name')
|
|
slug = bc.get('slug')
|
|
tid = bc.get('treeNodeId') or bc.get('treeId')
|
|
front_lines.append(' -')
|
|
front_lines.append(f" name: {json.dumps(name)}")
|
|
front_lines.append(f" slug: {json.dumps(slug)}")
|
|
front_lines.append(f" treeNodeId: {json.dumps(tid)}")
|
|
|
|
# Misc summary fields
|
|
for key in ('category','cmeTopicId','documentVersionId','imageCount','lastUpdated','pageDescription','pageKeywords','pageTitle','enhancedTitle','type'):
|
|
val = safe_get(key)
|
|
if val is not None:
|
|
# booleans and numbers should be represented without JSON quoting
|
|
if isinstance(val, (bool,int)):
|
|
front_lines.append(f"{key}: {json.dumps(val)}")
|
|
else:
|
|
front_lines.append(f"{key}: {json.dumps(str(val))}")
|
|
# 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)
|
|
|
|
image_data_to_add = {}
|
|
|
|
if isinstance(data_peek, dict) and "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:
|
|
image_data = IMAGE_GROUPS[group_id]
|
|
image_data_to_add[image_data["name"]] = image_data["images"]
|
|
|
|
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)}")
|
|
|
|
# If we have image groups to add, copy matching files and append a formatted
|
|
# Images section to the markdown. We search the capture input directory
|
|
# recursively for filenames containing the imageId and copy any matches.
|
|
try:
|
|
if image_data_to_add:
|
|
images_dir = os.path.join(target_dir, "images")
|
|
os.makedirs(images_dir, exist_ok=True)
|
|
appended_images = []
|
|
images_md_lines = ["\n\n## Images\n"]
|
|
# map imageId -> (rel_path, caption) for replacement in-md
|
|
image_map: dict = {}
|
|
|
|
search_root = CAPTURE_INPUT_DIR or os.path.dirname(path) or "."
|
|
|
|
for group_name, images in image_data_to_add.items():
|
|
images_md_lines.append(f"\n### {group_name}\n")
|
|
# render each image in the group
|
|
for img in images:
|
|
try:
|
|
iid = img.get('imageId') or img.get('id')
|
|
caption = img.get('caption') or ""
|
|
thumb = img.get('thumbnailUrl') or ""
|
|
image_title = img.get('imageTitle') or img.get('title') or img.get('enhancedTitle') or ""
|
|
# find matching files under the capture input dir recursively
|
|
matches = []
|
|
if iid:
|
|
matches = glob.glob(os.path.join(search_root, "**", f"*{iid}*"), recursive=True)
|
|
# also consider thumbnail URL id in case different naming
|
|
if not matches and isinstance(thumb, str) and thumb:
|
|
m = re.search(r"/thumbnail/([0-9a-fA-F-\-]+)", thumb)
|
|
if m:
|
|
tid = m.group(1)
|
|
matches = glob.glob(os.path.join(search_root, "**", f"*{tid}*"), recursive=True)
|
|
|
|
rel_paths = []
|
|
for match in sorted(set(matches)):
|
|
# skip JSON metadata files; prefer binary images
|
|
b = os.path.basename(match)
|
|
if b.endswith('.json') or b.endswith('.meta.json'):
|
|
# still copy metadata alongside image when present
|
|
# we'll copy meta files later if we also find an image
|
|
continue
|
|
try:
|
|
dst = os.path.join(images_dir, os.path.basename(match))
|
|
if os.path.abspath(match) != os.path.abspath(dst):
|
|
shutil.copy2(match, dst)
|
|
rel = os.path.join('images', os.path.basename(match))
|
|
rel_paths.append(rel)
|
|
appended_images.append(rel)
|
|
# associate imageId -> rel + caption + title for later replacement
|
|
if iid:
|
|
# prefer explicit caption if available, and capture title
|
|
cval = caption or img.get('caption') or image_title or ''
|
|
image_map[iid] = (rel, cval, image_title)
|
|
except Exception:
|
|
continue
|
|
|
|
# If we found images, render them; otherwise, render the thumbnail URL if present
|
|
if rel_paths:
|
|
for rp in rel_paths:
|
|
cap = caption or os.path.basename(rp)
|
|
# render image and then visible title (if any) and caption on subsequent lines
|
|
images_md_lines.append(f"")
|
|
if image_title:
|
|
images_md_lines.append(f"**{image_title}**")
|
|
images_md_lines.append(f"*{cap}*")
|
|
images_md_lines.append("")
|
|
else:
|
|
# try to render a thumbnail path (as-is) if no local file
|
|
if thumb:
|
|
# normalize thumb path to a relative filename when possible
|
|
thumb_name = os.path.basename(thumb.split('?')[0])
|
|
images_md_lines.append(f"")
|
|
if image_title:
|
|
images_md_lines.append(f"**{image_title}**")
|
|
images_md_lines.append(f"*{caption}*")
|
|
images_md_lines.append("")
|
|
else:
|
|
# at minimum show caption text
|
|
if caption:
|
|
images_md_lines.append(f"*{caption}*")
|
|
images_md_lines.append("")
|
|
except Exception:
|
|
continue
|
|
|
|
# attach Images section to markdown
|
|
# Before attaching, replace inline references to these images so
|
|
# their alt text shows the caption (not original alt).
|
|
try:
|
|
for iid, (rel, cap, title) in image_map.items():
|
|
# replace markdown image links that include the imageId or filename
|
|
bn = os.path.basename(rel)
|
|
# 1) replace markdown image links where the URL contains the imageId
|
|
if title:
|
|
md = re.sub(r'!\[[^\]]*\]\([^)]*' + re.escape(iid) + r'[^)]*\)', f'\n\n**{title}**\n\n*{cap}*', md)
|
|
else:
|
|
md = re.sub(r'!\[[^\]]*\]\([^)]*' + re.escape(iid) + r'[^)]*\)', f'\n\n*{cap}*', md)
|
|
# 2) replace markdown image links where URL contains the basename
|
|
if title:
|
|
md = re.sub(r'!\[[^\]]*\]\([^)]*' + re.escape(bn) + r'[^)]*\)', f'\n\n**{title}**\n\n*{cap}*', md)
|
|
else:
|
|
md = re.sub(r'!\[[^\]]*\]\([^)]*' + re.escape(bn) + r'[^)]*\)', f'\n\n*{cap}*', md)
|
|
# 3) replace HTML <img ... src="...basename..."> with markdown image
|
|
if title:
|
|
md = re.sub(r'<img[^>]+src=["\"][^"\']*' + re.escape(bn) + r'[^"\']*["\"][^>]*>', f'\n\n**{title}**\n\n*{cap}*', md)
|
|
else:
|
|
md = re.sub(r'<img[^>]+src=["\"][^"\']*' + re.escape(bn) + r'[^"\']*["\"][^>]*>', f'\n\n*{cap}*', md)
|
|
# 4) replace full URLs ending with basename
|
|
md = re.sub(r'https?://[^)\s]*' + re.escape(bn), rel, md)
|
|
# 5) replace URL-encoded variants
|
|
try:
|
|
q = urllib.parse.quote(bn)
|
|
md = md.replace(q, rel)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|
|
|
|
md = md.rstrip() + "\n" + "\n".join(images_md_lines) + "\n"
|
|
|
|
# update a top-level images index so we can see associations across runs
|
|
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 = {}
|
|
images_index[out_path] = appended_images
|
|
with open(images_index_path, 'w', encoding='utf-8') as ixf:
|
|
json.dump(images_index, ixf, indent=2, ensure_ascii=False)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
# non-fatal; continue without images
|
|
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
|
|
|
|
# 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 ("_media_" not in base):
|
|
continue
|
|
|
|
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)}")
|
|
|
|
# Then cache document summary data
|
|
for path in files:
|
|
base = os.path.basename(path)
|
|
# only process files that match the desired prefix
|
|
if "document_summary" not in base or "meta" in base:
|
|
continue
|
|
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
summary_data = json.load(f)
|
|
|
|
doc_id = base.split("_document_summary_")[1].split("_")[0]
|
|
|
|
DOCUMENT_SUMMARYS[doc_id] = summary_data #logger.debug(f"Document summary {n}: {pformat(doc)}")
|
|
|
|
logger.debug(f"Cached {len(IMAGE_GROUPS)} image groups from media files.")
|
|
logger.debug(f"Cached {len(DOCUMENT_SUMMARYS)} document summaries from summary 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") or ("_media_" in base) or ("breadcrumbs" in base) or ("_summary_" in base) or ("_tables_" in base):
|
|
#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
|
|
if args.verbose:
|
|
print(f"WROTE: {info}")
|
|
else:
|
|
if args.verbose:
|
|
print(f"SKIP: {path} -> {info}")
|
|
continue
|
|
|
|
print(f"Converted {ok}/{len(files)} files to Markdown in {output_dir}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|