claudeandClaude Opus 4.8
cd30d0dc7a
HVM (HPE Morpheus VM Essentials) 9.0.0 shipped June 2026; the MCP had
nothing about it. Add it and stop hand-adding a docId per release.
scrape/bundles.py now AUTO-DISCOVERS new versions (discover_specs, on by
default): it scans the DocPortal docId range above the highest known
bundle and classifies each abstract. On-by-default in CI's refresh.yml,
so future releases flow into the corpus with no code change. Gotchas
encoded:
- classify on the abstract BODY, not the <h1> (a Release Notes h1 is
just "v9.0.0 Release Notes" with no product name)
- EXCLUDE the interleaved siblings that share the same version string:
Morpheus Enterprise, Morpheus Central, Storage Integration Pack
- failsoft: a scan error logs a warning and falls back to known bundles
- self-healing + idempotent: prior bundles.json persists discoveries;
the scan anchor rolls forward; re-runs add no duplicates
Discovered + scraped for 9.0.0:
- User Manual sd00008058en_us (414 pages; Deployment Guide now folded in)
- Release Notes sd00008079en_us
corpus: +412 pages (9.0), sidecar topic_cluster peers re-linked to 9.0
across the 8.1.x bundles (shared GUIDs). Local reindex: 3645 chunks
(956 are 9.0.0); 9.0 content verified retrievable end-to-end.
Co-Authored-By: Claude Opus 4.8 <[email protected]>
Claude-Session: https://claude.ai/code/session_01MNQRS6F7N9Uwh1SMn3xz2g
382 lines
16 KiB
Python
382 lines
16 KiB
Python
"""Discover HVM doc bundles on HPE Support DocPortal and write bundles.json.
|
|
|
|
HPE mints a NEW docId per product version rather than versioning a single
|
|
doc (see ~/.claude/.../reference_hpe_docs_portal_api.md). This script no
|
|
longer requires a human to hand-add each new version's docId: it
|
|
AUTO-DISCOVERS new HPE Morpheus VM Essentials versions by scanning the
|
|
docId numeric range above the highest known bundle and classifying each
|
|
hit by its abstract body.
|
|
|
|
The static BUNDLES list below is the evergreen floor — the unversioned /
|
|
special bundles (Deployment Guide, Qualification Matrix, QuickSpecs) plus
|
|
the versioned bundles known when this file was last hand-edited. Discovery
|
|
appends anything newer on top. So a new HVM release (e.g. 9.0) flows into
|
|
corpus/ through the weekly refresh with no code change.
|
|
|
|
For each bundle this script:
|
|
1. GETs /hpesc/public/api/document/{docId} → abstract HTML
|
|
2. GETs /hpesc/public/api/document/{docId}/toc → page tree (or 404 for single-doc)
|
|
3. Writes bundles.json at repo root with the schema PLAN.md Phase 1 documents.
|
|
|
|
Discovery (see discover_specs) is on by default; pass --no-discover to
|
|
build only from the static + prior-bundles.json set. It is FAILSOFT: a
|
|
network error during the range scan logs a warning and falls back to the
|
|
known bundles rather than breaking the refresh.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import re
|
|
import sys
|
|
import time
|
|
from concurrent.futures import ThreadPoolExecutor, as_completed
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
|
|
API = "https://support.hpe.com/hpesc/public/api/document"
|
|
DOC_URL = "https://support.hpe.com/hpesc/public/docDisplay?docId={doc_id}"
|
|
UA = "hvm-docs-mcp/0.1 (+https://git.jpaul.io/justin/hvm-docs; [email protected])"
|
|
ROOT = Path(__file__).resolve().parent.parent
|
|
BUNDLES_JSON = ROOT / "bundles.json"
|
|
|
|
|
|
@dataclass
|
|
class BundleSpec:
|
|
slug: str
|
|
doc_id: str
|
|
title: str
|
|
version: str | None
|
|
product: str # e.g. "User Manual", "Release Notes", "Deployment Guide"
|
|
mode: str # "toc", "single", or "html-file" (committed fixture under scrape/quickspecs/)
|
|
platform: str | None = None
|
|
language: str = "en-US"
|
|
source_url: str | None = None # overrides the default support.hpe.com URL
|
|
|
|
|
|
# Static floor. The unversioned/special bundles are authoritative here
|
|
# (Deployment Guide, Qualification Matrix, QuickSpecs); the versioned
|
|
# entries are the seed set. Newer versions are added automatically by
|
|
# discover_specs() — you should NOT need to hand-add a docId per release
|
|
# anymore. Versions confirmed 2026-05-22 by probing sd00007400..7740.
|
|
BUNDLES: list[BundleSpec] = [
|
|
BundleSpec("hvm_user_manual_8_1_0", "sd00007520en_us", "HPE Morpheus VM Essentials Software Documentation", "8.1.0", "User Manual", "toc"),
|
|
BundleSpec("hvm_user_manual_8_1_1", "sd00007620en_us", "HPE Morpheus VM Essentials Software Documentation", "8.1.1", "User Manual", "toc"),
|
|
BundleSpec("hvm_user_manual_8_1_2", "sd00007735en_us", "HPE Morpheus VM Essentials Software Documentation", "8.1.2", "User Manual", "toc"),
|
|
BundleSpec("hvm_release_notes_8_1_0", "sd00007497en_us", "HPE Morpheus VM Essentials Software Release Notes", "8.1.0", "Release Notes", "single"),
|
|
BundleSpec("hvm_release_notes_8_1_1", "sd00007609en_us", "HPE Morpheus VM Essentials Software Release Notes", "8.1.1", "Release Notes", "single"),
|
|
BundleSpec("hvm_release_notes_8_1_2", "sd00007734en_us", "HPE Morpheus VM Essentials Software Release Notes", "8.1.2", "Release Notes", "single"),
|
|
BundleSpec("hvm_deployment_guide", "sd00007332en_us", "HPE Morpheus VM Essentials Deployment Guide", None, "Deployment Guide","toc"),
|
|
BundleSpec("hvm_qualification_matrix","sd00006551en_us", "Qualification Matrix for HVM Clusters Managed by HPE Morpheus Software", None, "Qualification Matrix", "toc"),
|
|
# QuickSpecs is a static-HTML fixture (www.hpe.com edge drops automated
|
|
# connections — see scrape/quickspecs/README.md). doc_id = the QuickSpecs
|
|
# PSNow ref (a50004260enw). page_count is 1; source_url points at the
|
|
# public PSNow URL.
|
|
BundleSpec("hvm_quickspecs", "a50004260enw", "HPE Morpheus VM Essentials Software QuickSpecs",
|
|
"v4-2026-02-02", "QuickSpecs", "html-file",
|
|
source_url="https://www.hpe.com/psnow/doc/a50004260enw"),
|
|
]
|
|
|
|
|
|
def _session() -> requests.Session:
|
|
s = requests.Session()
|
|
s.headers.update({"User-Agent": UA, "Accept": "application/json, text/html"})
|
|
return s
|
|
|
|
|
|
def _get(s: requests.Session, url: str, expect_json: bool = False, retries: int = 4) -> Any:
|
|
delay = 1.0
|
|
for attempt in range(retries):
|
|
r = s.get(url, timeout=30)
|
|
if r.status_code == 200:
|
|
return r.json() if expect_json else r.text
|
|
if r.status_code == 404:
|
|
return None
|
|
if r.status_code in (429, 500, 502, 503, 504):
|
|
time.sleep(delay)
|
|
delay *= 2
|
|
continue
|
|
r.raise_for_status()
|
|
raise RuntimeError(f"GET failed after {retries} retries: {url}")
|
|
|
|
|
|
def _count_toc(toc: list[dict] | None) -> tuple[int, str | None]:
|
|
"""Returns (page_count, landing_page_guid)."""
|
|
if not toc:
|
|
return 0, None
|
|
landing = None
|
|
n = 0
|
|
|
|
def walk(nodes: list[dict] | None, depth: int) -> None:
|
|
nonlocal n, landing
|
|
for node in nodes or []:
|
|
link = node.get("topicLink")
|
|
if link:
|
|
n += 1
|
|
m = re.search(r"page=(GUID-[A-F0-9-]+)\.html", link)
|
|
if m and landing is None:
|
|
landing = m.group(1)
|
|
walk(node.get("children"), depth + 1)
|
|
|
|
walk(toc, 0)
|
|
return n, landing
|
|
|
|
|
|
def _parse_abstract(html: str) -> dict[str, str]:
|
|
"""Pull title / abstract text / published date out of the DITA abstract HTML."""
|
|
soup = BeautifulSoup(html, "html.parser")
|
|
out: dict[str, str] = {}
|
|
h1 = soup.select_one("h1.title.topictitle1")
|
|
if h1:
|
|
out["title"] = h1.get_text(" ", strip=True)
|
|
desc = soup.select_one("div.desc")
|
|
if desc:
|
|
out["abstract"] = desc.get_text(" ", strip=True)
|
|
pub = soup.select_one("div.publishedDate")
|
|
if pub:
|
|
out["published"] = pub.get_text(" ", strip=True).replace("Published:", "").strip()
|
|
return out
|
|
|
|
|
|
# --- Auto-discovery -------------------------------------------------------
|
|
#
|
|
# HPE interleaves sibling products in the same docId range and reuses the
|
|
# SAME version string across them: HVM 9.0.0, Morpheus *Enterprise* 9.0.0,
|
|
# and Morpheus *Central* all sit within ~30 docIds of each other. Worse,
|
|
# a Release Notes doc's <h1> is often just "v9.0.0 Release Notes" with no
|
|
# product name — so we MUST classify on the abstract body, never the title
|
|
# alone. INCLUDE requires the full "Morpheus VM Essentials" string; EXCLUDE
|
|
# drops the neighbours and the Storage Integration Pack add-on.
|
|
_INCLUDE_RE = re.compile(r"Morpheus VM Essentials", re.I)
|
|
_EXCLUDE_RE = re.compile(r"Storage Integration Pack|Morpheus Enterprise|Morpheus Central|OneView", re.I)
|
|
_VER_RE = re.compile(r"v?(\d+\.\d+(?:\.\d+)?)")
|
|
_DOCID_NUM_RE = re.compile(r"^sd0*(\d+)en_us$")
|
|
|
|
# Default forward scan window above the high-water docId. 9.0 landed ~344
|
|
# ids above the 8.1.2 seed; 600 leaves comfortable headroom and self-heals
|
|
# from the static floor if bundles.json is ever lost. Bump if a future
|
|
# release outruns it (a one-line change, still no per-version docId edits).
|
|
DISCOVER_WINDOW = 600
|
|
|
|
|
|
def _docid_num(doc_id: str) -> int | None:
|
|
m = _DOCID_NUM_RE.match(doc_id)
|
|
return int(m.group(1)) if m else None
|
|
|
|
|
|
def _slug_for(kind: str, version: str) -> str:
|
|
v = version.replace(".", "_").replace("-", "_")
|
|
base = "hvm_user_manual" if kind == "doc" else "hvm_release_notes"
|
|
return f"{base}_{v}"
|
|
|
|
|
|
def _classify_abstract(html: str) -> tuple[str, str, str] | None:
|
|
"""Return (kind, version, title) for an HVM bundle, else None.
|
|
|
|
kind is 'doc' (paginated User Manual / Software Documentation) or 'rn'
|
|
(single-doc Release Notes). Only these two well-understood shapes are
|
|
auto-added; anything else that mentions VM Essentials is logged for
|
|
manual review by the caller and skipped."""
|
|
meta = _parse_abstract(html)
|
|
title = meta.get("title", "")
|
|
blob = f"{title} {meta.get('abstract', '')}"
|
|
if not _INCLUDE_RE.search(blob) or _EXCLUDE_RE.search(blob):
|
|
return None
|
|
vm = _VER_RE.search(blob)
|
|
if not vm:
|
|
return None
|
|
version = vm.group(1)
|
|
if re.search(r"release note", blob, re.I):
|
|
return ("rn", version, title)
|
|
if re.search(r"software documentation", blob, re.I):
|
|
return ("doc", version, title)
|
|
return None # VM-Essentials-ish but not a UM/RN shape → caller logs it
|
|
|
|
|
|
def discover_specs(s: requests.Session, known: list[BundleSpec],
|
|
window: int = DISCOVER_WINDOW,
|
|
scan_start: int | None = None) -> list[BundleSpec]:
|
|
"""Scan the docId range above the known high-water mark for new HVM
|
|
versions and return BundleSpecs for any not already covered.
|
|
|
|
FAILSOFT: any per-docId fetch error is swallowed (the range is sparse
|
|
and full of auth-gated stubs); a wholesale failure is logged and returns
|
|
[] so the caller proceeds with the known bundles."""
|
|
known_ids = {b.doc_id for b in known}
|
|
known_slugs = {b.slug for b in known}
|
|
anchor = scan_start
|
|
if anchor is None:
|
|
nums = [n for n in (_docid_num(b.doc_id) for b in known) if n is not None]
|
|
anchor = max(nums) if nums else 7400
|
|
lo, hi = anchor + 1, anchor + window
|
|
print(f" discovery: scanning sd[{lo}..{hi}] for new VM Essentials versions ...",
|
|
file=sys.stderr)
|
|
|
|
def probe(n: int) -> tuple[int, str, str, str] | None:
|
|
doc_id = f"sd{n:08d}en_us"
|
|
try:
|
|
html = _get(s, f"{API}/{doc_id}", expect_json=False)
|
|
except Exception:
|
|
return None
|
|
if not html or len(html) < 200:
|
|
return None
|
|
# Log VM-Essentials docs that don't classify as UM/RN for manual review.
|
|
meta = _parse_abstract(html)
|
|
blob = f"{meta.get('title','')} {meta.get('abstract','')}"
|
|
res = _classify_abstract(html)
|
|
if res is None:
|
|
if _INCLUDE_RE.search(blob) and not _EXCLUDE_RE.search(blob):
|
|
print(f" discovery: {doc_id} mentions VM Essentials but is not a "
|
|
f"UM/RN shape — review manually: {meta.get('title','')!r}",
|
|
file=sys.stderr)
|
|
return None
|
|
kind, version, title = res
|
|
return (n, kind, version, title)
|
|
|
|
hits: list[tuple[int, str, str, str]] = []
|
|
try:
|
|
with ThreadPoolExecutor(max_workers=6) as pool:
|
|
for fut in as_completed(pool.submit(probe, n) for n in range(lo, hi + 1)):
|
|
r = fut.result()
|
|
if r:
|
|
hits.append(r)
|
|
except Exception as e: # network/pool blew up wholesale
|
|
print(f" ! discovery scan failed ({e}) — proceeding with known bundles",
|
|
file=sys.stderr)
|
|
return []
|
|
|
|
new: list[BundleSpec] = []
|
|
for n, kind, version, title in sorted(hits):
|
|
doc_id = f"sd{n:08d}en_us"
|
|
slug = _slug_for(kind, version)
|
|
if doc_id in known_ids or slug in known_slugs:
|
|
continue
|
|
known_ids.add(doc_id)
|
|
known_slugs.add(slug)
|
|
product = "User Manual" if kind == "doc" else "Release Notes"
|
|
mode = "toc" if kind == "doc" else "single"
|
|
new.append(BundleSpec(slug, doc_id, title or "", version, product, mode))
|
|
print(f" discovery: + {slug} ({doc_id}) v{version} [{product}]", file=sys.stderr)
|
|
|
|
if not new:
|
|
print(" discovery: no new versions found", file=sys.stderr)
|
|
return new
|
|
|
|
|
|
def _spec_from_json(entry: dict[str, Any]) -> BundleSpec:
|
|
"""Reconstruct a BundleSpec from a prior bundles.json entry so previously
|
|
discovered versions persist across runs (the scan anchor rolls forward)."""
|
|
return BundleSpec(
|
|
slug=entry["slug"], doc_id=entry["doc_id"], title=entry.get("title") or "",
|
|
version=entry.get("version"), product=entry.get("product") or "",
|
|
mode=entry.get("mode") or "toc", platform=entry.get("platform"),
|
|
language=entry.get("language", "en-US"), source_url=entry.get("source_url"),
|
|
)
|
|
|
|
|
|
def collect_specs(s: requests.Session, discover: bool = True,
|
|
window: int = DISCOVER_WINDOW,
|
|
scan_start: int | None = None) -> list[BundleSpec]:
|
|
"""Merge static floor + prior bundles.json + freshly discovered specs,
|
|
deduped by doc_id (static wins)."""
|
|
by_id: dict[str, BundleSpec] = {}
|
|
for spec in BUNDLES:
|
|
by_id.setdefault(spec.doc_id, spec)
|
|
# Retain previously-discovered versions recorded in bundles.json.
|
|
if BUNDLES_JSON.exists():
|
|
try:
|
|
for entry in json.loads(BUNDLES_JSON.read_text()):
|
|
by_id.setdefault(entry["doc_id"], _spec_from_json(entry))
|
|
except Exception as e:
|
|
print(f" ! could not read prior {BUNDLES_JSON.name} ({e})", file=sys.stderr)
|
|
specs = list(by_id.values())
|
|
if discover:
|
|
for spec in discover_specs(s, specs, window=window, scan_start=scan_start):
|
|
if spec.doc_id not in by_id:
|
|
by_id[spec.doc_id] = spec
|
|
specs.append(spec)
|
|
return specs
|
|
|
|
|
|
def discover_bundle(s: requests.Session, spec: BundleSpec) -> dict[str, Any]:
|
|
# html-file bundles are static fixtures — no upstream fetch.
|
|
if spec.mode == "html-file":
|
|
return {
|
|
"slug": spec.slug,
|
|
"doc_id": spec.doc_id,
|
|
"title": spec.title,
|
|
"version": spec.version,
|
|
"platform": spec.platform,
|
|
"product": spec.product,
|
|
"language": spec.language,
|
|
"page_count": 1,
|
|
"mode": "html-file",
|
|
"abstract": "",
|
|
"dates": {},
|
|
"landing_page": spec.doc_id,
|
|
"source_url": spec.source_url or f"https://www.hpe.com/psnow/doc/{spec.doc_id}",
|
|
}
|
|
|
|
abstract_html = _get(s, f"{API}/{spec.doc_id}", expect_json=False)
|
|
meta = _parse_abstract(abstract_html or "")
|
|
|
|
page_count: int
|
|
landing: str | None
|
|
if spec.mode == "toc":
|
|
toc = _get(s, f"{API}/{spec.doc_id}/toc", expect_json=True)
|
|
page_count, landing = _count_toc(toc)
|
|
if page_count == 0:
|
|
print(f" ! {spec.slug}: TOC empty — falling back to single-doc mode", file=sys.stderr)
|
|
spec.mode = "single"
|
|
page_count, landing = 1, spec.doc_id
|
|
else:
|
|
page_count, landing = 1, spec.doc_id
|
|
|
|
return {
|
|
"slug": spec.slug,
|
|
"doc_id": spec.doc_id,
|
|
"title": meta.get("title") or spec.title,
|
|
"version": spec.version,
|
|
"platform": spec.platform,
|
|
"product": spec.product,
|
|
"language": spec.language,
|
|
"page_count": page_count,
|
|
"mode": spec.mode,
|
|
"abstract": meta.get("abstract", ""),
|
|
"dates": {"Published": meta.get("published", "")},
|
|
"landing_page": landing,
|
|
"source_url": spec.source_url or DOC_URL.format(doc_id=spec.doc_id),
|
|
}
|
|
|
|
|
|
def main() -> int:
|
|
p = argparse.ArgumentParser(description="Build bundles.json (static floor + auto-discovered versions).")
|
|
p.add_argument("--out", default=str(BUNDLES_JSON))
|
|
p.add_argument("--no-discover", dest="discover", action="store_false",
|
|
help="skip the docId range scan; build only from static + prior bundles.json")
|
|
p.add_argument("--window", type=int, default=DISCOVER_WINDOW,
|
|
help=f"how many docIds above the high-water mark to scan (default {DISCOVER_WINDOW})")
|
|
p.add_argument("--scan-start", type=int, default=None,
|
|
help="override the scan anchor (numeric docId, e.g. 7735)")
|
|
args = p.parse_args()
|
|
|
|
s = _session()
|
|
specs = collect_specs(s, discover=args.discover, window=args.window,
|
|
scan_start=args.scan_start)
|
|
out: list[dict[str, Any]] = []
|
|
for spec in specs:
|
|
print(f" • {spec.slug} ({spec.doc_id}) ...", file=sys.stderr)
|
|
out.append(discover_bundle(s, spec))
|
|
|
|
Path(args.out).write_text(json.dumps(out, indent=2) + "\n")
|
|
print(f"wrote {args.out}: {len(out)} bundles, {sum(b['page_count'] for b in out)} pages total", file=sys.stderr)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|