app/services/external/keepa_client.py

"""Keepa Product API — optional PAID backfill (dormant without KEEPA_API_KEY).

Rankora FYP default is Rankora-only scrape history. Keepa has no free API tier.
This module is unused unless a paid key is configured.

Docs: https://keepa.com/#!discuss/t/product-object/116
"""
from __future__ import annotations

import logging
import time
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional, Tuple

import requests

from app.config import settings

logger = logging.getLogger(__name__)

KEEPA_EPOCH = datetime(2011, 1, 1, tzinfo=timezone.utc)
AMAZON_US_SELLER_ID = "ATVPDKIKX0DER"

# Product.csv indices (US)
CSV_AMAZON = 0
CSV_NEW = 1
CSV_SALES = 3
CSV_COUNT_NEW = 11
CSV_BUY_BOX_SHIPPING = 18

_cache: Dict[str, Tuple[float, dict]] = {}
_CACHE_TTL_SEC = 6 * 3600  # avoid burning tokens on every page load


def keepa_enabled() -> bool:
    return bool((settings.keepa_api_key or "").strip())


def keepa_minute_to_datetime(minutes: int | str) -> datetime:
    return KEEPA_EPOCH + timedelta(minutes=int(minutes))


def _cents_to_usd(cents: Optional[int]) -> Optional[float]:
    if cents is None or cents < 0:
        return None
    return round(cents / 100.0, 2)


def _parse_csv_pairs(csv_row: Optional[List[int]]) -> List[Tuple[datetime, Optional[float]]]:
    """Parse Keepa [time, value, time, value, ...] into (datetime, usd|None)."""
    if not csv_row or len(csv_row) < 2:
        return []
    out: List[Tuple[datetime, Optional[float]]] = []
    for i in range(0, len(csv_row) - 1, 2):
        try:
            ts = keepa_minute_to_datetime(csv_row[i])
            val = csv_row[i + 1]
            out.append((ts, _cents_to_usd(val)))
        except (TypeError, ValueError, IndexError):
            continue
    return out


def _parse_seller_history(history: Optional[List[str]]) -> List[Tuple[datetime, str]]:
    """Parse buyBoxSellerIdHistory [time, sellerId, ...]."""
    if not history or len(history) < 2:
        return []
    out: List[Tuple[datetime, str]] = []
    for i in range(0, len(history) - 1, 2):
        try:
            ts = keepa_minute_to_datetime(history[i])
            sid = str(history[i + 1] or "").strip()
            if sid in ("-1", ""):
                continue  # no qualified Buy Box
            if sid == "-2":
                out.append((ts, "Unknown seller"))
            elif sid == AMAZON_US_SELLER_ID:
                out.append((ts, "Amazon.com"))
            else:
                out.append((ts, sid))
        except (TypeError, ValueError, IndexError):
            continue
    return out


def _seller_display(seller_id: str) -> str:
    if seller_id == AMAZON_US_SELLER_ID or seller_id == "Amazon.com":
        return "Amazon.com"
    if seller_id == "Unknown seller":
        return seller_id
    # Keepa only returns seller IDs without names unless a separate seller lookup is used
    if len(seller_id) >= 10 and seller_id.isalnum():
        return f"Seller {seller_id[-6:]}"
    return seller_id


def fetch_keepa_product(asin: str, domain: int = 1, stats_days: int = 365) -> Optional[dict]:
    """Call Keepa Product API. Returns raw product dict or None."""
    if not keepa_enabled():
        return None

    asin = (asin or "").strip().upper()
    if not asin or len(asin) != 10:
        return None

    cached = _cache.get(asin)
    if cached and (time.time() - cached[0]) < _CACHE_TTL_SEC:
        return cached[1]

    params = {
        "key": settings.keepa_api_key.strip(),
        "domain": domain,
        "asin": asin,
        "history": 1,
        "buybox": 1,
        "stats": max(30, min(int(stats_days), 365)),
    }
    try:
        resp = requests.get("https://api.keepa.com/product", params=params, timeout=45)
        if resp.status_code != 200:
            logger.warning("[keepa] HTTP %s for %s: %s", resp.status_code, asin, resp.text[:200])
            return None
        payload = resp.json()
        products = payload.get("products") or []
        if not products:
            logger.info("[keepa] No product for %s (tokensLeft=%s)", asin, payload.get("tokensLeft"))
            return None
        product = products[0]
        _cache[asin] = (time.time(), product)
        logger.info(
            "[keepa] OK %s tokensLeft=%s tokensConsumed=%s",
            asin,
            payload.get("tokensLeft"),
            payload.get("tokensConsumed"),
        )
        return product
    except Exception as e:
        logger.warning("[keepa] Request failed for %s: %s", asin, e)
        return None


