#!/usr/bin/env python3 """Resume eBay catalog import: fetch only items missing from catalog.db. Skips items already present, handles error 518 (daily limit) gracefully, upserts into existing catalog.db without deleting prior data. """ import json import logging import sqlite3 import sys import time import xml.etree.ElementTree as ET from collections import Counter from pathlib import Path from agent_samochodowy.config import Settings from agent_samochodowy.matcher.catalog import ( _NS, _extract_brand_from_title, _extract_category_from_title, _extract_model_from_title, _extract_oe_from_title, _is_car_part, _parse_oe_string, _trading_call, _get_all_active_listings, _fetch_and_map_item, ) from agent_samochodowy.matcher.index import CatalogIndex from agent_samochodowy.models import Offer logging.basicConfig( level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s", ) logger = logging.getLogger(__name__) _SKIP_VALUES = {"nie dotyczy", "n/a", "does not apply", "-", ""} import re def _fetch_item_safe(item_id: str, settings: Settings) -> tuple[Offer | None, str]: """Fetch a single item via GetItem. Returns (offer_or_None, status). status is one of: "ok", "skip", "limit", "error" """ xml_body = f""" {item_id} ReturnAll true """ try: root = _trading_call("GetItem", xml_body, settings, timeout=30) except Exception as e: logger.error("GetItem HTTP error for %s: %s", item_id, e) return None, "error" ack = root.findtext("e:Ack", "", _NS) if ack not in ("Success", "Warning"): err_code = root.findtext(".//e:Errors/e:ErrorCode", "", _NS) err_msg = root.findtext(".//e:Errors/e:LongMessage", "", _NS) if err_code == "518": logger.warning("eBay: GetItem daily limit reached (error 518)") return None, "limit" logger.warning("GetItem failed for %s: [%s] %s", item_id, err_code, err_msg) return None, "error" item = root.find(".//e:Item", _NS) if item is None: return None, "skip" # Extract title, price, url from the GetItem response itself title = item.findtext("e:Title", "", _NS) price_el = item.find(".//e:SellingStatus/e:CurrentPrice", _NS) price = float(price_el.text) if price_el is not None and price_el.text else None currency = price_el.get("currencyID", "PLN") if price_el is not None else "PLN" qty = int(item.findtext("e:Quantity", "1", _NS)) url = item.findtext(".//e:ListingDetails/e:ViewItemURL", "", _NS) if not url: url = f"https://www.ebay.pl/itm/{item_id}" # --- ItemSpecifics --- specs: dict[str, list[str]] = {} for nv in item.findall(".//e:ItemSpecifics/e:NameValueList", _NS): name = nv.findtext("e:Name", "", _NS) values = [v.text for v in nv.findall("e:Value", _NS) if v.text] specs[name] = values # OE numbers oe_numbers: list[str] = [] for key in specs: if any(tok in key.lower() for tok in ("referencyjny oe", "oe/oem", "numer części oe")): for v in specs[key]: if v.lower().strip() not in _SKIP_VALUES: oe_numbers.extend(_parse_oe_string(v)) for v in specs.get("MPN", []): if v.lower().strip() not in _SKIP_VALUES: oe_numbers.extend(_parse_oe_string(v)) if not oe_numbers: title_oes = _extract_oe_from_title(title) oe_numbers.extend(title_oes) # Deduplicate OE seen_oe: set[str] = set() unique_oe: list[str] = [] for oe in oe_numbers: norm = re.sub(r"[\s.\-/]", "", oe).upper() if norm and norm not in seen_oe and len(norm) >= 4 and re.search(r"\d", norm): seen_oe.add(norm) unique_oe.append(norm) brand = _extract_brand_from_title(title) model = _extract_model_from_title(title, brand) category = _extract_category_from_title(title) is_part = _is_car_part(title, category) offer = Offer( offer_id=f"ebay_{item_id}", platform="ebay", url=url, title=title, price=price, currency=currency, qty=qty, brand=brand, model=model, category=category, engine_code=None, oe_numbers=unique_oe, is_part=is_part, ) return offer, "ok" def get_existing_ids(db_path: str) -> set[str]: """Get set of offer_ids already in catalog.db.""" conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT offer_id FROM offers WHERE platform = 'ebay'") ids = {row[0] for row in cur.fetchall()} conn.close() return ids def main() -> None: settings = Settings() db_path = settings.db_path Path(db_path).parent.mkdir(parents=True, exist_ok=True) # Step 1: Get existing IDs from DB existing_ids = get_existing_ids(db_path) logger.info("Already in catalog.db: %d eBay offers", len(existing_ids)) # Step 2: Get all active listings (GetMyeBaySelling — cheap calls) logger.info("Fetching active listing IDs via GetMyeBaySelling...") all_listings = list(_get_all_active_listings(settings)) logger.info("Total active listings on eBay: %d", len(all_listings)) # Step 3: Filter to missing only missing = [ item for item in all_listings if f"ebay_{item['item_id']}" not in existing_ids ] logger.info("Missing from catalog.db: %d (skipping %d already imported)", len(missing), len(all_listings) - len(missing)) if not missing: logger.info("Nothing to import — catalog is complete.") _print_db_report(db_path) return # Step 4: Fetch details for missing items, upsert into DB index = CatalogIndex(db_path) imported = 0 errors = 0 skipped = 0 hit_limit = False for i, basic in enumerate(missing): offer, status = _fetch_item_safe(basic["item_id"], settings) if status == "limit": hit_limit = True logger.warning( "Daily limit hit after %d new items. Stopping cleanly.", imported ) break elif status == "error": errors += 1 elif status == "skip" or offer is None: skipped += 1 else: index.upsert_offer(offer) imported += 1 if (i + 1) % 100 == 0: logger.info( "Progress: %d/%d checked | imported=%d errors=%d skipped=%d", i + 1, len(missing), imported, errors, skipped, ) # Throttle: ~10 req/s if (i + 1) % 10 == 0: time.sleep(1.0) # Rebuild FTS after all upserts index._rebuild_fts() index.close() logger.info( "Import done. New: %d, Errors: %d, Skipped: %d, Limit hit: %s, Remaining: %d", imported, errors, skipped, hit_limit, len(missing) - (imported + errors + skipped + (1 if hit_limit else 0)), ) # Step 5: Report _print_db_report(db_path) def _print_db_report(db_path: str) -> None: """Print stats directly from catalog.db.""" conn = sqlite3.connect(db_path) cur = conn.cursor() cur.execute("SELECT COUNT(*) FROM offers") total = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM offers WHERE oe_numbers_json != '[]'") has_oe = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM offers WHERE brand IS NOT NULL AND brand != ''") has_brand = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM offers WHERE model IS NOT NULL AND model != ''") has_model = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM offers WHERE category IS NOT NULL AND category != ''") has_category = cur.fetchone()[0] cur.execute("SELECT COUNT(*) FROM offers WHERE is_part = 0") not_part = cur.fetchone()[0] pct = lambda n: f"{n * 100 // total}%" if total else "0%" print("\n" + "=" * 60) print("RAPORT KATALOGU (z catalog.db)") print("=" * 60) print(f"Łącznie ofert: {total}") print(f"Z numerami OE: {has_oe} ({pct(has_oe)})") print(f"Z rozpoznaną marką: {has_brand} ({pct(has_brand)})") print(f"Z rozpoznanym modelem: {has_model} ({pct(has_model)})") print(f"Z kategorią: {has_category} ({pct(has_category)})") print(f"is_part=false: {not_part}") # Top brands cur.execute(""" SELECT brand, COUNT(*) as cnt FROM offers WHERE brand IS NOT NULL AND brand != '' GROUP BY brand ORDER BY cnt DESC LIMIT 10 """) print("\nTOP 10 marek:") for row in cur.fetchall(): print(f" {row[0]:20s} {row[1]:5d}") # Top categories cur.execute(""" SELECT category, COUNT(*) as cnt FROM offers WHERE category IS NOT NULL AND category != '' GROUP BY category ORDER BY cnt DESC LIMIT 10 """) print("\nTOP 10 kategorii:") for row in cur.fetchall(): print(f" {row[0]:20s} {row[1]:5d}") conn.close() print("=" * 60) if __name__ == "__main__": sys.exit(main() or 0)