feat: expose REST API with header-based X-Auth-Secret authentication

.env.example

@@ -0,0 +1 @@
+AUTH_SECRET="your-auth-secret-here"

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,
+    }