def keepa_product_to_history(
    product: dict,
    range_days: int = 365,
) -> Dict[str, Any]:
    """Convert Keepa product → Rankora-compatible price + Buy Box snapshot series."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=max(range_days, 1))
    csv = product.get("csv") or []

    def csv_at(idx: int) -> Optional[List[int]]:
        if idx < len(csv) and csv[idx] is not None:
            return csv[idx]
        return None

    amazon_prices = _parse_csv_pairs(csv_at(CSV_AMAZON))
    new_prices = _parse_csv_pairs(csv_at(CSV_NEW))
    bb_prices = _parse_csv_pairs(csv_at(CSV_BUY_BOX_SHIPPING))
    if not bb_prices:
        bb_prices = new_prices or amazon_prices

    sales = _parse_csv_pairs(csv_at(CSV_SALES))  # BSR values (not cents)
    # Sales rank is stored as integer in csv, not cents — re-parse
    sales_rank: List[Tuple[datetime, Optional[int]]] = []
    raw_sales = csv_at(CSV_SALES)
    if raw_sales:
        for i in range(0, len(raw_sales) - 1, 2):
            try:
                ts = keepa_minute_to_datetime(raw_sales[i])
                rank = raw_sales[i + 1]
                sales_rank.append((ts, None if rank is None or rank < 0 else int(rank)))
            except (TypeError, ValueError, IndexError):
                continue

    offer_counts: List[Tuple[datetime, Optional[int]]] = []
    raw_offers = csv_at(CSV_COUNT_NEW)
    if raw_offers:
        for i in range(0, len(raw_offers) - 1, 2):
            try:
                ts = keepa_minute_to_datetime(raw_offers[i])
                n = raw_offers[i + 1]
                offer_counts.append((ts, None if n is None or n < 0 else int(n)))
            except (TypeError, ValueError, IndexError):
                continue

    seller_hist = _parse_seller_history(product.get("buyBoxSellerIdHistory"))

    def value_at(series: List[Tuple[datetime, Any]], when: datetime) -> Any:
        best = None
        for ts, val in series:
            if ts <= when:
                best = val
            else:
                break
        return best

    # Build daily price history from NEW (or Amazon) series
    price_points = [(ts, p) for ts, p in (new_prices or amazon_prices) if ts >= cutoff and p is not None]
    price_history: List[dict] = []
    for ts, price in price_points:
        price_history.append(
            {
                "price": price,
                "bsr": value_at(sales_rank, ts),
                "rating": None,
                "review_count": None,
                "recorded_at": ts.isoformat(),
                "source": "keepa",
            }
        )

    # Buy Box snapshots: align seller changes with BB price
    snapshots: List[dict] = []
    for ts, seller_id in seller_hist:
        if ts < cutoff:
            continue
        price = value_at(bb_prices, ts)
        seller = _seller_display(seller_id)
        is_amazon = seller == "Amazon.com" or seller_id == AMAZON_US_SELLER_ID
        snapshots.append(
            {
                "winner": seller,
                "seller_id": seller_id if seller_id not in ("Amazon.com",) else AMAZON_US_SELLER_ID,
                "price": price,
                "is_fba": True if is_amazon else None,  # Amazon FBA; other sellers unknown without offers
                "is_amazon": is_amazon,
                "seller_count": value_at(offer_counts, ts),
                "has_buy_box": True,
                "recorded_at": ts.isoformat(),
                "source": "keepa",
            }
        )

    # If seller history missing but BB price exists, still emit price-only snapshots
    if not snapshots and bb_prices:
        for ts, price in bb_prices:
            if ts < cutoff or price is None:
                continue
            snapshots.append(
                {
                    "winner": "Buy Box",
                    "seller_id": None,
                    "price": price,
                    "is_fba": None,
                    "is_amazon": False,
                    "seller_count": value_at(offer_counts, ts),
                    "has_buy_box": True,
                    "recorded_at": ts.isoformat(),
                    "source": "keepa",
                }
            )

    return {
        "asin": product.get("asin"),
        "title": product.get("title"),
        "price_history": price_history,
        "buy_box_snapshots": snapshots,
        "source": "keepa",
        "tokens_note": "Keepa tokens are consumed per request; results are cached 6h in-process.",
    }


def fetch_keepa_history(asin: str, range_days: int = 365) -> Optional[Dict[str, Any]]:
    """High-level: fetch + parse Keepa history for an ASIN."""
    product = fetch_keepa_product(asin, stats_days=range_days)
    if not product:
        return None
    return keepa_product_to_history(product, range_days=range_days)