228 lines
8.9 KiB
Python
228 lines
8.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Talk to the two Gitea repos of the new library-manager layout.
|
|
|
|
Repos (from config/gitea.env):
|
|
SKILL_REPO - this skill's own files (pushed with push_to_gitea.sh, not here).
|
|
COMPONENTS_REPO - one folder per Class (Diode, IC, Transistor, ...); inside each Class,
|
|
one MPN_make_typeid/ folder per part holding { xlsx, datasheet, symbol,
|
|
footprint }.
|
|
|
|
Subcommands (all accept --local <dir> to run against a local folder instead of cloning, for
|
|
testing or offline work):
|
|
|
|
check-mpn --mpn BAT46WJ --make Nexperia
|
|
Is any part with this MPN+make already in the components repo? (typeid-agnostic - the
|
|
early duplicate gate, run BEFORE classifying.) Prints EXISTS/<tag> or ABSENT; exit 0
|
|
if absent, 3 if it already exists (so a shell can hard-stop).
|
|
|
|
checkout --dest work/
|
|
Clone the components repo to a working dir you can browse and edit in place.
|
|
|
|
list-type --typeid SCH [--root work/] [--json]
|
|
List every existing part of a typeid and the files in its folder (for backfill: the
|
|
datasheet to re-read is in there). Needs a checked-out --root (or --local).
|
|
|
|
place-part --folder staging/BAT46WJ_Nexperia_SCH --typeid SCH --root work/
|
|
Copy an assembled part folder into work/<Class>/<tag>/ (creates the Class folder if it
|
|
is missing). Then commit-push.
|
|
|
|
commit-push --root work/ [--message "..."]
|
|
Commit everything in the checked-out clone and push.
|
|
|
|
push-part --folder staging/BAT46WJ_Nexperia_SCH --typeid SCH [--message "..."]
|
|
Convenience: clone -> place-part -> commit-push in one go (the common single-part path).
|
|
|
|
Host unreachable (e.g. sandbox without the domain allowlisted) -> clones fail clearly and
|
|
nothing is written.
|
|
"""
|
|
import argparse, os, re, shutil, subprocess, sys, tempfile
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
from common import class_folder, load_taxonomy, mpn_make_prefix
|
|
|
|
CFG_DEFAULT = os.path.join(os.path.dirname(__file__), "..", "config", "gitea.env")
|
|
|
|
|
|
# ------------------------------------------------------------------ gitea plumbing
|
|
|
|
def load_env(cfg):
|
|
env = {}
|
|
if os.path.exists(cfg):
|
|
for line in open(cfg):
|
|
line = line.strip()
|
|
if not line or line.startswith("#") or "=" not in line:
|
|
continue
|
|
k, v = line.split("=", 1)
|
|
env[k.strip()] = v.strip()
|
|
for k in ("GIT_HOST", "GIT_USER", "GIT_TOKEN", "SKILL_REPO", "COMPONENTS_REPO"):
|
|
if os.environ.get(k):
|
|
env[k] = os.environ[k]
|
|
return env
|
|
|
|
|
|
def repo_url(env, repo):
|
|
host = re.sub(r"^https?://", "", env["GIT_HOST"]).rstrip("/")
|
|
cred = (env.get("GIT_USER", "") + ":" if env.get("GIT_USER") else "") + env["GIT_TOKEN"]
|
|
return f"https://{cred}@{host}/{repo}.git"
|
|
|
|
|
|
def clone(env, dest):
|
|
url = repo_url(env, env["COMPONENTS_REPO"])
|
|
r = subprocess.run(["git", "clone", url, dest], capture_output=True, text=True)
|
|
if r.returncode != 0:
|
|
err = (r.stderr or "").replace(env["GIT_TOKEN"], "***")
|
|
sys.exit(f"clone COMPONENTS failed (host reachable / token scope / repo exists?):\n{err[:400]}")
|
|
return dest
|
|
|
|
|
|
def commit_push(root, token, msg):
|
|
for args in (["config", "user.email", "datasheet-bot@local"],
|
|
["config", "user.name", "library-manager"],
|
|
["checkout", "-B", "main"], ["add", "-A"]):
|
|
subprocess.run(["git", "-C", root] + args, capture_output=True)
|
|
if subprocess.run(["git", "-C", root, "diff", "--cached", "--quiet"]).returncode == 0:
|
|
print("nothing new to push.")
|
|
return
|
|
subprocess.run(["git", "-C", root, "commit", "-m", msg], capture_output=True)
|
|
r = subprocess.run(["git", "-C", root, "push", "-u", "origin", "main"],
|
|
capture_output=True, text=True)
|
|
print((r.stdout + r.stderr).replace(token or "", "***").strip() or "pushed.")
|
|
|
|
|
|
def _root(env, args):
|
|
"""Return (root_dir, is_temp_clone). Honour --local / --root, else clone."""
|
|
if getattr(args, "local", None):
|
|
return args.local, False
|
|
if getattr(args, "root", None):
|
|
return args.root, False
|
|
tmp = tempfile.mkdtemp(prefix="components_")
|
|
return clone(env, tmp), True
|
|
|
|
|
|
# ------------------------------------------------------------------ folder logic
|
|
|
|
def find_part_dirs(root, prefix="", typeid=None):
|
|
"""Yield (class_folder, part_folder_name, abspath) for part folders under any Class dir
|
|
that match a name prefix and/or a typeid suffix."""
|
|
if not os.path.isdir(root):
|
|
return
|
|
for cls in sorted(os.listdir(root)):
|
|
cdir = os.path.join(root, cls)
|
|
if not os.path.isdir(cdir) or cls.startswith("."):
|
|
continue
|
|
for name in sorted(os.listdir(cdir)):
|
|
pdir = os.path.join(cdir, name)
|
|
if not os.path.isdir(pdir):
|
|
continue
|
|
if prefix and not name.startswith(prefix):
|
|
continue
|
|
if typeid and not name.endswith(f"_{typeid}"):
|
|
continue
|
|
yield cls, name, pdir
|
|
|
|
|
|
def cmd_check_mpn(env, args):
|
|
root, _ = _root(env, args)
|
|
prefix = mpn_make_prefix(args.mpn, args.make)
|
|
matches = [(cls, name) for cls, name, _ in find_part_dirs(root, prefix=prefix)]
|
|
if matches:
|
|
for cls, name in matches:
|
|
print(f"EXISTS\t{cls}/{name}")
|
|
sys.exit(3)
|
|
print("ABSENT")
|
|
|
|
|
|
def cmd_checkout(env, args):
|
|
dest = clone(env, args.dest)
|
|
print(f"components repo checked out at {dest}")
|
|
|
|
|
|
def cmd_list_type(env, args):
|
|
root, _ = _root(env, args)
|
|
parts = list(find_part_dirs(root, typeid=args.typeid))
|
|
if args.json:
|
|
import json
|
|
out = [{"class": cls, "tag": name,
|
|
"folder": pdir,
|
|
"files": sorted(os.listdir(pdir))} for cls, name, pdir in parts]
|
|
print(json.dumps(out, indent=2))
|
|
return
|
|
if not parts:
|
|
print(f"no existing parts of typeid {args.typeid}")
|
|
return
|
|
for cls, name, pdir in parts:
|
|
files = ", ".join(sorted(os.listdir(pdir)))
|
|
print(f"{cls}/{name} -> {files}")
|
|
|
|
|
|
def place_part(root, folder, typeid):
|
|
cls = class_folder(typeid)
|
|
if not cls:
|
|
sys.exit(f"unknown typeid '{typeid}' (not in taxonomy)")
|
|
tag = os.path.basename(os.path.normpath(folder))
|
|
cls_dir = os.path.join(root, cls)
|
|
# A repo may seed classes as 0-byte placeholder FILES (git can't store empty dirs).
|
|
# Replace such a placeholder with a real class directory before adding the part.
|
|
if os.path.exists(cls_dir) and not os.path.isdir(cls_dir):
|
|
os.remove(cls_dir)
|
|
os.makedirs(cls_dir, exist_ok=True)
|
|
dest = os.path.join(cls_dir, tag)
|
|
if os.path.isdir(dest):
|
|
shutil.rmtree(dest)
|
|
shutil.copytree(folder, dest)
|
|
return cls, tag, dest
|
|
|
|
|
|
def cmd_place_part(env, args):
|
|
root, _ = _root(env, args)
|
|
cls, tag, dest = place_part(root, args.folder, args.typeid)
|
|
print(f"placed {tag} -> {cls}/{tag}")
|
|
|
|
|
|
def cmd_commit_push(env, args):
|
|
commit_push(args.root, env.get("GIT_TOKEN", ""), args.message or "library-manager: update components")
|
|
|
|
|
|
def cmd_push_part(env, args):
|
|
tmp = tempfile.mkdtemp(prefix="components_")
|
|
clone(env, tmp)
|
|
cls, tag, _ = place_part(tmp, args.folder, args.typeid)
|
|
print(f"placed {tag} -> {cls}/{tag}")
|
|
commit_push(tmp, env.get("GIT_TOKEN", ""), args.message or f"library-manager: add {tag}")
|
|
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument("--config", default=CFG_DEFAULT)
|
|
sub = ap.add_subparsers(dest="cmd", required=True)
|
|
|
|
p = sub.add_parser("check-mpn"); p.add_argument("--mpn", required=True)
|
|
p.add_argument("--make", required=True); p.add_argument("--local"); p.add_argument("--root")
|
|
|
|
p = sub.add_parser("checkout"); p.add_argument("--dest", required=True)
|
|
|
|
p = sub.add_parser("list-type"); p.add_argument("--typeid", required=True)
|
|
p.add_argument("--root"); p.add_argument("--local"); p.add_argument("--json", action="store_true")
|
|
|
|
p = sub.add_parser("place-part"); p.add_argument("--folder", required=True)
|
|
p.add_argument("--typeid", required=True); p.add_argument("--root"); p.add_argument("--local")
|
|
|
|
p = sub.add_parser("commit-push"); p.add_argument("--root", required=True)
|
|
p.add_argument("--message")
|
|
|
|
p = sub.add_parser("push-part"); p.add_argument("--folder", required=True)
|
|
p.add_argument("--typeid", required=True); p.add_argument("--message")
|
|
|
|
args = ap.parse_args()
|
|
env = load_env(args.config)
|
|
if args.cmd in ("check-mpn", "checkout", "commit-push", "push-part") or \
|
|
(args.cmd in ("list-type", "place-part") and not getattr(args, "local", None) and not getattr(args, "root", None)):
|
|
for k in ("GIT_HOST", "GIT_TOKEN", "COMPONENTS_REPO"):
|
|
if not env.get(k):
|
|
sys.exit(f"missing {k} in env/config")
|
|
{"check-mpn": cmd_check_mpn, "checkout": cmd_checkout, "list-type": cmd_list_type,
|
|
"place-part": cmd_place_part, "commit-push": cmd_commit_push, "push-part": cmd_push_part}[args.cmd](env, args)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|