"""
TikTok Shop Cache Service - Cache-first access to TikTok Shop data via Apify.

Stores normalized API responses as JSON in SQLite (echotik_cache table).
Serves from cache when fresh, fetches from Apify only on miss/expiry.
Falls back to stale cache if API fails.

Also manages ProductSnapshot for trend charts (daily snapshots of monitored products).
"""

import json
import logging
from datetime import datetime, timedelta

from backend.database.connection import SessionLocal
from backend.models.echotik_cache import EchoTikCache
from backend.models.product_snapshot import ProductSnapshot
from backend.services.tiktok_shop_api import TikTokShopClient

logger = logging.getLogger(__name__)

# TTL in hours per cache type
TTL_HOURS = {
    "product_ranking": 12,      # Rankings refresh 2x/day
    "product_search": 12,       # Search results valid 12h
    "product_detail": 24,       # Detail valid 24h
    "seller_ranking": 24,       # Seller ranking daily
    "seller_search": 12,
    "seller_detail": 24,
    "influencer_ranking": 24,   # Built from product data
    "video_ranking": 24,        # Built from product data
    "product_trend": 1,         # Trends always from local snapshots (1h = force refresh on request)
}


class TikTokShopCacheService:
    """Cache-first TikTok Shop data service using Apify."""

    def __init__(self, api_token: str):
        self.client = TikTokShopClient(api_token)

    def _get_or_fetch(self, cache_type: str, cache_key: str, fetch_fn, ttl_hours: int = None):
        """Core cache pattern: serve from DB if fresh, otherwise fetch and store."""
        if ttl_hours is None:
            ttl_hours = TTL_HOURS.get(cache_type, 12)

        db = SessionLocal()
        try:
            row = db.query(EchoTikCache).filter(EchoTikCache.cache_key == cache_key).first()

            # Cache hit and still fresh
            if row and row.expires_at and datetime.utcnow() < row.expires_at:
                return json.loads(row.data_json)

            # Cache miss or expired - try to fetch
            try:
                data = fetch_fn()
                is_empty = (isinstance(data, list) and len(data) == 0)
                if not is_empty:
                    self._save_cache(db, cache_type, cache_key, data, ttl_hours)
                elif row:
                    logger.info(f"API returned empty for {cache_key}, serving stale cache")
                    return json.loads(row.data_json)
                return data
            except Exception as e:
                logger.warning(f"Apify fetch failed ({cache_type}/{cache_key}): {e}")
                if row:
                    logger.info(f"Serving stale cache for {cache_key}")
                    return json.loads(row.data_json)
                raise
        finally:
            db.close()

    def _get_cache_only(self, cache_key: str):
        """Get from cache only, no API call."""
        db = SessionLocal()
        try:
            row = db.query(EchoTikCache).filter(EchoTikCache.cache_key == cache_key).first()
            if row:
                return json.loads(row.data_json)
            return None
        finally:
            db.close()

    def _save_cache(self, db, cache_type: str, cache_key: str, data, ttl_hours: int):
        """Save or update cache entry."""
        now = datetime.utcnow()
        expires = now + timedelta(hours=ttl_hours)
        data_str = json.dumps(data, ensure_ascii=False)
        count = len(data) if isinstance(data, list) else 1

        row = db.query(EchoTikCache).filter(EchoTikCache.cache_key == cache_key).first()
        if row:
            row.data_json = data_str
            row.item_count = count
            row.fetched_at = now
            row.expires_at = expires
        else:
            db.add(EchoTikCache(
                cache_type=cache_type,
                cache_key=cache_key,
                data_json=data_str,
                item_count=count,
                fetched_at=now,
                expires_at=expires,
            ))
        db.commit()

    def _save_to_cache_direct(self, cache_type: str, cache_key: str, data, ttl_hours: int = 24):
        """Save data to cache from external context."""
        db = SessionLocal()
        try:
            self._save_cache(db, cache_type, cache_key, data, ttl_hours)
        finally:
            db.close()

    # ─── Product Endpoints ───────────────────────────────

    def product_ranking(self, sort_by="sales", rank_field=1, page=1, page_size=10, category_id=""):
        """Get product ranking (from cache or API, with fallback to all cached products)."""
        sort = "gmv" if rank_field == 2 else "sales"
        key = f"product_ranking:rf{rank_field}:p{page}:ps{page_size}:cat_{category_id}"
        try:
            result = self._get_or_fetch("product_ranking", key,
                lambda: self.client.product_ranking(sort, page_size, category_id))
            if result:
                return result
        except Exception:
            pass
        # Fallback: build ranking from ALL cached search/detail data
        return self._build_product_ranking_from_cache(rank_field, page_size)

    def product_search(self, keyword="", page=1, page_size=10):
        """Search products by keyword."""
        key = f"product_search:kw_{keyword}:p{page}:ps{page_size}"
        try:
            return self._get_or_fetch("product_search", key,
                lambda: self.client.product_search(keyword, page_size))
        except Exception:
            # Fallback: search in cached products locally
            return self._search_products_in_cache(keyword, page_size)

    def product_detail(self, product_id: str):
        """Get full product details."""
        key = f"product_detail:{product_id}"
        return self._get_or_fetch("product_detail", key,
            lambda: self.client.product_detail(product_id))

    def product_trend(self, product_id: str, days: int = 7):
        """Get product trend from local snapshots."""
        return self._get_local_trend(product_id, days)

    def product_videos(self, product_id: str, page=1, page_size=10):
        """Get videos related to product (from scanner DB cross-reference)."""
        return self._get_product_videos(product_id)

    def product_influencers(self, product_id: str, page=1, page_size=10):
        """Get influencers related to product (from scanner DB cross-reference)."""
        return self._get_product_influencers(product_id)

    # ─── Seller Endpoints ────────────────────────────────

    def seller_ranking(self, date="", rank_field=1, page=1, page_size=10):
        """Get seller ranking (aggregated from product data)."""
        key = f"seller_ranking:rf{rank_field}:p{page}:ps{page_size}"
        return self._get_or_fetch("seller_ranking", key,
            lambda: self._build_seller_ranking(rank_field))

    def seller_detail(self, seller_id: str):
        """Get seller details from cache."""
        key = f"seller_detail:{seller_id}"
        cached = self._get_cache_only(key)
        if cached:
            return cached
        # Try to build from product data
        return self._build_seller_detail(seller_id)

    def seller_products(self, seller_id: str, page=1, page_size=10):
        """Get seller products from cache."""
        key = f"seller_products:{seller_id}:p{page}"
        cached = self._get_cache_only(key)
        if cached:
            return cached
        return self._find_seller_products(seller_id)

    def seller_search(self, keyword="", page=1, page_size=10):
        """Search sellers (from cached product data)."""
        key = f"seller_search:kw_{keyword}:p{page}:ps{page_size}"
        return self._get_or_fetch("seller_search", key,
            lambda: self._search_sellers(keyword))

    # ─── Influencer Endpoints ────────────────────────────

    def influencer_ranking(self, date="", rank_field=1, page=1, page_size=10, days=30):
        """Get influencer ranking (built from scanner data, BR only)."""
        key = f"influencer_ranking:rf{rank_field}:d{days}:p{page}:ps{page_size}"
        return self._get_or_fetch("influencer_ranking", key,
            lambda: self._build_influencer_ranking(rank_field, days))

    # ─── Video Endpoints ─────────────────────────────────

    def video_ranking(self, date="", rank_field=1, page=1, page_size=10, days=30):
        """Get video ranking (built from scanner data, BR only)."""
        key = f"video_ranking:rf{rank_field}:d{days}:p{page}:ps{page_size}"
        return self._get_or_fetch("video_ranking", key,
            lambda: self._build_video_ranking(rank_field, days))

    # ─── Daily Trends (BR) ────────────────────────────────

    def daily_trends(self, days=30):
        """Get daily video trends for BR (chart data)."""
        key = f"daily_trends:d{days}"
        return self._get_or_fetch("video_ranking", key,
            lambda: self._build_daily_trends(days), ttl_hours=6)

    # ─── Trend (Local Snapshots) ─────────────────────────

    def _get_local_trend(self, product_id: str, days: int = 7) -> list:
        """Build trend data from ProductSnapshot table."""
        db = SessionLocal()
        try:
            since = datetime.utcnow() - timedelta(days=days)
            snapshots = db.query(ProductSnapshot).filter(
                ProductSnapshot.product_id == product_id,
                ProductSnapshot.captured_at >= since,
            ).order_by(ProductSnapshot.captured_at.asc()).all()

            return [{
                "date": s.captured_at.strftime("%Y-%m-%d"),
                "price": float(s.price) if s.price else 0,
                "sales_total": s.sold_count or 0,
                "sales_daily": 0,  # Calculated on frontend as delta
                "gmv_total": (s.sold_count or 0) * (float(s.price) if s.price else 0),
                "gmv_daily": 0,
                "stock": s.stock or 0,
                "rating": s.rating or 0,
                "review_count": s.review_count or 0,
            } for s in snapshots]
        finally:
            db.close()

    def take_snapshot(self, product_id: str) -> dict:
        """Take a snapshot of a product's current state for trend tracking.

        First checks cache, then fetches from API if needed.
        """
        # Get product data (from cache or API)
        try:
            detail = self.product_detail(product_id)
        except Exception:
            # Try cache only
            cached = self._get_cache_only(f"product_detail:{product_id}")
            if not cached:
                raise ValueError(f"Produto {product_id} nao encontrado")
            detail = cached

        db = SessionLocal()
        try:
            # Check if we already have a snapshot today
            today = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)
            existing = db.query(ProductSnapshot).filter(
                ProductSnapshot.product_id == product_id,
                ProductSnapshot.captured_at >= today,
            ).first()

            if existing:
                # Update existing snapshot
                existing.sold_count = detail.get("sales_total", 0)
                existing.stock = detail.get("stock", 0)
                existing.price = str(detail.get("price", "0"))
                existing.rating = detail.get("rating", 0)
                existing.review_count = detail.get("review_count", 0)
                db.commit()
                return existing.to_dict()

            # Create new snapshot
            snapshot = ProductSnapshot(
                product_id=product_id,
                sold_count=detail.get("sales_total", 0),
                stock=detail.get("stock", 0),
                price=str(detail.get("price", "0")),
                rating=detail.get("rating", 0),
                review_count=detail.get("review_count", 0),
            )
            db.add(snapshot)
            db.commit()
            return snapshot.to_dict()
        finally:
            db.close()

    def take_snapshots_for_monitored(self) -> int:
        """Take snapshots for monitored products using cached data only.

        Only snapshots products that already exist in our cache
        (from search results). Does NOT call the detail API for each product.
        """
        db = SessionLocal()
        try:
            # Build a map of product_id -> data from all cached search/ranking results
            cached_products = {}
            rows = db.query(EchoTikCache).filter(
                EchoTikCache.cache_type.in_(["product_ranking", "product_search"])
            ).all()
            for row in rows:
                products = json.loads(row.data_json)
                if not isinstance(products, list):
                    continue
                for p in products:
                    pid = p.get("product_id", "")
                    if pid:
                        cached_products[pid] = p

            if not cached_products:
                return 0

            count = 0
            today = datetime.utcnow().replace(hour=0, minute=0, second=0, microsecond=0)

            for pid, p in cached_products.items():
                # Check if we already have a snapshot today
                existing = db.query(ProductSnapshot).filter(
                    ProductSnapshot.product_id == pid,
                    ProductSnapshot.captured_at >= today,
                ).first()
                if existing:
                    continue

                snapshot = ProductSnapshot(
                    product_id=pid,
                    sold_count=int(p.get("sales_total", 0) or 0),
                    stock=int(p.get("stock", 0) or 0),
                    price=str(p.get("price", "0")),
                    rating=float(p.get("rating", 0) or 0),
                    review_count=int(p.get("review_count", 0) or 0),
                )
                db.add(snapshot)
                count += 1

            if count > 0:
                db.commit()
            return count
        finally:
            db.close()

    # ─── Product Fallbacks (offline mode) ─────────────────

    def _search_products_in_cache(self, keyword: str, limit: int = 10) -> list:
        """Search products by keyword in all cached data (offline fallback)."""
        db = SessionLocal()
        try:
            rows = db.query(EchoTikCache).filter(
                EchoTikCache.cache_type.in_(["product_search", "product_detail", "product_ranking"])
            ).all()

            kw_lower = keyword.lower()
            results = []
            seen = set()
            for row in rows:
                data = json.loads(row.data_json)
                items = data if isinstance(data, list) else [data]
                for p in items:
                    pid = p.get("product_id", "")
                    title = (p.get("title", "") or "").lower()
                    category = (p.get("category", "") or "").lower()
                    if pid and pid not in seen and (kw_lower in title or kw_lower in category):
                        seen.add(pid)
                        results.append(p)

            results.sort(key=lambda x: x.get("sales_total", 0) or 0, reverse=True)
            return results[:limit]
        except Exception:
            return []
        finally:
            db.close()

    def _build_product_ranking_from_cache(self, rank_field: int = 1, limit: int = 10) -> list:
        """Build product ranking from cached search data (BR products)."""
        db = SessionLocal()
        try:
            # Prioritize search results (these are from BR keyword searches)
            search_pids = set()
            products = {}

            # First pass: get all search products (these are BR)
            search_rows = db.query(EchoTikCache).filter(
                EchoTikCache.cache_type == "product_search"
            ).all()
            for row in search_rows:
                data = json.loads(row.data_json)
                if isinstance(data, list):
                    for p in data:
                        pid = p.get("product_id", "")
                        if pid:
                            search_pids.add(pid)
                            products[pid] = p

            # Second pass: enrich with detail data (only for products already in search)
            detail_rows = db.query(EchoTikCache).filter(
                EchoTikCache.cache_type == "product_detail"
            ).all()
            for row in detail_rows:
                data = json.loads(row.data_json)
                items = data if isinstance(data, list) else [data]
                for p in items:
                    pid = p.get("product_id", "")
                    if pid in search_pids:
                        # Merge: keep detail data but preserve search metrics
                        existing = products.get(pid, {})
                        merged = {**p}
                        # Preserve sales from search if detail has 0
                        if not merged.get("sales_total") and existing.get("sales_total"):
                            merged["sales_total"] = existing["sales_total"]
                            merged["sales_formatted"] = existing.get("sales_formatted", "0")
                        products[pid] = merged

            result = list(products.values())

            if rank_field == 2:  # GMV
                result.sort(key=lambda x: x.get("gmv_total", 0) or 0, reverse=True)
            elif rank_field == 3:  # Volume/rating
                result.sort(key=lambda x: x.get("rating", 0) or 0, reverse=True)
            else:  # Sales
                result.sort(key=lambda x: x.get("sales_total", 0) or 0, reverse=True)

            # Cache the result for next time
            if result:
                key = f"product_ranking:rf{rank_field}:p1:ps{limit}:cat_"
                self._save_to_cache_direct("product_ranking", key, result[:limit], ttl_hours=24)

            return result[:limit]
        except Exception as e:
            logger.warning(f"Error building product ranking from cache: {e}")
            return []
        finally:
            db.close()

    # ─── Cross-Reference: Videos & Influencers ───────────

    def _get_product_videos(self, product_id: str) -> list:
        """Find videos in scanner DB that mention this product."""
        from backend.models.video import Video
        db = SessionLocal()
        try:
            # Get product title for keyword matching
            detail_cache = self._get_cache_only(f"product_detail:{product_id}")
            title_words = []
            if detail_cache and isinstance(detail_cache, dict):
                title = detail_cache.get("title", "").lower()
                title_words = [w for w in title.split() if len(w) > 3][:5]

            # First try: videos with shop_product_url containing product_id
            videos = db.query(Video).filter(
                Video.platform == "tiktok",
                Video.shop_product_url.contains(product_id),
            ).order_by(Video.views.desc()).limit(10).all()

            # If none found, try keyword matching in top TikTok videos
            if not videos and title_words:
                all_videos = db.query(Video).filter(
                    Video.platform == "tiktok",
                ).order_by(Video.views.desc()).limit(100).all()

                videos = []
                for v in all_videos:
                    desc = (v.description or "").lower()
                    vtitle = (v.title or "").lower()
                    if any(w in desc or w in vtitle for w in title_words):
                        videos.append(v)
                        if len(videos) >= 10:
                            break

            results = []
            for v in videos:
                creator_name = ""
                if v.creator:
                    creator_name = v.creator.username if hasattr(v.creator, 'username') else ""
                results.append({
                    "video_id": v.external_id or str(v.id),
                    "description": v.title or v.description or "",
                    "thumbnail": v.thumbnail_url or "",
                    "duration": v.duration or 0,
                    "views": v.views or 0,
                    "views_formatted": _format_num(v.views),
                    "likes": v.likes or 0,
                    "comments": v.comments or 0,
                    "shares": v.shares or 0,
                    "sales": 0,
                    "sales_formatted": "0",
                    "gmv": 0,
                    "gmv_formatted": "R$ 0",
                    "creator_username": creator_name,
                    "creator_nickname": creator_name,
                    "creator_avatar": "",
                    "user_id": "",
                    "product_id": product_id,
                    "hashtags": "",
                    "created_at": v.published_at.isoformat() if v.published_at else "",
                    "region": "BR",
                })

            return results
        finally:
            db.close()

    def _get_product_influencers(self, product_id: str) -> list:
        """Find creators in DB that are associated with shop products."""
        from backend.models.video import Video
        from backend.models.creator import Creator
        db = SessionLocal()
        try:
            # Get creators who have TikTok videos with products
            creators = db.query(Creator).filter(
                Creator.platform == "tiktok",
            ).order_by(Creator.followers.desc()).limit(20).all()

            results = []
            for c in creators:
                results.append({
                    "user_id": str(c.id),
                    "username": c.username or "",
                    "nickname": c.display_name or c.username or "",
                    "avatar": c.avatar_url or "",
                    "category": "",
                    "followers": c.followers or 0,
                    "followers_formatted": _format_num(c.followers),
                    "likes": 0,
                    "sales_total": 0,
                    "sales_formatted": "0",
                    "gmv_total": 0,
                    "gmv_formatted": "R$ 0",
                    "ec_score": 0,
                    "video_count": c.total_videos or 0,
                    "product_count": 0,
                    "region": "BR",
                })

            return results
        except Exception:
            return []
        finally:
            db.close()

    # ─── Internal Builders ───────────────────────────────

    def _build_seller_ranking(self, rank_field: int = 1) -> list:
        """Build seller ranking from all cached product data."""
        db = SessionLocal()
        try:
            rows = db.query(EchoTikCache).filter(
                EchoTikCache.cache_type.in_(["product_ranking", "product_search"])
            ).all()

            sellers = {}
            for row in rows:
                products = json.loads(row.data_json)
                if not isinstance(products, list):
                    continue
                for p in products:
                    sid = p.get("seller_id", "")
                    if not sid:
                        continue
                    if sid not in sellers:
                        sellers[sid] = {
                            "seller_id": sid,
                            "name": p.get("seller_name", ""),
                            "logo": p.get("seller_logo", ""),
                            "rating": 0,
                            "category": p.get("category", ""),
                            "product_count": 0,
                            "sales_total": 0,
                            "gmv_total": 0,
                            "influencer_count": 0,
                            "video_count": 0,
                            "region": "BR",
                            "user_id": "",
                        }
                    s = sellers[sid]
                    s["product_count"] += 1
                    # Use seller-level data if available
                    seller_sales = p.get("seller_sales", 0) or 0
                    seller_gmv = p.get("seller_gmv", 0) or 0
                    if seller_sales > s["sales_total"]:
                        s["sales_total"] = int(seller_sales)
                    if seller_gmv > s["gmv_total"]:
                        s["gmv_total"] = seller_gmv
                    if not s["name"]:
                        s["name"] = p.get("seller_name", "")

            result = list(sellers.values())
            for s in result:
                s["sales_formatted"] = _format_num(s["sales_total"])
                s["gmv_formatted"] = _format_brl_num(s["gmv_total"])

            if rank_field == 2:
                result.sort(key=lambda x: x["gmv_total"], reverse=True)
            else:
                result.sort(key=lambda x: x["sales_total"], reverse=True)

            return result[:20]
        finally:
            db.close()

    def _build_seller_detail(self, seller_id: str) -> dict:
        """Build seller detail from cached product data."""
        db = SessionLocal()
        try:
            rows = db.query(EchoTikCache).filter(
                EchoTikCache.cache_type.in_(["product_ranking", "product_search"])
            ).all()

            seller_info = None
            products_count = 0
            total_sales = 0
            total_gmv = 0.0

            for row in rows:
                products = json.loads(row.data_json)
                if not isinstance(products, list):
                    continue
                for p in products:
                    if p.get("seller_id") == seller_id:
                        if not seller_info:
                            seller_info = {
                                "seller_id": seller_id,
                                "name": p.get("seller_name", ""),
                                "logo": p.get("seller_logo", ""),
                            }
                        products_count += 1
                        s_sales = p.get("seller_sales", 0) or 0
                        s_gmv = p.get("seller_gmv", 0) or 0
                        if s_sales > total_sales:
                            total_sales = int(s_sales)
                        if s_gmv > total_gmv:
                            total_gmv = s_gmv

            if not seller_info:
                raise ValueError("Loja nao encontrada no cache")

            return {
                **seller_info,
                "rating": 0,
                "category": "",
                "product_count": products_count,
                "sales_total": total_sales,
                "sales_formatted": _format_num(total_sales),
                "gmv_total": total_gmv,
                "gmv_formatted": _format_brl_num(total_gmv),
                "sales_1d": 0, "sales_7d": 0, "sales_30d": total_sales, "sales_90d": 0,
                "gmv_1d": 0, "gmv_7d": 0, "gmv_30d": total_gmv, "gmv_90d": 0,
                "gmv_1d_formatted": "R$ 0",
                "gmv_7d_formatted": "R$ 0",
                "gmv_30d_formatted": _format_brl_num(total_gmv),
                "avg_price": 0,
                "avg_price_formatted": "R$ 0",
                "influencer_count": 0,
                "video_count": 0,
                "region": "BR",
                "user_id": "",
                "total_products_crawled": products_count,
            }
        finally:
            db.close()

    def _find_seller_products(self, seller_id: str) -> list:
        """Find products by seller_id in cache."""
        db = SessionLocal()
        try:
            rows = db.query(EchoTikCache).filter(
                EchoTikCache.cache_type.in_(["product_ranking", "product_search", "product_detail"])
            ).all()

            products = []
            seen = set()
            for row in rows:
                data = json.loads(row.data_json)
                items = data if isinstance(data, list) else [data]
                for p in items:
                    pid = p.get("product_id", "")
                    if p.get("seller_id") == seller_id and pid not in seen:
                        seen.add(pid)
                        products.append(p)

            return products
        finally:
            db.close()

    def _search_sellers(self, keyword: str) -> list:
        """Search sellers by keyword in cached data."""
        db = SessionLocal()
        try:
            rows = db.query(EchoTikCache).filter(
                EchoTikCache.cache_type.in_(["product_ranking", "product_search", "seller_ranking"])
            ).all()

            sellers = {}
            kw_lower = keyword.lower()
            for row in rows:
                data = json.loads(row.data_json)
                items = data if isinstance(data, list) else [data]
                for p in items:
                    sname = (p.get("seller_name") or p.get("name", "")).lower()
                    sid = p.get("seller_id", "")
                    if sid and kw_lower in sname:
                        if sid not in sellers:
                            sellers[sid] = {
                                "seller_id": sid,
                                "name": p.get("seller_name") or p.get("name", ""),
                                "logo": p.get("seller_logo") or p.get("logo", ""),
                                "rating": 0,
                                "category": p.get("category", ""),
                                "product_count": 1,
                                "sales_total": int(p.get("seller_sales", 0) or p.get("sales_total", 0) or 0),
                                "gmv_total": float(p.get("seller_gmv", 0) or p.get("gmv_total", 0) or 0),
                                "influencer_count": 0,
                                "video_count": 0,
                                "region": "BR",
                                "user_id": "",
                            }
                        else:
                            sellers[sid]["product_count"] += 1

            result = list(sellers.values())
            for s in result:
                s["sales_formatted"] = _format_num(s["sales_total"])
                s["gmv_formatted"] = _format_brl_num(s["gmv_total"])

            result.sort(key=lambda x: x["sales_total"], reverse=True)
            return result
        finally:
            db.close()

    def _build_influencer_ranking(self, rank_field: int = 1, days: int = 30) -> list:
        """Build influencer ranking from scanner data (BR only, with enriched metrics)."""
        from backend.models.collection import VideoCollection  # noqa: F401 - needed for Video relationship
        from backend.models.creator import Creator
        from backend.models.video import Video
        from sqlalchemy import func, case, or_
        db = SessionLocal()
        try:
            cutoff = datetime.utcnow() - timedelta(days=days)
            br_filter = or_(Video.region == 'BR', Video.language == 'pt')

            results = db.query(
                Creator.id,
                Creator.username,
                Creator.display_name,
                Creator.avatar_url,
                Creator.followers,
                Creator.avg_engagement,
                Creator.size_tier,
                func.count(Video.id).label('video_count_br'),
                func.coalesce(func.sum(Video.views), 0).label('total_views'),
                func.coalesce(func.sum(Video.likes), 0).label('total_likes'),
                func.count(case((Video.viral_score > 60, 1))).label('viral_count'),
                func.count(case((Video.shop_product_url != '', 1))).label('shop_video_count'),
                func.coalesce(func.avg(Video.engagement_rate), 0).label('avg_eng'),
            ).join(Video, Video.creator_id == Creator.id).filter(
                Creator.platform == 'tiktok',
                Video.platform == 'tiktok',
                br_filter,
                Video.published_at >= cutoff,
            ).group_by(Creator.id).having(
                func.count(Video.id) > 0
            )

            # Sort by rank_field: 1=views, 2=viral_count, 3=engagement
            if rank_field == 2:
                results = results.order_by(func.count(case((Video.viral_score > 60, 1))).desc())
            elif rank_field == 3:
                results = results.order_by(func.avg(Video.engagement_rate).desc())
            else:
                results = results.order_by(func.sum(Video.views).desc())

            rows = results.limit(20).all()

            return [{
                "user_id": str(r.id),
                "username": r.username or "",
                "nickname": r.display_name or r.username or "",
                "avatar": r.avatar_url or "",
                "category": r.size_tier or "",
                "followers": r.followers or 0,
                "followers_formatted": _format_num(r.followers),
                "likes": int(r.total_likes),
                "total_views": int(r.total_views),
                "total_views_formatted": _format_num(r.total_views),
                "viral_count": int(r.viral_count),
                "shop_video_count": int(r.shop_video_count),
                "sales_total": int(r.shop_video_count),
                "sales_formatted": str(int(r.shop_video_count)),
                "gmv_total": 0,
                "gmv_formatted": "R$ 0",
                "ec_score": round(float(r.avg_eng), 1) if r.avg_eng else 0,
                "avg_engagement": round(float(r.avg_eng), 1) if r.avg_eng else 0,
                "video_count": int(r.video_count_br),
                "video_count_br": int(r.video_count_br),
                "product_count": int(r.shop_video_count),
                "region": "BR",
            } for r in rows]
        except Exception as e:
            logger.warning(f"Error building influencer ranking: {e}")
            return []
        finally:
            db.close()

    def _build_video_ranking(self, rank_field: int = 1, days: int = 30) -> list:
        """Build video ranking from scanner data (BR only, with time filter)."""
        from backend.models.collection import VideoCollection  # noqa: F401
        from backend.models.video import Video
        from sqlalchemy import or_
        db = SessionLocal()
        try:
            cutoff = datetime.utcnow() - timedelta(days=days)
            br_filter = or_(Video.region == 'BR', Video.language == 'pt')

            # Sort by rank_field: 1=views, 2=viral_score, 3=engagement
            if rank_field == 2:
                order = Video.viral_score.desc()
            elif rank_field == 3:
                order = Video.engagement_rate.desc()
            else:
                order = Video.views.desc()

            videos = db.query(Video).filter(
                Video.platform == "tiktok",
                br_filter,
                Video.published_at >= cutoff,
            ).order_by(order).limit(20).all()

            results = []
            for v in videos:
                creator_name = ""
                creator_avatar = ""
                if v.creator:
                    creator_name = v.creator.username or ""
                    creator_avatar = v.creator.avatar_url or ""
                has_shop = bool(v.shop_product_url and v.shop_product_url.strip())
                results.append({
                    "video_id": v.external_id or str(v.id),
                    "db_id": v.id,
                    "description": v.title or v.description or "",
                    "thumbnail": f"/api/thumbnail/{v.id}",
                    "duration": v.duration or 0,
                    "views": v.views or 0,
                    "views_formatted": _format_num(v.views),
                    "likes": v.likes or 0,
                    "comments": v.comments or 0,
                    "shares": v.shares or 0,
                    "viral_score": round(v.viral_score or 0, 1),
                    "engagement_rate": round(v.engagement_rate or 0, 2),
                    "has_shop": has_shop,
                    "shop_product_url": v.shop_product_url or "",
                    "sales": 0,
                    "sales_formatted": "0",
                    "gmv": 0,
                    "gmv_formatted": "R$ 0",
                    "creator_username": creator_name,
                    "creator_nickname": creator_name,
                    "creator_avatar": creator_avatar,
                    "user_id": str(v.creator_id) if v.creator_id else "",
                    "product_id": "",
                    "hashtags": ",".join(v.hashtags) if isinstance(v.hashtags, list) else "",
                    "created_at": v.published_at.isoformat() if v.published_at else "",
                    "region": "BR",
                })
            return results
        except Exception as e:
            logger.warning(f"Error building video ranking: {e}")
            return []
        finally:
            db.close()

    def _build_daily_trends(self, days: int = 30) -> list:
        """Build daily trend data from BR videos for chart display."""
        from backend.models.collection import VideoCollection  # noqa: F401
        from backend.models.video import Video
        from sqlalchemy import func, case, or_
        db = SessionLocal()
        try:
            cutoff = datetime.utcnow() - timedelta(days=days)
            br_filter = or_(Video.region == 'BR', Video.language == 'pt')
            date_col = func.date(Video.published_at)

            rows = db.query(
                date_col.label('date'),
                func.count(Video.id).label('video_count'),
                func.coalesce(func.sum(Video.views), 0).label('total_views'),
                func.coalesce(func.avg(Video.viral_score), 0).label('avg_viral_score'),
                func.coalesce(func.avg(Video.engagement_rate), 0).label('avg_engagement'),
                func.count(case((Video.shop_product_url != '', 1))).label('shop_video_count'),
                func.coalesce(func.sum(Video.likes), 0).label('total_likes'),
            ).filter(
                Video.platform == 'tiktok',
                br_filter,
                Video.published_at >= cutoff,
                Video.published_at.isnot(None),
            ).group_by(date_col).order_by(date_col.asc()).all()

            return [{
                "date": str(r.date),
                "video_count": int(r.video_count),
                "total_views": int(r.total_views),
                "total_views_formatted": _format_num(r.total_views),
                "avg_viral_score": round(float(r.avg_viral_score), 1),
                "avg_engagement": round(float(r.avg_engagement), 2),
                "shop_video_count": int(r.shop_video_count),
                "total_likes": int(r.total_likes),
            } for r in rows]
        except Exception as e:
            logger.warning(f"Error building daily trends: {e}")
            return []
        finally:
            db.close()

    # ─── Sync & Maintenance ──────────────────────────────

    def sync_rankings(self, keywords: list = None) -> int:
        """Sync TikTok Shop data from Apify.

        Searches popular keywords and caches results.
        Also takes snapshots for monitored products.
        """
        if keywords is None:
            keywords = ["mais vendido", "tendencia", "moda", "beleza", "eletronico"]

        synced = 0
        for kw in keywords:
            try:
                products = self.product_search(kw, page_size=10)
                if products:
                    synced += 1
                    logger.info(f"Synced {len(products)} products for keyword '{kw}'")

                    # Cache seller data from products
                    self._extract_and_cache_sellers(products)
            except Exception as e:
                logger.warning(f"Sync failed for keyword '{kw}': {e}")

        # Take snapshots for monitored products
        try:
            snap_count = self.take_snapshots_for_monitored()
            logger.info(f"Took {snap_count} product snapshots")
        except Exception as e:
            logger.warning(f"Snapshot sync failed: {e}")

        return synced

    def _extract_and_cache_sellers(self, products: list):
        """Extract seller data from products and cache as seller ranking."""
        sellers = {}
        for p in products:
            sid = p.get("seller_id", "")
            if not sid:
                continue
            if sid not in sellers or (p.get("seller_sales", 0) or 0) > (sellers[sid].get("sales_total", 0)):
                sellers[sid] = {
                    "seller_id": sid,
                    "name": p.get("seller_name", ""),
                    "logo": p.get("seller_logo", ""),
                    "rating": 0,
                    "category": p.get("category", ""),
                    "product_count": 1,
                    "sales_total": int(p.get("seller_sales", 0) or 0),
                    "sales_formatted": _format_num(p.get("seller_sales", 0)),
                    "gmv_total": float(p.get("seller_gmv", 0) or 0),
                    "gmv_formatted": _format_brl_num(p.get("seller_gmv", 0)),
                    "influencer_count": 0,
                    "video_count": 0,
                    "region": "BR",
                    "user_id": "",
                }

        if sellers:
            seller_list = sorted(sellers.values(), key=lambda x: x["sales_total"], reverse=True)
            self._save_to_cache_direct("seller_ranking", "seller_ranking:rf1:p1:ps10:cat_", seller_list, 24)

    @staticmethod
    def clear_expired():
        """Delete expired cache entries older than 48h."""
        db = SessionLocal()
        try:
            deleted = db.query(EchoTikCache).filter(
                EchoTikCache.expires_at < datetime.utcnow() - timedelta(hours=48)
            ).delete()
            db.commit()
            if deleted:
                logger.info(f"Cleared {deleted} expired cache entries")
        finally:
            db.close()

    @staticmethod
    def get_cache_stats() -> dict:
        """Get cache statistics."""
        db = SessionLocal()
        try:
            total = db.query(EchoTikCache).count()
            fresh = db.query(EchoTikCache).filter(
                EchoTikCache.expires_at > datetime.utcnow()
            ).count()
            snapshots = db.query(ProductSnapshot).count()
            monitored = db.query(ProductSnapshot.product_id).distinct().count()
            return {
                "total_entries": total,
                "fresh": fresh,
                "expired": total - fresh,
                "snapshots": snapshots,
                "monitored_products": monitored,
            }
        finally:
            db.close()


# ─── Utility formatters ──────────────────────────────────

def _format_num(n) -> str:
    try:
        n = float(n or 0)
        if n >= 1_000_000:
            return f"{n/1_000_000:.1f}M"
        if n >= 1_000:
            return f"{n/1_000:.1f}K"
        return str(int(n))
    except (ValueError, TypeError):
        return "0"


def _format_brl_num(value) -> str:
    try:
        v = float(value or 0)
        if v >= 1_000_000:
            return f"R$ {v/1_000_000:.1f}M"
        if v >= 1_000:
            return f"R$ {v/1_000:.1f}K"
        return f"R$ {v:.2f}"
    except (ValueError, TypeError):
        return "R$ 0"
