molecule-publish

#!/usr/bin/env python3
"""component_graph.py - ENFORCE the board -> components dependency graph.

Stages 2.5/2.6 were documented but not enforced, so boards kept publishing with
`dependencies: {}` - the publish looks successful (page renders, 3D + schematics
show) while Nucleus/APM shows an empty BOM panel (molecule-pipeline #4).
This script turns that convention into a gate a publish has to pass.

The graph is TWO layers - board -> components (#18: the auto-generated
intermediate `<board>-library` page was dropped). A board declares a dependency
directly on each component page. The gate enforces two distinct properties (#17):

  - DECLARED  (coverage):  every electrical BOM part has a dependency naming it.
                           A non-empty `dependencies:{}` is NOT enough - it must
                           actually cover the parts.
  - RESOLVED  (existence): every declared dependency resolves to a real wiki
                           page. A dep on a page that does not exist installs
                           broken in APM/Nucleus (core owner-pinning fragility);
                           "declared" alone never caught this.

  plan   <board> [--sch S] [--owner O]   parse BOM -> classify -> probe the wiki
                                          namespace -> emit plan.json + a table
                                          of reused / create / FLAGGED parts
  gate   <package.json> [--plan P]        HARD GATE, non-zero exit on failure:
                                          - electrical BOM rows but no deps
                                          - electrical parts not covered by any dep
                                            (DECLARED check, #17)
                                          - a declared dep that resolves to no page
                                            (RESOLVED check, #17; needs --resolve)
                                          - any slug keyed on an LCSC code or a
                                            value string instead of an MPN (#142)
                                          opt out ONLY with --allow-no-components

Stdlib only. Wiki probing shells to `adom-wiki` (skipped with --offline).
"""
import argparse
import json
import os
import re
import subprocess
import sys
import time

# ── identity rules (#142: key by MPN, never by value or distributor code) ──
LCSC_RE = re.compile(r'^c\d{3,}$', re.I)
VALUE_RE = re.compile(
    r'^(?:[rcl]_)?\d+(?:\.\d+)?\s*(?:[kmunp]|meg)?(?:ohm|r|f|h)?(?:_\d{4})?$', re.I)
VALUE_NAMED_RE = re.compile(r'^[rcl]_[\d.]+[kmunp]?_?\d{4}$', re.I)

# ── classification: what actually belongs in the dependency graph ──
NON_ELECTRICAL_REFS = re.compile(r'^(LOGO|FID|MH|H|G|TP|NT|REF\*\*)\d*$', re.I)
NON_ELECTRICAL_FP = re.compile(
    r'logo|fiducial|mounting\s*hole|test\s*point|silk|graphic|icon|keepout', re.I)
FURNITURE_REFS = re.compile(r'^(MP|MC)\d+$', re.I)  # Adom machine pins/contacts


def sh(cmd, timeout=60):
    try:
        r = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
        return r.returncode, r.stdout, r.stderr
    except Exception as e:
        return 1, "", str(e)


def parse_bom(board_path):
    """Rows from a .kicad_pcb (footprint blocks) or an EAGLE .brd (XML elements).
    Returns [{ref, value, footprint, lcsc, mpn}] with whatever the file carries."""
    txt = open(board_path, encoding="utf-8", errors="replace").read()
    rows = []
    if board_path.lower().endswith((".kicad_pcb", ".kicad_mod")):
        for m in re.finditer(r'\(footprint "([^"]+)"', txt):
            start = m.start()
            depth, i = 0, start
            while i < len(txt):
                if txt[i] == '(':
                    depth += 1
                elif txt[i] == ')':
                    depth -= 1
                    if depth == 0:
                        break
                i += 1
            blk = txt[start:i + 1]
            def prop(name):
                mm = re.search(r'\(property "%s" "([^"]*)"' % name, blk)
                return mm.group(1) if mm else ""
            rows.append({
                "ref": prop("Reference"),
                "value": prop("Value"),
                "footprint": m.group(1),
                "lcsc": prop("LCSC") or prop("LCSC Part") or prop("JLCPCB"),
                "mpn": prop("MPN") or prop("Manufacturer Part Number") or prop("MFR#"),
                "dnp": "(dnp" in blk or '"dnp"' in blk,
            })
    else:  # EAGLE .brd XML
        for m in re.finditer(r'<element\s+([^>]*?)/?>', txt):
            attrs = dict(re.findall(r'(\w+)="([^"]*)"', m.group(1)))
            rows.append({
                "ref": attrs.get("name", ""),
                "value": attrs.get("value", ""),
                "footprint": attrs.get("package", ""),
                "lcsc": "", "mpn": "", "dnp": False,
            })
    return [r for r in rows if r["ref"]]


