molecule-publish

#!/usr/bin/env python3
"""Pack and verify fabrication sources. Standard library only; never publishes."""
import argparse
import hashlib
import json
import re
import sys
import zipfile
from pathlib import Path, PurePosixPath

VERSION = "1.0.0"
REQUIRED = {"copper_top", "copper_bottom", "mask_top", "mask_bottom",
            "silk_top", "silk_bottom", "outline"}
DRILLS = {"drill_plated", "drill_unplated", "drill_mixed"}
EXT_ROLES = {"gtl": "copper_top", "gbl": "copper_bottom", "gts": "mask_top",
             "gbs": "mask_bottom", "gto": "silk_top", "gbo": "silk_bottom",
             "gtp": "paste_top", "gbp": "paste_bottom", "gko": "outline",
             "gm1": "outline", "gbrjob": "gerber_job"}


def safe_path(name):
    p = PurePosixPath(name)
    if (not name or "\\" in name or ":" in name or p.is_absolute()
            or any(v in ("", ".", "..") for v in name.split("/"))):
        raise ValueError(f"not a relative file path: {name!r}")
    return p


def fingerprint(name, role, data):
    return {"path": name, "role": role, "bytes": len(data),
            "sha256": hashlib.sha256(data).hexdigest()}


def role_for(name, data):
    ext = Path(name).suffix.lower().lstrip(".")
    text = data.decode("utf-8", errors="replace")
    # X2 is authoritative, including KiCad's .gbr files and named imported layers.
    match = re.search(r"TF\.FileFunction,([^*%\r\n]+)", text)
    if match:
        parts = match[1].split(",")
        kind = parts[0]
        side = "top" if "Top" in parts else "bottom" if "Bot" in parts else None
        if kind == "Profile":
            return "outline"
        if kind in ("Copper", "Soldermask", "Legend", "Paste") and side:
            return {"Copper": "copper", "Soldermask": "mask", "Legend": "silk",
                    "Paste": "paste"}[kind] + "_" + side
        if kind == "Copper":
            layer = next((p for p in parts if re.fullmatch(r"L[2-9][0-9]*", p)), None)
            return "copper_inner_" + str(int(layer[1:]) - 1) if layer else "copper_inner"
    if ext in ("drl", "xln", "exc", "txt") and re.search(r"(?m)^M48\s*$", text):
        stem = Path(name).stem.upper()
        if "NPTH" in stem or "NONPLATED" in stem or "NonPlated" in text:
            return "drill_unplated"
        if "PTH" in stem or stem.endswith("-PLATED") or re.search(r"FileFunction,Plated,", text):
            return "drill_plated"
        return "drill_mixed"
    if re.fullmatch(r"g[1-9][0-9]*", ext):
        return "copper_inner_" + ext[1:]
    return EXT_ROLES.get(ext, "other")


def read_zip(path):
    out = {}
    with zipfile.ZipFile(path) as z:
        for item in z.infolist():
            if item.is_dir():
                continue
            safe_path(item.filename)
            if item.filename in out:
                raise ValueError(f"duplicate zip entry: {item.filename}")
            out[item.filename] = z.read(item)
    return out


def layer_entries(files):
    entries = [fingerprint(n, role_for(n, b), b) for n, b in sorted(files.items())]
    roles = {e["role"] for e in entries}
    missing = REQUIRED - roles
    if missing:
        raise ValueError("missing fabrication layers: " + ", ".join(sorted(missing)))
    if not roles & DRILLS:
        raise ValueError("missing Excellon drill file (M48 header)")
    # An empty mechanical-layer plot is not a board outline (Altium can emit one).
    outlines = [files[e["path"]].decode("utf-8", errors="replace")
                for e in entries if e["role"] == "outline"]
    if not any(re.search(r"[XY][+-]?\d+[^*]*D0?1\*", t) for t in outlines):
        raise ValueError("outline Gerber has no drawn segments; export the actual board boundary")
    return entries


def source_entry(root, value):
    role, sep, name = value.partition("=")
    if not sep or not role:
        raise ValueError("source must be ROLE=relative/path")
    safe_path(name)
    file = (root / name).resolve()
    if not file.is_relative_to(root.resolve()):
        raise ValueError(f"source escapes bundle: {name}")
    return fingerprint(name, role, file.read_bytes())


