Support the current jlcparts cache schema

.gitignore

@@ -2,6 +2,7 @@ __pycache__/
 *.pyc
 *.pyo
 .venv/
+.venv
 .env
 *.egg-info/
 build/

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.

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))

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")

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()