enrich_runtime.py

"""
On-demand TM scraper for the Space. Self-contained (requests + bs4) — does not
depend on corner_kick_pipeline. Mirrors the parsers from scripts/scrape_search_v2.py
and adds the JSON transfer-history endpoint + the injury HTML page.

Used by the POST /api/enrich-by-url route to enrich a single player by their
TM profile URL, persist into tm_enrichment.parquet, and push the updated
parquet back to the HF Dataset repo so the change survives Space restarts.
"""

import json as _json
import os
import re
import threading
import time
import unicodedata
from datetime import datetime
from pathlib import Path
from urllib.parse import urlparse

import pandas as pd
import requests
from bs4 import BeautifulSoup

UA = ('Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 '
      '(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36')
HDRS = {'User-Agent': UA, 'Accept-Language': 'en-US,en;q=0.9'}
TIMEOUT = 12
TM_HOST = 'transfermarkt.com'

_lock = threading.Lock()


# ============================================================================
# parsers
# ============================================================================

def parse_profile(html: str) -> dict:
    soup = BeautifulSoup(html, 'html.parser')
    out = {'market_value': None, 'age': None, 'nationality': None,
           'foot': None, 'height': None, 'contract_until': None,
           'photo': None, 'position': None, 'name': None}

    h = soup.select_one('h1.data-header__headline-wrapper')
    if h:
        out['name'] = h.get_text(' ', strip=True)

    mv = soup.select_one('a.data-header__market-value-wrapper, [class*=market-value]')
    if mv:
        m = re.search(r'[€$£]\s*[\d.,]+\s*[mkMK]?', mv.get_text(' ', strip=True))
        if m:
            out['market_value'] = re.sub(r'\s+', '', m.group(0))

    photo = soup.select_one('img.data-header__profile-image')
    if photo:
        out['photo'] = photo.get('src')

    for li in soup.select('.data-header li, .data-header__details li'):
        text = li.get_text(' ', strip=True)
        if ':' not in text:
            continue
        label, _, value = text.partition(':')
        label = label.lower().strip(); value = value.strip()
        if ('age' in label or 'birth' in label) and not out['age']:
            m = re.search(r'\((\d+)\)', value)
            if m: out['age'] = m.group(1)
        elif 'height' in label and not out['height']:
            out['height'] = value
        elif ('citizenship' in label or 'nationality' in label) and not out['nationality']:
            flags = li.select('img')
            nats = [im.get('title') or im.get('alt') for im in flags
                    if im.get('title') or im.get('alt')]
            out['nationality'] = ', '.join(nats) if nats else value
        elif 'position' in label and not out['position']:
            out['position'] = value

    cur = None
    for span in soup.select('span.info-table__content'):
        text = span.get_text(' ', strip=True)
        cls = span.get('class') or []
        if any('regular' in c for c in cls) and text.endswith(':'):
            cur = text.rstrip(':').lower().strip()
        elif cur:
            if 'foot' in cur and not out['foot']:
                out['foot'] = text.lower()
            elif 'contract' in cur and 'expires' in cur and not out['contract_until']:
                out['contract_until'] = text
            elif 'height' in cur and not out['height']:
                out['height'] = text
            cur = None
    return out


def parse_transfers_json(raw: str) -> list:
    try:
        data = _json.loads(raw)
    except Exception:
        return []
    out = []
    for t in data.get('transfers', []):
        out.append({
            'season': t.get('season', ''),
            'date': t.get('date', '') or t.get('dateUnformatted', ''),
            'from_club': (t.get('from') or {}).get('clubName', ''),
            'to_club': (t.get('to') or {}).get('clubName', ''),
            'fee': t.get('fee', ''),
            'market_value': t.get('marketValue', ''),
            'upcoming': bool(t.get('upcoming', False)),
        })
    return out


def parse_injuries(html: str) -> list:
    soup = BeautifulSoup(html, 'html.parser')
    table = soup.select_one('table.items')
    if not table:
        return []
    out = []
    for row in table.select('tbody tr'):
        cells = row.select('td')
        if len(cells) >= 6:
            out.append({
                'season': cells[0].get_text(strip=True),
                'injury': cells[1].get_text(strip=True),
                'from_date': cells[2].get_text(strip=True),
                'to_date': cells[3].get_text(strip=True),
                'days_out': cells[4].get_text(strip=True),
                'games_missed': cells[5].get_text(strip=True),
            })
        elif len(cells) >= 5:
            out.append({
                'season': '',
                'injury': cells[0].get_text(strip=True),
                'from_date': cells[1].get_text(strip=True),
                'to_date': cells[2].get_text(strip=True),
                'days_out': cells[3].get_text(strip=True),
                'games_missed': cells[4].get_text(strip=True),
            })
    return out


# ============================================================================
# orchestration
# ============================================================================

def extract_tm_id(url: str) -> str | None:
    m = re.search(r'/spieler/(\d+)', url or '')
    return m.group(1) if m else None