def check(root, slug):
    root = Path(root)
    rec = json.loads((root / "export.json").read_text(encoding="utf-8"))
    if rec.get("schema") != "adom.export/1":
        raise ValueError("export.json needs schema adom.export/1")
    for key in ("tool", "toolVersion", "exporter", "exporterVersion"):
        if not isinstance(rec.get(key), str) or not rec[key].strip():
            raise ValueError(f"export.json needs {key}")
    entries = rec.get("files", [])
    if not isinstance(entries, list) or not entries:
        raise ValueError("export.json needs files[]")
    names = set()
    for entry in entries:
        name = entry["path"]
        safe_path(name)
        if name in names:
            raise ValueError(f"duplicate file record: {name}")
        names.add(name)
        path = (root / name).resolve()
        if not path.is_relative_to(root.resolve()):
            raise ValueError(f"file escapes bundle: {name}")
        actual = fingerprint(name, entry["role"], path.read_bytes())
        if any(entry.get(k) != actual[k] for k in ("sha256", "bytes")):
            raise ValueError(f"source changed since export: {name}")
    roles = {e["role"] for e in entries}
    if not roles & {"board_kicad_pcb", "board_eagle_brd", "board_altium_pcbdoc"}:
        raise ValueError("export.json needs the board source and its hash")
    if not rec.get("backfill") and "step" not in roles:
        raise ValueError("export.json needs the STEP source and its hash")
    zips = [e for e in entries if e["role"] == "fab_zip"]
    if len(zips) != 1 or zips[0]["path"] != f"{slug}-fab.zip":
        raise ValueError(f"exactly one fab_zip named {slug}-fab.zip is required")
    files = read_zip(root / zips[0]["path"])
    actual = layer_entries(files)
    expected = zips[0].get("entries", [])
    by_name = {e["path"]: e for e in expected}
    if len(by_name) != len(expected) or set(by_name) != set(files):
        raise ValueError("zip entries and export.json disagree")
    for entry in actual:
        saved = by_name[entry["path"]]
        if any(saved.get(k) != entry[k] for k in ("sha256", "bytes", "role")):
            raise ValueError(f"zip entry changed or has wrong role: {entry['path']}")
    return rec


def create(args):
    root = Path(args.root).resolve()
    if (root / "export.json").exists():
        raise ValueError("export.json already exists; verify it or use a fresh export folder")
    source = Path(args.fab)
    if source.is_dir():
        files = {}
        for p in sorted(source.rglob("*")):
            if p.is_symlink():
                raise ValueError(f"symlink in fabrication folder: {p}")
            if p.is_file():
                name = p.relative_to(source).as_posix()
                safe_path(name)
                files[name] = p.read_bytes()
    else:
        files = read_zip(source)
    layers = layer_entries(files)
    entries = [source_entry(root, s) for s in args.source]
    target = root / f"{args.slug}-fab.zip"
    if target.exists():
        raise ValueError(f"output exists: {target}; use a fresh export folder")
    # Fixed ZIP metadata; the Gerber/drill bytes remain exactly as exported.
    with zipfile.ZipFile(target, "w", compression=zipfile.ZIP_DEFLATED) as z:
        for name, data in sorted(files.items()):
            item = zipfile.ZipInfo(name, date_time=(1980, 1, 1, 0, 0, 0))
            item.compress_type = zipfile.ZIP_DEFLATED
            item.external_attr = 0o644 << 16
            z.writestr(item, data)
    fab = fingerprint(target.name, "fab_zip", target.read_bytes())
    fab["entries"] = layers
    entries.append(fab)
    stackup = json.loads(Path(args.stackup).read_text(encoding="utf-8")) if args.stackup else {}
    rec = {"schema": "adom.export/1", "tool": args.tool, "toolVersion": args.tool_version,
           "exporter": args.exporter, "exporterVersion": args.exporter_version,
           "design": args.slug, "files": entries, "stackup": stackup,
           "gerbers": {"included": True},
           "drill": {"plating": "mixed" if any(e["role"] == "drill_mixed" for e in layers) else "split"}}
    (root / "export.json").write_text(json.dumps(rec, indent=2) + "\n", encoding="utf-8")
    return check(root, args.slug)


def main():
    p = argparse.ArgumentParser(description=__doc__)
    sub = p.add_subparsers(dest="command", required=True)
    for verb in ("create", "check"):
        q = sub.add_parser(verb)
        q.add_argument("root", help="export folder; all recorded source paths are relative to it")
        q.add_argument("--slug", required=True)
        if verb == "create":
            q.add_argument("--fab", required=True, help="EDA fabrication output directory or zip")
            q.add_argument("--tool", required=True)
            q.add_argument("--tool-version", required=True)
            q.add_argument("--exporter", required=True)
            q.add_argument("--exporter-version", required=True)
            q.add_argument("--source", action="append", required=True, help="ROLE=relative/path (repeat)")
            q.add_argument("--stackup", help="JSON with known thicknessMm, maskColor, silkColor, finish")
    args = p.parse_args()
    try:
        if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", args.slug):
            raise ValueError("slug must be lowercase words separated by hyphens")
        rec = create(args) if args.command == "create" else check(args.root, args.slug)
        print(json.dumps({"ok": True, "files": [e["path"] for e in rec["files"]] + ["export.json"]}))
    except (ValueError, OSError, KeyError, TypeError, zipfile.BadZipFile) as e:
        print(json.dumps({"ok": False, "error": str(e)}), file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())