hiffi-artist-intelligence/collectors/browser_collector.py

"""
collectors/browser_collector.py
Multi-platform Signal Collector:
- Spotify: Direct static fetch via Kworb/Songstats mirror (instant, zero JavaScript required)
- Instagram: Headless Playwright meta-tag extraction
"""

import re
import requests
from typing import Dict, Any, Optional
from playwright.sync_api import sync_playwright, Page


def parse_shorthand_number(val_str: Optional[str]) -> Optional[int]:
    """Converts strings like '18.6M', '500K', '70,053,200' into standard integers."""
    if not val_str:
        return None
    val_str = str(val_str).strip().upper().replace(",", "")
    try:
        if val_str.endswith("M"):
            return int(float(val_str[:-1]) * 1_000_000)
        elif val_str.endswith("K"):
            return int(float(val_str[:-1]) * 1_000)
        elif val_str.endswith("B"):
            return int(float(val_str[:-1]) * 1_000_000_000)
        return int(float(val_str))
    except (ValueError, TypeError):
        return None


class MultiPlatformSignalCollector:
    def __init__(self, headless: bool = True):
        self.headless = headless
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"
        })

    def fetch_spotify_listeners(self, spotify_url: str) -> Optional[int]:
        """
        Fetches Spotify monthly listeners directly via public static mirror.
        Bypasses client-side JavaScript, cookie walls, and 403 API gates.
        """
        artist_id = spotify_url.split("?")[0].rstrip("/").split("/")[-1]
        if not artist_id:
            return None

        # Strategy 1: Kworb static artist index
        try:
            url = f"https://kworb.net/spotify/artist/{artist_id}.html"
            res = self.session.get(url, timeout=10)
            if res.status_code == 200:
                # Look for monthly listeners in the table/header
                m = re.search(r"([\d,]+)\s+monthly listeners", res.text, re.IGNORECASE)
                if m:
                    return parse_shorthand_number(m.group(1))
                
                # Check table cells with large numbers
                m2 = re.search(r"Monthly Listeners</td><td>([\d,]+)</td>", res.text, re.IGNORECASE)
                if m2:
                    return parse_shorthand_number(m2.group(1))
        except Exception:
            pass

        # Strategy 2: Direct public embed oEmbed query
        try:
            oembed_url = f"https://open.spotify.com/oembed?url=https://open.spotify.com/artist/{artist_id}"
            res = self.session.get(oembed_url, timeout=10)
            if res.status_code == 200:
                pass # Verified artist exists
        except Exception:
            pass

        return None

    def fetch_instagram_followers(self, page: Page, ig_url: str) -> Optional[int]:
        """Extracts follower count via meta tags on Instagram."""
        try:
            page.goto(ig_url, timeout=25000, wait_until="domcontentloaded")
            page.wait_for_timeout(2000)

            content = page.content()

            # Meta description check
            meta_match = re.search(
                r'content="([0-9.,KMBkmb]+)\s+Followers', content, re.IGNORECASE
            )
            if meta_match:
                return parse_shorthand_number(meta_match.group(1))

            # OpenGraph check
            og_match = re.search(
                r'property="og:description"\s+content="([0-9.,KMBkmb]+)\s+Followers',
                content,
                re.IGNORECASE,
            )
            if og_match:
                return parse_shorthand_number(og_match.group(1))

            # Rendered text fallback
            body_text = page.inner_text("body")
            m = re.search(r"([\d.,]+[KMBkmb]?)\s+followers", body_text, re.IGNORECASE)
            if m:
                return parse_shorthand_number(m.group(1))

        except Exception:
            pass
        return None

    def collect_artist(self, artist_info: Dict[str, str], page: Page) -> Dict[str, Any]:
        """Collects metrics for a single artist from both platforms."""
        name = artist_info.get("name", "Unknown")
        spotify_url = artist_info.get("spotify_url")
        ig_url = artist_info.get("ig_url")

        spotify_listeners = self.fetch_spotify_listeners(spotify_url) if spotify_url else None
        ig_followers = self.fetch_instagram_followers(page, ig_url) if ig_url else None

        return {
            "name": name,
            "spotify_monthly_listeners": spotify_listeners,
            "instagram_followers": ig_followers,
            "spotify_status": "SUCCESS" if spotify_listeners is not None else "FAILED",
            "ig_status": "SUCCESS" if ig_followers is not None else "FAILED",
        }