hiffi-artist-intelligence/scripts/audit_youtube_channels.py

"""
scripts/audit_youtube_channels.py
Audits YouTube URLs across all 500 cohort artists in hiffi_catalog.db.
Extracts handles/channel IDs, checks link formatting, and flags potential mismatches.
"""

import sqlite3
import re
import csv
from pathlib import Path
from urllib.parse import urlparse

BASE_DIR = Path(__file__).resolve().parent.parent
DB_PATH = BASE_DIR / "data" / "hiffi_catalog.db"
OUTPUT_CSV = BASE_DIR / "data" / "youtube_audit_report.csv"

def extract_channel_identifier(url: str) -> tuple[str, str]:
    """Returns (identifier_type, identifier_value) from a YouTube URL."""
    if not url or not isinstance(url, str):
        return ("MISSING", "")
    
    url = url.strip()
    parsed = urlparse(url)
    path = parsed.path.strip("/")
    
    if not path:
        return ("INVALID", url)
    
    parts = path.split("/")
    
    # Handle /@username
    if parts[0].startswith("@"):
        return ("HANDLE", parts[0])
    
    # Handle /channel/UC...
    if len(parts) >= 2 and parts[0] == "channel":
        return ("CHANNEL_ID", parts[1])
    
    # Handle /c/custom_name or /user/legacy_name
    if len(parts) >= 2 and parts[0] in ["c", "user"]:
        return (f"LEGACY_{parts[0].upper()}", parts[1])
        
    return ("CUSTOM_PATH", path)

def main():
    if not DB_PATH.exists():
        print(f"[-] Database not found at: {DB_PATH}")
        return

    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    cursor.execute("""
        SELECT artist_id, artist_name, youtube_url, spotify_url
        FROM artists
        ORDER BY artist_id ASC
    """)
    rows = cursor.fetchall()
    conn.close()

    total = len(rows)
    print("\n" + "=" * 65)
    print(f"      AUDITING YOUTUBE URLS FOR {total} COHORT ARTISTS")
    print("=" * 65 + "\n")

    report_data = []
    missing_count = 0
    handle_count = 0
    channel_id_count = 0
    other_count = 0

    for r in rows:
        artist_id = r["artist_id"]
        artist_name = r["artist_name"]
        yt_url = r["youtube_url"] or ""

        id_type, id_val = extract_channel_identifier(yt_url)
        
        # Heuristic check for possible mismatches:
        # e.g., handle letters differ dramatically from artist name
        flag = "OK"
        if id_type == "MISSING":
            flag = "MISSING_URL"
            missing_count += 1
        elif id_type == "INVALID":
            flag = "INVALID_URL"
            other_count += 1
        else:
            if id_type == "HANDLE":
                handle_count += 1
            elif id_type == "CHANNEL_ID":
                channel_id_count += 1
            else:
                other_count += 1

            # Simple name token overlap check
            norm_name = re.sub(r"[^a-zA-Z0-9]", "", artist_name.lower())
            norm_id = re.sub(r"[^a-zA-Z0-9]", "", id_val.lower())
            
            # If the artist name is totally absent from handle/path
            if len(norm_name) > 3 and norm_name not in norm_id and norm_id not in norm_name:
                flag = "CHECK_NAME_MISMATCH"

        report_data.append({
            "artist_id": artist_id,
            "artist_name": artist_name,
            "youtube_url": yt_url,
            "identifier_type": id_type,
            "identifier_value": id_val,
            "audit_status": flag
        })

    # Save to CSV
    OUTPUT_CSV.parent.mkdir(parents=True, exist_ok=True)
    with open(OUTPUT_CSV, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=[
            "artist_id", "artist_name", "youtube_url", 
            "identifier_type", "identifier_value", "audit_status"
        ])
        writer.writeheader()
        writer.writerows(report_data)

    print(f"[+] Audit summary saved to: {OUTPUT_CSV.name}")
    print(f"    - Total Artists:     {total}")
    print(f"    - Handles (@...):    {handle_count}")
    print(f"    - Channel IDs:       {channel_id_count}")
    print(f"    - Missing URLs:      {missing_count}")
    print(f"    - Flagged / Review:  {sum(1 for r in report_data if r['audit_status'] != 'OK')}")
    print("\n" + "=" * 65 + "\n")

if __name__ == "__main__":
    main()