def classify(row):
    """electrical | furniture | skip - only `electrical` is graph-mandatory."""
    ref, fp = row["ref"], row["footprint"]
    if row.get("dnp"):
        return "skip"
    if NON_ELECTRICAL_REFS.match(ref) or NON_ELECTRICAL_FP.search(fp):
        return "skip"
    if FURNITURE_REFS.match(ref):
        return "furniture"
    return "electrical"


def identity(row):
    """(mpn, problem). An LCSC code or a value string is NOT an identity (#142)."""
    mpn = (row.get("mpn") or "").strip()
    if mpn:
        return mpn, None
    val = (row.get("value") or "").strip()
    if not val:
        return None, "no MPN and no value - unresolvable from the board file"
    if LCSC_RE.match(val) or LCSC_RE.match((row.get("lcsc") or "").strip()):
        return None, (f"only an LCSC/distributor code ({val or row.get('lcsc')}) - "
                      "resolve the manufacturer part number "
                      "(adom-jlcpcb search <code>) before creating a page")
    if VALUE_RE.match(val) or VALUE_NAMED_RE.match(val):
        return None, (f"value string '{val}' is a property, not an identity - "
                      "resolve the real MPN (#142)")
    return val, None


def probe_wiki(mpn, offline=False):
    """Global-namespace probe by resolved MPN -> existing '<owner>/<slug>' or None.
    Reuse beats recreate: the component namespace is shared across owners."""
    if offline or not mpn:
        return None
    slug = re.sub(r'[^a-z0-9]+', '-', mpn.lower()).strip('-')
    code, out, _ = sh(["adom-wiki", "api", f"/api/v1/search?q={slug}&type=component"])
    if code != 0:
        return None
    try:
        res = (json.loads(out).get("response") or json.loads(out)).get("results") or []
    except Exception:
        return None
    for r in res:
        if (r.get("slug") or "").lower() == slug:
            return f"{r.get('owner')}/{r.get('slug')}"
    return None


def slugify(s):
    return re.sub(r'[^a-z0-9]+', '-', (s or "").lower()).strip('-')


def dep_slug(ref):
    """Slug portion of a dependency key: 'adom/lm358'->'lm358', 'lm358'->'lm358'.
    A caret/range value lives in the dict VALUE, never the key, so keys are clean."""
    return slugify(ref.split("/")[-1])


def page_exists(ref, offline=False):
    """Does a dependency resolve to a real wiki page? (#17 RESOLVED check.)
    'owner/slug' is looked up directly; a bare slug is searched by slug. Returns
    True/False, or None when we could not tell (offline / probe error) so the
    caller can degrade to a warning instead of a false FAIL."""
    if offline or not ref:
        return None
    if "/" in ref:
        code, out, err = sh(["adom-wiki", "api", f"/api/pages/{ref}"])
        if code != 0:
            # a definitive not-found is reported on stderr (and echoed in the JSON
            # body); anything else (network/auth) is indeterminate -> None
            return False if "PAGE_NOT_FOUND" in ((out or "") + (err or "")) else None
        try:
            j = json.loads(out)
            return bool(j.get("page") or j.get("slug") or j.get("id"))
        except Exception:
            return None
    # bare slug: accept any owner that actually owns a page with this slug
    slug = slugify(ref)
    code, out, _ = sh(["adom-wiki", "api", f"/api/v1/search?q={slug}"])
    if code != 0:
        return None
    try:
        res = (json.loads(out).get("response") or json.loads(out)).get("results") or []
    except Exception:
        return None
    return any((r.get("slug") or "").lower() == slug for r in res)


