diff --git a/.gitignore b/.gitignore index fb3a9e8..27f4e88 100644 --- a/.gitignore +++ b/.gitignore @@ -2,6 +2,7 @@ __pycache__/ *.pyc *.pyo .venv/ +.venv .env *.egg-info/ build/ diff --git a/INSTALL.md b/INSTALL.md index 288c19b..9adb8e8 100644 --- a/INSTALL.md +++ b/INSTALL.md @@ -63,6 +63,25 @@ The LCSC client is offline during normal construction. Populate or refresh its jlcparts database only through the explicit refresh operation documented by `scripts/lcsc_client.py`; importing the client never downloads data. +### Current LCSC evidence + +On 2026-07-31, 7-Zip 23.01 was installed from the Ubuntu package repository. +The explicit refresh reused the existing 13-part archive and extracted a +5.6 GB SQLite database. The refreshed upstream schema uses +`jlc_components`, not the legacy `components`/`manufacturers` tables. + +Validation on that database reported: + +- SQLite `PRAGMA quick_check`: `ok` +- 7,157,071 component rows +- C8734 ID lookup: found +- exact-MPN keyword lookup: three results at a limit of three +- C8734 price parsing: six quantity tiers + +These counts are dated cache evidence, not a stable upstream invariant. The +client supports both the current and legacy table layouts through fixture +tests and fails clearly when it encounters an unknown schema. + Nexar Supply and Design APIs use different OAuth scopes and separate caches. The Design API operates on Altium 365 projects; it does not accept KiCad uploads. Pure-KiCad rendering must use KiCad's own export tools. diff --git a/scripts/lcsc_client.py b/scripts/lcsc_client.py index b6eb21c..a204c03 100644 --- a/scripts/lcsc_client.py +++ b/scripts/lcsc_client.py @@ -31,6 +31,7 @@ import json import os +import re import sqlite3 import subprocess import time @@ -98,18 +99,39 @@ def _ensure_db(db_path: Path = DB_PATH) -> None: def _parse_price_tiers(raw: str | None) -> list[dict[str, Any]]: - """Parse JSON price string -> [{qty, price_usd}].""" + """Parse legacy JSON or current jlcparts range-price text.""" if not raw: return [] try: tiers = json.loads(raw) - return [ - {"qty": t.get("qFrom", 1), "price_usd": round(float(t.get("price", 0)), 6)} - for t in tiers - if isinstance(t, dict) - ] - except (json.JSONDecodeError, TypeError, ValueError): - return [] + except (json.JSONDecodeError, TypeError): + tiers = None + if isinstance(tiers, list): + try: + return [ + { + "qty": int(t.get("qFrom", 1)), + "price_usd": round(float(t.get("price", 0)), 6), + } + for t in tiers + if isinstance(t, dict) + ] + except (TypeError, ValueError): + return [] + + parsed: list[dict[str, Any]] = [] + for entry in raw.split(","): + match = re.fullmatch( + r"\s*(\d+)(?:-\d*)?\s*:\s*(\d+(?:\.\d+)?)\s*", + entry, + ) + if match is None: + return [] + parsed.append({ + "qty": int(match.group(1)), + "price_usd": round(float(match.group(2)), 6), + }) + return parsed def _row_to_record(row: tuple) -> dict[str, Any]: @@ -161,6 +183,34 @@ def _row_to_record(row: tuple) -> dict[str, Any]: WHERE c.lcsc = ? """ +_CURRENT_QUERY_BY_MFR = """ + SELECT lcsc, mfr, manufacturer, package, + CASE WHEN library_type = 'base' THEN 1 ELSE 0 END AS basic, + preferred, stock, price, datasheet + FROM jlc_components + WHERE present = 1 AND mfr LIKE ? + ORDER BY preferred DESC, basic DESC, stock DESC + LIMIT ? +""" + +_CURRENT_QUERY_BY_DESC = """ + SELECT lcsc, mfr, manufacturer, package, + CASE WHEN library_type = 'base' THEN 1 ELSE 0 END AS basic, + preferred, stock, price, datasheet + FROM jlc_components + WHERE present = 1 AND (description LIKE ? OR mfr LIKE ?) + ORDER BY preferred DESC, basic DESC, stock DESC + LIMIT ? +""" + +_CURRENT_QUERY_BY_LCSC = """ + SELECT lcsc, mfr, manufacturer, package, + CASE WHEN library_type = 'base' THEN 1 ELSE 0 END AS basic, + preferred, stock, price, datasheet + FROM jlc_components + WHERE present = 1 AND lcsc = ? +""" + class LcscClient: """LCSC component lookup backed by a local jlcparts SQLite cache. @@ -182,6 +232,7 @@ class LcscClient: def __init__(self, db_path: Path = DB_PATH) -> None: self._db_path = db_path self._conn: sqlite3.Connection | None = None + self._schema: str | None = None @classmethod def from_env( @@ -216,8 +267,77 @@ def _get_conn(self) -> sqlite3.Connection: if self._conn is None: self._conn = sqlite3.connect(str(self._db_path)) self._conn.row_factory = sqlite3.Row + self._conn.execute("PRAGMA query_only = ON") return self._conn + def _get_schema(self) -> str: + if self._schema is not None: + return self._schema + connection = self._get_conn() + tables = { + str(row[0]) + for row in connection.execute( + "SELECT name FROM sqlite_master WHERE type = 'table'" + ) + } + if "jlc_components" in tables: + columns = { + str(row[1]) + for row in connection.execute( + "PRAGMA table_info(jlc_components)" + ) + } + required = { + "lcsc", + "mfr", + "manufacturer", + "package", + "library_type", + "preferred", + "present", + "description", + "stock", + "price", + "datasheet", + } + if required <= columns: + self._schema = "current" + elif {"components", "manufacturers"} <= tables: + component_columns = { + str(row[1]) + for row in connection.execute("PRAGMA table_info(components)") + } + manufacturer_columns = { + str(row[1]) + for row in connection.execute( + "PRAGMA table_info(manufacturers)" + ) + } + required_components = { + "lcsc", + "mfr", + "manufacturer_id", + "package", + "basic", + "preferred", + "description", + "stock", + "price", + "datasheet", + } + if ( + required_components <= component_columns + and {"id", "name"} <= manufacturer_columns + ): + self._schema = "legacy" + if self._schema is None: + raise RuntimeError( + "unsupported jlcparts schema: expected jlc_components or " + "legacy components/manufacturers tables; refresh the cache " + "or update Electronics Stack" + ) + return self._schema + def keyword_search(self, mpn: str, limit: int = 5) -> list[dict[str, Any]]: """Search by MPN / keyword. Returns up to `limit` PartRecord dicts. @@ -226,10 +346,20 @@ def keyword_search(self, mpn: str, limit: int = 5) -> list[dict[str, Any]]: """ conn = self._get_conn() pattern = f"%{mpn}%" + schema = self._get_schema() + query_by_mfr = ( + _CURRENT_QUERY_BY_MFR if schema == "current" else _QUERY_BY_MFR + ) + query_by_desc = ( + _CURRENT_QUERY_BY_DESC if schema == "current" else _QUERY_BY_DESC + ) # Try exact mfr first - rows = conn.execute(_QUERY_BY_MFR, (pattern, limit)).fetchall() + rows = conn.execute(query_by_mfr, (pattern, limit)).fetchall() if not rows: - rows = conn.execute(_QUERY_BY_DESC, (pattern, pattern, limit)).fetchall() + rows = conn.execute( + query_by_desc, + (pattern, pattern, limit), + ).fetchall() return [_row_to_record(tuple(r)) for r in rows] def lookup_lcsc_id(self, c_code: str) -> dict[str, Any] | None: @@ -243,7 +373,12 @@ def lookup_lcsc_id(self, c_code: str) -> dict[str, Any] | None: except ValueError: return None conn = self._get_conn() - row = conn.execute(_QUERY_BY_LCSC, (lcsc_int,)).fetchone() + query = ( + _CURRENT_QUERY_BY_LCSC + if self._get_schema() == "current" + else _QUERY_BY_LCSC + ) + row = conn.execute(query, (lcsc_int,)).fetchone() if row is None: return None return _row_to_record(tuple(row)) diff --git a/tests/test_lcsc_client.py b/tests/test_lcsc_client.py index a8ab310..d69d2d6 100644 --- a/tests/test_lcsc_client.py +++ b/tests/test_lcsc_client.py @@ -52,6 +52,57 @@ def _create_fixture_database(path: Path) -> None: connection.close() +def _create_current_fixture_database(path: Path) -> None: + connection = sqlite3.connect(path) + connection.executescript( + """ + CREATE TABLE jlc_components ( + lcsc INTEGER PRIMARY KEY NOT NULL, + fetched_at INTEGER NOT NULL, + present INTEGER NOT NULL, + sync_seen INTEGER NOT NULL DEFAULT 0, + category TEXT NOT NULL, + subcategory TEXT NOT NULL, + mfr TEXT NOT NULL, + package TEXT NOT NULL, + joints INTEGER NOT NULL, + manufacturer TEXT NOT NULL, + library_type TEXT NOT NULL, + preferred INTEGER NOT NULL, + last_on_stock INTEGER NOT NULL, + description TEXT NOT NULL, + datasheet TEXT NOT NULL, + stock INTEGER NOT NULL, + price TEXT NOT NULL, + attributes TEXT NOT NULL, + rohs INTEGER, + eccn TEXT NOT NULL, + assembly INTEGER, + assembly_process TEXT, + assembly_mode TEXT, + website_component_id TEXT, + attrition TEXT NOT NULL + ); + INSERT INTO jlc_components ( + lcsc, fetched_at, present, sync_seen, category, subcategory, + mfr, package, joints, manufacturer, library_type, preferred, + last_on_stock, description, datasheet, stock, price, attributes, + rohs, eccn, assembly, assembly_process, assembly_mode, + website_component_id, attrition + ) VALUES ( + 8734, 0, 1, 1, 'Integrated Circuits', 'MCU', + 'STM32F103C8T6', 'LQFP-48', 48, 'STMicroelectronics', + 'expand', 1, 0, 'MCU', + 'https://example.invalid/datasheet.pdf', 270767, + '1-9:1.4415,10-29:1.2527,1000-:0.9192', '{}', + 1, '', 1, NULL, NULL, 'fixture', '' + ); + """ + ) + connection.commit() + connection.close() + + def test_missing_cache_fails_without_network( tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: @@ -81,3 +132,39 @@ def test_local_database_supports_keyword_and_id_lookup(tmp_path: Path) -> None: assert matches[0]["stock"] == 42 assert matches[0]["price_tiers"] == [{"qty": 1, "price_usd": 2.5}] assert by_id == matches[0] + + +def test_current_jlcparts_schema_supports_keyword_and_id_lookup( + tmp_path: Path, +) -> None: + database = tmp_path / "cache.sqlite3" + _create_current_fixture_database(database) + + with LcscClient.from_env(db_path=database) as client: + matches = client.keyword_search("STM32F103", limit=3) + by_id = client.lookup_lcsc_id("C8734") + + assert by_id == matches[0] + assert by_id["mpn"] == "STM32F103C8T6" + assert by_id["manufacturer"] == "STMicroelectronics" + assert by_id["basic_extended"] == "Preferred" + assert by_id["stock"] == 270767 + assert by_id["price_tiers"] == [ + {"qty": 1, "price_usd": 1.4415}, + {"qty": 10, "price_usd": 1.2527}, + {"qty": 1000, "price_usd": 0.9192}, + ] + + +def test_unknown_jlcparts_schema_fails_with_actionable_message( + tmp_path: Path, +) -> None: + database = tmp_path / "cache.sqlite3" + connection = sqlite3.connect(database) + connection.execute("CREATE TABLE jlc_components (lcsc INTEGER PRIMARY KEY)") + connection.commit() + connection.close() + + with LcscClient.from_env(db_path=database) as client: + with pytest.raises(RuntimeError, match="unsupported jlcparts schema"): + client.lookup_lcsc_id("C8734") diff --git a/tests/test_repository_contract.py b/tests/test_repository_contract.py index 14210a4..8bfc149 100644 --- a/tests/test_repository_contract.py +++ b/tests/test_repository_contract.py @@ -44,6 +44,10 @@ def test_python_environment_is_declared(self) -> None: self.assertIn(dependency, text) self.assertIn("pytest", text) + def test_local_virtual_environment_is_ignored(self) -> None: + gitignore = (ROOT / ".gitignore").read_text(encoding="utf-8") + self.assertIn(".venv", gitignore.splitlines()) + def test_ci_runs_repository_contract_and_behavior_tests(self) -> None: workflow = ROOT / ".github" / "workflows" / "validate.yml" self.assertTrue(workflow.is_file(), "validation workflow is required") @@ -136,6 +140,13 @@ def test_public_docs_explain_sourcing_network_boundaries(self) -> None: self.assertIn("default", adr.lower()) self.assertIn("offline", adr.lower()) + def test_install_docs_record_current_lcsc_schema_evidence(self) -> None: + install = (ROOT / "INSTALL.md").read_text(encoding="utf-8") + + self.assertIn("jlc_components", install) + self.assertIn("7-Zip", install) + self.assertIn("C8734", install) + if __name__ == "__main__": unittest.main()