loyaly-catalogue/backend/app/electronics/price_lookup.py

"""Fill missing prices on search-only platforms (Amazon.in, Flipkart, Croma...)
from Google Programmable Search, without fetching those sites.

For each listing that has no price (or an unconfirmed one), search Google for
that product on that site. A price is taken only when:
  * the result is the SAME product page (its site product id equals the
    listing's), and
  * Google's structured data for the page (pagemap offer / product:price meta)
    states an INR price.
The listing is updated through the normal path, so the price is stored with
its evidence, appended to price_history, and outlier-checked.
"""
from __future__ import annotations

import logging
from typing import Callable, Dict, Optional

from app.electronics.collector import Collector, RunOptions, RunStats, source_sku
from app.electronics.db import repository as repo
from app.electronics.db.connection import connect
from app.electronics.normalise.title_parser import parse_title
from app.electronics.reference import load_reference, site_for_url
from app.electronics.search.engine import SearchEngine

logger = logging.getLogger(__name__)


def _listings_needing_price(limit: int, category: Optional[str]) -> list:
    with connect() as conn:
        return conn.execute(
            """
            SELECT l.id, l.title, l.source_sku, l.source_url, s.domain, b.slug AS brand_slug, c.slug AS category,
                   (p.verification_status = 'verified') AS verified
            FROM elec.source_listing l
            JOIN elec.site s ON s.id = l.site_id
            JOIN elec.brand b ON b.id = l.brand_id
            JOIN elec.category c ON c.id = l.category_id
            JOIN elec.product_listing_map m ON m.listing_id = l.id AND m.review_status IN ('auto','approved')
            JOIN elec.product p ON p.id = m.product_id
            WHERE l.source_type = 'search_snippet'
              AND (l.price IS NULL OR l.price_outlier)
              AND (s.policy = 'serp_only' OR coalesce(s.probe_outcome, 'C') = 'C')
              AND (%(category)s::text IS NULL OR c.slug = %(category)s)
            ORDER BY (p.verification_status = 'verified') DESC, l.last_seen_at DESC
            LIMIT %(limit)s
            """,
            {"limit": limit, "category": category},
        ).fetchall()


def lookup_prices(limit: int = 40, category: Optional[str] = None,
                  progress: Callable[[str], None] = logger.info) -> Dict[str, object]:
    stats: Dict[str, object] = {"checked": 0, "priced": 0, "no_same_page": 0, "no_structured_price": 0}
    engine = SearchEngine(budget=limit)
    if not engine.google.enabled:
        stats["error"] = "Google Programmable Search is not configured (GOOGLE_API_KEY / GOOGLE_CSE_ID)"
        return stats
    ids = repo.id_maps()
    ref = load_reference()
    run_id = repo.start_run("price_lookup", {"limit": limit, "category": category})
    try:
        for row in _listings_needing_price(limit, category):
            if not engine.google.enabled:
                break
            site = ref.sites[row["domain"]]
            query = f"site:{row['domain']} {row['title'][:110]}"
            hits = engine.text(query, max_results=10, providers="google")
            stats["checked"] += 1
            if hits is None:
                continue
            same = [h for h in hits
                    if (s := site_for_url(h.url)) is not None and s.domain == site.domain
                    and source_sku(site, h.url) == row["source_sku"]]
            if not same:
                stats["no_same_page"] += 1
                continue
            hit = next((h for h in same if h.offer), None)
            if hit is None:
                stats["no_structured_price"] += 1
                continue
            collector = Collector.__new__(Collector)  # only its listing builder is used
            collector.opt = RunOptions(category=row["category"], brands=[row["brand_slug"]])
            collector.stats = RunStats()
            parsed = parse_title(row["title"], row["category"], expected_brand=row["brand_slug"])
            if parsed.brand is None:
                continue
            listing = collector.listing_from_search(hit, site, parsed, query)
            listing.source_sku = row["source_sku"]
            if listing.price is None:
                stats["no_structured_price"] += 1
                continue
            repo.upsert_listing(listing, ids, run_id)
            stats["priced"] += 1
            progress(f"{site.name}: {row['title'][:70]} -> Rs {listing.price}")
        stats["products"] = repo.refresh_verification()
        if engine.google.error:
            stats["error"] = engine.google.error
        repo.finish_run(run_id, "done", {k: v for k, v in stats.items() if k != "products"})
    except Exception as exc:
        repo.finish_run(run_id, "failed", {}, repr(exc))
        raise
    return stats