179 lines
8.8 KiB
Python
179 lines
8.8 KiB
Python
#!/usr/bin/env python3
|
|
"""Reconcile a datasheet-extractor run against the DFS + Parameters Gitea repos.
|
|
|
|
Only DFS and Parameters are MPN-indexed (per-part folders / per-part rows). Skill_Assets
|
|
holds the skill + templates, not per-part data, so it is never reconciled here - push it
|
|
with push_to_gitea.sh as before.
|
|
|
|
Identity = the part tag <MPN>_<make>_<typeid> (see fill_templates.part_tag). Same tag =
|
|
same part, so re-processing an MPN that already lives in Gitea is a conflict.
|
|
|
|
Two passes:
|
|
1) CHECK - clone both repos, report which run MPNs already exist. No writes, no push.
|
|
python scripts/gitea_reconcile.py --parts parts.json --dfs-src <dfs_stage> \
|
|
--template assets/template/template.xlsx --report conflicts.json
|
|
2) APPLY - with --decisions (a JSON map tag -> "replace"|"discard") + --push:
|
|
python scripts/gitea_reconcile.py --parts parts.json --dfs-src <dfs_stage> \
|
|
--template assets/template/template.xlsx --decisions decisions.json --push
|
|
|
|
replace -> overwrite that MPN's DFS folder AND its Parameters row (a fresh datasheet /
|
|
footprint / symbol / values fully supersede the old ones).
|
|
discard -> leave the copy already in Gitea untouched; drop the newly-extracted one.
|
|
New (non-conflicting) MPNs are always added. Every OTHER row/folder is preserved -
|
|
the Parameters sheet is merged, never wholesale-overwritten.
|
|
|
|
Refuses to push if any conflict has no decision, so nothing is ever resolved silently.
|
|
"""
|
|
import argparse, json, os, re, shutil, subprocess, sys, tempfile
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import fill_templates as ft
|
|
|
|
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","DFS_REPO","PARAMS_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(url, dest, token):
|
|
r=subprocess.run(["git","clone","--depth","1",url,dest], capture_output=True, text=True)
|
|
return r.returncode==0, (r.stderr or "").replace(token,"***")
|
|
|
|
def type_filename(ctype): return re.sub(r'\s+','_',ctype)+".xlsx"
|
|
|
|
def run_rows(parts):
|
|
return [(ft.part_tag(p), p["type"], p.get("mpn",""), p) for p in parts]
|
|
|
|
def scan_conflicts(parts, dfs_clone, params_clone):
|
|
conflicts=[]; ptags={}
|
|
for tag,ctype,mpn,_ in run_rows(parts):
|
|
in_dfs=os.path.isdir(os.path.join(dfs_clone, tag))
|
|
if ctype not in ptags:
|
|
f=os.path.join(params_clone, type_filename(ctype))
|
|
ptags[ctype]=set(ft.read_type_rows(f)[1].keys()) if os.path.exists(f) else set()
|
|
in_params=tag in ptags[ctype]
|
|
if in_dfs or in_params:
|
|
conflicts.append({"tag":tag,"type":ctype,"mpn":mpn,
|
|
"in_dfs":in_dfs,"in_params":in_params})
|
|
return conflicts
|
|
|
|
def apply_dfs(parts, dfs_src, dfs_clone, decisions):
|
|
for tag,_,_,_ in run_rows(parts):
|
|
src=os.path.join(dfs_src, tag)
|
|
if not os.path.isdir(src): continue
|
|
dst=os.path.join(dfs_clone, tag); dec=decisions.get(tag)
|
|
if os.path.isdir(dst):
|
|
if dec=="discard": print(f" DFS keep existing : {tag}")
|
|
elif dec=="replace":
|
|
shutil.rmtree(dst); shutil.copytree(src,dst); print(f" DFS replaced : {tag}")
|
|
else: print(f" DFS SKIP (undecided): {tag}")
|
|
else:
|
|
shutil.copytree(src,dst); print(f" DFS added : {tag}")
|
|
|
|
def apply_params(parts, params_clone, template, decisions, version):
|
|
by={}
|
|
for p in parts: by.setdefault(p["type"], []).append(p)
|
|
for ctype, plist in by.items():
|
|
headers=ft.template_headers(template, ctype)
|
|
if headers is None:
|
|
print(f" ! no template sheet for '{ctype}' - skipping Parameters"); continue
|
|
f=os.path.join(params_clone, type_filename(ctype))
|
|
rows=dict(ft.read_type_rows(f)[1]) if os.path.exists(f) else {}
|
|
for p in plist:
|
|
tag,row=ft.part_to_row(p, headers); dec=decisions.get(tag)
|
|
if tag in rows:
|
|
if dec=="discard": print(f" PARAMS keep row : {tag}")
|
|
elif dec=="replace": rows[tag]=row; print(f" PARAMS replaced : {tag}")
|
|
else: print(f" PARAMS SKIP (undecided): {tag}")
|
|
else:
|
|
rows[tag]=row; print(f" PARAMS added : {tag}")
|
|
ft.write_type_workbook(ctype, headers, rows, version, params_clone)
|
|
|
|
def commit_push(clone_dir, msg, token):
|
|
for args in (["config","user.email","datasheet-bot@local"],
|
|
["config","user.name","datasheet-extractor"],
|
|
["checkout","-B","main"], ["add","-A"]):
|
|
subprocess.run(["git","-C",clone_dir]+args, capture_output=True)
|
|
if subprocess.run(["git","-C",clone_dir,"diff","--cached","--quiet"]).returncode==0:
|
|
print(f" nothing new for {os.path.basename(clone_dir)}"); return
|
|
subprocess.run(["git","-C",clone_dir,"commit","-m",msg], capture_output=True)
|
|
r=subprocess.run(["git","-C",clone_dir,"push","-u","origin","main"],
|
|
capture_output=True, text=True)
|
|
print(" "+((r.stdout+r.stderr).replace(token,"***").strip() or "pushed."))
|
|
|
|
def undecided(conflicts, decisions):
|
|
return [c["tag"] for c in conflicts if decisions.get(c["tag"]) not in ("replace","discard")]
|
|
|
|
def main():
|
|
ap=argparse.ArgumentParser()
|
|
ap.add_argument("--parts", required=True)
|
|
ap.add_argument("--dfs-src", required=True, help="staging dir of MPN_make_typeid/ folders")
|
|
ap.add_argument("--template", required=True)
|
|
ap.add_argument("--decisions", default=None, help="JSON map: tag -> replace|discard")
|
|
ap.add_argument("--report", default=None, help="write the conflict report here")
|
|
ap.add_argument("--push", action="store_true", help="apply decisions and push")
|
|
ap.add_argument("--on-conflict", choices=["ask","replace","discard"], default="ask",
|
|
help="what to do when an MPN already exists: ask the user (default), or "
|
|
"apply one policy to every conflict for an unattended run")
|
|
ap.add_argument("--config",
|
|
default=os.path.join(os.path.dirname(__file__),"..","config","gitea.env"))
|
|
a=ap.parse_args()
|
|
|
|
parts=json.load(open(a.parts))["parts"]
|
|
env=load_env(a.config)
|
|
for k in ("GIT_HOST","GIT_TOKEN","DFS_REPO","PARAMS_REPO"):
|
|
if not env.get(k): sys.exit(f"missing {k} in env/config")
|
|
token=env["GIT_TOKEN"]
|
|
version=ft.resolve_version(a.template, None)
|
|
decisions=json.load(open(a.decisions)) if a.decisions else {}
|
|
|
|
tmp=tempfile.mkdtemp(prefix="reconcile_")
|
|
dfs_clone=os.path.join(tmp,"dfs"); params_clone=os.path.join(tmp,"params")
|
|
ok,err=clone(repo_url(env,env["DFS_REPO"]), dfs_clone, token)
|
|
if not ok: sys.exit(f"clone DFS failed (host reachable / token scope?):\n{err[:400]}")
|
|
ok,err=clone(repo_url(env,env["PARAMS_REPO"]), params_clone, token)
|
|
if not ok: sys.exit(f"clone Parameters failed:\n{err[:400]}")
|
|
|
|
conflicts=scan_conflicts(parts, dfs_clone, params_clone)
|
|
report={"has_conflicts":bool(conflicts),"conflicts":conflicts}
|
|
if a.report: json.dump(report, open(a.report,"w"), indent=2)
|
|
print(json.dumps(report, indent=2))
|
|
|
|
if not a.push:
|
|
if conflicts:
|
|
print(f"\n{len(conflicts)} existing MPN(s) found — ask the user discard/replace, "
|
|
"then re-run with --decisions/--push (or set --on-conflict for unattended).")
|
|
else:
|
|
print("\nNo conflicts — re-run with --push to add everything (no prompt needed).")
|
|
return
|
|
|
|
# A standing policy resolves every conflict without a prompt; otherwise conflicts must
|
|
# each carry an explicit decision so nothing gets overwritten behind the user's back.
|
|
if a.on_conflict in ("replace","discard"):
|
|
for c in conflicts:
|
|
decisions.setdefault(c["tag"], a.on_conflict)
|
|
miss=undecided(conflicts, decisions)
|
|
if miss:
|
|
sys.exit("Refusing to push — need a discard/replace decision (or --on-conflict) for: "
|
|
+", ".join(miss))
|
|
print("Applying to DFS:")
|
|
apply_dfs(parts, a.dfs_src, dfs_clone, decisions)
|
|
print("Applying to Parameters:")
|
|
apply_params(parts, params_clone, a.template, decisions, version)
|
|
print("Pushing:")
|
|
commit_push(dfs_clone, "datasheet-extractor: reconcile DFS", token)
|
|
commit_push(params_clone, "datasheet-extractor: reconcile Parameters", token)
|
|
print("\nDone — DFS + Parameters reconciled and pushed.")
|
|
|
|
if __name__=="__main__":
|
|
main()
|