""" Tribunale dei tipster — collector Telegram (via Telethon, account utente). Legge i canali configurati (anche privati/VIP a cui il TUO account è iscritto, per ID numerico o @username), salva i messaggi nuovi in telegram_messages e poi estrae le pick con l'LLM nella tabella tips (stesso vocabolario e stesso scoring del resto del tribunale). Il timestamp della pick è quello ESATTO di pubblicazione del messaggio Telegram: il migliore possibile per il CLV. Setup una tantum (solo la prima volta, va fatto dal proprietario dell'account): 1. Prendi api_id e api_hash su https://my.telegram.org -> API development tools e aggiungili al file .env: TELEGRAM_API_ID=1234567 TELEGRAM_API_HASH=abcdef... 2. Esegui: python scripts/telegram_collector.py --login (ti chiede il numero di telefono e il codice che arriva su Telegram; la sessione resta salvata in telegram.session e non serve più) Uso quotidiano (automatizzabile): python scripts/telegram_collector.py python scripts/telegram_collector.py --no-extract # solo raccolta messaggi """ import argparse import asyncio import json import re import sys from pathlib import Path import anthropic import pymysql import config PROJECT_ROOT = Path(__file__).resolve().parent.parent SESSION_PATH = PROJECT_ROOT / "telegram" # -> telegram.session MEDIA_DIR = PROJECT_ROOT / "data" / "raw" / "telegram" sys.path.insert(0, str(Path(__file__).resolve().parent)) from agent_formazioni import load_api_key # Canali da monitorare: ID numerico (da web.telegram.org: -100...) o "@username". # Il nome del tipster viene risolto automaticamente dal titolo del canale # alla prima raccolta (poi si può rinominare nella tabella tipsters). CHANNELS = [ -1001004557063, ] DEFAULT_MODEL = "claude-opus-4-8" SCHEMA = """ CREATE TABLE IF NOT EXISTS telegram_messages ( id INT UNSIGNED AUTO_INCREMENT PRIMARY KEY, channel_id BIGINT NOT NULL, msg_id BIGINT NOT NULL, tipster_id SMALLINT UNSIGNED NULL, sent_at DATETIME NOT NULL, text TEXT NULL, has_media TINYINT(1) NOT NULL DEFAULT 0, processed TINYINT(1) NOT NULL DEFAULT 0, UNIQUE KEY uq_msg (channel_id, msg_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """ EXTRACT_PROMPT = """Questo è un messaggio di un canale Telegram di pronostici calcistici. Se contiene UNA O PIÙ pick concrete PRE-partita (partita identificabile + selezione precisa, con o senza quota), estraile. Ignora: resoconti di vincite, promozioni, chiacchiere, schedine già giocate/risultati. Rispondi SOLO con un blocco ```json```: {"picks": [ {"home": "...", "away": "...", "kickoff": "YYYY-MM-DD HH:MM" o null, "raw": "testo esatto della pick", "market": "1x2"|"double_chance"|"over_under"|"btts"|"multigol_total"|"multigol_home"|"multigol_away"|"other", "selection": "...", "line": null o 2.5, "odds": null o 1.85} ]} Se non ci sono pick: {"picks": []}. Non inventare quote né partite. ATTENZIONE alle schedine MULTIPLE (più partite con una quota complessiva): estrai le singole selezioni solo se ciascuna ha la PROPRIA quota; altrimenti estraile con odds null. Non attribuire MAI la quota complessiva della schedina a una singola selezione. Se è allegata un'immagine (screenshot di schedina), leggi le pick dall'immagine con le stesse regole; se immagine e testo ripetono la stessa pick, estraila una volta sola. MESSAGGIO (inviato il {sent_at}): """ def db_connect(args): return pymysql.connect(host=args.host, port=args.port, user=args.user, password=args.password, database="tigertips") def load_tg_credentials() -> tuple[int, str] | None: import os values = {} env_file = PROJECT_ROOT / ".env" if env_file.exists(): for line in env_file.read_text().splitlines(): if "=" in line: k, v = line.split("=", 1) values[k.strip()] = v.strip() api_id = os.environ.get("TELEGRAM_API_ID") or values.get("TELEGRAM_API_ID") api_hash = os.environ.get("TELEGRAM_API_HASH") or values.get("TELEGRAM_API_HASH") if api_id and api_hash: return int(api_id), api_hash return None async def collect(args) -> None: from telethon import TelegramClient creds = load_tg_credentials() if creds is None: print("Mancano TELEGRAM_API_ID / TELEGRAM_API_HASH nel file .env " "(prendili su https://my.telegram.org)") return api_id, api_hash = creds conn = db_connect(args) cur = conn.cursor() cur.execute(SCHEMA) conn.commit() client = TelegramClient(str(SESSION_PATH), api_id, api_hash) await client.start() # interattivo solo al primo avvio (--login) print("Connesso a Telegram.") if args.login: print("Login completato: sessione salvata. Ora puoi eseguire senza --login.") await client.disconnect() return for channel in CHANNELS: entity = await client.get_entity(channel) title = getattr(entity, "title", str(channel)) cur.execute("INSERT IGNORE INTO tipsters (name, source_url, notes) " "VALUES (%s, %s, 'telegram')", (title[:50], f"tg:{channel}")) conn.commit() cur.execute("SELECT id FROM tipsters WHERE source_url = %s", (f"tg:{channel}",)) tipster_id = cur.fetchone()[0] cur.execute("SELECT COALESCE(MAX(msg_id), 0) FROM telegram_messages " "WHERE channel_id = %s", (int(channel) if isinstance(channel, int) else 0,)) last_id = cur.fetchone()[0] n_new = 0 async for msg in client.iter_messages(entity, min_id=int(last_id), limit=args.max_messages): cur.execute(""" INSERT IGNORE INTO telegram_messages (channel_id, msg_id, tipster_id, sent_at, text, has_media) VALUES (%s, %s, %s, %s, %s, %s) """, (int(channel) if isinstance(channel, int) else entity.id, msg.id, tipster_id, msg.date.replace(tzinfo=None), msg.text or "", 1 if msg.media else 0)) n_new += cur.rowcount > 0 conn.commit() print(f" {title}: {n_new} messaggi nuovi") # Scarica le foto dei messaggi non ancora processati (per il pass vision) MEDIA_DIR.mkdir(parents=True, exist_ok=True) chan_key = int(channel) if isinstance(channel, int) else entity.id cur.execute(""" SELECT msg_id FROM telegram_messages WHERE channel_id = %s AND has_media = 1 AND processed = 0 """, (chan_key,)) want = [int(r[0]) for r in cur.fetchall()] missing = [mid for mid in want if not (MEDIA_DIR / f"{chan_key}_{mid}.jpg").exists()] n_img = 0 for batch_start in range(0, len(missing), 50): batch = missing[batch_start:batch_start + 50] for m in await client.get_messages(entity, ids=batch): if m and m.photo: await m.download_media( file=str(MEDIA_DIR / f"{chan_key}_{m.id}.jpg")) n_img += 1 if missing: print(f" {title}: scaricate {n_img} immagini " f"({len(missing) - n_img} media non-foto)") await client.disconnect() conn.close() def extract(args) -> None: api_key = load_api_key() if not api_key: print("Manca ANTHROPIC_API_KEY: estrazione saltata") return client = anthropic.Anthropic(api_key=api_key) conn = db_connect(args) cur = conn.cursor() cur.execute(""" SELECT id, tipster_id, sent_at, text, channel_id, msg_id, has_media FROM telegram_messages WHERE processed = 0 AND (CHAR_LENGTH(COALESCE(text, '')) >= 25 OR has_media = 1) ORDER BY sent_at LIMIT %s """, (args.max_extract,)) rows = cur.fetchall() print(f"Messaggi da esaminare: {len(rows)}") n_picks = 0 n_img = 0 for msg_pk, tipster_id, sent_at, text, channel_id, msg_id, has_media in rows: text = text or "" image_path = MEDIA_DIR / f"{channel_id}_{msg_id}.jpg" content = [] if has_media and image_path.exists(): import base64 data = base64.standard_b64encode(image_path.read_bytes()).decode() content.append({"type": "image", "source": {"type": "base64", "media_type": "image/jpeg", "data": data}}) n_img += 1 elif len(text) < 25: # media non-foto (video ecc.) senza testo utile: chiudi e salta cur.execute("UPDATE telegram_messages SET processed = 1 WHERE id = %s", (msg_pk,)) conn.commit() continue prompt = EXTRACT_PROMPT.replace("{sent_at}", str(sent_at)) + text[:4000] content.append({"type": "text", "text": prompt}) response = client.messages.create(model=args.model, max_tokens=2000, messages=[{"role": "user", "content": content}]) out = "".join(b.text for b in response.content if b.type == "text") m = re.search(r"```json\s*(\{.*?\})\s*```", out, re.DOTALL) raw = m.group(1) if m else out[out.find("{"):out.rfind("}") + 1] try: picks = json.loads(raw).get("picks") or [] except json.JSONDecodeError: picks = [] for p in picks: tip = p if isinstance(p, dict) else {} cur.execute(""" INSERT IGNORE INTO tips (tipster_id, article_url, published_date, scraped_at, home_team, away_team, kickoff, market, selection, line, odds, raw_text) VALUES (%s, %s, %s, NOW(), %s, %s, %s, %s, %s, %s, %s, %s) """, (tipster_id, f"tg-msg:{msg_pk}", sent_at.date(), tip.get("home"), tip.get("away"), tip.get("kickoff"), tip.get("market"), tip.get("selection"), tip.get("line"), tip.get("odds"), (tip.get("raw") or "")[:300])) n_picks += 1 cur.execute("UPDATE telegram_messages SET processed = 1 WHERE id = %s", (msg_pk,)) conn.commit() print(f"Pick estratte: {n_picks} (messaggi con immagine esaminati: {n_img})") conn.close() def main() -> int: parser = argparse.ArgumentParser(description="Collector Telegram per il tribunale dei tipster") parser.add_argument("--login", action="store_true", help="Solo login interattivo iniziale") parser.add_argument("--no-extract", action="store_true") parser.add_argument("--max-messages", type=int, default=200, help="Max messaggi per canale a raccolta") parser.add_argument("--max-extract", type=int, default=50) parser.add_argument("--model", default=DEFAULT_MODEL) config.add_db_args(parser) args = parser.parse_args() asyncio.run(collect(args)) if not args.login and not args.no_extract: extract(args) return 0 if __name__ == "__main__": sys.exit(main())