nearxos/kicad-mcp-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #204 from mixelpixx/fix/issue-199-jlcpcb-prebuilt-download

Browse Source 
fix(jlcpcb): download prebuilt catalog instead of broken JLCSearch offset loop (#199)
This commit is contained in:
mixelpixx
2026-05-30 09:29:48 -04:00
committed by GitHub
parent 00618befb6 a18e729076
commit 034ffb79bd
6 changed files with 1061 additions and 377 deletions
Download Patch File Download Diff File Expand all files Collapse all files

331
download_jlcpcb.py
View File

@@ -1,314 +1,69 @@
#!/usr/bin/env python3
"""
Download JLCPCB parts database from yaqwsx/jlcparts pre-built cache.
CLI to download the JLCPCB parts database into the MCP server's local DB.
This downloads the full JLCPCB catalog (~421MB compressed, ~1.5GB SQLite)
from GitHub Pages in ~5 minutes instead of the broken JLCSearch API approach.
This is a thin wrapper around ``commands.jlcpcb_downloader`` (the same code the
``download_jlcpcb_database`` MCP tool uses). It downloads a prebuilt catalog and
converts it into ``jlcpcb_parts.db``:
The cache.sqlite3 file contains all JLCPCB parts with stock, pricing,
and category data. We then convert it into the format expected by the
KiCad MCP server's JLCPCBPartsManager.
CDFER single-file SQLite (primary, no 7z) ->
yaqwsx split-7z (fallback, needs 7z CLI) ->
official JLCPCB API (optional, if JLCPCB_APP_ID/API_KEY/API_SECRET set)
Usage:
python download_jlcpcb.py [--source cdfer|yaqwsx|official] [--force]
"""
import json
import os
import shutil
import sqlite3
import subprocess
import argparse
import sys
import time
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent / "python"))
from utils.platform_helper import PlatformHelper # noqa: E402
DATA_DIR = PlatformHelper.get_data_dir()
DATA_DIR.mkdir(parents=True, exist_ok=True)
CACHE_DIR = DATA_DIR / "jlcparts_cache"
CACHE_DIR.mkdir(parents=True, exist_ok=True)
BASE_URL = "https://yaqwsx.github.io/jlcparts/data"
MAX_PARTS = 30 # probe up to this many split volumes
TARGET_DB = DATA_DIR / "jlcpcb_parts.db"
def curl_download(url: str, dest: Path) -> bool:
"""Download url to dest with curl. Returns False on 4xx/5xx or network error."""
result = subprocess.run(
["curl", "-L", "-f", "-o", str(dest), "--progress-bar", url], capture_output=False
)
return result.returncode == 0
def download_files() -> bool:
"""Download split archive volumes (z01..zNN) then cache.zip, stopping volumes at 404 or MAX_PARTS."""
print("Downloading jlcparts database (~421MB)...")
for i in range(1, MAX_PARTS + 1):
part = f"cache.z{i:02d}"
dest = CACHE_DIR / part
if dest.exists() and dest.stat().st_size > 1000:
print(f" {part} already exists, skipping")
continue
print(f" Downloading {part}...")
if not curl_download(f"{BASE_URL}/{part}", dest):
# -f causes non-zero exit on 4xx/5xx; treat as end of volumes
if dest.exists():
dest.unlink()
print(f" {part} not found — {i - 1} volumes total")
break
dest = CACHE_DIR / "cache.zip"
if dest.exists() and dest.stat().st_size > 1000:
print(" cache.zip already exists, skipping")
else:
print(" Downloading cache.zip...")
if not curl_download(f"{BASE_URL}/cache.zip", dest):
print(" ERROR downloading cache.zip")
return False
return True
def extract_database() -> bool:
"""Extract the split 7z archive to get cache.sqlite3."""
cache_sqlite = CACHE_DIR / "cache.sqlite3"
if cache_sqlite.exists() and cache_sqlite.stat().st_size > 100_000_000:
print(f"cache.sqlite3 already extracted ({cache_sqlite.stat().st_size // (1024*1024)}MB)")
return True
print("Extracting archive (requires 7z or p7zip)...")
# Try 7z first, then 7zz (homebrew)
for cmd in ["7z", "7zz", "7za"]:
try:
result = subprocess.run(
[cmd, "x", "-y", "-o" + str(CACHE_DIR), str(CACHE_DIR / "cache.zip")],
capture_output=True,
text=True,
)
if result.returncode == 0:
print(f"Extracted with {cmd}")
return True
else:
print(f" {cmd} failed: {result.stderr[:200]}")
except FileNotFoundError:
continue
print("\nERROR: 7z not found. Install with: brew install p7zip")
return False
def convert_to_mcp_format() -> bool:
"""Convert jlcparts cache.sqlite3 to the MCP server's expected format."""
source = CACHE_DIR / "cache.sqlite3"
if not source.exists():
print("ERROR: cache.sqlite3 not found")
return False
print(f"Reading source database...")
src = sqlite3.connect(str(source))
src.row_factory = sqlite3.Row
# Check schema
tables = [
r[0] for r in src.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
]
print(f" Source tables: {tables}")
# Find the main components table
comp_table = None
for t in tables:
count = src.execute(f"SELECT COUNT(*) FROM [{t}]").fetchone()[0]
print(f" {t}: {count:,} rows")
if count > 10000 and comp_table is None:
comp_table = t
if not comp_table:
# Try 'components' specifically
comp_table = "components" if "components" in tables else tables[0]
# Get column names
cols = [r[1] for r in src.execute(f"PRAGMA table_info([{comp_table}])").fetchall()]
print(f" Using table '{comp_table}' with columns: {cols[:10]}...")
# Remove old target DB
if TARGET_DB.exists():
TARGET_DB.unlink()
# Create target DB in MCP format
dst = sqlite3.connect(str(TARGET_DB))
dst.execute(
"""
CREATE TABLE components (
lcsc TEXT PRIMARY KEY,
category TEXT,
subcategory TEXT,
mfr_part TEXT,
package TEXT,
solder_joints INTEGER,
manufacturer TEXT,
library_type TEXT,
description TEXT,
datasheet TEXT,
stock INTEGER,
price_json TEXT,
last_updated INTEGER
)
"""
)
dst.execute("CREATE INDEX idx_category ON components(category, subcategory)")
dst.execute("CREATE INDEX idx_package ON components(package)")
dst.execute("CREATE INDEX idx_manufacturer ON components(manufacturer)")
dst.execute("CREATE INDEX idx_library_type ON components(library_type)")
dst.execute("CREATE INDEX idx_mfr_part ON components(mfr_part)")
# Map source columns to our schema
# jlcparts schema varies but commonly has:
# lcsc, mfr, description, joint, manufacturer, basic, preferred, stock, price, url, etc.
print(f"\nConverting parts to MCP format...")
now = int(time.time())
batch = []
count = 0
for row in src.execute(f"SELECT * FROM [{comp_table}]"):
row_dict = dict(row)
# Adapt column names (jlcparts uses various schemas)
lcsc = row_dict.get("lcsc") or row_dict.get("LCSC_Part") or row_dict.get("lcsc_id")
if lcsc is None:
continue
if isinstance(lcsc, int):
lcsc = f"C{lcsc}"
elif not str(lcsc).startswith("C"):
lcsc = f"C{lcsc}"
mfr_part = row_dict.get("mfr") or row_dict.get("MFR_Part") or row_dict.get("mfr_part") or ""
package = row_dict.get("package") or row_dict.get("Package") or ""
manufacturer = row_dict.get("manufacturer") or row_dict.get("Manufacturer") or ""
description = row_dict.get("description") or row_dict.get("Description") or ""
stock = row_dict.get("stock") or row_dict.get("Stock") or 0
category = row_dict.get("category") or row_dict.get("First Category") or ""
subcategory = row_dict.get("subcategory") or row_dict.get("Second Category") or ""
datasheet = row_dict.get("datasheet") or row_dict.get("url") or ""
# Library type
is_basic = row_dict.get("basic") or row_dict.get("is_basic") or row_dict.get("Basic")
is_preferred = (
row_dict.get("preferred") or row_dict.get("is_preferred") or row_dict.get("Preferred")
)
if is_basic:
lib_type = "Basic"
elif is_preferred:
lib_type = "Preferred"
else:
lib_type = "Extended"
# Price
price = row_dict.get("price") or row_dict.get("Price") or 0
price_json = json.dumps([{"qty": 1, "price": price}] if price else [])
batch.append(
(
str(lcsc),
category,
subcategory,
mfr_part,
package,
0,
manufacturer,
lib_type,
description,
datasheet,
int(stock) if stock else 0,
price_json,
now,
)
)
if len(batch) >= 10000:
dst.executemany(
"""
INSERT OR REPLACE INTO components
(lcsc, category, subcategory, mfr_part, package,
solder_joints, manufacturer, library_type, description,
datasheet, stock, price_json, last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
batch,
)
count += len(batch)
batch = []
if count % 100000 == 0:
print(f" Converted {count:,} parts...")
if batch:
dst.executemany(
"""
INSERT OR REPLACE INTO components
(lcsc, category, subcategory, mfr_part, package,
solder_joints, manufacturer, library_type, description,
datasheet, stock, price_json, last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
batch,
)
count += len(batch)
# Build FTS index
print(f" Building full-text search index...")
dst.execute(
"""
CREATE VIRTUAL TABLE IF NOT EXISTS components_fts USING fts5(
lcsc, description, mfr_part, manufacturer,
content=components
)
"""
)
dst.execute("INSERT INTO components_fts(components_fts) VALUES('rebuild')")
dst.commit()
# Stats
total = dst.execute("SELECT COUNT(*) FROM components").fetchone()[0]
basic = dst.execute("SELECT COUNT(*) FROM components WHERE library_type='Basic'").fetchone()[0]
extended = dst.execute(
"SELECT COUNT(*) FROM components WHERE library_type='Extended'"
).fetchone()[0]
dst.close()
src.close()
print("Cleaning up source database...")
shutil.rmtree(CACHE_DIR)
print(f" Removed {CACHE_DIR}")
db_size = TARGET_DB.stat().st_size / (1024 * 1024)
print(f"\nDatabase ready: {TARGET_DB}")
print(f" Total parts: {total:,}")
print(f" Basic parts: {basic:,}")
print(f" Extended parts: {extended:,}")
print(f" DB size: {db_size:.1f} MB")
return True
from commands import jlcpcb_downloader # noqa: E402
def main() -> None:
parser = argparse.ArgumentParser(description="Download the JLCPCB parts database")
parser.add_argument(
"--source",
choices=["cdfer", "yaqwsx", "official"],
default=None,
help="Force a single source (default: try cdfer, then yaqwsx, then official)",
)
parser.add_argument(
"--force", action="store_true", help="Re-download even if a database already exists"
)
args = parser.parse_args()
print("=" * 60)
print("JLCPCB Parts Database Downloader (jlcparts method)")
print("JLCPCB Parts Database Downloader")
print("=" * 60)
start = time.time()
if not download_files():
sys.exit(1)
result = jlcpcb_downloader.download_database(
force=args.force,
prefer_source=args.source,
progress=lambda msg: print(f" {msg}"),
)
if not extract_database():
sys.exit(1)
if not convert_to_mcp_format():
if not result.get("success"):
print(f"\nERROR: {result.get('message', 'download failed')}")
for err in result.get("errors", []):
print(f" - {err}")
sys.exit(1)
elapsed = time.time() - start
print(f"\nTotal time: {elapsed/60:.1f} minutes")
print(f"\nDatabase ready: {result['db_path']}")
print(f" Source: {result['source']}")
if result.get("catalog_last_modified"):
print(f" Catalog dated: {result['catalog_last_modified']}")
print(f" Total parts: {result['total_parts']:,}")
print(f" Basic parts: {result['basic_parts']:,}")
print(f" Extended parts: {result['extended_parts']:,}")
print(f" DB size: {result['db_size_mb']} MB")
print(f"\nTotal time: {elapsed / 60:.1f} minutes")
print("Done! Restart the MCP server (/mcp) to use the new database.")

645
python/commands/jlcpcb_downloader.py Normal file
View File

@@ -0,0 +1,645 @@
"""
JLCPCB parts catalog downloader (issue #199).
The old approach paged the community JLCSearch API with an ``offset`` parameter,
but that endpoint is a search front-end that ignores ``offset`` and returns the
same first 100 parts on every page, so a full catalog download was impossible.
This module replaces that with a layered download of a *prebuilt* catalog and
converts it into the schema expected by ``JLCPCBPartsManager``
(``get_data_dir()/jlcpcb_parts.db``):
1. CDFER (primary) -- https://github.com/CDFER/jlcpcb-parts-database
A single, raw, uncompressed SQLite file on GitHub Pages. No 7z/zip
needed, so this is the reliable cross-platform (incl. Windows) path.
2. yaqwsx/jlcparts (fallback) -- https://github.com/yaqwsx/jlcparts
The canonical upstream, published as a split 7z archive. Used only if
CDFER fails AND a 7z CLI is available.
3. Official JLCPCB API (optional) -- requires JLCPCB_APP_ID / JLCPCB_API_KEY /
JLCPCB_API_SECRET. Uses cursor pagination (works correctly).
Data is MIT-licensed via CDFER and yaqwsx/jlcparts; the underlying catalog facts
originate from JLCPCB / LCSC and are subject to their terms.
"""
import json
import logging
import os
import shutil
import sqlite3
import time
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Tuple
from utils.platform_helper import PlatformHelper
logger = logging.getLogger("kicad_interface")
CDFER_SQLITE_URL = "https://cdfer.github.io/jlcpcb-parts-database/jlcpcb-components.sqlite3"
YAQWSX_BASE_URL = "https://yaqwsx.github.io/jlcparts/data"
YAQWSX_MAX_VOLUMES = 30
ProgressFn = Optional[Callable[[str], None]]
def _progress(progress: ProgressFn, message: str) -> None:
if progress:
try:
progress(message)
except Exception: # pragma: no cover - progress must never break a download
pass
logger.info(message)
# ---------------------------------------------------------------------------
# Conversion (shared by the CDFER and yaqwsx paths)
# ---------------------------------------------------------------------------
def normalize_price_json(raw: Any) -> str:
"""Normalize a source price value into the manager's ``[{"qty","price"}]`` shape.
Handles both CDFER/jlcparts JSON arrays (``[{"qFrom","qTo","price"}, ...]``)
and a bare scalar price. Returns a JSON string (``"[]"`` when no price).
"""
if raw is None or raw == "":
return json.dumps([])
# JSON array string (CDFER / jlcparts store price as TEXT JSON)
if isinstance(raw, str) and raw.strip().startswith("["):
try:
arr = json.loads(raw)
except (ValueError, TypeError):
return json.dumps([])
out: List[Dict[str, Any]] = []
for item in arr if isinstance(arr, list) else []:
if not isinstance(item, dict):
continue
price = item.get("price")
if price is None:
continue
qty = item.get("qFrom", item.get("qty", 1)) or 1
out.append({"qty": qty, "price": price})
return json.dumps(out)
# Scalar numeric price
try:
price_val = float(raw)
except (TypeError, ValueError):
return json.dumps([])
return json.dumps([{"qty": 1, "price": price_val}]) if price_val else json.dumps([])
def _library_type(is_basic: Any, is_preferred: Any) -> str:
if is_basic:
return "Basic"
if is_preferred:
return "Preferred"
return "Extended"
def _relation_names(src: sqlite3.Connection) -> List[str]:
# Include temp objects so the v_components view built by _ensure_components_view
# (a TEMP view) is visible to _pick_source_relation.
return [
r[0]
for r in src.execute(
"SELECT name FROM sqlite_master WHERE type IN ('table','view') "
"UNION SELECT name FROM sqlite_temp_master WHERE type IN ('table','view')"
).fetchall()
]
def _ensure_components_view(src: sqlite3.Connection) -> None:
"""Create a denormalized ``v_components`` view for yaqwsx-style sources.
CDFER ships a ``v_components`` view, but yaqwsx's raw ``cache.sqlite3`` (the
FULL catalog) stores category/manufacturer as IDs in sibling ``categories``
and ``manufacturers`` tables. Without this join the full-catalog convert
would leave category/subcategory/manufacturer blank. We build the same view
shape CDFER exposes so the rest of the converter is source-agnostic.
"""
names = _relation_names(src)
if "v_components" in names:
return
if not ({"components", "categories", "manufacturers"} <= set(names)):
return
comp_cols = {r[1] for r in src.execute("PRAGMA table_info([components])").fetchall()}
if not ({"category_id", "manufacturer_id"} <= comp_cols):
return
try:
src.execute("""
CREATE TEMP VIEW v_components AS
SELECT c.*, cat.category AS category, cat.subcategory AS subcategory,
m.name AS manufacturer
FROM components c
LEFT JOIN categories cat ON cat.id = c.category_id
LEFT JOIN manufacturers m ON m.id = c.manufacturer_id
""")
logger.info("Built v_components join view for yaqwsx-style source")
except sqlite3.Error as exc: # pragma: no cover - defensive
logger.debug(f"could not build v_components view: {exc}")
def _pick_source_relation(src: sqlite3.Connection) -> str:
"""Choose which table/view to read from the source database.
Prefers a denormalized ``v_components`` view (CDFER ships one; for yaqwsx we
build an equivalent in ``_ensure_components_view``). Falls back to the
largest table.
"""
names = _relation_names(src)
if "v_components" in names:
return "v_components"
if "components" in names:
return "components"
# Fallback: largest table
best, best_count = None, -1
for name in names:
try:
count = src.execute(f"SELECT COUNT(*) FROM [{name}]").fetchone()[0]
except sqlite3.Error:
continue
if count > best_count:
best, best_count = name, count
if not best:
raise RuntimeError("No usable table/view found in source database")
return best
def convert_source_sqlite(
source_path: Path, target_path: Path, progress: ProgressFn = None
) -> Dict[str, int]:
"""Convert a prebuilt source SQLite (CDFER or yaqwsx) into the MCP schema.
Writes a fresh ``target_path`` (deleting any existing file) with the
``components`` table + FTS index that ``JLCPCBPartsManager`` expects.
Returns ``{"total","basic","extended"}`` counts.
"""
source_path = Path(source_path)
target_path = Path(target_path)
if not source_path.exists():
raise FileNotFoundError(f"source database not found: {source_path}")
src = sqlite3.connect(str(source_path))
src.row_factory = sqlite3.Row
try:
_ensure_components_view(src)
relation = _pick_source_relation(src)
_progress(progress, f"Converting from source relation '{relation}'...")
if target_path.exists():
target_path.unlink()
dst = sqlite3.connect(str(target_path))
try:
dst.execute("""
CREATE TABLE components (
lcsc TEXT PRIMARY KEY,
category TEXT,
subcategory TEXT,
mfr_part TEXT,
package TEXT,
solder_joints INTEGER,
manufacturer TEXT,
library_type TEXT,
description TEXT,
datasheet TEXT,
stock INTEGER,
price_json TEXT,
last_updated INTEGER
)
""")
dst.execute("CREATE INDEX idx_category ON components(category, subcategory)")
dst.execute("CREATE INDEX idx_package ON components(package)")
dst.execute("CREATE INDEX idx_manufacturer ON components(manufacturer)")
dst.execute("CREATE INDEX idx_library_type ON components(library_type)")
dst.execute("CREATE INDEX idx_mfr_part ON components(mfr_part)")
now = int(time.time())
batch: List[Tuple] = []
count = 0
insert_sql = """
INSERT OR REPLACE INTO components
(lcsc, category, subcategory, mfr_part, package, solder_joints,
manufacturer, library_type, description, datasheet, stock,
price_json, last_updated)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
"""
for row in src.execute(f"SELECT * FROM [{relation}]"):
rd = dict(row)
lcsc = rd.get("lcsc") or rd.get("LCSC_Part") or rd.get("lcsc_id")
if lcsc is None:
continue
if isinstance(lcsc, int):
lcsc = f"C{lcsc}"
elif not str(lcsc).startswith("C"):
lcsc = f"C{lcsc}"
mfr_part = rd.get("mfr") or rd.get("MFR_Part") or rd.get("mfr_part") or ""
package = rd.get("package") or rd.get("Package") or ""
manufacturer = rd.get("manufacturer") or rd.get("Manufacturer") or ""
description = rd.get("description") or rd.get("Description") or ""
stock = rd.get("stock") or rd.get("Stock") or 0
category = rd.get("category") or rd.get("First Category") or ""
subcategory = rd.get("subcategory") or rd.get("Second Category") or ""
datasheet = rd.get("datasheet") or rd.get("url") or ""
joints = rd.get("joints") or rd.get("solder_joints") or 0
lib_type = _library_type(
rd.get("basic") or rd.get("is_basic") or rd.get("Basic"),
rd.get("preferred") or rd.get("is_preferred") or rd.get("Preferred"),
)
price_json = normalize_price_json(rd.get("price") or rd.get("Price"))
batch.append(
(
str(lcsc),
category,
subcategory,
mfr_part,
package,
int(joints) if joints else 0,
manufacturer,
lib_type,
description,
datasheet,
int(stock) if stock else 0,
price_json,
now,
)
)
if len(batch) >= 10000:
dst.executemany(insert_sql, batch)
count += len(batch)
batch = []
if count % 100000 == 0:
_progress(progress, f"Converted {count:,} parts...")
if batch:
dst.executemany(insert_sql, batch)
count += len(batch)
_progress(progress, "Building full-text search index...")
dst.execute("""
CREATE VIRTUAL TABLE IF NOT EXISTS components_fts USING fts5(
lcsc, description, mfr_part, manufacturer,
content=components
)
""")
dst.execute("INSERT INTO components_fts(components_fts) VALUES('rebuild')")
dst.commit()
total = dst.execute("SELECT COUNT(*) FROM components").fetchone()[0]
basic = dst.execute(
"SELECT COUNT(*) FROM components WHERE library_type='Basic'"
).fetchone()[0]
extended = dst.execute(
"SELECT COUNT(*) FROM components WHERE library_type='Extended'"
).fetchone()[0]
finally:
dst.close()
finally:
src.close()
return {"total": total, "basic": basic, "extended": extended}
# ---------------------------------------------------------------------------
# Source downloads
# ---------------------------------------------------------------------------
def download_cdfer(
cache_dir: Path, progress: ProgressFn = None, max_retries: int = 5
) -> Tuple[Path, Optional[str]]:
"""Stream-download CDFER's single uncompressed SQLite, with resume + retry.
Stalls/drops were a recurring complaint, so this uses a read timeout and, on
a network error, retries while *resuming* from the partial file via an HTTP
Range request (GitHub Pages supports ranged GETs) instead of restarting.
Returns (path, last_modified).
"""
import requests
cache_dir.mkdir(parents=True, exist_ok=True)
dest = cache_dir / "cdfer.sqlite3"
# HEAD once for the freshness date AND the total size, so we can detect a
# cache file that is already complete (e.g. a prior run downloaded it but
# died before conversion/cleanup) and skip re-downloading. Without this,
# resuming a complete file sends Range: bytes=<size>- which the server
# answers with HTTP 416, and the retry loop would spin on it until failure.
last_modified: Optional[str] = None
total_size: Optional[int] = None
try:
head = requests.head(CDFER_SQLITE_URL, allow_redirects=True, timeout=30)
if head.ok:
last_modified = head.headers.get("Last-Modified")
cl = head.headers.get("Content-Length")
total_size = int(cl) if cl and cl.isdigit() else None
except requests.RequestException as exc:
logger.debug(f"HEAD {CDFER_SQLITE_URL} failed: {exc}")
if total_size and dest.exists() and dest.stat().st_size == total_size:
_progress(progress, "CDFER SQLite already fully downloaded; skipping download.")
return dest, last_modified
_progress(
progress,
"Downloading CDFER prebuilt SQLite (~1.5 GB)"
+ (f" [catalog dated {last_modified}]" if last_modified else "")
+ "...",
)
attempt = 0
while True:
attempt += 1
resume_from = dest.stat().st_size if dest.exists() else 0
headers = {"Range": f"bytes={resume_from}-"} if resume_from else {}
try:
# (connect timeout, read timeout): a stalled socket raises after 120s
# of no data, so we can resume rather than hang forever.
with requests.get(
CDFER_SQLITE_URL, stream=True, timeout=(30, 120), headers=headers
) as resp:
# HTTP 416: our partial is at/past the resource end. If it matches
# the known total it's genuinely complete — done. Otherwise the
# cached partial is stale/corrupt, so drop it and restart fresh.
if resume_from and resp.status_code == 416:
if total_size and dest.exists() and dest.stat().st_size == total_size:
_progress(progress, "Existing file already complete; download done.")
break
_progress(progress, "Cached partial invalid (range rejected); restarting.")
dest.unlink(missing_ok=True)
continue
# If we asked to resume but the server ignored Range (200, not
# 206), start the file over to avoid a corrupt append.
if resume_from and resp.status_code == 200:
resume_from = 0
resp.raise_for_status()
last_modified = last_modified or resp.headers.get("Last-Modified")
mode = "ab" if resume_from else "wb"
written = resume_from
next_mark = ((written // (50 * 1024 * 1024)) + 1) * 50 * 1024 * 1024
with open(dest, mode) as fh:
for chunk in resp.iter_content(chunk_size=1024 * 1024):
if not chunk:
continue
fh.write(chunk)
written += len(chunk)
if written >= next_mark:
_progress(progress, f"Downloaded {written // (1024 * 1024)} MB...")
next_mark += 50 * 1024 * 1024
break # clean end of stream
except requests.RequestException as exc:
if attempt > max_retries:
raise RuntimeError(
f"CDFER download failed after {max_retries} retries: {exc}"
) from exc
have_mb = (dest.stat().st_size // (1024 * 1024)) if dest.exists() else 0
_progress(
progress,
f"Download interrupted ({exc}); resuming from {have_mb} MB "
f"(attempt {attempt}/{max_retries})...",
)
time.sleep(min(2 * attempt, 10))
if dest.stat().st_size < 1_000_000:
raise RuntimeError("CDFER download too small to be valid")
return dest, last_modified
def _find_7z() -> Optional[str]:
for cmd in ("7z", "7zz", "7za"):
if shutil.which(cmd):
return cmd
return None
def download_yaqwsx(cache_dir: Path, progress: ProgressFn = None) -> Path:
"""Download + extract the yaqwsx split 7z archive. Requires a 7z CLI."""
import subprocess
seven_zip = _find_7z()
if not seven_zip:
raise RuntimeError("yaqwsx fallback requires a 7z CLI (7z/7zz/7za) which was not found")
if not shutil.which("curl"):
raise RuntimeError("yaqwsx fallback requires curl which was not found")
cache_dir.mkdir(parents=True, exist_ok=True)
_progress(progress, "Downloading yaqwsx split archive (~421 MB)...")
def _curl(url: str, dst: Path) -> bool:
# -C - resumes a partial file; --retry handles transient stalls/drops.
return (
subprocess.run(
[
"curl",
"-L",
"-f",
"-C",
"-",
"--retry",
"5",
"--retry-delay",
"2",
"--retry-connrefused",
"-o",
str(dst),
"--progress-bar",
url,
]
).returncode
== 0
)
for i in range(1, YAQWSX_MAX_VOLUMES + 1):
part = f"cache.z{i:02d}"
dst = cache_dir / part
if dst.exists() and dst.stat().st_size > 1000:
continue
if not _curl(f"{YAQWSX_BASE_URL}/{part}", dst):
if dst.exists():
dst.unlink()
break
zip_dst = cache_dir / "cache.zip"
if not (zip_dst.exists() and zip_dst.stat().st_size > 1000):
if not _curl(f"{YAQWSX_BASE_URL}/cache.zip", zip_dst):
raise RuntimeError("failed to download yaqwsx cache.zip")
_progress(progress, f"Extracting archive with {seven_zip}...")
result = subprocess.run(
[seven_zip, "x", "-y", "-o" + str(cache_dir), str(zip_dst)],
capture_output=True,
text=True,
)
if result.returncode != 0:
raise RuntimeError(f"7z extraction failed: {result.stderr[:300]}")
sqlite_path = cache_dir / "cache.sqlite3"
if not sqlite_path.exists():
raise RuntimeError("yaqwsx archive did not yield cache.sqlite3")
return sqlite_path
def _download_official(target: Path, force: bool, progress: ProgressFn) -> Dict[str, Any]:
"""Optional fallback: official JLCPCB API (cursor pagination). Requires creds."""
from commands.jlcpcb import JLCPCBClient
from commands.jlcpcb_parts import JLCPCBPartsManager
client = JLCPCBClient()
if not (client.app_id and client.access_key and client.secret_key):
raise RuntimeError("official API credentials not set")
_progress(progress, "Downloading via official JLCPCB API...")
parts = client.download_full_database(
callback=lambda page, total, msg: _progress(progress, msg)
)
if force and target.exists():
target.unlink()
mgr = JLCPCBPartsManager(db_path=str(target))
try:
mgr.import_parts(parts, progress_callback=lambda c, t, m: _progress(progress, m))
stats = mgr.get_database_stats()
finally:
mgr.close()
return _result(
"official-api",
target,
stats["total_parts"],
stats["basic_parts"],
stats["extended_parts"],
None,
)
# Warn if the catalog is older than this (CDFER's pipeline has stalled for
# weeks at a time, so a "fresh download" can still be stale data).
STALE_AFTER_DAYS = 14
def _catalog_age_days(last_modified: Optional[str]) -> Optional[int]:
"""Parse an HTTP Last-Modified date and return its age in whole days."""
if not last_modified:
return None
try:
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone
dt = parsedate_to_datetime(last_modified)
if dt is None:
return None
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return max(0, (datetime.now(timezone.utc) - dt).days)
except (TypeError, ValueError):
return None
def _result(
source: str,
target: Path,
total: int,
basic: int,
extended: int,
last_modified: Optional[str],
) -> Dict[str, Any]:
db_size_mb = round(target.stat().st_size / (1024 * 1024), 2) if target.exists() else 0
result: Dict[str, Any] = {
"success": True,
"source": source,
"total_parts": total,
"basic_parts": basic,
"extended_parts": extended,
"db_size_mb": db_size_mb,
"db_path": str(target),
"catalog_last_modified": last_modified,
}
age_days = _catalog_age_days(last_modified)
if age_days is not None:
result["catalog_age_days"] = age_days
if age_days >= STALE_AFTER_DAYS:
warning = (
f"Catalog from '{source}' is ~{age_days} days old (dated {last_modified}). "
"Its upstream pipeline may have stalled. For fresher data, re-run with "
"source='yaqwsx' (needs a 7z CLI) or source='official' (needs JLCPCB API creds)."
)
result["stale"] = True
result["warning"] = warning
logger.warning(warning)
return result
# ---------------------------------------------------------------------------
# Orchestration
# ---------------------------------------------------------------------------
def download_database(
force: bool = False,
prefer_source: Optional[str] = None,
progress: ProgressFn = None,
) -> Dict[str, Any]:
"""Download the JLCPCB catalog into ``jlcpcb_parts.db`` using a layered strategy.
Order: CDFER (primary) -> yaqwsx (if 7z available) -> official API (if creds).
``prefer_source`` (``"cdfer"``/``"yaqwsx"``/``"official"``) forces a single source.
Returns a result dict (see ``_result``) or ``{"success": False, "message": ...}``.
NOTE: callers that hold an open ``JLCPCBPartsManager`` on the target file must
close it before calling (the prebuilt paths recreate the file) and reopen after.
"""
data_dir = PlatformHelper.get_data_dir()
data_dir.mkdir(parents=True, exist_ok=True)
target = data_dir / "jlcpcb_parts.db"
cache_dir = data_dir / "jlcparts_cache"
order = [prefer_source] if prefer_source else ["cdfer", "yaqwsx", "official"]
errors: List[str] = []
for src in order:
try:
if src == "cdfer":
src_path, last_mod = download_cdfer(cache_dir, progress)
stats = convert_source_sqlite(src_path, target, progress)
_safe_rmtree(cache_dir)
return _result(
"cdfer", target, stats["total"], stats["basic"], stats["extended"], last_mod
)
if src == "yaqwsx":
src_path = download_yaqwsx(cache_dir, progress)
stats = convert_source_sqlite(src_path, target, progress)
_safe_rmtree(cache_dir)
return _result(
"yaqwsx", target, stats["total"], stats["basic"], stats["extended"], None
)
if src == "official":
return _download_official(target, force, progress)
errors.append(f"{src}: unknown source")
except Exception as exc:
logger.warning(f"JLCPCB source '{src}' failed: {exc}")
errors.append(f"{src}: {exc}")
continue
return {
"success": False,
"message": "All JLCPCB download sources failed. "
"Install a 7z CLI for the yaqwsx fallback, or set JLCPCB_APP_ID/"
"JLCPCB_API_KEY/JLCPCB_API_SECRET for the official API.",
"errors": errors,
}
def _safe_rmtree(path: Path) -> None:
try:
if path.exists():
shutil.rmtree(path)
except OSError as exc: # pragma: no cover - cleanup is best-effort
logger.debug(f"cache cleanup failed for {path}: {exc}")

59
python/commands/jlcsearch.py
View File

@@ -137,59 +137,12 @@ class JLCSearchClient:
logger.error(f"Failed to get part C{lcsc_number}: {e}")
return None
def download_all_components(
self, callback: Optional[Callable[[int, str], None]] = None, batch_size: int = 100
) -> List[Dict]:
"""
Download all components from jlcsearch database
Note: tscircuit API has a hard-coded 100 result limit per request.
Full catalog download requires ~25,000 paginated requests (~40-60 minutes).
Args:
callback: Optional progress callback function(parts_count, status_msg)
batch_size: Number of parts per batch (max 100 due to API limit)
Returns:
List of all parts
"""
all_parts = []
offset = 0
logger.info("Starting full jlcsearch parts database download...")
while True:
try:
batch = self.search_components("components", limit=batch_size, offset=offset)
# Stop if no results returned (end of catalog)
if not batch or len(batch) == 0:
break
all_parts.extend(batch)
offset += len(batch)
if callback:
callback(len(all_parts), f"Downloaded {len(all_parts)} parts...")
else:
logger.info(f"Downloaded {len(all_parts)} parts so far...")
# Continue pagination - API returns exactly 100 results per page until exhausted
# Only stop when we get 0 results (handled above)
# Rate limiting - be nice to the API
time.sleep(0.1)
except Exception as e:
logger.error(f"Error downloading parts at offset {offset}: {e}")
if len(all_parts) > 0:
logger.warning(f"Partial download available: {len(all_parts)} parts")
return all_parts
else:
raise
logger.info(f"Download complete: {len(all_parts)} parts retrieved")
return all_parts
# NOTE: bulk catalog download was removed (issue #199). The JLCSearch
# endpoint is a *search front-end* that ignores the ``offset`` parameter,
# so offset-paged "download everything" loops returned the same first 100
# parts forever. Full-catalog download now uses a prebuilt source via
# ``commands.jlcpcb_downloader.download_database()``. This client remains
# for interactive/parametric lookups only (search_components, etc.).
def test_jlcsearch_connection() -> bool:

60
python/kicad_interface.py
View File

@@ -6314,12 +6314,18 @@ print("ok")
# JLCPCB API handlers
def _handle_download_jlcpcb_database(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Download JLCPCB parts database from JLCSearch API"""
"""Download the JLCPCB parts catalog from a prebuilt source (issue #199).
Layered strategy (see commands.jlcpcb_downloader): CDFER single-file
SQLite (primary, no 7z) -> yaqwsx split-7z (fallback) -> official JLCPCB
API (optional, if credentials set). Replaces the broken JLCSearch
offset-pagination download.
"""
from commands import jlcpcb_downloader
try:
force = params.get("force", False)
# Check if database exists
import os
prefer_source = params.get("source") # optional: cdfer|yaqwsx|official
stats = self.jlcpcb_parts.get_database_stats()
if stats["total_parts"] > 0 and not force:
@@ -6329,36 +6335,38 @@ print("ok")
"stats": stats,
}
logger.info("Downloading JLCPCB parts database from JLCSearch...")
# The prebuilt paths recreate jlcpcb_parts.db on disk, so the open
# manager connection must be released first (Windows file locking),
# then reopened on the freshly written database.
self.jlcpcb_parts.close()
# Download parts from JLCSearch public API (no auth required)
parts = self.jlcsearch_client.download_all_components(
callback=lambda total, msg: logger.info(f"{msg}")
result = jlcpcb_downloader.download_database(
force=force,
prefer_source=prefer_source,
progress=lambda msg: logger.info(msg),
)
# Import into database
logger.info(f"Importing {len(parts)} parts into database...")
self.jlcpcb_parts.import_jlcsearch_parts(
parts, progress_callback=lambda curr, total, msg: logger.info(msg)
)
# Reopen the manager on the new database regardless of outcome.
self.jlcpcb_parts = JLCPCBPartsManager()
# Get final stats
if not result.get("success"):
return result
# Refresh counts from the reopened manager (authoritative).
stats = self.jlcpcb_parts.get_database_stats()
# Calculate database size
db_size_mb = os.path.getsize(self.jlcpcb_parts.db_path) / (1024 * 1024)
return {
"success": True,
"total_parts": stats["total_parts"],
"basic_parts": stats["basic_parts"],
"extended_parts": stats["extended_parts"],
"db_size_mb": round(db_size_mb, 2),
"db_path": stats["db_path"],
}
result["total_parts"] = stats["total_parts"]
result["basic_parts"] = stats["basic_parts"]
result["extended_parts"] = stats["extended_parts"]
result["db_path"] = stats["db_path"]
return result
except Exception as e:
logger.error(f"Error downloading JLCPCB database: {e}", exc_info=True)
# Best-effort: ensure the manager is usable after a failure.
try:
self.jlcpcb_parts = JLCPCBPartsManager()
except Exception:
pass
return {
"success": False,
"message": f"Failed to download database: {str(e)}",

39
src/tools/jlcpcb-api.ts
View File

@@ -10,21 +10,33 @@ export function registerJLCPCBApiTools(server: McpServer, callKicadScript: Funct
// Download JLCPCB parts database
server.tool(
"download_jlcpcb_database",
`Download the complete JLCPCB parts catalog to local database.
`Download the JLCPCB parts catalog to a local SQLite database for fast offline search.
This is a one-time setup that downloads ~2.5M+ parts from JLCSearch API.
No API credentials required - uses public JLCSearch API.
Sources (no API credentials required by default):
- cdfer (default): in-stock subset (~600k parts, ~1.5 GB download). Single file,
no extra tools — the most reliable path, especially on Windows.
- yaqwsx: the FULL catalog (all parts incl. out-of-stock, ~10 GB extracted).
Use source="yaqwsx" if you specifically want everything. Requires a 7z CLI.
- official: official JLCPCB API, used only if JLCPCB_APP_ID/JLCPCB_API_KEY/
JLCPCB_API_SECRET are set.
The download takes 5-10 minutes and creates a local SQLite database
for fast offline searching.`,
One-time setup; downloads resume automatically if interrupted. Re-run with
force=true to refresh.`,
{
force: z
.boolean()
.optional()
.default(false)
.describe("Force re-download even if database exists"),
source: z
.enum(["cdfer", "yaqwsx", "official"])
.optional()
.describe(
'Force one source. "cdfer" (default) = in-stock subset, no 7z needed. ' +
'"yaqwsx" = FULL ~10GB catalog (needs a 7z CLI). "official" = JLCPCB API (needs creds).',
),
},
async (args: { force?: boolean }) => {
async (args: { force?: boolean; source?: "cdfer" | "yaqwsx" | "official" }) => {
const result = await callKicadScript("download_jlcpcb_database", args);
if (result.success) {
return {
@@ -33,11 +45,20 @@ for fast offline searching.`,
type: "text",
text:
`✓ Successfully downloaded JLCPCB parts database\n\n` +
`Source: ${result.source}\n` +
(result.catalog_last_modified
? `Catalog dated: ${result.catalog_last_modified}` +
(typeof result.catalog_age_days === "number"
? ` (~${result.catalog_age_days} days old)`
: "") +
`\n`
: "") +
`Total parts: ${result.total_parts}\n` +
`Basic parts: ${result.basic_parts}\n` +
`Extended parts: ${result.extended_parts}\n` +
`Database size: ${result.db_size_mb} MB\n` +
`Database path: ${result.db_path}`,
`Database path: ${result.db_path}` +
(result.warning ? `\n\n⚠ ${result.warning}` : ""),
},
],
};
@@ -46,9 +67,7 @@ for fast offline searching.`,
content: [
{
type: "text",
text:
`✗ Failed to download JLCPCB database: ${result.message || "Unknown error"}\n\n` +
`Make sure JLCPCB_API_KEY and JLCPCB_API_SECRET environment variables are set.`,
text: `✗ Failed to download JLCPCB database: ${result.message || "Unknown error"}`,
},
],
};

304
tests/test_jlcpcb_downloader.py Normal file
View File

@@ -0,0 +1,304 @@
"""
Regression tests for the JLCPCB prebuilt-catalog downloader (issue #199).
These tests do NOT hit the network. They build a small SQLite source that
mimics CDFER's schema (incl. the v_components view and bare-integer lcsc) and
verify that convert_source_sqlite() produces a database in the schema that
JLCPCBPartsManager consumes, plus that download_database() falls through
sources gracefully when none are usable.
"""
import json
import sqlite3
import sys
from pathlib import Path
import pytest
# Match the import root used elsewhere in the test suite (python/ on sys.path).
PYTHON_DIR = Path(__file__).resolve().parent.parent / "python"
if str(PYTHON_DIR) not in sys.path:
sys.path.insert(0, str(PYTHON_DIR))
from commands import jlcpcb_downloader # noqa: E402
def _make_cdfer_like_source(path: Path) -> None:
"""Create a tiny SQLite mirroring CDFER: components + categories + manufacturers + view."""
con = sqlite3.connect(str(path))
con.executescript("""
CREATE TABLE manufacturers (id INTEGER PRIMARY KEY, name TEXT);
CREATE TABLE categories (id INTEGER PRIMARY KEY, category TEXT, subcategory TEXT);
CREATE TABLE components (
lcsc INTEGER PRIMARY KEY,
category_id INTEGER,
mfr TEXT,
package TEXT,
joints INTEGER,
manufacturer_id INTEGER,
basic INTEGER,
preferred INTEGER,
description TEXT,
datasheet TEXT,
stock INTEGER,
price TEXT
);
CREATE VIEW v_components AS
SELECT c.lcsc, c.category_id, cat.category, cat.subcategory, c.mfr,
c.package, c.joints, m.name AS manufacturer, c.basic, c.preferred,
c.description, c.datasheet, c.stock, c.price
FROM components c
JOIN categories cat ON cat.id = c.category_id
JOIN manufacturers m ON m.id = c.manufacturer_id;
""")
con.execute("INSERT INTO manufacturers VALUES (1, 'Yageo')")
con.execute("INSERT INTO categories VALUES (1, 'Resistors', 'Chip Resistor')")
# basic part, with CDFER-style price JSON (qFrom/qTo/price)
con.execute(
"INSERT INTO components VALUES (25804, 1, 'RC0603FR-071KL', '0603', 2, 1, 1, 0, "
"'1k 1% 0603 resistor', 'http://ds/25804', 5000, ?)",
(
json.dumps(
[
{"qFrom": 1, "qTo": 99, "price": 0.0042},
{"qFrom": 100, "qTo": None, "price": 0.0021},
]
),
),
)
# extended part, empty price
con.execute(
"INSERT INTO components VALUES (11111, 1, 'EXT-PART', '0402', 2, 1, 0, 0, "
"'extended resistor', '', 12, '[]')"
)
con.commit()
con.close()
def test_convert_source_sqlite_produces_manager_schema(tmp_path):
source = tmp_path / "cdfer.sqlite3"
target = tmp_path / "jlcpcb_parts.db"
_make_cdfer_like_source(source)
stats = jlcpcb_downloader.convert_source_sqlite(source, target)
assert stats["total"] == 2
assert stats["basic"] == 1
assert stats["extended"] == 1
con = sqlite3.connect(str(target))
con.row_factory = sqlite3.Row
rows = {r["lcsc"]: dict(r) for r in con.execute("SELECT * FROM components")}
# bare-int lcsc must become C-prefixed
assert "C25804" in rows and "C11111" in rows
basic = rows["C25804"]
assert basic["library_type"] == "Basic"
assert basic["mfr_part"] == "RC0603FR-071KL" # mfr -> mfr_part
assert basic["manufacturer"] == "Yageo" # via v_components view
assert basic["category"] == "Resistors"
assert basic["subcategory"] == "Chip Resistor"
assert basic["stock"] == 5000
# CDFER price JSON normalized to [{"qty","price"}]
breaks = json.loads(basic["price_json"])
assert breaks[0] == {"qty": 1, "price": 0.0042}
assert breaks[1] == {"qty": 100, "price": 0.0021}
assert rows["C11111"]["library_type"] == "Extended"
assert json.loads(rows["C11111"]["price_json"]) == []
# FTS index exists and is queryable
fts_hit = con.execute(
"SELECT lcsc FROM components_fts WHERE components_fts MATCH '1k*'"
).fetchall()
assert any(r[0] == "C25804" for r in fts_hit)
con.close()
def _make_yaqwsx_like_source(path: Path) -> None:
"""Create a SQLite mirroring yaqwsx's FULL-catalog schema: normalized, NO view.
category/manufacturer are IDs in sibling tables (no v_components view), which
is what the real ~10GB cache.sqlite3 looks like.
"""
con = sqlite3.connect(str(path))
con.executescript("""
CREATE TABLE manufacturers (id INTEGER PRIMARY KEY, name TEXT);
CREATE TABLE categories (id INTEGER PRIMARY KEY, category TEXT, subcategory TEXT);
CREATE TABLE components (
lcsc INTEGER PRIMARY KEY,
category_id INTEGER,
mfr TEXT,
package TEXT,
joints INTEGER,
manufacturer_id INTEGER,
basic INTEGER,
preferred INTEGER,
description TEXT,
datasheet TEXT,
stock INTEGER,
price TEXT
);
""")
con.execute("INSERT INTO manufacturers VALUES (7, 'Texas Instruments')")
con.execute("INSERT INTO categories VALUES (3, 'ICs', 'LDO Regulators')")
con.execute(
"INSERT INTO components VALUES (12345, 3, 'TLV70033', 'SOT-23-5', 5, 7, 0, 0, "
"'3.3V LDO', 'http://ds/12345', 0, ?)",
(json.dumps([{"qFrom": 1, "qTo": None, "price": 0.12}]),),
)
con.commit()
con.close()
def test_convert_yaqwsx_schema_populates_category_and_manufacturer(tmp_path):
"""Full-catalog (yaqwsx) source has no v_components view; the join must be built."""
source = tmp_path / "cache.sqlite3"
target = tmp_path / "jlcpcb_parts.db"
_make_yaqwsx_like_source(source)
stats = jlcpcb_downloader.convert_source_sqlite(source, target)
assert stats["total"] == 1
con = sqlite3.connect(str(target))
con.row_factory = sqlite3.Row
row = dict(con.execute("SELECT * FROM components WHERE lcsc='C12345'").fetchone())
con.close()
# These would be blank without the built v_components join:
assert row["category"] == "ICs"
assert row["subcategory"] == "LDO Regulators"
assert row["manufacturer"] == "Texas Instruments"
assert row["mfr_part"] == "TLV70033"
assert row["library_type"] == "Extended"
assert json.loads(row["price_json"]) == [{"qty": 1, "price": 0.12}]
def test_normalize_price_json_handles_scalar_and_array_and_empty():
assert json.loads(jlcpcb_downloader.normalize_price_json(None)) == []
assert json.loads(jlcpcb_downloader.normalize_price_json("")) == []
assert json.loads(jlcpcb_downloader.normalize_price_json("[]")) == []
assert json.loads(jlcpcb_downloader.normalize_price_json(0.05)) == [{"qty": 1, "price": 0.05}]
arr = json.loads(jlcpcb_downloader.normalize_price_json('[{"qFrom":10,"qTo":50,"price":1.5}]'))
assert arr == [{"qty": 10, "price": 1.5}]
def test_download_database_falls_through_when_no_source_available(tmp_path, monkeypatch):
"""With CDFER download failing, no 7z present, and no API creds -> clean failure dict."""
monkeypatch.setattr(
jlcpcb_downloader.PlatformHelper, "get_data_dir", staticmethod(lambda: tmp_path)
)
def _boom(*a, **k):
raise RuntimeError("network down")
monkeypatch.setattr(jlcpcb_downloader, "download_cdfer", _boom)
monkeypatch.setattr(jlcpcb_downloader, "_find_7z", lambda: None)
# Ensure no official creds leak in from the environment.
for var in ("JLCPCB_APP_ID", "JLCPCB_API_KEY", "JLCPCB_API_SECRET"):
monkeypatch.delenv(var, raising=False)
result = jlcpcb_downloader.download_database()
assert result["success"] is False
assert "errors" in result and len(result["errors"]) >= 1
assert any("cdfer" in e for e in result["errors"])
def test_download_database_prefer_cdfer_converts(tmp_path, monkeypatch):
"""prefer_source='cdfer' downloads then converts via the real conversion path."""
monkeypatch.setattr(
jlcpcb_downloader.PlatformHelper, "get_data_dir", staticmethod(lambda: tmp_path)
)
def _fake_cdfer(cache_dir, progress=None):
cache_dir.mkdir(parents=True, exist_ok=True)
src = cache_dir / "cdfer.sqlite3"
_make_cdfer_like_source(src)
return src, "Thu, 02 Apr 2026 06:22:46 GMT"
monkeypatch.setattr(jlcpcb_downloader, "download_cdfer", _fake_cdfer)
result = jlcpcb_downloader.download_database(prefer_source="cdfer")
assert result["success"] is True
assert result["source"] == "cdfer"
assert result["total_parts"] == 2
assert result["basic_parts"] == 1
assert result["catalog_last_modified"].startswith("Thu, 02 Apr 2026")
assert (tmp_path / "jlcpcb_parts.db").exists()
def _http_date(days_ago: int) -> str:
from datetime import datetime, timedelta, timezone
from email.utils import format_datetime
return format_datetime(datetime.now(timezone.utc) - timedelta(days=days_ago))
def test_result_flags_stale_catalog_and_leaves_fresh_alone(tmp_path, monkeypatch):
"""A catalog older than STALE_AFTER_DAYS must be flagged; a fresh one must not."""
monkeypatch.setattr(
jlcpcb_downloader.PlatformHelper, "get_data_dir", staticmethod(lambda: tmp_path)
)
def _fake_cdfer_dated(http_date):
def _impl(cache_dir, progress=None):
cache_dir.mkdir(parents=True, exist_ok=True)
src = cache_dir / "cdfer.sqlite3"
_make_cdfer_like_source(src)
return src, http_date
return _impl
# Stale: 60 days old
monkeypatch.setattr(jlcpcb_downloader, "download_cdfer", _fake_cdfer_dated(_http_date(60)))
stale = jlcpcb_downloader.download_database(prefer_source="cdfer")
assert stale.get("stale") is True
assert "warning" in stale
assert stale["catalog_age_days"] >= jlcpcb_downloader.STALE_AFTER_DAYS
# Fresh: 1 day old
monkeypatch.setattr(jlcpcb_downloader, "download_cdfer", _fake_cdfer_dated(_http_date(1)))
fresh = jlcpcb_downloader.download_database(prefer_source="cdfer")
assert fresh.get("stale") is not True
assert "warning" not in fresh
assert fresh["catalog_age_days"] <= 1
def test_download_cdfer_skips_when_file_already_complete(tmp_path, monkeypatch):
"""A complete cache file must short-circuit, not re-download (no HTTP 416 loop).
Regression for the force/resume bug: resuming a complete file sends
Range: bytes=<size>- which the server answers with 416; the old code retried
that until failure. With a HEAD size check we return immediately and never
call requests.get.
"""
import requests
cache_dir = tmp_path / "cache"
cache_dir.mkdir()
dest = cache_dir / "cdfer.sqlite3"
payload = b"x" * 4096
dest.write_bytes(payload)
class _Head:
ok = True
headers = {
"Content-Length": str(len(payload)),
"Last-Modified": "Wed, 01 Apr 2026 00:00:00 GMT",
}
monkeypatch.setattr(requests, "head", lambda *a, **k: _Head())
def _no_get(*a, **k):
raise AssertionError("requests.get must not run when the file is already complete")
monkeypatch.setattr(requests, "get", _no_get)
path, last_mod = jlcpcb_downloader.download_cdfer(cache_dir)
assert path == dest
assert path.stat().st_size == len(payload) # untouched
assert last_mod.startswith("Wed, 01 Apr 2026")