"""SQLite + FTS5 catalog index for fast offer lookup.""" from __future__ import annotations import json import sqlite3 from pathlib import Path from agent_samochodowy.models import Offer class CatalogIndex: """Local index of seller offers backed by SQLite + FTS5.""" def __init__(self, db_path: str = ":memory:") -> None: if db_path != ":memory:": Path(db_path).parent.mkdir(parents=True, exist_ok=True) self.conn = sqlite3.connect(db_path) self.conn.row_factory = sqlite3.Row self._init_schema() def _init_schema(self) -> None: cur = self.conn.cursor() cur.executescript(""" CREATE TABLE IF NOT EXISTS offers ( offer_id TEXT PRIMARY KEY, platform TEXT NOT NULL, url TEXT NOT NULL, title TEXT NOT NULL, price REAL, currency TEXT DEFAULT 'PLN', qty INTEGER DEFAULT 1, brand TEXT, model TEXT, category TEXT, engine_code TEXT, oe_numbers_json TEXT DEFAULT '[]', is_part INTEGER DEFAULT 1 ); CREATE TABLE IF NOT EXISTS oe_index ( oe_number TEXT NOT NULL, offer_id TEXT NOT NULL, UNIQUE(oe_number, offer_id) ); CREATE INDEX IF NOT EXISTS idx_oe ON oe_index(oe_number); CREATE TABLE IF NOT EXISTS engine_index ( engine_code TEXT NOT NULL, offer_id TEXT NOT NULL, UNIQUE(engine_code, offer_id) ); CREATE INDEX IF NOT EXISTS idx_engine ON engine_index(engine_code); """) # FTS5 for fuzzy text search (standalone, not content-synced) cur.execute(""" CREATE VIRTUAL TABLE IF NOT EXISTS offers_fts USING fts5( offer_id UNINDEXED, title, category, brand, model, engine_code, tokenize='unicode61 remove_diacritics 2' ) """) self.conn.commit() def upsert_offer(self, offer: Offer) -> None: """Insert or update a single offer and its index entries.""" cur = self.conn.cursor() oe_json = json.dumps(offer.oe_numbers) cur.execute(""" INSERT INTO offers (offer_id, platform, url, title, price, currency, qty, brand, model, category, engine_code, oe_numbers_json, is_part) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT(offer_id) DO UPDATE SET platform=excluded.platform, url=excluded.url, title=excluded.title, price=excluded.price, currency=excluded.currency, qty=excluded.qty, brand=excluded.brand, model=excluded.model, category=excluded.category, engine_code=excluded.engine_code, oe_numbers_json=excluded.oe_numbers_json, is_part=excluded.is_part """, ( offer.offer_id, offer.platform, offer.url, offer.title, offer.price, offer.currency, offer.qty, offer.brand, offer.model, offer.category, offer.engine_code, oe_json, int(offer.is_part), )) # Rebuild OE index for this offer cur.execute("DELETE FROM oe_index WHERE offer_id = ?", (offer.offer_id,)) for oe in offer.oe_numbers: cur.execute( "INSERT OR IGNORE INTO oe_index (oe_number, offer_id) VALUES (?, ?)", (oe, offer.offer_id), ) # Rebuild engine code index cur.execute("DELETE FROM engine_index WHERE offer_id = ?", (offer.offer_id,)) if offer.engine_code: cur.execute( "INSERT OR IGNORE INTO engine_index (engine_code, offer_id) VALUES (?, ?)", (offer.engine_code.upper(), offer.offer_id), ) self.conn.commit() def build_from_offers(self, offers: list[Offer]) -> None: """Bulk-load offers into the index.""" for offer in offers: if offer.is_part: self.upsert_offer(offer) self._rebuild_fts() def _rebuild_fts(self) -> None: """Rebuild the FTS5 index from the offers table.""" cur = self.conn.cursor() cur.execute("DELETE FROM offers_fts") cur.execute(""" INSERT INTO offers_fts(offer_id, title, category, brand, model, engine_code) SELECT offer_id, title, category, brand, model, engine_code FROM offers WHERE is_part = 1 """) self.conn.commit() def _rows_to_offers(self, rows: list[sqlite3.Row]) -> list[Offer]: result = [] for row in rows: result.append(Offer( offer_id=row["offer_id"], platform=row["platform"], url=row["url"], title=row["title"], price=row["price"], currency=row["currency"], qty=row["qty"], brand=row["brand"], model=row["model"], category=row["category"], engine_code=row["engine_code"], oe_numbers=json.loads(row["oe_numbers_json"]), is_part=bool(row["is_part"]), )) return result def search_by_oe(self, oe: str) -> list[Offer]: """Exact OE number lookup.""" cur = self.conn.cursor() cur.execute(""" SELECT o.* FROM offers o JOIN oe_index i ON o.offer_id = i.offer_id WHERE i.oe_number = ? AND o.is_part = 1 """, (oe,)) return self._rows_to_offers(cur.fetchall()) def search_by_engine_code(self, code: str) -> list[Offer]: """Exact engine code lookup.""" cur = self.conn.cursor() cur.execute(""" SELECT o.* FROM offers o JOIN engine_index i ON o.offer_id = i.offer_id WHERE i.engine_code = ? AND o.is_part = 1 """, (code.upper(),)) return self._rows_to_offers(cur.fetchall()) def search_by_attributes( self, brand: str | None = None, model: str | None = None, category: str | None = None, ) -> list[Offer]: """Attribute-based search (brand, model, category). Category matching uses synonym expansion: a query for "sprężarka powietrza" also matches offers with category "compressor" or title containing "SPRĘŻARKA". """ from agent_samochodowy.matcher.normalize import expand_category_synonyms conditions = ["is_part = 1"] params: list[str] = [] if brand: conditions.append("LOWER(brand) = LOWER(?)") params.append(brand) if model: # Prefix matching: "LF" matches "LF45"/"LF55", "R" matches "R142" # Also reverse: "R410" matches "R" (query is more specific than catalog) conditions.append( "(LOWER(model) = LOWER(?)" " OR LOWER(model) LIKE LOWER(?) || '%'" # catalog "LF45" matches query "LF" " OR LOWER(?) LIKE LOWER(model) || '%'" # query "R410" matches catalog "R" ")" ) params.extend([model, model, model]) if category: synonyms = expand_category_synonyms(category) cat_clauses = [] for syn in synonyms: cat_clauses.append("LOWER(category) LIKE LOWER(?)") params.append(f"%{syn}%") # Also match synonyms in title (compact forms ≤2 words). # Searches title regardless of category value — covers cases where # category is set to a different term (e.g. "zawór" for an air dryer). for syn in synonyms: if len(syn.split()) <= 2: cat_clauses.append("LOWER(title) LIKE LOWER(?)") params.append(f"% {syn} %") conditions.append(f"({' OR '.join(cat_clauses)})") if len(conditions) == 1: return [] cur = self.conn.cursor() cur.execute( f"SELECT * FROM offers WHERE {' AND '.join(conditions)} LIMIT 50", params ) return self._rows_to_offers(cur.fetchall()) def search_fuzzy(self, text: str) -> list[Offer]: """FTS5 full-text search on title, category, brand, model.""" # Clean text for FTS5 query: keep alphanumeric and spaces import re tokens = re.findall(r"\w+", text, re.UNICODE) if not tokens: return [] # FTS5 query: OR-join tokens for broad matching fts_query = " OR ".join(tokens[:10]) # limit tokens cur = self.conn.cursor() try: cur.execute(""" SELECT o.* FROM offers_fts f JOIN offers o ON f.offer_id = o.offer_id WHERE offers_fts MATCH ? AND o.is_part = 1 LIMIT 20 """, (fts_query,)) return self._rows_to_offers(cur.fetchall()) except sqlite3.OperationalError: return [] def count(self) -> int: cur = self.conn.cursor() cur.execute("SELECT COUNT(*) FROM offers WHERE is_part = 1") return cur.fetchone()[0] def close(self) -> None: self.conn.close()