def cmd_plan(a):
    rows = parse_bom(a.board)
    plan = {"board": os.path.basename(a.board), "parts": [],
            "counts": {"electrical": 0, "furniture": 0, "skip": 0,
                       "reuse": 0, "create": 0, "flagged": 0}}
    for row in rows:
        kind = classify(row)
        plan["counts"][kind] = plan["counts"].get(kind, 0) + 1
        if kind == "skip":
            continue
        mpn, problem = identity(row)
        existing = probe_wiki(mpn, a.offline)
        status = ("flagged" if problem else "reuse" if existing else "create")
        if kind == "electrical":
            plan["counts"][status] += 1
        plan["parts"].append({"ref": row["ref"], "kind": kind, "mpn": mpn,
                              "value": row.get("value"), "footprint": row.get("footprint"),
                              "status": status, "page": existing, "problem": problem})

    w = max([len(p["ref"]) for p in plan["parts"]] + [4])
    print(f"{'REF'.ljust(w)}  {'STATUS'.ljust(8)}  {'KIND'.ljust(10)}  IDENTITY")
    for p in sorted(plan["parts"], key=lambda x: (x["status"] != "flagged", x["ref"])):
        ident = p["page"] or p["mpn"] or (p["problem"] or "")
        print(f"{p['ref'].ljust(w)}  {p['status'].ljust(8)}  {p['kind'].ljust(10)}  {ident}")
    c = plan["counts"]
    print(f"\nelectrical={c['electrical']} furniture={c['furniture']} skipped={c['skip']}"
          f"  ->  reuse={c['reuse']} create={c['create']} FLAGGED={c['flagged']}")
    if c["flagged"]:
        print("\nFLAGGED parts have no resolvable MPN. Resolve them (or explicitly drop\n"
              "them from the graph) before publishing - the gate refuses them.")
    if a.out:
        json.dump(plan, open(a.out, "w"), indent=2)
        print(f"\nplan -> {a.out}")
    return 1 if c["flagged"] else 0


