nitishKumar
/
Skill
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Skill/scripts/gitea_components.py

458 lines
20 KiB
Python
Executable File
Raw Blame History

#!/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).
LIBRARY_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 library 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).
find-part --mpn BAT46WJ --make Nexperia [--root work/] [--json]
Locate an existing part so it can be REVISED (the update path — the mirror of
check-mpn, which gates *new* parts). Reports the part's Class, tag, typeid and the
files in its folder. Point --root at a `checkout` clone and the folder it returns is
the real, editable one to change in place before `commit-push`. Exit 0 if found, 4 if
not (so a shell can branch: found -> update, not found -> fall through to add).
checkout --dest work/
Clone the library 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).
push-skill [--src <skill-dir>] [--message "..."]
Push this skill's files to the SKILL repo (token blanked out) and MERGE the changelog:
the new rows in the local CHANGELOG.xlsx are appended onto the repo's existing one, so
previous entries in Gitea are preserved, not overwritten. The merged changelog is
written back locally so local and Gitea stay in sync.
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,
SKILL_ROOT, TEMPLATE_XLSX, VERSIONS_JSON, CHANGELOG_XLSX,
CHANGELOG_HEADERS, resolve_operator)
CFG_DEFAULT = os.path.join(os.path.dirname(__file__), "..", "config", "gitea.env")
# The skill's stateful files (accumulate over time) — kept in the SKILL repo and pulled at the
# start of a run so template/version/changelog changes always build on Gitea's latest state.
# Paths are derived from common so the writer (append_parameter) and the sync can never drift.
STATE_FILES = [os.path.relpath(p, SKILL_ROOT)
for p in (TEMPLATE_XLSX, VERSIONS_JSON, CHANGELOG_XLSX)]
CHANGELOG_REL = os.path.relpath(CHANGELOG_XLSX, SKILL_ROOT)
# ------------------------------------------------------------------ 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", "LIBRARY_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, repo=None):
repo = repo or env["LIBRARY_REPO"]
url = repo_url(env, 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 {repo} failed (host reachable / token scope / repo exists?):\n{err[:400]}")
return dest
def commit_push(root, token, msg, author=None):
"""Commit + push. `author` = (name, email) of the operator; when given, the commit is
authored by that real person (visible in git log / Gitea commit view / blame). Without it,
a generic bot identity is used."""
name, email = author or (None, None)
name = name or "library-manager"
email = email or "datasheet-bot@local"
for args in (["config", "user.email", email],
["config", "user.name", name],
["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 _typeid_of(tag):
"""Recover the typeid from a part tag <MPN>_<make>_<typeid>. The typeid is the last
underscore-delimited token, so this is robust even if the MPN itself contains underscores."""
return tag.rsplit("_", 1)[1] if "_" in tag else ""
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_find_part(env, args):
"""Locate an existing part by MPN+make so it can be REVISED — the mirror image of
check-mpn (which gates *new* parts). Reports the folder, its typeid and its files.
Point --root at a checkout (from `checkout`) and the folder it returns is the real,
editable, committable one — that's the update path: checkout -> find-part -> edit in
place -> commit-push. Without --root/--local it clones a throwaway copy for a read-only
look. Exit 0 if found, 4 if not (so a shell can branch)."""
root, _ = _root(env, args)
prefix = mpn_make_prefix(args.mpn, args.make)
matches = list(find_part_dirs(root, prefix=prefix))
if not matches:
print("[]" if args.json else "NOT FOUND")
sys.exit(4)
out = [{"class": cls, "tag": name, "typeid": _typeid_of(name),
"folder": pdir, "files": sorted(os.listdir(pdir))}
for cls, name, pdir in matches]
if args.json:
import json
print(json.dumps(out, indent=2))
else:
for o in out:
print(f'{o["class"]}/{o["tag"]} (typeid {o["typeid"]}) -> {", ".join(o["files"])}')
def cmd_checkout(env, args):
dest = clone(env, args.dest)
print(f"library 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 _operator(args):
"""(name, email) of who's running this — from --author, else env, else identity file."""
return resolve_operator(getattr(args, "author", None))
def _by(name):
return f" (by {name})" if name else ""
def cmd_commit_push(env, args):
name, email = _operator(args)
commit_push(args.root, env.get("GIT_TOKEN", ""),
args.message or f"library-manager: update components{_by(name)}", author=(name, email))
def cmd_push_part(env, args):
name, email = _operator(args)
tmp = tempfile.mkdtemp(prefix="components_")
clone(env, tmp)
cls, tag, _ = place_part(tmp, args.folder, args.typeid)
print(f"placed {tag} -> {cls}/{tag}" + (f" (by {name})" if name else ""))
commit_push(tmp, env.get("GIT_TOKEN", ""),
args.message or f"library-manager: add {tag}{_by(name)}", author=(name, email))
# ------------------------------------------------------------------ skill repo + changelog
def _changelog_rows(path):
"""(workbook, sheet, [row-tuples]) for a changelog xlsx; ([]) if the file is absent. Rows are
padded/truncated to the canonical column count so old 5-column changelogs (no 'By') merge
cleanly with new 6-column ones."""
import openpyxl
if not os.path.exists(path):
return None, None, []
n = len(CHANGELOG_HEADERS)
wb = openpyxl.load_workbook(path)
ws = wb["Changelog"] if "Changelog" in wb.sheetnames else wb.active
rows = []
for r in ws.iter_rows(min_row=2, values_only=True):
vals = ["" if c is None else str(c) for c in r]
vals = (vals + [""] * n)[:n]
if any(v for v in vals):
rows.append(tuple(vals))
return wb, ws, rows
def merge_changelog(local_path, repo_path):
"""Union the local changelog's rows into the repo's changelog and keep the result sorted by
date (chronological, oldest first). This keeps Gitea's changelog as the growing source of
truth — appended, never overwritten — and always in order regardless of which machine
pushed when. Writes the merged file to repo_path and returns how many new rows were added."""
import shutil
from openpyxl.styles import Border, Side, Alignment
if not os.path.exists(local_path):
return 0
if not os.path.exists(repo_path): # repo has none yet -> seed from local
os.makedirs(os.path.dirname(repo_path), exist_ok=True)
shutil.copy(local_path, repo_path)
_, _, rows = _changelog_rows(repo_path)
return len(rows)
from openpyxl.styles import Font, PatternFill
repo_wb, repo_ws, repo_rows = _changelog_rows(repo_path)
_, _, local_rows = _changelog_rows(local_path)
seen = set(repo_rows)
merged = list(repo_rows)
added = 0
for row in local_rows:
if row not in seen:
merged.append(row)
seen.add(row)
added += 1
merged.sort(key=lambda r: r[0]) # chronological by Date (column A)
n = len(CHANGELOG_HEADERS)
desc_col = CHANGELOG_HEADERS.index("Description") + 1
s = Side(style="thin", color="BFBFBF")
border = Border(left=s, right=s, top=s, bottom=s)
if repo_ws.max_row > 1: # clear everything, rewrite header + rows
repo_ws.delete_rows(1, repo_ws.max_row)
for c, h in enumerate(CHANGELOG_HEADERS, start=1): # canonical header (migrates 5->6 cols)
cell = repo_ws.cell(1, c, h)
cell.font = Font(name="Calibri", bold=True)
cell.fill = PatternFill("solid", fgColor="B6D7A8")
cell.border = border
cell.alignment = Alignment(horizontal="center", vertical="center")
for row in merged:
rr = repo_ws.max_row + 1
for c in range(1, n + 1):
cell = repo_ws.cell(rr, c, row[c - 1] if c - 1 < len(row) else "")
cell.border = border
cell.alignment = Alignment(vertical="center", wrap_text=(c == desc_col), horizontal="left")
repo_wb.freeze_panes = None
repo_ws.freeze_panes = "A2"
repo_wb.save(repo_path)
return added
def _copy_skill_files(src, dst):
"""Copy the skill's own files into a repo clone, minus junk and the changelog (merged
separately) and WITHOUT the real token (config/gitea.env is copied with GIT_TOKEN blanked)."""
skip = {".git", "__pycache__"}
for root, dirs, files in os.walk(src):
dirs[:] = [d for d in dirs if d not in skip]
rel = os.path.relpath(root, src)
for f in files:
if f.endswith((".bak.xlsx",)) or "OLD-18class" in f:
continue
relpath = os.path.normpath(os.path.join(rel, f))
if relpath == CHANGELOG_REL: # changelog handled by merge, not copy
continue
s = os.path.join(root, f)
d = os.path.join(dst, relpath)
os.makedirs(os.path.dirname(d), exist_ok=True)
if relpath == os.path.join("config", "gitea.env"):
text = open(s, encoding="utf-8").read()
text = re.sub(r"(?m)^(GIT_TOKEN=).*$", r"\1", text) # blank the secret
open(d, "w", encoding="utf-8").write(text)
else:
shutil.copy2(s, d)
def cmd_push_skill(env, args):
"""Push this skill's files to the SKILL repo and MERGE the changelog into the repo's
existing one (append-only, so previous entries are preserved). The merged changelog is
also written back to the local skill assets so local and Gitea stay in sync."""
src = os.path.abspath(args.src)
tmp = tempfile.mkdtemp(prefix="skill_")
clone(env, tmp, env["SKILL_REPO"])
_copy_skill_files(src, tmp)
local_cl = os.path.join(src, CHANGELOG_REL)
repo_cl = os.path.join(tmp, CHANGELOG_REL)
added = merge_changelog(local_cl, repo_cl)
if os.path.exists(repo_cl): # sync merged history back to local
shutil.copy(repo_cl, local_cl)
print(f"changelog: merged (+{added} new row(s)) into the skill repo's CHANGELOG.xlsx")
name, email = _operator(args)
commit_push(tmp, env.get("GIT_TOKEN", ""),
args.message or f"library-manager: sync skill files + changelog{_by(name)}",
author=(name, email))
def cmd_pull_skill(env, args):
"""Pull the skill's stateful files (template.xlsx, versions.json, CHANGELOG.xlsx) from the
SKILL repo into the local skill BEFORE making changes. This is what makes template/version
changes cumulative: without it, a fresh run starts from the packaged v1 state, so a second
change wouldn't build on the first (versions would restart at v1 and the changelog would
look like it only has the latest change). Run this at the start of every run."""
src = os.path.abspath(args.src)
tmp = tempfile.mkdtemp(prefix="skillpull_")
clone(env, tmp, env["SKILL_REPO"])
pulled = []
for rel in STATE_FILES:
s = os.path.join(tmp, rel)
d = os.path.join(src, rel)
if os.path.exists(s):
os.makedirs(os.path.dirname(d), exist_ok=True)
shutil.copy2(s, d)
pulled.append(rel)
print("pulled from skill repo: " + (", ".join(pulled) if pulled else
"nothing yet (skill repo has no template/version/changelog state)"))
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("find-part"); p.add_argument("--mpn", required=True)
p.add_argument("--make", required=True); p.add_argument("--local"); p.add_argument("--root")
p.add_argument("--json", action="store_true")
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.add_argument("--author", help="'Name <email>' of the operator")
p = sub.add_parser("push-part"); p.add_argument("--folder", required=True)
p.add_argument("--typeid", required=True); p.add_argument("--message")
p.add_argument("--author", help="'Name <email>' of the operator (who added the part)")
p = sub.add_parser("push-skill")
p.add_argument("--author", help="'Name <email>' of the operator (who made the change)")
p.add_argument("--src", default=os.path.join(os.path.dirname(__file__), ".."),
help="skill directory to push (defaults to this skill)")
p.add_argument("--message")
p = sub.add_parser("pull-skill")
p.add_argument("--src", default=os.path.join(os.path.dirname(__file__), ".."),
help="skill directory to sync into (defaults to this skill)")
args = ap.parse_args()
env = load_env(args.config)
needs = {"GIT_HOST", "GIT_TOKEN"}
if args.cmd in ("push-skill", "pull-skill"):
needs.add("SKILL_REPO")
elif args.cmd in ("check-mpn", "checkout", "commit-push", "push-part") or \
(args.cmd in ("list-type", "place-part", "find-part") and not getattr(args, "local", None) and not getattr(args, "root", None)):
needs.add("LIBRARY_REPO")
for k in sorted(needs):
if not env.get(k):
sys.exit(f"missing {k} in env/config")
{"check-mpn": cmd_check_mpn, "find-part": cmd_find_part, "checkout": cmd_checkout,
"list-type": cmd_list_type, "place-part": cmd_place_part, "commit-push": cmd_commit_push,
"push-part": cmd_push_part, "push-skill": cmd_push_skill, "pull-skill": cmd_pull_skill}[args.cmd](env, args)
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink