diff --git a/api.py b/api.py index 6973fe1..9bc72f0 100644 --- a/api.py +++ b/api.py @@ -10,10 +10,12 @@ import pandas as pd from fastapi import FastAPI, Header, HTTPException, Query, Security, status from fastapi.middleware.cors import CORSMiddleware +from routes.sales import sales_router # Production auth secret required for Go backend header integration AUTH_SECRET = "hiffi_live_secret_key_prod_2026" + def verify_auth_secret(x_auth_secret: Optional[str] = Header(None)) -> str: """ Validates the incoming X-Auth-Secret header for backend-to-backend communication. @@ -57,6 +59,9 @@ app.add_middleware( allow_headers=["*"], ) +# Mount decoupled commercial sales router +app.include_router(sales_router) + def load_dataset() -> pd.DataFrame: """Loads artist dataset with fallback paths.""" @@ -122,6 +127,7 @@ def list_artists( ) -> List[Dict[str, Any]]: """ Returns ranked artist list respecting data governance scopes. + Strictly locked to the 500-artist telemetry cohort. Zero commercial sales leakage. """ df = load_dataset() if df.empty: diff --git a/core/sales_ranking_engine.py b/core/sales_ranking_engine.py new file mode 100644 index 0000000..108af90 --- /dev/null +++ b/core/sales_ranking_engine.py @@ -0,0 +1,51 @@ +""" +core/sales_ranking_engine.py +Calculates dedicated commercial sales rank across all records in commercial_sales. +Completely isolated from 500-cohort telemetry XP ranking logic. +""" + +import sqlite3 +import time + +DB_PATH = "data/hiffi_catalog.db" + +def recalculate_17k_sales_ranks(): + conn = sqlite3.connect(DB_PATH) + cur = conn.cursor() + + print("[*] Calculating global commercial sales rankings...") + start_time = time.time() + + # Join with telemetry_metrics to cleanly break ties for 0-sales artists + rank_query = """ + SELECT + cs.sales_id, + cs.artist_id, + cs.audited_sales_units, + COALESCE(tm.total_views, 0) as total_views, + COALESCE(tm.spotify_monthly_listeners, 0) as listeners + FROM commercial_sales cs + LEFT JOIN telemetry_metrics tm ON cs.artist_id = tm.artist_id + ORDER BY + cs.audited_sales_units DESC, + total_views DESC, + listeners DESC, + cs.artist_id ASC + """ + rows = cur.execute(rank_query).fetchall() + total = len(rows) + + updates = [] + for rank, row in enumerate(rows, start=1): + sales_id = row[0] + updates.append((rank, sales_id)) + + cur.executemany("UPDATE commercial_sales SET sales_rank = ? WHERE sales_id = ?", updates) + conn.commit() + conn.close() + + elapsed = round(time.time() - start_time, 2) + print(f"✓ Recalculated sales ranks for {total} artists in {elapsed}s.") + +if __name__ == "__main__": + recalculate_17k_sales_ranks() diff --git a/data/hiffi_catalog.db b/data/hiffi_catalog.db index 2de6dd8..b09d761 100644 Binary files a/data/hiffi_catalog.db and b/data/hiffi_catalog.db differ diff --git a/routes/__init__.py b/routes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/routes/sales.py b/routes/sales.py new file mode 100644 index 0000000..544b3ce --- /dev/null +++ b/routes/sales.py @@ -0,0 +1,137 @@ +""" +routes/sales.py +Dedicated modular APIRouter serving commercial sales records and catalog sales rankings. +Completely decoupled from the live 500-cohort telemetry endpoint (/artists). +""" + +import sqlite3 +from typing import Any, Dict, List, Optional +from fastapi import APIRouter, Depends, HTTPException, Query, Header, status + +DB_PATH = "data/hiffi_catalog.db" +AUTH_SECRET = "hiffi_live_secret_key_prod_2026" + +sales_router = APIRouter(prefix="/sales", tags=["Commercial Sales"]) + + +def verify_sales_auth(x_auth_secret: Optional[str] = Header(None)) -> str: + """Verifies incoming X-Auth-Secret header for protected sales endpoints.""" + if x_auth_secret != AUTH_SECRET: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Invalid or missing X-Auth-Secret header" + ) + return x_auth_secret + + +def get_db(): + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + try: + yield conn + finally: + conn.close() + + +@sales_router.get("/summary", dependencies=[Depends(verify_sales_auth)]) +def get_sales_summary(db: sqlite3.Connection = Depends(get_db)) -> Dict[str, Any]: + """Returns aggregate commercial sales totals across all ingested catalog records.""" + cur = db.cursor() + summary_sql = """ + SELECT + COUNT(*) as total_catalog_artists, + SUM(CASE WHEN audited_sales_units > 0 THEN 1 ELSE 0 END) as commercial_active_artists, + COALESCE(SUM(audited_sales_units), 0) as total_audited_units, + COALESCE(SUM(pure_album_sales), 0) as total_pure_albums, + COALESCE(SUM(digital_singles_sales), 0) as total_digital_singles, + COALESCE(SUM(physical_singles_sales), 0) as total_physical_singles, + COALESCE(SUM(streaming_eas), 0) as total_streaming_eas, + COALESCE(SUM(awards_count), 0) as total_certifications + FROM commercial_sales + """ + row = cur.execute(summary_sql).fetchone() + return dict(row) + + +@sales_router.get("/artists", dependencies=[Depends(verify_sales_auth)]) +def list_sales_artists( + limit: int = Query(50, ge=1, le=500, description="Records per page"), + offset: int = Query(0, ge=0, description="Offset for pagination"), + search: Optional[str] = Query(None, description="Search artist by name"), + provenance: Optional[str] = Query(None, description="Filter by provenance flag"), + min_sales: Optional[float] = Query(None, description="Minimum audited sales units"), + db: sqlite3.Connection = Depends(get_db) +) -> List[Dict[str, Any]]: + """Returns paginated catalog commercial sales rankings ordered strictly by sales_rank.""" + cur = db.cursor() + base_sql = """ + SELECT + cs.sales_rank, + cs.artist_id, + a.artist_name, + a.country_origin, + a.image_url, + cs.audited_sales_units, + cs.pure_album_sales, + cs.digital_singles_sales, + cs.physical_singles_sales, + cs.streaming_eas, + cs.awards_count, + cs.provenance_flag + FROM commercial_sales cs + JOIN artists a ON cs.artist_id = a.artist_id + WHERE 1=1 + """ + params = [] + if search: + base_sql += " AND a.artist_name LIKE ?" + params.append(f"%{search}%") + if provenance: + base_sql += " AND cs.provenance_flag = ?" + params.append(provenance) + if min_sales is not None: + base_sql += " AND cs.audited_sales_units >= ?" + params.append(min_sales) + + base_sql += " ORDER BY cs.sales_rank ASC LIMIT ? OFFSET ?" + params.extend([limit, offset]) + + rows = cur.execute(base_sql, params).fetchall() + return [dict(r) for r in rows] + + +@sales_router.get("/artists/{artist_id}", dependencies=[Depends(verify_sales_auth)]) +def get_artist_sales_dossier( + artist_id: str, + db: sqlite3.Connection = Depends(get_db) +) -> Dict[str, Any]: + """Returns commercial sales dossier for a single catalog artist.""" + cur = db.cursor() + sql = """ + SELECT + cs.sales_rank, + cs.artist_id, + a.artist_name, + a.country_origin, + a.image_url, + a.youtube_url, + a.spotify_url, + a.instagram_url, + cs.audited_sales_units, + cs.pure_album_sales, + cs.digital_singles_sales, + cs.physical_singles_sales, + cs.streaming_eas, + cs.awards_count, + cs.provenance_flag + FROM commercial_sales cs + JOIN artists a ON cs.artist_id = a.artist_id + WHERE cs.artist_id = ? + """ + row = cur.execute(sql, (artist_id,)).fetchone() + if not row: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Commercial sales record for artist ID '{artist_id}' not found." + ) + return dict(row) diff --git a/scripts/fast_sales_ingestion_17k.py b/scripts/fast_sales_ingestion_17k.py new file mode 100644 index 0000000..656e0fc --- /dev/null +++ b/scripts/fast_sales_ingestion_17k.py @@ -0,0 +1,217 @@ +""" +scripts/fast_sales_ingestion_17k.py +High-throughput multithreaded commercial sales ingestion. +Uses ThreadPoolExecutor (8-10 workers) to ingest 2,500-3,500 artists per hour. +Safe SQLite transactions via threading Lock. +""" + +import sqlite3 +import json +import time +import re +import urllib.parse +import urllib.request +from concurrent.futures import ThreadPoolExecutor, as_completed +import threading + +DB_PATH = "data/hiffi_catalog.db" +WIKI_API_URL = "https://en.wikipedia.org/w/api.php" +HEADERS = {"User-Agent": "HIFFI-FastScraper/1.0 (sri.lekha@superlabs.co)"} +MAX_WORKERS = 8 +db_lock = threading.Lock() + +def clean_artist_query(name: str) -> str: + clean = re.sub(r'[,$\'\"]', ' ', name) + return re.sub(r'\s+', ' ', clean).strip() + +def search_best_discography_title(artist_name: str) -> str: + clean_name = clean_artist_query(artist_name) + queries = [ + f"{artist_name} albums discography", + f"{clean_name} discography", + f"{artist_name} discography", + artist_name + ] + for q in queries: + params = { + "action": "opensearch", + "search": q, + "limit": "4", + "namespace": "0", + "format": "json" + } + url = f"{WIKI_API_URL}?{urllib.parse.urlencode(params)}" + try: + req = urllib.request.Request(url, headers=HEADERS) + with urllib.request.urlopen(req, timeout=5) as resp: + data = json.loads(resp.read().decode("utf-8")) + titles = data[1] if len(data) > 1 else [] + for t in titles: + if "albums discography" in t.lower(): + return t + for t in titles: + if "discography" in t.lower(): + return t + if titles and (q == artist_name or q == clean_name): + return titles[0] + except Exception: + continue + return f"{artist_name} discography" + +def fetch_wiki_content(exact_title: str) -> str: + params = { + "action": "query", + "prop": "revisions", + "rvprop": "content", + "rvslots": "main", + "titles": exact_title, + "format": "json", + "redirects": "1" + } + url = f"{WIKI_API_URL}?{urllib.parse.urlencode(params)}" + req = urllib.request.Request(url, headers=HEADERS) + try: + with urllib.request.urlopen(req, timeout=6) as resp: + data = json.loads(resp.read().decode("utf-8")) + pages = data.get("query", {}).get("pages", {}) + for page_id, page_data in pages.items(): + if page_id != "-1": + slots = page_data.get("revisions", [{}])[0].get("slots", {}) + return slots.get("main", {}).get("*", "") + except Exception: + pass + return "" + +def clean_wiki_markup(text: str) -> str: + text = re.sub(r']*>.*?', '', text, flags=re.DOTALL | re.IGNORECASE) + text = re.sub(r']*/>', '', text, flags=re.IGNORECASE) + text = re.sub(r'\[\[(?:[^|\]]*\|)?([^\]]+)\]\]', r'\1', text) + return text + +def parse_studio_sales(wikitext: str): + clean_text = clean_wiki_markup(wikitext) + match = re.search( + r'={2,4}\s*(?:Studio albums|Commercial mixtapes|Albums)\s*={2,4}.*?(?===={2,4}|\Z)', + clean_text, + re.DOTALL | re.IGNORECASE + ) + album_section = match.group(0) if match else clean_text + + pure_album_sales = 0.0 + awards_count = 0 + + sales_matches = re.findall( + r'(?:US:\s*|sales:\s*|US sales:\s*)([0-9]{1,3}(?:,[0-9]{3})+)', + album_section, + re.IGNORECASE + ) + seen_sales = set() + for s in sales_matches: + num = float(s.replace(",", "")) + if 40_000 <= num <= 25_000_000 and num not in seen_sales: + pure_album_sales += num + seen_sales.add(num) + + multi_plat = re.findall(r'RIAA[:\s|=]*(\d+)[\xd7x\s]*Platinum', album_section, re.IGNORECASE) + for m in multi_plat: + mult = float(m) + pure_album_sales += mult * 1_000_000 + awards_count += int(mult) + + cleaned_multi = re.sub(r'\d+[\xd7x\s]*Platinum', '', album_section, flags=re.IGNORECASE) + single_plat = len(re.findall(r'RIAA[:\s|=]*Platinum', cleaned_multi, re.IGNORECASE)) + pure_album_sales += single_plat * 1_000_000 + awards_count += single_plat + + gold = len(re.findall(r'RIAA[:\s|=]*Gold', album_section, re.IGNORECASE)) + pure_album_sales += gold * 500_000 + awards_count += gold + + if pure_album_sales == 0.0 and awards_count == 0: + general_multi = re.findall(r'(\d+)[\xd7x]\s*Platinum', album_section, re.IGNORECASE) + for m in general_multi[:10]: + mult = float(m) + pure_album_sales += mult * 1_000_000 + awards_count += int(mult) + + return pure_album_sales, awards_count + +def process_single_artist(artist_row): + aid = artist_row["artist_id"] + name = artist_row["artist_name"] + title = search_best_discography_title(name) + wikitext = fetch_wiki_content(title) if title else "" + if wikitext: + sales_units, awards = parse_studio_sales(wikitext) + prov = "WIKI_DISCOGRAPHY_AUDIT" if sales_units > 0 else "UNAUDITED" + else: + sales_units, awards = 0.0, 0 + prov = "UNAUDITED" + return (aid, sales_units, awards, prov, 99999, sales_units, 0.0, 0.0, 0.0) + +def run_fast_ingestion(limit=3000): + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + cur = conn.cursor() + + rows = cur.execute(""" + SELECT a.artist_id, a.artist_name + FROM artists a + LEFT JOIN commercial_sales cs ON a.artist_id = cs.artist_id + WHERE cs.sales_id IS NULL + LIMIT ? + """, (limit,)).fetchall() + + total = len(rows) + if total == 0: + print("✓ All catalog artists already ingested.") + conn.close() + return + + print(f"🚀 Starting high-speed ingestion for {total} artists with {MAX_WORKERS} concurrent threads...") + start_time = time.time() + + insert_sql = """ + INSERT INTO commercial_sales ( + artist_id, audited_sales_units, awards_count, provenance_flag, + sales_rank, pure_album_sales, physical_singles_sales, + digital_singles_sales, streaming_eas + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """ + + completed = 0 + buffer = [] + + with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor: + futures = {executor.submit(process_single_artist, dict(r)): r for r in rows} + for future in as_completed(futures): + try: + record = future.result() + buffer.append(record) + completed += 1 + + if len(buffer) >= 50: + with db_lock: + cur.executemany(insert_sql, buffer) + conn.commit() + elapsed = time.time() - start_time + rate = round(completed / elapsed, 1) + print(f" ⚡ Ingested {completed}/{total} | {rate} artists/sec | Elapsed: {int(elapsed)}s") + buffer.clear() + + except Exception as e: + pass + + if buffer: + with db_lock: + cur.executemany(insert_sql, buffer) + conn.commit() + + conn.close() + total_time = round(time.time() - start_time, 1) + print(f"✓ Fast batch complete: {completed} artists processed in {total_time}s.") + +if __name__ == "__main__": + import sys + count = int(sys.argv[1]) if len(sys.argv) > 1 else 3000 + run_fast_ingestion(count) diff --git a/scripts/resolve_missing_platform_urls.py b/scripts/resolve_missing_platform_urls.py index 2a97cc1..5f5c547 100644 --- a/scripts/resolve_missing_platform_urls.py +++ b/scripts/resolve_missing_platform_urls.py @@ -113,6 +113,7 @@ def main(): print(f"Valid Instagram URLs : {tally[3]} / 500") print("=" * 65) + conn.close() if __name__ == "__main__": diff --git a/scripts/run_sales_ingestion_17k.py b/scripts/run_sales_ingestion_17k.py new file mode 100644 index 0000000..cfc3c04 --- /dev/null +++ b/scripts/run_sales_ingestion_17k.py @@ -0,0 +1,215 @@ +""" +scripts/run_sales_ingestion_17k.py +Automated ingestion of pure commercial sales and certifications across the 17K catalog. +Directly updates `commercial_sales` in data/hiffi_catalog.db. +""" + +import sqlite3 +import json +import time +import re +import urllib.parse +import urllib.request +from typing import Dict, Any, Tuple + +DB_PATH = "data/hiffi_catalog.db" +WIKI_API_URL = "https://en.wikipedia.org/w/api.php" +HEADERS = {"User-Agent": "HIFFI-Artist-Intelligence/1.0 (sri.lekha@superlabs.co)"} + + +def clean_artist_query(name: str) -> str: + clean = re.sub(r'[,$\'\"]', ' ', name) + return re.sub(r'\s+', ' ', clean).strip() + + +def search_best_discography_title(artist_name: str) -> str: + clean_name = clean_artist_query(artist_name) + queries = [ + f"{artist_name} albums discography", + f"{clean_name} discography", + f"{artist_name} discography", + artist_name + ] + for q in queries: + params = { + "action": "opensearch", + "search": q, + "limit": "4", + "namespace": "0", + "format": "json" + } + url = f"{WIKI_API_URL}?{urllib.parse.urlencode(params)}" + try: + req = urllib.request.Request(url, headers=HEADERS) + with urllib.request.urlopen(req, timeout=6) as resp: + data = json.loads(resp.read().decode("utf-8")) + titles = data[1] if len(data) > 1 else [] + for t in titles: + if "albums discography" in t.lower(): + return t + for t in titles: + if "discography" in t.lower(): + return t + if titles and (q == artist_name or q == clean_name): + return titles[0] + except Exception: + continue + return f"{artist_name} discography" + + +def fetch_wiki_content(exact_title: str) -> str: + params = { + "action": "query", + "prop": "revisions", + "rvprop": "content", + "rvslots": "main", + "titles": exact_title, + "format": "json", + "redirects": "1" + } + url = f"{WIKI_API_URL}?{urllib.parse.urlencode(params)}" + req = urllib.request.Request(url, headers=HEADERS) + try: + with urllib.request.urlopen(req, timeout=8) as resp: + data = json.loads(resp.read().decode("utf-8")) + pages = data.get("query", {}).get("pages", {}) + for page_id, page_data in pages.items(): + if page_id != "-1": + slots = page_data.get("revisions", [{}])[0].get("slots", {}) + return slots.get("main", {}).get("*", "") + except Exception: + pass + return "" + + +def clean_wiki_markup(text: str) -> str: + text = re.sub(r']*>.*?', '', text, flags=re.DOTALL | re.IGNORECASE) + text = re.sub(r']*/>', '', text, flags=re.IGNORECASE) + text = re.sub(r'\[\[(?:[^|\]]*\|)?([^\]]+)\]\]', r'\1', text) + return text + + +def parse_studio_sales(wikitext: str) -> Tuple[float, int]: + clean_text = clean_wiki_markup(wikitext) + + # Matches == Studio albums ==, === Studio albums ===, or Commercial mixtapes + match = re.search( + r'={2,4}\s*(?:Studio albums|Commercial mixtapes|Albums)\s*={2,4}.*?(?===={2,4}|\Z)', + clean_text, + re.DOTALL | re.IGNORECASE + ) + album_section = match.group(0) if match else clean_text + + pure_album_sales = 0.0 + awards_count = 0 + + # 1. Reported US Sales Figures + sales_matches = re.findall( + r'(?:US:\s*|sales:\s*|US sales:\s*)([0-9]{1,3}(?:,[0-9]{3})+)', + album_section, + re.IGNORECASE + ) + seen_sales = set() + for s in sales_matches: + num = float(s.replace(",", "")) + if 40_000 <= num <= 25_000_000 and num not in seen_sales: + pure_album_sales += num + seen_sales.add(num) + + # 2. Multi-Platinum RIAA + multi_plat = re.findall(r'RIAA[:\s|=]*(\d+)[\xd7x\s]*Platinum', album_section, re.IGNORECASE) + for m in multi_plat: + mult = float(m) + pure_album_sales += mult * 1_000_000 + awards_count += int(mult) + + # 3. Single Platinum RIAA + cleaned_multi = re.sub(r'\d+[\xd7x\s]*Platinum', '', album_section, flags=re.IGNORECASE) + single_plat = len(re.findall(r'RIAA[:\s|=]*Platinum', cleaned_multi, re.IGNORECASE)) + pure_album_sales += single_plat * 1_000_000 + awards_count += single_plat + + # 4. Gold RIAA + gold = len(re.findall(r'RIAA[:\s|=]*Gold', album_section, re.IGNORECASE)) + pure_album_sales += gold * 500_000 + awards_count += gold + + # 5. General Platinum fallback if RIAA tag was omitted + if pure_album_sales == 0.0 and awards_count == 0: + general_multi = re.findall(r'(\d+)[\xd7x]\s*Platinum', album_section, re.IGNORECASE) + for m in general_multi[:10]: + mult = float(m) + pure_album_sales += mult * 1_000_000 + awards_count += int(mult) + + return pure_album_sales, awards_count + + +def ingest_batch(limit: int = 500): + conn = sqlite3.connect(DB_PATH) + conn.row_factory = sqlite3.Row + cur = conn.cursor() + + # Query only artists not yet in commercial_sales + unprocessed_query = """ + SELECT a.artist_id, a.artist_name + FROM artists a + LEFT JOIN commercial_sales cs ON a.artist_id = cs.artist_id + WHERE cs.sales_id IS NULL + LIMIT ? + """ + rows = cur.execute(unprocessed_query, (limit,)).fetchall() + total_to_process = len(rows) + + if total_to_process == 0: + print("✓ All 17,135 artists already exist in commercial_sales table.") + conn.close() + return + + print(f"[*] Commencing sales ingestion for next batch of {total_to_process} artists...") + + insert_sql = """ + INSERT INTO commercial_sales ( + artist_id, audited_sales_units, awards_count, provenance_flag, + sales_rank, pure_album_sales, physical_singles_sales, + digital_singles_sales, streaming_eas + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + """ + + processed_count = 0 + for row in rows: + aid = row["artist_id"] + name = row["artist_name"] + + title = search_best_discography_title(name) + wikitext = fetch_wiki_content(title) if title else "" + + if wikitext: + sales_units, awards = parse_studio_sales(wikitext) + prov = "WIKI_DISCOGRAPHY_AUDIT" if sales_units > 0 else "UNAUDITED" + else: + sales_units, awards = 0.0, 0 + prov = "UNAUDITED" + + # Temporary sales_rank placeholder (recalculated globally afterwards) + cur.execute(insert_sql, ( + aid, sales_units, awards, prov, 99999, + sales_units, 0.0, 0.0, 0.0 + )) + + processed_count += 1 + if processed_count % 25 == 0: + conn.commit() + print(f" -> Ingested {processed_count}/{total_to_process} records...") + + time.sleep(0.15) + + conn.commit() + conn.close() + print(f"✓ Batch completed. Processed {processed_count} artists.") + + +if __name__ == "__main__": + import sys + batch_size = int(sys.argv[1]) if len(sys.argv) > 1 else 100 + ingest_batch(batch_size) \ No newline at end of file