fix(scrape): auto-discover new Morpheus versions; pin 9.0.0 (#4)

Co-authored-by: claude <[email protected]>
2026-07-15 11:17:27 -04:00
parent 7b2c39bb0a
commit 2922161cc7
1 changed files with 125 additions and 6 deletions
@@ -15,9 +15,11 @@ from __future__ import annotations

 import argparse
 import json
+import os
 import re
 import sys
 import time
+from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any
@@ -31,6 +33,23 @@ UA = "morpheus-docs-mcp/0.1 (+https://git.jpaul.io/justin/morpheus-docs; admin@j
 ROOT = Path(__file__).resolve().parent.parent
 BUNDLES_JSON = ROOT / "bundles.json"

+# --- Auto-discovery of new versions -------------------------------------
+# The pinned BUNDLES list below is a known-good baseline. On top of it we
+# sweep the HPE docId range for NEWLY published Morpheus Enterprise User
+# Manuals / Release Notes (e.g. a 9.0.1 that lands after this file was last
+# edited) so the weekly refresh picks them up without a code change. New
+# versions always get higher docIds, so we scan from just below the lowest
+# pinned bundle up to the highest pinned docId plus a forward window.
+#
+# Set DISCOVER_VERSIONS=0 to disable (reproducible/offline runs). Widen
+# DISCOVER_WINDOW if HPE publishes a new version far beyond the newest pin.
+DISCOVER = os.environ.get("DISCOVER_VERSIONS", "1") != "0"
+DISCOVER_WINDOW = int(os.environ.get("DISCOVER_WINDOW", "500"))
+DISCOVER_WORKERS = int(os.environ.get("DISCOVER_WORKERS", "12"))
+
+_DOCID_NUM_RE = re.compile(r"sd0*(\d+)en_us")
+_VERSION_RE = re.compile(r"v?(\d+\.\d+\.\d+)")
+

@dataclass
 class BundleSpec:
@@ -45,14 +64,17 @@ class BundleSpec:
    source_url: str | None = None   # overrides the default support.hpe.com URL


-# Declared bundles. Versions confirmed 2026-05-22 by probing the docId
-# range sd00006500..7740 for `Morpheus Enterprise` matches in the abstract.
+# Pinned baseline bundles. docIds confirmed by probing the portal for
+# `Morpheus Enterprise` matches in the abstract (8.1.x on 2026-05-22,
+# 9.0.0 on 2026-07-15). This list is a known-good floor; _discover_enterprise()
+# sweeps forward from here each run and appends any NEWER versions HPE has
+# published, so a fresh 9.0.1 lands in the corpus without editing this file.
 #
 # Notes:
 #   - Morpheus Enterprise has User Manuals dating back to 8.0.10
-#     (sd00006774en_us, Sep 2025) but we only ship the 8.1.x line for
-#     now. Add the 8.0.x bundles here if you need older versions in the
-#     corpus.
+#     (sd00006774en_us, Sep 2025) but we ship the 8.1.x + 9.0.0 line.
+#     Add the 8.0.x bundles here if you need older versions in the corpus
+#     (they predate the discovery floor, so they won't auto-appear).
 #   - No dedicated Deployment Guide or Qualification Matrix for Morpheus
 #     Enterprise on HPE Support — the only QM (sd00006551en_us) covers
 #     HVM clusters managed by Morpheus, which lives in hvm-docs.
@@ -65,6 +87,11 @@ BUNDLES: list[BundleSpec] = [
    BundleSpec("morpheus_release_notes_8_1_0", "sd00007496en_us", "HPE Morpheus Enterprise Software Release Notes",  "8.1.0", "Release Notes", "single"),
    BundleSpec("morpheus_release_notes_8_1_1", "sd00007610en_us", "HPE Morpheus Enterprise Software Release Notes",  "8.1.1", "Release Notes", "single"),
    BundleSpec("morpheus_release_notes_8_1_2", "sd00007733en_us", "HPE Morpheus Enterprise Software Release Notes",  "8.1.2", "Release Notes", "single"),
+    # 9.0.0 (May 2026). Confirmed 2026-07-15 by probing the docId range;
+    # also auto-discovered by _discover_enterprise() below, but pinned here
+    # so it ships even if discovery is disabled or the portal is down.
+    BundleSpec("morpheus_user_manual_9_0_0",   "sd00008014en_us", "HPE Morpheus Enterprise Software Documentation", "9.0.0", "User Manual",   "toc"),
+    BundleSpec("morpheus_release_notes_9_0_0", "sd00008017en_us", "HPE Morpheus Enterprise Software Release Notes",  "9.0.0", "Release Notes", "single"),
    BundleSpec("morpheus_quickspecs",          "a50009231enw",    "HPE Morpheus Enterprise Software QuickSpecs",
               "v1", "QuickSpecs", "html-file",
               source_url="https://www.hpe.com/psnow/doc/a50009231enw"),
@@ -180,14 +207,106 @@ def discover_bundle(s: requests.Session, spec: BundleSpec) -> dict[str, Any]:
    }


+def _docid_num(doc_id: str) -> int | None:
+    m = _DOCID_NUM_RE.fullmatch(doc_id)
+    return int(m.group(1)) if m else None
+
+
+def _classify_enterprise(title: str, abstract: str) -> tuple[str, str, str] | None:
+    """Decide whether an abstract is a Morpheus Enterprise User Manual or
+    Release Notes and return (product, mode, version); else None.
+
+    Filters out the sibling products that share the HPE portal (Morpheus VM
+    Essentials → hvm-docs, Morpheus Central). The Release Notes abstract puts
+    the "Morpheus Enterprise" signal in the description (its <h1> is just
+    "vX.Y.Z Release Notes"), so we match against title + abstract combined."""
+    blob = f"{title}  {abstract}"
+    if "Morpheus Enterprise" not in blob:
+        return None
+    if "VM Essentials" in blob or "Morpheus Central" in blob:
+        return None
+    m = _VERSION_RE.search(title) or _VERSION_RE.search(abstract)
+    if not m:
+        return None
+    version = m.group(1)
+    if "Release Notes" in blob or "Release notes" in blob:
+        return ("Release Notes", "single", version)
+    if "Documentation" in blob or "User Manual" in blob:
+        return ("User Manual", "toc", version)
+    return None
+
+
+def _discover_enterprise(s: requests.Session, pinned: list[BundleSpec]) -> list[BundleSpec]:
+    """Sweep the docId range for Morpheus Enterprise UM/RN bundles not already
+    pinned. Best-effort and per-id fault tolerant: a probe that errors just
+    yields nothing, and the caller falls back to the pinned list on any
+    top-level failure. The pinned bundles always ship regardless."""
+    nums = [n for n in (_docid_num(b.doc_id) for b in pinned) if n is not None]
+    if not nums:
+        return []
+    lo, hi = min(nums), max(nums) + DISCOVER_WINDOW
+    pinned_docids = {b.doc_id for b in pinned}
+    pinned_pv = {(b.product, b.version) for b in pinned}
+
+    def probe(n: int) -> BundleSpec | None:
+        doc_id = f"sd{n:08d}en_us"
+        if doc_id in pinned_docids:
+            return None
+        try:
+            html = _get(s, f"{API}/{doc_id}", expect_json=False)
+        except Exception:
+            return None
+        if not html:
+            return None
+        meta = _parse_abstract(html)
+        cls = _classify_enterprise(meta.get("title", ""), meta.get("abstract", ""))
+        if not cls:
+            return None
+        product, mode, version = cls
+        if (product, version) in pinned_pv:
+            return None
+        vslug = version.replace(".", "_")
+        if product == "Release Notes":
+            slug = f"morpheus_release_notes_{vslug}"
+            title = "HPE Morpheus Enterprise Software Release Notes"
+        else:
+            slug = f"morpheus_user_manual_{vslug}"
+            title = "HPE Morpheus Enterprise Software Documentation"
+        return BundleSpec(slug, doc_id, title, version, product, mode)
+
+    found: dict[tuple[str, str], BundleSpec] = {}
+    print(f"  discovery: sweeping docId {lo}..{hi} for new Enterprise bundles", file=sys.stderr)
+    with ThreadPoolExecutor(max_workers=DISCOVER_WORKERS) as pool:
+        # pool.map preserves input (ascending docId) order, so the first hit
+        # for a (product, version) is the lowest docId. HPE mirrors some docs
+        # under several docIds; the lowest is the original and tends to be the
+        # most complete, so keep it and ignore later duplicates.
+        for spec in pool.map(probe, range(lo, hi + 1)):
+            if spec and (spec.product, spec.version) not in found:
+                found[(spec.product, spec.version)] = spec
+    return list(found.values())
+
+
 def main() -> int:
    p = argparse.ArgumentParser(description="Build bundles.json from BUNDLES list.")
    p.add_argument("--out", default=str(BUNDLES_JSON))
+    p.add_argument("--no-discover", action="store_true",
+                   help="skip the docId-range sweep; use the pinned BUNDLES list only")
    args = p.parse_args()

    s = _session()
+
+    specs = list(BUNDLES)
+    if DISCOVER and not args.no_discover:
+        try:
+            for d in sorted(_discover_enterprise(s, BUNDLES), key=lambda b: b.slug):
+                print(f"  + discovered {d.slug} ({d.doc_id})", file=sys.stderr)
+                specs.append(d)
+        except Exception as e:
+            print(f"  ! discovery failed ({e}); using pinned bundles only", file=sys.stderr)
+
    out: list[dict[str, Any]] = []
-    for spec in BUNDLES:
+    for spec in specs:
        print(f"  • {spec.slug} ({spec.doc_id}) ...", file=sys.stderr)
        out.append(discover_bundle(s, spec))