"""Fetch, normalise, persist and report one source's catalog. Every write and delete is scoped by (source_kind, source_ref). The crawler, a Shopify store and an HTTP API share one table, so an unscoped delete here would erase another producer's products. """ import logging import statistics from collections import Counter from datetime import datetime, timezone from psycopg2 import sql from psycopg2.extras import Json from app.services.catalog import crawl as catalog_crawl from app.services.catalog import http as catalog_http from app.services.catalog import shopify as catalog_shopify from app.services.catalog.crawl import CrawlSourceError from app.services.catalog.http import HttpSourceError from app.services.catalog.normalise import ( content_hash, normalise, quality_score, record_hash, ) from app.services.infra.quotas import BRANDFORGE, check_quota from app.services.infra.database import ( count_new_products, get_db_connection, get_setting, migrate_products_table, update_setting, ) from app.services.catalog.sources import ( KIND_CRAWL, get_credentials, mark_source_status, touch_last_synced, ) from app.services.integrations.shopify_oauth import ShopQueryError logger = logging.getLogger(__name__) REJECTION_ALERT_RATIO = 0.05 def _is_auth_failure(ex: Exception) -> bool: # Both ShopQueryError and HttpSourceError carry their HTTP status as # "HTTP {status}" in the message; only 401/403 mean the credential itself # is bad rather than a transient throttle or outage. message = str(ex) return "HTTP 401" in message or "HTTP 403" in message def delete_stale(tenant_id: str, source_kind: str, source_ref: str, run_started) -> int: if not source_ref: # A blank ref would match every row of this kind across all sources. raise ValueError("delete_stale requires a source_ref") conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(sql.SQL( "DELETE FROM {}.strategist_products " "WHERE source_kind = %s AND source_ref = %s " "AND (synced_at IS NULL OR synced_at < %s)" ).format(sql.Identifier(tenant_id)), (source_kind, source_ref, run_started)) removed = cur.rowcount conn.commit() return removed except Exception: conn.rollback() logger.error("Could not delete stale products for %s", tenant_id, exc_info=True) raise finally: conn.close() def has_any_products(tenant_id: str, source_kind: str, source_ref: str) -> bool: """Whether this source currently has any product rows -- used to stop the crawl throttle from skipping a source whose data was wiped out from under it (e.g. a manual cleanup) without also resetting last_synced_at. A throttle that only looks at the timestamp can otherwise report a source as "recently synced" while its catalogue is actually empty.""" conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(sql.SQL( "SELECT EXISTS (SELECT 1 FROM {}.strategist_products " "WHERE source_kind = %s AND source_ref = %s)" ).format(sql.Identifier(tenant_id)), (source_kind, source_ref)) return cur.fetchone()[0] except Exception: logger.error("Could not check product presence for %s:%s:%s", tenant_id, source_kind, source_ref, exc_info=True) # Fail open toward crawling rather than toward silently skipping -- # an unnecessary crawl costs time, a wrongly-skipped one costs a # visibly empty catalogue. return False finally: conn.close() def reap_unregistered_crawl_rows(tenant_id: str) -> int: """Retire crawled products written before the website became a real source. Those rows carry source_kind 'crawl' with a NULL source_ref, so delete_stale -- which matches on source_ref -- can never see them. Left alone they would sit beside the properly-synced rows forever, duplicating every product under an old key and with no price, category or content hash. Scoped to NULL refs only: a row belonging to a registered website source has a ref and is left to the ordinary stale-delete. """ conn = get_db_connection() try: with conn.cursor() as cur: cur.execute(sql.SQL( "DELETE FROM {}.strategist_products " "WHERE source_kind = %s AND source_ref IS NULL" ).format(sql.Identifier(tenant_id)), (KIND_CRAWL,)) removed = cur.rowcount conn.commit() if removed: logger.info("Retired %d unregistered crawl row(s) for %s", removed, tenant_id) return removed except Exception: conn.rollback() logger.error("Could not retire unregistered crawl rows for %s", tenant_id, exc_info=True) raise finally: conn.close() def persist(tenant_id: str, products: list, source_kind: str, source_ref: str, run_started) -> dict: saved = 0 conn = get_db_connection() try: with conn.cursor() as cur: for p in products: cur.execute(sql.SQL(""" INSERT INTO {}.strategist_products (product_key, name, description, image_url, product_url, category, source_kind, source_ref, external_id, brand, taxonomy_path, taxonomy_source, raw_category, price_cents, price_max_cents, compare_at_cents, currency, on_sale, in_stock, status, attributes, quality_score, content_hash, record_hash, tenant_relations, rating, review_count, featured_rank, missing_fields, price_reference_cents, fx_rate_used, ctas, synced_at, extracted_at) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s, %s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s, CURRENT_TIMESTAMP) ON CONFLICT (product_key) DO UPDATE SET name = EXCLUDED.name, description = EXCLUDED.description, image_url = EXCLUDED.image_url, product_url = EXCLUDED.product_url, category = EXCLUDED.category, source_kind = EXCLUDED.source_kind, source_ref = EXCLUDED.source_ref, external_id = EXCLUDED.external_id, brand = EXCLUDED.brand, taxonomy_path = EXCLUDED.taxonomy_path, taxonomy_source = EXCLUDED.taxonomy_source, raw_category = EXCLUDED.raw_category, price_cents = EXCLUDED.price_cents, price_max_cents = EXCLUDED.price_max_cents, compare_at_cents = EXCLUDED.compare_at_cents, currency = EXCLUDED.currency, on_sale = EXCLUDED.on_sale, in_stock = EXCLUDED.in_stock, status = EXCLUDED.status, attributes = EXCLUDED.attributes, quality_score = EXCLUDED.quality_score, content_hash = EXCLUDED.content_hash, record_hash = EXCLUDED.record_hash, tenant_relations = EXCLUDED.tenant_relations, rating = EXCLUDED.rating, review_count = EXCLUDED.review_count, featured_rank = EXCLUDED.featured_rank, missing_fields = EXCLUDED.missing_fields, price_reference_cents = EXCLUDED.price_reference_cents, fx_rate_used = EXCLUDED.fx_rate_used, ctas = EXCLUDED.ctas, synced_at = EXCLUDED.synced_at, extracted_at = CURRENT_TIMESTAMP """).format(sql.Identifier(tenant_id)), ( p["product_key"], p["name"], p["description"], p["image_url"], p["product_url"], p["category"], source_kind, source_ref, p["external_id"], p["brand"], p["taxonomy_path"], p["taxonomy_source"], p["raw_category"], p["price_cents"], p["price_max_cents"], p["compare_at_cents"], p["currency"], p["on_sale"], p["in_stock"], p["status"], Json(p["attributes"]), quality_score(p), content_hash(p), record_hash(p), Json(p["tenant_relations"]), p["rating"], p["review_count"], p["featured_rank"], p["missing_fields"], p["price_reference_cents"], p["fx_rate_used"], Json(p.get("ctas") or []), run_started, )) saved += 1 conn.commit() return {"saved": saved} except Exception: conn.rollback() logger.error("Could not persist products for %s", tenant_id, exc_info=True) raise finally: conn.close() def _report(progress, percent: int, step: str) -> None: """Progress is best-effort. The job is spending real money on model calls; a reporting failure must never abort it.""" if progress is None: return try: progress(int(percent), step) except Exception: logger.warning("Progress callback failed", exc_info=True) def _scaled(progress, start: int, end: int): """Maps a sub-call's own 0-100 into this stage's slice of it.""" if progress is None: return None return lambda percent, step: _report(progress, start + (end - start) * int(percent) // 100, step) async def sync_source(tenant_id: str, source: dict, progress=None) -> dict: run_started = datetime.now(timezone.utc) kind, ref, config = source["kind"], source["external_ref"], source["config"] # A public website needs no credential to read. Every other kind speaks to # an account, so a missing one there is a real fault rather than a mode. credentials = get_credentials(kind, ref) if not credentials and kind != KIND_CRAWL: raise ValueError(f"no credentials stored for {kind}:{ref}") _report(progress, 10, f"syncing {ref} (1 of 1)") try: if kind == "shopify": nodes = await catalog_shopify.fetch_products( ref, credentials["access_token"]) source_products = [catalog_shopify.to_source_product(n, ref) for n in nodes] context = { "shop_name": config.get("shop_name"), "primary_domain": config.get("primary_domain"), "url_template": None, "currency": config.get("currency_code") or config.get("currency"), "known_brands": [], } elif kind == "http_api": records = await catalog_http.fetch_products(config, credentials, ref) # Currency is optional at connect time, but the only legitimate # source is the merchant's own override. Falling back to the # tenant's reference_currency was tried and caused a live # mis-pricing: it assumes an unrelated source happens to price in # the same currency as whichever source connected first, which is # a guess wearing a fallback's clothes, and a wrong currency is # invisible in the output -- exactly the failure this code exists # to prevent. Unknown must stay unknown; to_source_product is # told not to raise, and normalise()'s missing_fields flags the # product incomplete instead (the same rule used for a missing # URL or an unconvertible price). currency = config.get("currency") source_products = [ catalog_http.to_source_product( r, config, ref, require_currency=False) for r in records] context = { "shop_name": config.get("shop_name"), "primary_domain": None, "url_template": config.get("url_template"), "currency": currency, "known_brands": [], } elif kind == KIND_CRAWL: source_products = await catalog_crawl.fetch_products( config, ref, tenant_id, progress=_scaled(progress, 10, 60)) context = { "shop_name": config.get("shop_name"), "primary_domain": None, # A crawled product already carries the page it was read from, # so there is nothing to build a URL out of. "url_template": None, "currency": config.get("currency"), "known_brands": [], } else: raise ValueError(f"unsupported source kind: {kind}") except (ShopQueryError, HttpSourceError, CrawlSourceError) as ex: # Only an auth failure (401/403) means the credential itself no longer # works. A throttle or an outage (e.g. Shopify's routine 429 at # first: 250) must not permanently disable the source and force the # merchant to redo OAuth -- it must simply be retried next run. if _is_auth_failure(ex): mark_source_status(kind, ref, "error") raise # The first connected source defines the tenant's reference currency. # Recomputing it per sync would let it drift the moment a second source # with a different currency is added, silently invalidating every price # comparison already made against the original currency. reference_currency = get_setting(tenant_id, "reference_currency") if not reference_currency: reference_currency = context["currency"] update_setting(tenant_id, "reference_currency", lambda _: reference_currency) context["reference_currency"] = reference_currency normalised = [n for n in (normalise(sp, context) for sp in source_products) if n is not None] fetched, kept = len(source_products), len(normalised) rejected = fetched - kept rate = round(rejected / fetched, 4) if fetched else 0.0 incomplete = sum(1 for p in normalised if p["missing_fields"]) _report(progress, 60, f"syncing {ref} (1 of 1)") # Deferred until after a successful fetch: a table migration ahead of a # fetch that never completes is wasted work, and it would turn every # fetch failure into a spurious DB round trip. # Same rule as the CSV path: refuse the whole source before writing any of # it. Persisting what fits would leave delete_stale to reap the rest, # turning a plan limit into deleted products. _sync_sources catches this # per source, so one capped source does not stop the others. check_quota(tenant_id, "product_count", product_name=BRANDFORGE, requested_amount=count_new_products( tenant_id, [p["product_key"] for p in normalised])) migrate_products_table(tenant_id) persist(tenant_id, normalised, kind, ref, run_started) # Only reached when every page fetched and every write succeeded, so a # partial run can never be mistaken for "these products are gone". removed = delete_stale(tenant_id, kind, ref, run_started) if kind == KIND_CRAWL: # The same products have just been written properly, under a key that # includes the source. Retiring the old rows here rather than at connect # time means it only ever happens after a crawl that actually succeeded. removed += reap_unregistered_crawl_rows(tenant_id) _report(progress, 100, f"syncing {ref} (1 of 1)") scores = [quality_score(p) for p in normalised] if rate > REJECTION_ALERT_RATIO: logger.warning("Rejection rate %.1f%% for %s:%s suggests a normaliser " "bug rather than bad merchant data", rate * 100, kind, ref) if incomplete: # A source producing unlinkable products must be obvious in the sync # response, not discovered later by a merchant wondering why nothing # is recommended. most_common = Counter( f for p in normalised for f in p["missing_fields"]).most_common(1)[0][0] logger.warning("%d of %d products for %s:%s are incomplete; most common " "missing field is %r", incomplete, kept, kind, ref, most_common) try: # The catalog write already committed; a bookkeeping timestamp that # fails to save must not be reported as a failed sync. touch_last_synced(kind, ref, run_started) except Exception: logger.error("Could not record last_synced_at for %s:%s", kind, ref, exc_info=True) return { "fetched": fetched, "normalised": kept, "rejected": rejected, "rejection_rate": rate, "incomplete": incomplete, "saved": kept, "deleted": removed, "quality_median": round(statistics.median(scores), 2) if scores else None, }