Sync datasheet-extractor skill assets

main
datasheet-extractor 2026-07-07 11:26:49 +00:00
parent 10fc380a0f
commit 1537551e36
5 changed files with 373 additions and 181 deletions

112
SKILL.md
View File

@ -5,8 +5,10 @@ description: >-
template (18 component types incl. resistors, capacitors, inductors, diodes,
transistors, ICs, protection, power modules, connectors, sensors, and more).
Column A = MPN_make_typeid (make = first word of manufacturer, typeid from the
taxonomy). Writes one Excel file per type with a Meta sheet holding the template
version, assembles a per-part DFS folder (datasheet, Altium footprint, Altium symbol), and pushes
taxonomy). Writes ONE master workbook (Components_Master.xlsx) with sheets only for the
types whose datasheets were provided, plus a Meta sheet at the end (template version,
total components, date, time), assembles
a per-part DFS folder (datasheet, Altium footprint, Altium symbol), and pushes
design files to the Gitea DFS repo, Excel outputs to the Parameters repo, and skill
files plus templates to the Skill_Assets repo. Also appends new parameters to a
template as a new version. Use WHENEVER the user uploads component datasheets and
@ -55,9 +57,11 @@ Column A of every sheet (header `MPN_make_type`) and each DFS folder name is:
3. **Extract values** into that type's columns (headers come from
`assets/template/template.xlsx`). Convert to each header's unit; leave blank if the
datasheet doesn't state it (an honest blank beats a guess).
4. **Build the Excel outputs** with `scripts/fill_templates.py` — one `<Type>.xlsx` per
type present, column A = `MPN_make_typeid`, plus a **Meta** sheet holding the template
version. (See Producing outputs.)
4. **Build the master workbook** with `scripts/fill_templates.py` — a single
`Components_Master.xlsx` that mirrors the template but keeps **only the sheets for the
types you actually extracted** (the components whose datasheets were given), each with
column A = `MPN_make_typeid`, and a single **Meta** sheet appended at the end. (See
Producing outputs.)
5. **Assemble each part's DFS folder** `MPN_make_typeid/` containing `MPN_data` (the
datasheet), `MPN_fp` (footprint), `MPN_sym` (symbol). See Footprint & symbol.
6. **Push everything to Gitea — automatically.** At the end of a run, push all outputs to
@ -83,9 +87,17 @@ python scripts/fill_templates.py parts.json \
--template assets/template/template.xlsx --dest <outputs-dir>
```
Produces one `<Type>.xlsx` per type present (only types with parts), each with a **Meta**
sheet (Template Version, date, type, part count). Deliver these files to the user, and
push them to the Parameters repo.
Produces a **single** `Components_Master.xlsx`. It starts from a copy of the reference
template (so every kept sheet keeps its exact headers, styling, widths and freeze), fills
this run's parts into the matching sheet (column A = `MPN_make_typeid`), then **drops every
type sheet with no parts** — so the file holds only the components whose datasheets were
provided. A single **Meta** sheet is appended **at the end** with: Template Version, Total
Components, Date, Time, and a per-sheet count breakdown. Deliver this one file to the user;
the Parameters repo stores the same master, accumulated across runs (see Pushing to Gitea).
Sheet naming follows the template (Diode, Resistor, …), not `Sheet1/Sheet2`. If the user
supplies a new reference template, drop it in at `assets/template/template.xlsx` and the
master mirrors whatever sheets/headers it has.
## Footprint & symbol (for the DFS folder)
@ -95,26 +107,34 @@ For each part, the DFS folder needs `MPN_fp` (footprint) and `MPN_sym` (symbol)
that carries both — because that's what the design team actually consumes. Get them in
this order:
1. **Scout which sources actually hold this part first — before any login.** Search
across SnapEDA, Ultra Librarian, Component Search Engine, DigiKey (its "PCB Symbol,
Footprint & 3D Model" links), and the manufacturer's own CAD downloads, and for each
note whether *this exact MPN* has a downloadable **Altium** symbol + footprint, and
whether grabbing it needs an account. The point is to find where the model lives before
asking the user to do anything.
1. **Auto-download via the user's pre-logged-in Chrome** (`scripts/fetch_cad.py`). The user
keeps signed-in CAD-site windows open; the script attaches to that running Chrome and
reuses those live sessions — it never logs in, and no password is ever handled. One-time
setup: the user quits Chrome, relaunches it with `--remote-debugging-port=9222
--user-data-dir="$HOME/cad-chrome"`, and signs into the sites. Then, per part:
2. **Surface what you found, then handle the login.** Tell the user which sources have the
part in Altium form — e.g. "BAT46WJ: SnapEDA and Ultra Librarian both have an Altium
symbol + footprint; DigiKey lists a model too" — so they can pick. Then get the file:
- If you're driving the user's own signed-in browser (e.g. Claude in Chrome) and it's
already logged into that source, just download — no password ever changes hands.
- Otherwise **ask the user to sign in to the source they picked** (or to grab the file
and drop it in), then continue. Never type, store, or ask for their password, and
never write credentials into this skill — it's pushed to Skill_Assets, and a secret in
a repo is a real leak (same reason the `GIT_TOKEN` is blanked before syncing).
```bash
python scripts/fetch_cad.py --mpn BAT46WJ --dest <DFS_folder> \
--sites snapeda,ultralibrarian,componentsearchengine,digikey
```
Once you have a candidate, **cross-check** its pad/pin count and key package dimensions
against the datasheet's package drawing before trusting it — library models are
sometimes wrong, or drawn for a different variant of the series.
It tries each site in order, downloads the Altium model, and drops it into the DFS folder
named `MPN_fp` / `MPN_sym` (or `MPN_cad` for a bundle). **It never solves CAPTCHAs or
defeats bot-detection** — if a site shows a login wall, CAPTCHA, or bot-check, that
adapter stops and reports `manual` with the URL. Note: this runs on the *user's* machine
(where the browsers are), and some sites' terms restrict automated downloads — treat it
as a convenience with the manual fallback below.
2. **Manual fallback** (script returned `manual`/`notfound`, or no browser session): **share
the user the direct link** where the model lives and ask them to download it and send it
back; then attach it into the DFS folder with the agreed naming (`MPN_fp` / `MPN_sym`).
Never type, store, or ask for their password, and never write credentials into this skill
— it's pushed to Skill_Assets, and a secret in a repo is a real leak (same reason the
`GIT_TOKEN` is blanked before syncing).
Whichever path, once you have a candidate **cross-check** its pad/pin count and key
package dimensions against the datasheet's package drawing before trusting it — library
models are sometimes wrong, or drawn for a different variant of the series.
3. **If no good model exists, generate the footprint/symbol yourself** as Altium files
from the datasheet's package drawing and pinout — but only when you can do it reliably
@ -142,10 +162,11 @@ default.
The three repos split into two kinds, and they're pushed differently:
- **DFS** and **Parameters** are **MPN-indexed** — DFS has one `MPN_make_typeid/` folder
per part, Parameters has one row per part (keyed on column A) inside each `<Type>.xlsx`.
Because a re-run can collide with a part that's already there, these go through
`scripts/gitea_reconcile.py`, which is dedup-aware and **merge-safe** (it never blows
away parts it isn't touching). See *Handling parts already in Gitea* below.
per part; Parameters holds a **single master workbook** (`Components_Master.xlsx`) whose
type sheets each have one row per part (keyed on column A). Because a re-run can collide
with a part that's already there, these go through `scripts/gitea_reconcile.py`, which is
dedup-aware and **merge-safe** (it never blows away parts it isn't touching). See
*Handling parts already in Gitea* below.
- **Skill_Assets** holds the skill itself + templates — it is **not** per-part, so there's
nothing MPN-level to reconcile. Push it with `scripts/push_to_gitea.sh` as before.
@ -163,10 +184,11 @@ domain allowlisted) it fails clearly and leaves the staged files for a manual pu
real token. Before syncing skill assets, blank the `GIT_TOKEN` line in the copy you push,
or push everything except `config/gitea.env`.
> Why not push Parameters with the flat script? `fill_templates.py` writes a `<Type>.xlsx`
> holding **only this run's** rows. Copied flat over the repo, it would overwrite the file
> and delete every previously-stored part of that type. `gitea_reconcile.py` merges into
> the repo's existing sheet instead, so only the parts you decided on change.
> Why not push Parameters with the flat script? `fill_templates.py` writes a master
> workbook holding **only this run's** rows. Copied flat over the repo, it would overwrite
> the stored master and delete every previously-stored part. `gitea_reconcile.py` opens the
> repo's master, merges this run's rows into the right sheets, and writes it back — so only
> the parts you decided on change, and the Meta sheet (totals, date, time) refreshes.
## Handling parts already in Gitea (discard vs replace)
@ -222,21 +244,27 @@ python scripts/append_parameter.py --type Diode --param "Reverse Recovery Time(n
This adds the column at the end and increments `assets/template/VERSION` (v1 → v2 → …).
Then **push the updated `assets/template/template.xlsx` + `VERSION` to the Skill_Assets
repo** so the new template version is archived. New outputs will record the new version in
their Meta sheet automatically.
repo** so the new template version is archived. New masters record the new version in the
Meta sheet automatically.
## Resources
- `assets/template/template.xlsx` — master template, one sheet per type (source of headers).
- `assets/template/template.xlsx` — master template, one sheet per type (source of headers,
styling, and sheet order the master workbook mirrors).
- `assets/template/Type_ID.xlsx` + `references/taxonomy.md` — Class → Subclass → Type ID.
- `assets/template/VERSION` — current template version (integer; Meta sheet shows `vN`).
- `scripts/fill_templates.py` — build per-type Excel outputs (+ Meta sheet); also exposes
the row/style helpers (`part_to_row`, `read_type_rows`, `write_type_workbook`) that the
reconcile step reuses so merged sheets look identical to freshly generated ones.
- `assets/template/VERSION` — current template version (integer; the Meta sheet shows `vN`).
- `scripts/fill_templates.py` — build the single master workbook `Components_Master.xlsx`
(only sheets for the extracted types + a trailing Meta sheet); also exposes the shared
helpers (`part_to_row`, `template_headers`, `sheet_rows`, `read_all_rows`, `build_master`,
`MASTER_NAME`) the reconcile step reuses so the Gitea master is built and merged identically.
- `scripts/gitea_reconcile.py` — add new MPNs and push automatically; only stops on an MPN
that already exists (discard/replace), or run unattended with `--on-conflict replace`.
Merge-safe: other rows/folders untouched.
Merge-safe: merges into the repo's master workbook, other rows/sheets/folders untouched.
- `scripts/push_to_gitea.sh` — push a folder's contents to a Gitea repo (flat). Used for
Skill_Assets (not MPN-indexed).
- `scripts/append_parameter.py` — append a parameter to a template + bump version.
- `scripts/fetch_cad.py` — download Altium footprint/symbol by attaching to the user's
pre-logged-in Chrome (remote-debugging port); tries sites in order, defers on
CAPTCHA/login (never bypasses), and names files `MPN_fp`/`MPN_sym`. Runs on the user's
machine; needs `selenium`.
- `config/gitea.env` — host, user, token, and the DFS / Parameters / Skill_Assets repos (**secret**).

155
scripts/fetch_cad.py Normal file
View File

@ -0,0 +1,155 @@
#!/usr/bin/env python3
"""Download Altium footprints/symbols by driving the user's ALREADY-LOGGED-IN Chrome.
Runs on the USER'S machine (where the signed-in browser windows live) - NOT in the sandbox.
The user starts Chrome once with a remote-debugging port and logs into the CAD sites
(SnapEDA, Ultra Librarian, Component Search Engine, DigiKey, ...). This script ATTACHES to
that running Chrome and reuses those live sessions, so NO password is ever typed, stored,
or seen here.
# user, once: fully quit Chrome, then relaunch with a debug port + own profile
# macOS: "/Applications/Google Chrome.app/Contents/MacOS/Google Chrome" \
# --remote-debugging-port=9222 --user-data-dir="$HOME/cad-chrome"
# Linux: google-chrome --remote-debugging-port=9222 --user-data-dir="$HOME/cad-chrome"
# then sign into the CAD sites in that window, then:
python fetch_cad.py --mpn BAT46WJ --dest <DFS_folder> --sites snapeda,ultralibrarian
SAFETY: this NEVER solves CAPTCHAs or defeats bot-detection. If a site shows a login wall,
CAPTCHA, or bot-check, the adapter STOPS and returns ('manual', url) so the caller can fall
back to asking the user to grab the file by hand. It also does not create accounts or enter
passwords. Per-site selectors live in SITES below and may need tweaking when a site changes
its layout (they can't be verified from the sandbox).
"""
import argparse, glob, os, re, shutil, sys, time
# ----- pure helpers (no browser) -------------------------------------------------
BLOCK_MARKERS=("recaptcha","g-recaptcha","hcaptcha","captcha","are you a robot",
"please sign in","log in to download","sign in to download","cf-challenge")
def looks_blocked(page_source_lower):
"""Heuristic: does the page look like a CAPTCHA / login wall / bot-check?
If so we DEFER to the human - we never try to get past it."""
return any(m in page_source_lower for m in BLOCK_MARKERS)
def newest_download(dl_dir, known_before, timeout=60):
"""Return the path of a file that appeared in dl_dir after `known_before`, once it has
finished downloading (no .crdownload/.part). None on timeout."""
end=time.time()+timeout
while time.time()<end:
now=set(glob.glob(os.path.join(dl_dir,"*")))
new=[f for f in now-known_before if not f.endswith((".crdownload",".part",".tmp"))]
if new:
f=max(new, key=os.path.getmtime)
if os.path.getsize(f)>0: return f
time.sleep(1)
return None
def place(downloaded, dest, mpn, kind):
"""Move a downloaded CAD file into the DFS folder named <mpn>_<kind><ext>
(kind = 'fp' | 'sym' | 'cad' for a bundle/archive). Returns the new path."""
os.makedirs(dest, exist_ok=True)
ext=os.path.splitext(downloaded)[1] or ""
out=os.path.join(dest, f"{mpn}_{kind}{ext}")
shutil.move(downloaded, out); return out
# ----- browser attach ------------------------------------------------------------
def attach(port, dl_dir):
"""Attach to a Chrome already running with --remote-debugging-port=<port> and point
downloads at dl_dir. Selenium is imported lazily so this file stays importable without
a browser present."""
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
opts=Options(); opts.add_experimental_option("debuggerAddress", f"127.0.0.1:{port}")
drv=webdriver.Chrome(options=opts)
try: # route downloads to our folder via CDP
drv.execute_cdp_cmd("Page.setDownloadBehavior",
{"behavior":"allow","downloadPath":os.path.abspath(dl_dir)})
except Exception:
pass
return drv
# ----- per-site adapters ---------------------------------------------------------
# Each returns ('ok', path) | ('manual', url) | ('notfound', url). Adapters open the site's
# search for the MPN, prefer the Altium export, click download, and wait for the file.
# Selectors are best-effort starting points - verify/adjust on the first real run.
def _snapeda(drv, mpn, dl_dir):
from selenium.webdriver.common.by import By
url=f"https://www.snapeda.com/search/?q={mpn}&has_symbol=true"
drv.get(url); time.sleep(3)
if looks_blocked(drv.page_source.lower()): return ("manual", drv.current_url)
hits=drv.find_elements(By.CSS_SELECTOR, "a[href*='/parts/']")
if not hits: return ("notfound", url)
hits[0].click(); time.sleep(3)
if looks_blocked(drv.page_source.lower()): return ("manual", drv.current_url)
before=set(glob.glob(os.path.join(dl_dir,"*")))
# try to pick an Altium download control; fall back to any 'download' button
btns=drv.find_elements(By.XPATH,
"//*[contains(translate(text(),'ALTIUM','altium'),'altium')]"
"//ancestor-or-self::a | //a[contains(translate(.,'DOWNLOAD','download'),'download')]")
if not btns: return ("manual", drv.current_url)
btns[0].click()
f=newest_download(dl_dir, before)
return ("ok", f) if f else ("manual", drv.current_url)
def _searchonly(base):
"""Adapters not yet wired for auto-download: land on the site's search and defer to the
user (returns 'manual' with the URL). Fill in the click/download flow to automate."""
def adapter(drv, mpn, dl_dir):
url=base.format(mpn=mpn)
try: drv.get(url); time.sleep(2)
except Exception: pass
return ("manual", url)
return adapter
SITES={
"snapeda": _snapeda,
"ultralibrarian": _searchonly("https://app.ultralibrarian.com/search?queryText={mpn}"),
"componentsearchengine": _searchonly("https://componentsearchengine.com/search?term={mpn}"),
"digikey": _searchonly("https://www.digikey.com/en/products/result?keywords={mpn}"),
}
def main():
ap=argparse.ArgumentParser()
ap.add_argument("--mpn", required=True)
ap.add_argument("--dest", required=True, help="the part's DFS folder (MPN_make_typeid/)")
ap.add_argument("--sites", default="snapeda",
help="comma list, tried in order: "+", ".join(SITES))
ap.add_argument("--port", type=int, default=9222, help="Chrome remote-debugging port")
ap.add_argument("--kind", default="cad", choices=["fp","sym","cad"],
help="name the file <mpn>_<kind>; 'cad' for a bundle/archive")
ap.add_argument("--download-dir", default=None,
help="Chrome's download folder (default: <dest>/.dl)")
a=ap.parse_args()
dl_dir=a.download_dir or os.path.join(a.dest, ".dl"); os.makedirs(dl_dir, exist_ok=True)
try:
drv=attach(a.port, dl_dir)
except Exception as e:
sys.exit(f"Could not attach to Chrome on port {a.port}. Start Chrome with "
f"--remote-debugging-port={a.port} and sign in first.\n{e}")
manual=[]
for site in [s.strip() for s in a.sites.split(",") if s.strip()]:
adapter=SITES.get(site)
if not adapter: print(f" ! unknown site '{site}' - skipping"); continue
try:
status,info=adapter(drv, a.mpn, dl_dir)
except Exception as e:
print(f" {site}: error ({e}) -> manual"); manual.append((site, drv.current_url)); continue
if status=="ok":
out=place(info, a.dest, a.mpn, a.kind)
print(f" {site}: downloaded -> {out}")
print("\nDONE. Cross-check the model's pad/pin count and dimensions vs the datasheet.")
return
if status=="manual":
print(f" {site}: login/CAPTCHA/manual -> {info}"); manual.append((site, info))
else:
print(f" {site}: not found ({info})")
print("\nNo automatic download. These need you to grab the file by hand (then hand it "
"back and I'll place it as the footprint/symbol):")
for site,url in manual: print(f" - {site}: {url}")
sys.exit(2)
if __name__=="__main__":
main()

View File

@ -1,131 +1,143 @@
#!/usr/bin/env python3
"""Fill the master multi-type template from parts.json (or feed the merge/dedup path).
"""Fill the reference template into ONE master workbook (Components_Master.xlsx).
Per component type present, writes one <Type>.xlsx containing the filled type sheet plus
a Meta sheet (template version, date). Column A ("MPN_make_type") = <MPN>_<make>_<typeid>,
where make = first word of the manufacturer and typeid comes from the taxonomy.
The master contains ONLY the type sheets that actually have parts (the components whose
datasheets were provided) - built from a copy of the template so each kept sheet keeps its
exact headers, styling, widths and freeze - plus a single **Meta** sheet appended at the
END with: Template Version, Total Components, Date, Time, and a per-sheet count breakdown.
Empty template sheets are dropped. Column A = <MPN>_<make>_<typeid>.
The row/style helpers (part_tag, part_to_row, read_type_rows, write_type_workbook,
template_headers, resolve_version) are imported by gitea_reconcile.py so the merge path
produces byte-for-byte the same styling as a fresh generate - no duplicated logic.
Shared helpers (part_tag, part_to_row, template_headers, sheet_rows, read_all_rows,
build_master, resolve_version, MASTER_NAME) are imported by gitea_reconcile.py so the Gitea
master is built and merged the exact same way.
Fresh generate:
python fill_templates.py parts.json --template <template.xlsx> --dest <dir> [--version v1]
"""
import argparse, json, os, re, datetime
import openpyxl
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.styles import Font, PatternFill, Border, Side, Alignment
from openpyxl.utils import get_column_letter
GREEN="B6D7A8"; GRAY="BFBFBF"
MASTER_NAME="Components_Master.xlsx"; META_SHEET="Meta"
def norm(s): return re.sub(r'[^a-z0-9]', '', str(s).lower())
ALIAS={"induactanceuh":"inductanceuh"} # template header typo -> value key
def make_tag(manufacturer):
w = re.split(r'\s+', str(manufacturer).strip())
first = w[0] if w and w[0] else "NA"
w=re.split(r'\s+', str(manufacturer).strip()); first=w[0] if w and w[0] else "NA"
return re.sub(r'[^A-Za-z0-9]', '', first) or "NA"
def part_tag(part):
"""The identity of a part: <MPN>_<make>_<typeid>. Same MPN tag = same part."""
make = part.get("make") or make_tag(part.get("manufacturer",""))
"""Identity of a part: <MPN>_<make>_<typeid>. Same tag = same part."""
make=part.get("make") or make_tag(part.get("manufacturer",""))
return f'{part.get("mpn","")}_{make}_{part.get("typeid","NA")}'
def style_header(cell):
cell.font=Font(name="Calibri", bold=True); cell.fill=PatternFill("solid", fgColor=GREEN)
s=Side(style="thin", color=GRAY); cell.border=Border(left=s,right=s,top=s,bottom=s)
cell.alignment=Alignment(horizontal="center", vertical="center", wrap_text=True)
def template_headers(template_path, ctype):
"""Canonical header list for a type sheet, or None if the template has no such sheet."""
master=openpyxl.load_workbook(template_path)
if ctype not in master.sheetnames: return None
ms=master[ctype]
return [ms.cell(1,c).value for c in range(1, ms.max_column+1)]
"""Canonical data headers for a type sheet, or None if the template has no such sheet."""
wb=openpyxl.load_workbook(template_path)
if ctype not in wb.sheetnames: return None
ws=wb[ctype]; return [ws.cell(1,c).value for c in range(1, ws.max_column+1)]
def template_types(template_path):
return [s for s in openpyxl.load_workbook(template_path).sheetnames if s!=META_SHEET]
def part_to_row(part, headers):
"""Return (tag, {header: value}) for one part, following the column rules."""
vals={norm(k):v for k,v in part.get("values", {}).items()}
tag=part_tag(part)
row={}
vals={norm(k):v for k,v in part.get("values",{}).items()}; tag=part_tag(part); row={}
for c,h in enumerate(headers, start=1):
key=norm(h); key=ALIAS.get(key,key)
key=ALIAS.get(norm(h),norm(h))
if c==1: row[h]=tag
elif key=="class": row[h]=part.get("subclass","")
elif key=="manufacturer":row[h]=part.get("manufacturer", vals.get("manufacturer",""))
else: row[h]=vals.get(key,"")
return tag, row
return tag,row
def read_type_rows(path):
"""Read an existing <Type>.xlsx -> (headers, {tag: {header: value}}); Meta sheet ignored."""
wb=openpyxl.load_workbook(path)
ws=next((wb[n] for n in wb.sheetnames if n!="Meta"), None)
if ws is None: return [], {}
headers=[ws.cell(1,c).value for c in range(1, ws.max_column+1)]
rows={}
def sheet_rows(ws, width):
"""Existing data rows in a sheet -> (headers, {tag: {header: value}}), bounded to `width`."""
headers=[ws.cell(1,c).value for c in range(1,width+1)]; rows={}
for r in range(2, ws.max_row+1):
tag=ws.cell(r,1).value
if tag in (None,""): continue
rows[tag]={headers[c-1]: ws.cell(r,c).value for c in range(1, len(headers)+1)}
rows[tag]={headers[c-1]: ws.cell(r,c).value for c in range(1,width+1)}
return headers, rows
def write_type_workbook(ctype, headers, rows_by_tag, version, dest_dir):
"""Write one styled <Type>.xlsx (type sheet + Meta) from an ordered {tag: rowdict}.
Rows are pulled by header name, so callers can pass the canonical template headers
and any missing field simply comes out blank. Returns the filename written."""
os.makedirs(dest_dir, exist_ok=True)
out=openpyxl.Workbook(); ws=out.active; ws.title=ctype[:31]
for c,h in enumerate(headers, start=1):
ws.cell(1,c,h); style_header(ws.cell(1,c))
ws.row_dimensions[1].height=30
def read_all_rows(master_path, template_path):
"""Read every type sheet of an existing master -> {ctype: {tag: rowdict}} (skips Meta)."""
wb=openpyxl.load_workbook(master_path); types=set(template_types(template_path)); out={}
for name in wb.sheetnames:
if name not in types: continue
th=template_headers(template_path,name); w=len(th) if th else 0
if not w: continue
_,rows=sheet_rows(wb[name], w)
if rows: out[name]=rows
return out
def _thin(): s=Side(style="thin",color=GRAY); return Border(left=s,right=s,top=s,bottom=s)
def _hdr(cell):
cell.font=Font(name="Calibri",bold=True); cell.fill=PatternFill("solid",fgColor=GREEN)
cell.border=_thin(); cell.alignment=Alignment(horizontal="center",vertical="center")
def _write_rows(ws, headers, rows_by_tag):
for r in range(2, ws.max_row+1):
for c in range(1, len(headers)+1): ws.cell(r,c).value=None
r=2
for tag,row in rows_by_tag.items():
for c,h in enumerate(headers, start=1):
ws.cell(r,c, row.get(h,""))
for c,h in enumerate(headers, start=1): ws.cell(r,c, row.get(h,""))
r+=1
for c,h in enumerate(headers, start=1):
w=32 if norm(h)=="description" else (24 if c==1 else max(11,min(18,len(str(h))+2)))
ws.column_dimensions[get_column_letter(c)].width=w
ws.freeze_panes="A2"
meta=out.create_sheet("Meta")
mrows=[("Template Version", version),
("Generated", datetime.date.today().isoformat()),
("Component Type", ctype),
("Parts in file", len(rows_by_tag)),
("Source", "datasheet-extractor skill")]
for i,(k,v) in enumerate(mrows, start=1):
meta.cell(i,1,k).font=Font(bold=True); meta.cell(i,2,v)
meta.column_dimensions["A"].width=20; meta.column_dimensions["B"].width=40
fn=re.sub(r'\s+','_',ctype)+".xlsx"
out.save(os.path.join(dest_dir, fn))
return fn
def _build_meta_sheet(wb, version, counts, when):
if META_SHEET in wb.sheetnames: del wb[META_SHEET]
ws=wb.create_sheet(META_SHEET) # appended at the end
for r,(k,v) in enumerate([("Template Version",version),
("Total Components",sum(counts.values())),
("Date",when.strftime("%Y-%m-%d")),
("Time",when.strftime("%H:%M:%S"))], start=1):
a=ws.cell(r,1,k); a.font=Font(name="Calibri",bold=True)
a.fill=PatternFill("solid",fgColor=GREEN); a.border=_thin()
ws.cell(r,2,v).border=_thin()
r=6
for c,t in ((1,"Sheet"),(2,"Components")): _hdr(ws.cell(r,c,t))
for ctype,n in counts.items():
r+=1; ws.cell(r,1,ctype).border=_thin(); ws.cell(r,2,n).border=_thin()
ws.column_dimensions["A"].width=20; ws.column_dimensions["B"].width=16
return ws
def resolve_version(template_path, version):
if version: return version
vf=os.path.join(os.path.dirname(template_path), "VERSION")
vf=os.path.join(os.path.dirname(template_path),"VERSION")
return "v"+open(vf).read().strip() if os.path.exists(vf) else "v1"
def build_master(template_path, rows_by_type, version, dest_dir):
"""Write the single master workbook from a copy of the template. `rows_by_type` =
{ctype: {tag: rowdict}} is the COMPLETE desired contents. Only sheets present (with at
least one row) are kept; all other template sheets are dropped. A Meta sheet is appended
last. Returns the output path."""
os.makedirs(dest_dir, exist_ok=True)
wb=openpyxl.load_workbook(template_path); when=datetime.datetime.now(); counts={}
for ctype in template_types(template_path):
ws=wb[ctype]; rows=rows_by_type.get(ctype) or {}
if rows:
th=template_headers(template_path,ctype); width=len(th) if th else ws.max_column
_write_rows(ws, [ws.cell(1,c).value for c in range(1,width+1)], rows)
counts[ctype]=len(rows)
_build_meta_sheet(wb, version, counts, when) # append first so a sheet always remains
for ctype in template_types(template_path):
if ctype not in counts: del wb[ctype] # drop empty type sheets
out=os.path.join(dest_dir, MASTER_NAME); wb.save(out); return out
def main():
ap=argparse.ArgumentParser()
ap.add_argument("parts_json"); ap.add_argument("--template", required=True)
ap.add_argument("--dest", required=True); ap.add_argument("--version", default=None)
a=ap.parse_args()
data=json.load(open(a.parts_json, encoding="utf-8"))
version=resolve_version(a.template, a.version)
os.makedirs(a.dest, exist_ok=True)
by={}
version=resolve_version(a.template, a.version); by={}
for p in data.get("parts", []):
by.setdefault(p["type"], []).append(p)
for ctype, parts in by.items():
headers=template_headers(a.template, ctype)
if headers is None:
print(f"! no template sheet for type '{ctype}' - skipping"); continue
rows_by_tag={}
for p in parts:
tag,row=part_to_row(p, headers)
rows_by_tag[tag]=row
fn=write_type_workbook(ctype, headers, rows_by_tag, version, a.dest)
print(f"{ctype}: {len(rows_by_tag)} row(s) -> {fn} (template {version})")
headers=template_headers(a.template, p["type"])
if headers is None: print(f"! no template sheet for type '{p['type']}' - skipping"); continue
tag,row=part_to_row(p, headers); by.setdefault(p["type"],{})[tag]=row
out=build_master(a.template, by, version, a.dest)
total=sum(len(v) for v in by.values())
print(f"master: {out} ({total} part(s) across {len(by)} sheet(s) + Meta, template {version})")
if __name__=="__main__":
main()

View File

@ -1,31 +1,30 @@
#!/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.
Only DFS and Parameters are MPN-indexed. DFS has one MPN_make_typeid/ folder per part;
Parameters holds a SINGLE master workbook (Components_Master.xlsx) whose type sheets each
have one row per part (column A). Skill_Assets holds the skill + templates (not per-part),
so it is never reconciled here - push it with push_to_gitea.sh.
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.
Identity = the part tag <MPN>_<make>_<typeid> (fill_templates.part_tag). Same tag = same
part, so re-processing an MPN already in Gitea is a conflict.
Two passes:
1) CHECK - clone both repos, report which run MPNs already exist. No writes, no push.
CHECK (no writes/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:
APPLY + push:
python scripts/gitea_reconcile.py --parts parts.json --dfs-src <dfs_stage> \
--template assets/template/template.xlsx --decisions decisions.json --push
# or fully unattended: --on-conflict replace (or discard)
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.
replace -> overwrite that MPN's DFS folder AND its row in the master sheet.
discard -> keep the copy already in Gitea; drop the new one.
New MPNs are always added; every other folder/row/sheet is preserved (the master is
merged, never wholesale-overwritten). Refuses to push if a conflict has no decision.
"""
import argparse, json, os, re, shutil, subprocess, sys, tempfile
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import openpyxl
import fill_templates as ft
def load_env(cfg):
@ -48,19 +47,23 @@ 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 master_path(params_clone): return os.path.join(params_clone, ft.MASTER_NAME)
def _width(template, ctype):
h=ft.template_headers(template, ctype); return len(h) if h else 0
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={}
def scan_conflicts(parts, dfs_clone, params_clone, template):
conflicts=[]; tags_by_type={}
mp=master_path(params_clone)
wb=openpyxl.load_workbook(mp) if os.path.exists(mp) else None
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]
in_dfs=os.path.isdir(os.path.join(dfs_clone,tag))
if ctype not in tags_by_type:
s=set(); w=_width(template,ctype)
if wb is not None and ctype in wb.sheetnames and w:
_,rows=ft.sheet_rows(wb[ctype], w); s=set(rows.keys())
tags_by_type[ctype]=s
in_params=tag in tags_by_type[ctype]
if in_dfs or in_params:
conflicts.append({"tag":tag,"type":ctype,"mpn":mpn,
"in_dfs":in_dfs,"in_params":in_params})
@ -80,23 +83,23 @@ def apply_dfs(parts, dfs_src, dfs_clone, decisions):
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:
"""Merge run rows into the master workbook, honouring decisions, and write it back.
Reads the full existing master so every other sheet/row is preserved; empty sheets stay
out (build_master keeps only sheets that have parts)."""
mp=master_path(params_clone)
complete={k:dict(v) for k,v in ft.read_all_rows(mp, template).items()} if os.path.exists(mp) else {}
for p in parts:
headers=ft.template_headers(template, p["type"])
if headers is None: print(f" ! no sheet for '{p['type']}' - skipping"); continue
tag,row=ft.part_to_row(p, headers); sheet=complete.setdefault(p["type"], {})
if tag in sheet:
dec=decisions.get(tag)
if dec=="discard": print(f" PARAMS keep row : {tag}")
elif dec=="replace": rows[tag]=row; print(f" PARAMS replaced : {tag}")
elif dec=="replace": sheet[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)
sheet[tag]=row; print(f" PARAMS added : {tag}")
ft.build_master(template, complete, version, params_clone)
def commit_push(clone_dir, msg, token):
for args in (["config","user.email","datasheet-bot@local"],
@ -132,8 +135,7 @@ def main():
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)
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_")
@ -143,7 +145,7 @@ def main():
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)
conflicts=scan_conflicts(parts, dfs_clone, params_clone, a.template)
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))
@ -156,23 +158,18 @@ def main():
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)
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("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.")
commit_push(params_clone, "datasheet-extractor: reconcile master parameters", token)
print("\nDone — DFS + master Parameters reconciled and pushed.")
if __name__=="__main__":
main()