def is_valid_tm_url(url: str) -> bool:
    if not isinstance(url, str):
        return False
    try:
        u = urlparse(url)
    except Exception:
        return False
    if not u.scheme.startswith('http'):
        return False
    if TM_HOST not in u.netloc.lower():
        return False
    return extract_tm_id(url) is not None


def scrape_one(tm_url: str) -> dict:
    """Scrape profile + transfers + injuries for a single TM URL.
    Returns a dict ready to be merged into a tm_enrichment row."""
    tm_id = extract_tm_id(tm_url)
    if not tm_id:
        raise ValueError(f'invalid TM url: {tm_url}')

    s = requests.Session()
    s.headers.update(HDRS)

    profile_resp = s.get(tm_url, timeout=TIMEOUT)
    if profile_resp.status_code != 200:
        raise RuntimeError(f'profile fetch HTTP {profile_resp.status_code}')
    prof = parse_profile(profile_resp.text)

    transfers = []
    try:
        tr = s.get(f'https://www.transfermarkt.com/ceapi/transferHistory/list/{tm_id}',
                   timeout=TIMEOUT)
        if tr.status_code == 200:
            transfers = parse_transfers_json(tr.text)
    except Exception:
        pass

    injuries = []
    try:
        ij = s.get(f'https://www.transfermarkt.com/x/verletzungen/spieler/{tm_id}',
                   timeout=TIMEOUT)
        if ij.status_code == 200:
            injuries = parse_injuries(ij.text)
    except Exception:
        pass

    return {
        'tm_id': tm_id,
        'tm_url': tm_url,
        'tm_resolve_path': 'manual_link',
        'tm_age': prof.get('age') or '',
        'tm_nationality': prof.get('nationality') or '',
        'tm_market_value': prof.get('market_value') or '',
        'tm_contract_until': prof.get('contract_until') or '',
        'tm_foot': prof.get('foot') or '',
        'tm_height': prof.get('height') or '',
        'tm_position': prof.get('position') or '',
        'tm_photo': prof.get('photo') or '',
        'tm_injuries': _json.dumps(injuries, ensure_ascii=False) if injuries else '',
        'tm_transfers': _json.dumps(transfers, ensure_ascii=False) if transfers else '',
        '_scraped_name': prof.get('name'),
    }


# ============================================================================
# persist + push back to HF dataset
# ============================================================================

def _data_dir() -> Path:
    return Path(os.environ.get('DATA_DIR', '/data'))


def upsert_enrichment_row(player_id: str, scraped: dict):
    """Append-or-replace the player's row in tm_enrichment.parquet."""
    parquet = _data_dir() / 'tm_enrichment.parquet'
    cols = ['playerId', 'tm_age', 'tm_nationality', 'tm_market_value',
            'tm_contract_until', 'tm_foot', 'tm_height', 'tm_position',
            'tm_photo', 'tm_injuries', 'tm_transfers', 'tm_id', 'tm_url',
            'tm_resolve_path']

    if parquet.exists():
        df = pd.read_parquet(parquet)
    else:
        df = pd.DataFrame(columns=cols)

    new_row = {c: scraped.get(c, '') for c in cols if c != 'playerId'}
    new_row['playerId'] = str(player_id)

    # ensure all cols exist
    for c in cols:
        if c not in df.columns:
            df[c] = ''

    df = df[df['playerId'] != str(player_id)]
    df = pd.concat([df, pd.DataFrame([new_row])[cols]], ignore_index=True)
    df.to_parquet(parquet, index=False)
    return parquet


def push_to_hf(parquet: Path):
    """Upload the updated parquet to the HF dataset repo (overwrites the file)."""
    repo = os.environ.get('DATASET_REPO_ID')
    token = os.environ.get('HF_TOKEN')
    if not repo or not token:
        return False, 'DATASET_REPO_ID or HF_TOKEN not set — skipping push'
    try:
        from huggingface_hub import HfApi
        api = HfApi(token=token)
        api.upload_file(
            path_or_fileobj=str(parquet),
            path_in_repo=parquet.name,
            repo_id=repo,
            repo_type='dataset',
            commit_message=f'manual enrich {parquet.name}',
        )
        return True, ''
    except Exception as e:
        return False, str(e)


def enrich_and_persist(player_id: str, tm_url: str) -> dict:
    """Full pipeline: scrape → upsert → push back. Returns the scraped dict
    plus a small status payload."""
    if not is_valid_tm_url(tm_url):
        raise ValueError('not a valid Transfermarkt profile URL')

    with _lock:
        scraped = scrape_one(tm_url)
        parquet = upsert_enrichment_row(player_id, scraped)
        ok, push_err = push_to_hf(parquet)

    return {
        'ok': True,
        'scraped': scraped,
        'persisted_local': True,
        'pushed_to_hf': ok,
        'push_error': push_err,
    }