hiffi-artist-intelligence/scripts/cron_weekly_ranking.py

import os
import sys
import sqlite3
import logging
from datetime import datetime, timezone
import pandas as pd

BASE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if BASE_DIR not in sys.path:
    sys.path.insert(0, BASE_DIR)

from core.ranking_engine import HIFFIRankingEngine

DB_PATH = os.path.join(BASE_DIR, "data", "hiffi_catalog.db")
CSV_PATH = os.path.join(BASE_DIR, "data", "hiffi_multiplatform_500.csv")
LOG_PATH = os.path.join(BASE_DIR, "cron_ranking.log")

logging.basicConfig(
    filename=LOG_PATH,
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)
logger = logging.getLogger("HIFFI_CRON")

def run_weekly_ranking_pipeline():
    start_time = datetime.now(timezone.utc)
    print("\n========================================================")
    print("  HIFFI AUTOMATED WEEKLY RANKING PIPELINE (TOP 500 ONLY)")
    print(f"  Run Timestamp (UTC): {start_time.isoformat()}")
    print("========================================================\n")
    logger.info("Starting automated 500-cohort weekly ranking pipeline...")

    if not os.path.exists(DB_PATH):
        err = f"Database not found at {DB_PATH}"
        logger.error(err)
        print(f"ERROR: {err}")
        return False

    if not os.path.exists(CSV_PATH):
        err = f"Cohort CSV not found at {CSV_PATH}"
        logger.error(err)
        print(f"ERROR: {err}")
        return False

    conn = sqlite3.connect(DB_PATH)
    cur = conn.cursor()

    try:
        # Read the verified 500 cohort IDs directly from the baseline CSV
        base_cohort_df = pd.read_csv(CSV_PATH)
        cohort_ids = [str(x) for x in base_cohort_df['artist_id'].dropna().tolist()][:500]
        placeholders = ",".join([f"'{aid}'" for aid in cohort_ids])

        # 1. Fetch telemetry strictly for the 500 cohort
        query = f"""
            WITH latest_telemetry AS (
                SELECT t.*, ROW_NUMBER() OVER (
                    PARTITION BY t.artist_id 
                    ORDER BY t.last_synced_at DESC, t.telemetry_id DESC
                ) as rn
                FROM telemetry_metrics t
                WHERE t.artist_id IN ({placeholders})
            )
            SELECT 
                a.artist_id,
                a.artist_name,
                a.country_origin,
                a.image_url,
                a.spotify_url,
                a.youtube_url,
                a.instagram_url,
                COALESCE(t.spotify_monthly_listeners, 0) as spotify_monthly_listeners,
                COALESCE(t.subscribers_count, 0) as subscriber_count,
                COALESCE(t.daily_view_velocity, 0) as daily_view_velocity,
                COALESCE(t.instagram_followers, 0) as instagram_followers,
                r.catalog_rank as previous_rank,
                r.catalog_intelligence_tier as current_tier
            FROM artists a
            LEFT JOIN latest_telemetry t ON a.artist_id = t.artist_id AND t.rn = 1
            LEFT JOIN rankings r ON a.artist_id = r.artist_id
            WHERE a.artist_id IN ({placeholders});
        """
        df = pd.read_sql_query(query, conn)
        total_artists = len(df)
        print(f"Loaded verified cohort: {total_artists} artists.")
        logger.info(f"Loaded {total_artists} cohort artists for recalculation.")

        # 2. Run Core Scoring Engine (45% SP / 35% YT / 20% IG)
        engine = HIFFIRankingEngine()
        ranked_df = engine.calculate_hiffi_xp(df)

        if 'hiffi_xp_score' in ranked_df.columns:
            score_col = 'hiffi_xp_score'
        elif 'score' in ranked_df.columns:
            score_col = 'score'
        elif 'production_hiffi_score' in ranked_df.columns:
            score_col = 'production_hiffi_score'
        else:
            numeric_cols = ranked_df.select_dtypes(include=['float64', 'float32']).columns
            score_col = numeric_cols[0]

        # Sort descending by score
        ranked_df = ranked_df.sort_values(by=score_col, ascending=False).reset_index(drop=True)
        ranked_df['new_rank'] = ranked_df.index + 1

        # Track rank deltas
        ranked_df['rank_delta'] = ranked_df.apply(
            lambda row: int(row['previous_rank'] - row['new_rank'])
            if pd.notna(row['previous_rank']) and row['previous_rank'] > 0
            else 0,
            axis=1
        )

        def compute_tier(rank_val):
            if rank_val <= 50:
                return "TIER_1_ICON"
            elif rank_val <= 200:
                return "TIER_2_MAINSTREAM"
            else:
                return "TIER_3_RISING"

        ranked_df['tier_rank'] = ranked_df['new_rank']
        ranked_df['catalog_intelligence_tier'] = ranked_df['new_rank'].apply(compute_tier)

        # 3. Update SQLite database rankings table for the 500 cohort
        print("Committing 500 cohort ranks to SQLite database...")
        update_data = []
        for _, row in ranked_df.iterrows():
            aid = str(row['artist_id'])
            c_rank = int(row['new_rank'])
            t_rank = int(row['tier_rank'])
            tier_str = str(row['catalog_intelligence_tier'])
            update_data.append((c_rank, t_rank, tier_str, aid))

        cur.executemany("""
            UPDATE rankings 
            SET catalog_rank = ?,
                tier_rank = ?,
                catalog_intelligence_tier = ?
            WHERE artist_id = ?;
        """, update_data)
        conn.commit()
        print(f"Updated database ranking records for {len(update_data)} artists.")

        # 4. Save updated 500 cohort CSV
        ranked_df['catalog_rank'] = ranked_df['new_rank']
        if 'name' not in ranked_df.columns and 'artist_name' in ranked_df.columns:
            ranked_df['name'] = ranked_df['artist_name']

        ranked_df.to_csv(CSV_PATH, index=False)
        print(f"Exported Top 500 cohort CSV to: {CSV_PATH}")

        # Telemetry summary
        print("\n--- TOP 5 RANKING SNAPSHOT ---")
        for _, r in ranked_df.head(5).iterrows():
            delta_str = f"+{r['rank_delta']}" if r['rank_delta'] > 0 else str(r['rank_delta'])
            print(f"#{r['new_rank']} {r.get('name', r.get('artist_name'))} | Score: {r[score_col]:.2f} | Tier: {r['catalog_intelligence_tier']} | Delta: {delta_str}")

        elapsed = (datetime.now(timezone.utc) - start_time).total_seconds()
        print(f"\nWeekly ranking pipeline completed successfully in {elapsed:.2f}s.")
        logger.info(f"Weekly ranking pipeline finished successfully in {elapsed:.2f}s.")
        return True

    except Exception as e:
        conn.rollback()
        logger.error(f"Weekly ranking run failed: {str(e)}", exc_info=True)
        print(f"\nERROR running ranking pipeline: {e}")
        return False
    finally:
        conn.close()

if __name__ == "__main__":
    success = run_weekly_ranking_pipeline()
    sys.exit(0 if success else 1)