def cmd_gate(a):
    pkg = json.load(open(a.package))
    deps = pkg.get("dependencies") or {}
    slug = pkg.get("slug", "")
    fails, warns = [], []

    tags = [str(t).lower() for t in (pkg.get("tags") or [])]
    is_library = "library" in tags or slug.endswith("-library")

    # #18: the auto-generated `<board>-library` layer was dropped - boards depend
    # on component pages directly. A hand-made library page is still allowed, but
    # it must NOT carry a board/molecule KIND tag (#5): Nucleus keys on the kind
    # (project > molecule > circuit > board > library), so a stray 'board'/
    # 'molecule' wins and the library renders as the wrong kind.
    if is_library:
        stray = [t for t in ("board", "molecule", "circuit", "project") if t in tags]
        if stray:
            fails.append(f"library page carries kind tag(s) {stray} - a library is none of "
                         "these; tag it ['library', <topical>...] only (#5)")

    # the board's OWN slug must be MPN-shaped for components (#142) - NOT libraries
    elif pkg.get("type") == "component" and pkg.get("component", {}).get("category") != "board":
        if LCSC_RE.match(slug):
            fails.append(f"slug '{slug}' is an LCSC/distributor code, not an MPN (#142)")
        if VALUE_NAMED_RE.match(slug) or VALUE_RE.match(slug):
            fails.append(f"slug '{slug}' is a value string, not an MPN (#142)")

    plan = json.load(open(a.plan)) if a.plan and os.path.exists(a.plan) else None
    # A plan from a DIFFERENT board silently "passes" a board it never inspected -
    # the exact class of mistake this gate exists to stop. Compare stems.
    if plan and a.expect_board:
        stem = re.sub(r'[^a-z0-9]+', '', os.path.splitext(plan.get("board", ""))[0].lower())
        want = re.sub(r'[^a-z0-9]+', '', a.expect_board.lower())
        if stem and want and stem not in want and want not in stem:
            fails.append(f"plan is for '{plan.get('board')}' but this package is "
                         f"'{a.expect_board}' - re-run `plan` for THIS board")
    electrical = [p for p in (plan or {}).get("parts", []) if p["kind"] == "electrical"]
    flagged = [p for p in electrical if p["status"] == "flagged"]

    if a.allow_no_components:
        warns.append("--allow-no-components: dependency graph intentionally skipped")
    elif plan is None:
        warns.append("no --plan given: cannot verify the graph covers the BOM "
                     "(run `component_graph.py plan` first)")
        if not deps:
            fails.append("board publishes with dependencies:{} and no plan to justify it")
    else:
        if electrical and not deps:
            fails.append(f"{len(electrical)} electrical BOM parts but dependencies:{{}} - "
                         "APM/Nucleus will show an empty BOM panel (molecule-pipeline #4)")
        if flagged:
            fails.append(f"{len(flagged)} part(s) have no resolved MPN: "
                         + ", ".join(f"{p['ref']}({p['problem'][:40]})" for p in flagged[:4]))

        # ── DECLARED (coverage, #17): every RESOLVED electrical part must have a
        # dependency naming it. board -> components directly (#18): the expected
        # dep slug is the part's MPN slug (or the slug of the page probe found).
        dep_slugs = {dep_slug(d) for d in deps}
        uncovered = []
        for p in electrical:
            if p["status"] == "flagged":
                continue  # already failed above; no identity to match on
            want = {slugify(p["mpn"])} if p.get("mpn") else set()
            if p.get("page"):
                want.add(dep_slug(p["page"]))
            if want and not (want & dep_slugs):
                uncovered.append(p)
        if uncovered:
            fails.append(
                f"{len(uncovered)} electrical part(s) are NOT covered by any dependency "
                "(declared-coverage, #17): "
                + ", ".join(f"{p['ref']}={p['mpn'] or p.get('value')}" for p in uncovered[:5])
                + " - add a dependency on each part's component page")

        # ── RESOLVED (existence, #17): every declared dep must resolve to a real
        # page. Skipped unless --resolve (it hits the network); a probe that
        # cannot tell (offline/error) degrades to a WARN, never a false FAIL.
        if a.resolve and deps:
            dangling, unknown = [], []
            for d in deps:
                ex = page_exists(d, offline=a.offline)
                if ex is False:
                    dangling.append(d)
                elif ex is None:
                    unknown.append(d)
            if dangling:
                fails.append(
                    f"{len(dangling)} declared dependency/-ies resolve to NO wiki page "
                    "(resolved-existence, #17): " + ", ".join(dangling[:5])
                    + " - publish the component page(s) first, or fix the dep key")
            if unknown:
                warns.append("could not verify existence of: " + ", ".join(unknown[:5])
                             + " (probe unavailable; re-run with network to confirm)")
        elif deps and not a.resolve:
            warns.append("dependency existence not checked - pass --resolve to verify "
                         "each dep resolves to a real page (#17)")

    for f in fails:
        print(f"FAIL: {f}")
    for w in warns:
        print(f"WARN: {w}")
    if fails:
        print("\nGATE FAILED - do not publish. Fix the graph, or pass --allow-no-components "
              "if this board genuinely has no component pages.")
        return 1
    print("GATE PASSED" + (" (with warnings)" if warns else ""))
    return 0


def main():
    ap = argparse.ArgumentParser(description=__doc__,
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = ap.add_subparsers(dest="cmd", required=True)

    p = sub.add_parser("plan", help="parse the BOM, classify, probe the wiki namespace")
    p.add_argument("board")
    p.add_argument("--sch")
    p.add_argument("--owner")
    p.add_argument("--out", default="component-plan.json")
    p.add_argument("--offline", action="store_true", help="skip wiki namespace probing")
    p.set_defaults(fn=cmd_plan)

    g = sub.add_parser("gate", help="HARD GATE before publish (non-zero exit = do not publish)")
    g.add_argument("package", help="the board's package.json")
    g.add_argument("--plan", default="component-plan.json")
    g.add_argument("--expect-board", help="board filename this package should match "
                   "(guards against passing a stale plan from another board)")
    g.add_argument("--allow-no-components", action="store_true",
                   help="explicit opt-out for a board that genuinely has no component pages")
    g.add_argument("--resolve", action="store_true",
                   help="also verify each declared dependency resolves to a real wiki "
                        "page (#17 RESOLVED check; hits the network)")
    g.add_argument("--offline", action="store_true",
                   help="never probe the wiki (resolution checks degrade to warnings)")
    g.set_defaults(fn=cmd_gate)

    a = ap.parse_args()
    sys.exit(a.fn(a))


if __name__ == "__main__":
    main()