diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..79ac35f --- /dev/null +++ b/.env.example @@ -0,0 +1 @@ +AUTH_SECRET="your-auth-secret-here" diff --git a/api/main.py b/api/main.py new file mode 100644 index 0000000..39a5702 --- /dev/null +++ b/api/main.py @@ -0,0 +1,133 @@ +""" +api/main.py +REST API Service for HIFFI Hip-Hop 500 Multi-Platform Intelligence. +Secured with header-based auth secret (X-Auth-Secret). +""" + +import os +from pathlib import Path +import sys +from typing import Optional + +from dotenv import load_dotenv +from fastapi import FastAPI, Header, HTTPException, Query, Security, status +from fastapi.middleware.cors import CORSMiddleware +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parent.parent +if str(PROJECT_ROOT) not in sys.path: + sys.path.insert(0, str(PROJECT_ROOT)) + +load_dotenv(PROJECT_ROOT / ".env") + +AUTH_SECRET = os.getenv("AUTH_SECRET", "hiffi_live_secret_key_prod_2026").strip() +DATA_FILE = PROJECT_ROOT / "data" / "hiffi_multiplatform_500.csv" + +app = FastAPI( + title="HIFFI Artist Intelligence API", + version="1.0.0", + description="Programmatic query and ranking endpoints for the HIFFI Hip-Hop 500 cohort.", +) + +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + + +def verify_auth_secret(x_auth_secret: Optional[str] = Header(None)): + """Validates the auth secret passed in the request header.""" + if not x_auth_secret or x_auth_secret != AUTH_SECRET: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Unauthorized: Invalid or missing X-Auth-Secret header.", + ) + return x_auth_secret + + +def get_dataset() -> pd.DataFrame: + """Loads the current master cohort dataset.""" + if not DATA_FILE.exists(): + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Cohort dataset not found. Please run run_pipeline.py first.", + ) + return pd.read_csv(DATA_FILE) + + +@app.get("/health", tags=["System"]) +def health_check(): + """Public health probe.""" + return {"status": "HEALTHY", "dataset_exists": DATA_FILE.exists()} + + +@app.get("/api/v1/cohort", tags=["Rankings"]) +def get_cohort_rankings( + tier: Optional[int] = Query(None, ge=0, le=3, description="Filter by platform evidence tier (0-3)"), + limit: int = Query(50, ge=1, le=500, description="Max records to return"), + offset: int = Query(0, ge=0, description="Offset for pagination"), + auth: str = Security(verify_auth_secret), +): + """Fetches published cohort rankings with pagination and tier filtering.""" + df = get_dataset() + + if tier is not None and "platform_evidence_tier" in df.columns: + df = df[df["platform_evidence_tier"] == tier] + + if "hiffi_xp_published_rank" in df.columns: + df = df.sort_values(by="hiffi_xp_published_rank", ascending=True) + + paginated = df.iloc[offset : offset + limit] + + # Safe replacement of NaN/inf with None for valid JSON serialization + clean_df = paginated.replace({float('nan'): None, float('inf'): None, float('-inf'): None}) + records = clean_df.to_dict(orient="records") + + return { + "total": len(df), + "limit": limit, + "offset": offset, + "results": records, + } + +@app.get("/api/v1/artists/{artist_name}", tags=["Artists"]) +def get_artist_dossier( + artist_name: str, + auth: str = Security(verify_auth_secret), +): + """Retrieves full telemetry, evidence tiers, and ranks for a single artist.""" + df = get_dataset() + match = df[df["name"].str.strip().str.lower() == artist_name.strip().lower()] + + if match.empty: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail=f"Artist '{artist_name}' not found in cohort.", + ) + + row = match.iloc[0].where(pd.notna(match.iloc[0]), None).to_dict() + return {"status": "SUCCESS", "artist": row} + + +@app.post("/api/v1/pipeline/run", tags=["Pipeline"]) +def trigger_pipeline_run(auth: str = Security(verify_auth_secret)): + """Triggers deterministic re-scoring, anomaly detection, and publication gating.""" + from core.ranking_engine import HIFFIRankingEngine + + df = get_dataset() + engine = HIFFIRankingEngine() + + scored = engine.calculate_hiffi_xp(df) + flagged = engine.detect_metric_anomalies(scored) + gated, audit = engine.apply_publication_eligibility_gates(flagged, return_metadata=True) + + gated.to_csv(DATA_FILE, index=False) + + return { + "status": "SUCCESS", + "message": "Pipeline recalculated and dataset refreshed.", + "shortfall_audit": audit, + } \ No newline at end of file