data.py

"""
Data layer for the browsable MV database app.

Loads data/clean/mv_database.parquet (one row per player-season, MV-matched)
into DuckDB and exposes filter / detail helpers. No scoring — pure browse.
"""

import os
import sys
import time
from pathlib import Path
from threading import Lock

import duckdb

_PARENT = Path(__file__).resolve().parent.parent
if str(_PARENT) not in sys.path:
    sys.path.insert(0, str(_PARENT))

APP_DIR = Path(__file__).parent
DATA_DIR = Path(os.environ.get('DATA_DIR', APP_DIR.parent / 'data' / 'clean')).resolve()
DB_PARQUET = DATA_DIR / 'mv_database.parquet'

DATASET_REPO = os.environ.get('DATASET_REPO_ID')
HF_TOKEN = os.environ.get('HF_TOKEN')
MAX_AGE_HOURS = float(os.environ.get('DATA_MAX_AGE_HOURS', '24'))

# columns shown in the browse table (compact payload)
TABLE_COLS = ['playerId', 'tm_id', 'Jugador', 'Posicion', 'Temporada', 'age',
              'nationality', 'Equipo', 'Competencia', '"Minutos totales"',
              'mv_eos', 'mv_eos_prev', 'tm_mv_now', 'mv_source', 'match_conf',
              'tm_url', 'tm_contract_until', 'tm_foot']

_conn = None
_conn_lock = Lock()


def _is_stale(p):
    if not p.exists():
        return True
    return (time.time() - p.stat().st_mtime) / 3600 > MAX_AGE_HOURS


def bootstrap_data():
    DATA_DIR.mkdir(parents=True, exist_ok=True)
    if not _is_stale(DB_PARQUET):
        print(f'[data] using local {DB_PARQUET.name}', flush=True)
        return
    if not DATASET_REPO:
        if DB_PARQUET.exists():
            print(f'[data] DATASET_REPO_ID not set; using local {DB_PARQUET.name}', flush=True)
            return
        raise RuntimeError(f'No data at {DB_PARQUET} and DATASET_REPO_ID not set.')
    print(f'[data] pulling {DATASET_REPO} -> {DATA_DIR}', flush=True)
    from huggingface_hub import snapshot_download
    snapshot_download(repo_id=DATASET_REPO, repo_type='dataset', local_dir=str(DATA_DIR),
                      local_dir_use_symlinks=False, token=HF_TOKEN,
                      allow_patterns=['mv_database.parquet'])  # app only needs this
    global _conn
    _conn = None
    print('[data] download complete', flush=True)


def get_conn():
    global _conn
    with _conn_lock:
        if _conn is None:
            _conn = duckdb.connect(database=':memory:')
            tmp = os.environ.get('DUCKDB_TEMP_DIR', '/tmp/duckdb')
            os.makedirs(tmp, exist_ok=True)
            _conn.execute(f"PRAGMA memory_limit='{os.environ.get('DUCKDB_MEMORY_LIMIT', '1GB')}'")
            _conn.execute(f"PRAGMA threads={os.environ.get('DUCKDB_THREADS', '2')}")
            _conn.execute(f"PRAGMA temp_directory='{tmp}'")
            _conn.execute(f"CREATE OR REPLACE VIEW base AS SELECT * FROM read_parquet('{DB_PARQUET}');")
        return _conn


def season_sort_key(s):
    s = str(s).strip()
    if '-' in s:
        a, b = s.split('-')
        return int(a) * 100 + int(b)
    try:
        return int(s) * 100
    except ValueError:
        return 0


def get_meta():
    conn = get_conn()
    seasons = sorted([r[0] for r in conn.execute(
        "SELECT DISTINCT Temporada FROM base WHERE Temporada IS NOT NULL").fetchall()],
        key=season_sort_key)
    leagues = sorted([r[0] for r in conn.execute(
        "SELECT DISTINCT Competencia FROM base WHERE Competencia IS NOT NULL").fetchall()])
    positions = sorted([r[0] for r in conn.execute(
        "SELECT DISTINCT Posicion FROM base WHERE Posicion IS NOT NULL").fetchall()])
    n_players = conn.execute("SELECT COUNT(DISTINCT playerId) FROM base").fetchone()[0]
    n_rows = conn.execute("SELECT COUNT(*) FROM base").fetchone()[0]
    return {'seasons': seasons, 'latest_season': seasons[-1] if seasons else None,
            'leagues': leagues, 'positions': positions,
            'n_rows': n_rows, 'n_players': n_players}


def get_rows(seasons=None, leagues=None, positions=None, age_min=None, age_max=None,
             min_minutes=0, foot=None, mv_min_millions=None, mv_max_millions=None,
             nationality=None, conf=None, sort='mv_eos', sort_dir='desc', limit=2000):
    conn = get_conn()
    where, params = ['Jugador IS NOT NULL'], []
    if seasons:
        where.append(f"Temporada IN ({','.join(['?']*len(seasons))})"); params += seasons
    if leagues:
        where.append(f"Competencia IN ({','.join(['?']*len(leagues))})"); params += leagues
    if positions:
        where.append(f"Posicion IN ({','.join(['?']*len(positions))})"); params += positions
    if min_minutes:
        where.append('"Minutos totales" >= ?'); params.append(float(min_minutes))
    if age_min is not None:
        where.append('age >= ?'); params.append(age_min)
    if age_max is not None:
        where.append('age <= ?'); params.append(age_max)
    if foot:
        where.append('lower(tm_foot) = ?'); params.append(foot.lower())
    if mv_min_millions is not None:
        where.append('mv_eos >= ?'); params.append(mv_min_millions * 1e6)
    if mv_max_millions is not None:
        where.append('mv_eos <= ?'); params.append(mv_max_millions * 1e6)
    if nationality:
        where.append('lower(nationality) LIKE ?'); params.append(f'%{nationality.lower()}%')
    if conf:
        where.append('match_conf = ?'); params.append(conf)

    safe_sort = {'mv_eos', 'mv_eos_prev', 'age', 'Minutos totales', 'Jugador',
                 'Temporada', 'Competencia', 'Posicion', 'nationality'}
    sort_col = sort if sort in safe_sort else 'mv_eos'
    direction = 'DESC' if str(sort_dir).lower() == 'desc' else 'ASC'
    sql = (f"SELECT {', '.join(TABLE_COLS)} FROM base WHERE {' AND '.join(where)} "
           f'ORDER BY "{sort_col}" {direction} NULLS LAST LIMIT {int(limit)}')
    return conn.execute(sql, params).df()


def get_player(player_id, season):
    conn = get_conn()
    df = conn.execute("SELECT * FROM base WHERE playerId = ? AND Temporada = ? LIMIT 1",
                      [player_id, season]).df()
    if df.empty:
        return None
    history = conn.execute(
        """SELECT Temporada, Competencia, "Minutos totales", "Partidos jugados",
                  Goles_totales, asistencias, mv_eos, mv_eos_prev
           FROM base WHERE playerId = ? ORDER BY Temporada""",
        [player_id]).df()
    return {'row': df.iloc[0].to_dict(), 'history': history.to_dict('records')}