#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Clase FootballClub equivalente a FootballClub.php en Python. Permite gestionar la autenticación, sesiones, calendario de partidos, extracción y resolución de streams HLS con proxy/selección de calidad y notificaciones a Telegram. """ from __future__ import annotations import base64 import json import os import pathlib import re import sys import time import urllib.parse from datetime import datetime, timedelta from zoneinfo import ZoneInfo from typing import Any, Dict, List, Optional import requests import urllib3 # Desactivar advertencias de certificados no verificados en requests urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) try: import m3u8 except ImportError: m3u8 = None class FootballClub: CONFIG_FILE_NAME = 'footballclubconfig.json' API_BASE = 'https://www.footballclub.pro/api' PRODUCTS_ENDPOINT = f"{API_BASE}/fcmanager/purchased-products" CONTENTS_ENDPOINT = f"{API_BASE}/fcmanager/purchased-products-contents" HLS_MANIFEST = f"{API_BASE}/hls/manifest" SESSION_CHECK = f"{API_BASE}/session/check" SESSION_LIST = f"{API_BASE}/session/list" SESSION_REVOKE = f"{API_BASE}/session/revoke" LOGIN_URL = f"{API_BASE}/login" DEFAULT_HEADERS = { 'Accept': '*/*', 'Accept-Language': 'es-ES,es;q=0.5', 'Content-Type': 'application/json', 'User-Agent': 'Dalvik/2.1.0 (Linux; U; Android 15; SM-S926B Build/AP3A.240905.015.A2)', 'Accept-Encoding': 'gzip', 'x-cache-scope': 'main', } # Al solicitar la url de un evento, en type hay que poner el queryId para especificar la competicion SLUGS = [ {'slug': 'primera_federacion', 'name': 'Primera Federación', 'division': 'primera', 'queryId': 1}, {'slug': 'segunda_federacion', 'name': 'Segunda Federación', 'division': 'segunda', 'queryId': 2}, {'slug': 'tercera_federacion', 'name': 'Tercera Federación', 'division': 'tercera', 'queryId': 3}, ] MONTHS_ES = [ '', 'enero', 'febrero', 'marzo', 'abril', 'mayo', 'junio', 'julio', 'agosto', 'septiembre', 'octubre', 'noviembre', 'diciembre', ] def __init__(self, remote: bool = False): self.base_dir = pathlib.Path(__file__).resolve().parent self.config_path = self.base_dir / self.CONFIG_FILE_NAME self.matches_today = self.base_dir / 'matchestoday.json' self.login_file = self.base_dir / 'login.json' self.bot_token: str = '' self.telegram_users: List[str] = [] self.username: str = '' self.password: str = '' self.messages_img_primera: str = '' self.messages_img_segunda: str = '' self.messages_img_tercera: str = '' self.use_remote_token: bool = remote self.remote_token_url: str = '' self.session_checked: Optional[bool] = None self.cached_session: Dict[str, Any] = {} self.last_http_code: int = 0 self.http_session = requests.Session() self._load_config(remote) def _load_config(self, remote: bool): if not self.config_path.exists(): msg = f'[FOOTBALL CLUB TV] No encuentro el fichero de configuracion {self.config_path}, revisa' self.notifyTelegram(msg) raise FileNotFoundError(msg) try: with open(self.config_path, 'r', encoding='utf-8') as f: conf = json.load(f) except Exception as e: msg = f'[FOOTBALL CLUB TV] Error leyendo el fichero de configuracion {self.config_path}: {e}' self.notifyTelegram(msg) raise ValueError(msg) from e bot = conf.get('bot', {}) cuenta = conf.get('cuenta', {}) if ( 'token' in bot and 'receivers' in bot and 'imagensegunda' in bot and 'imagentercera' in bot and 'user' in cuenta and 'pass' in cuenta ): self.bot_token = bot['token'] self.telegram_users = [str(r) for r in bot['receivers']] self.username = cuenta['user'] self.password = cuenta['pass'] self.messages_img_primera = str(self.base_dir / bot.get('imagenprimera', 'primera.png')) self.messages_img_segunda = str(self.base_dir / bot.get('imagensegunda', 'segunda.png')) self.messages_img_tercera = str(self.base_dir / bot.get('imagentercera', 'tercera.jpg')) self.use_remote_token = remote if remote and 'tokenurl' in conf: self.remote_token_url = conf['tokenurl'] else: msg = f'[FOOTBALL CLUB TV] No encuentro los datos necesario en el fichero de configuracion {self.config_path}, revisa' if 'token' in bot and 'receivers' in bot: self.bot_token = bot['token'] self.telegram_users = [str(r) for r in bot['receivers']] self.notifyTelegram(msg) raise ValueError(msg) def slugForCompetition(self, competition: int) -> str: for tournament in self.SLUGS: if tournament['queryId'] == competition: return tournament['slug'] return 'primera_federacion' def queryIdForSlug(self, slug: str) -> int: for tournament in self.SLUGS: if tournament['slug'] == slug: return tournament['queryId'] return 1 def slugFromContentName(self, content_name: str) -> str: norm = (content_name or '').lower() if 'primera' in norm: return 'primera_federacion' if 'segunda' in norm: return 'segunda_federacion' if 'tercera' in norm: return 'tercera_federacion' return 'primera_federacion' def getEventByChannelId(self, chan_number: int, competition: int) -> Dict[str, Any]: matches_array = self.getMatchesToday(0) slug = self.slugForCompetition(competition) matches_list = matches_array.get('matches', {}).get(slug, []) for item in matches_list: if chan_number == item.get('canal'): return item return {} def getChannelUrl(self, chan_number: int, competition: int, quality: str = 'max') -> Dict[str, Any]: match = self.getEventByChannelId(chan_number, competition) if not match: return { 'error': 1, 'message': f'Hoy no hay partidos o no hay nada en el canal {chan_number} del tipo {competition} {self.matches_today}' } date_start_str = (match.get('date_start') or '').strip() if not date_start_str: return { 'error': 1, 'message': f'El evento {match.get("name", "")} no tiene fecha de inicio definida' } # Timestamp de inicio match_hour = int(datetime.fromisoformat(date_start_str.replace('Z', '+00:00')).timestamp()) date_end_str = (match.get('date_end') or '').strip() if date_end_str: match_end = int(datetime.fromisoformat(date_end_str.replace('Z', '+00:00')).timestamp()) else: match_end = match_hour + 7200 tz_madrid = ZoneInfo('Europe/Madrid') now = int(datetime.now(tz_madrid).timestamp()) time_diff = abs(now - match_hour) if ((now < match_hour and time_diff < 300) or (match_end and now >= match_hour and now < match_end)): event_id = str(match.get('id', '')) match_url = self.getEventUrl(event_id, quality) if len(match_url) > 0: return { 'error': 0, 'url': match_url } return { 'error': 2, 'message': f'No se ha encontrado url para el evento {match.get("name", "")}' } else: if now < match_hour: message = f'Aun faltan {time_diff} segundos para el {match.get("name", "")}' else: end_dt_str = datetime.fromtimestamp(match_end, tz=tz_madrid).strftime('%Y-%m-%d %H:%M:%S') message = f'El {match.get("name", "")} termino el {end_dt_str}' return { 'error': 1, 'message': message } def getEventUrl(self, event_id: str, quality: str = 'max') -> str: session = self.getSession() if not session: return '' try: schedule_id = int(event_id) except (ValueError, TypeError): return '' schedule = self.findScheduleById(schedule_id) if not schedule: return '' source = '' if schedule.get('m3u8Url'): source = schedule['m3u8Url'] elif schedule.get('videoURL'): m = re.search(r'/broadcast/embed/(\d+)', schedule['videoURL']) if m: source = f'https://vcloud.hudl.com/file/broadcast/{m.group(1)}.m3u8?hfr=1' elif schedule.get('streamUrl'): source = schedule['streamUrl'] if source: return self.getQualityManifestUrl(source, quality) return '' def getQualityManifestUrl(self, master_source: str, quality: str = 'max') -> str: variants = [] headers = { 'User-Agent': 'Mozilla/5.0 (Linux; Android 15; SM-S926B Build/AP3A.240905.015.A2; wv) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/154.0.8025.0 Mobile Safari/537.36', 'Origin': 'https://localhost', 'X-Requested-With': 'com.rfaf.playrfaf', 'Referer': 'https://localhost/', 'Accept': '*/*', } try: resp = self.http_session.get(master_source, headers=headers, timeout=10, verify=False) if resp.status_code == 200: if m3u8 is not None: parsed = m3u8.loads(resp.text, uri=master_source) if parsed.is_variant and parsed.playlists: # Extraer pistas de video for pl in parsed.playlists: uri = urllib.parse.urljoin(master_source, pl.uri) res = pl.stream_info.resolution if (pl.stream_info and pl.stream_info.resolution) else (0, 0) bandwidth = pl.stream_info.bandwidth if (pl.stream_info and pl.stream_info.bandwidth) else 0 variants.append({ 'URI': uri, 'resolution': res[0] * res[1], 'bandwidth': bandwidth }) else: # Fallback si m3u8 no estuviera: parsear #EXT-X-STREAM-INF lines = resp.text.splitlines() for i, line in enumerate(lines): if line.startswith('#EXT-X-STREAM-INF:'): res_m = re.search(r'RESOLUTION=(\d+)x(\d+)', line) res_val = int(res_m.group(1)) * int(res_m.group(2)) if res_m else 0 bw_m = re.search(r'BANDWIDTH=(\d+)', line) bw_val = int(bw_m.group(1)) if bw_m else 0 if i + 1 < len(lines): uri_line = lines[i + 1].strip() if uri_line and not uri_line.startswith('#'): variants.append({ 'URI': urllib.parse.urljoin(master_source, uri_line), 'resolution': res_val, 'bandwidth': bw_val }) except Exception: variants = [] if not variants: if self.isHlsManifestUrl(master_source): return master_source return f"{self.HLS_MANIFEST}?src={urllib.parse.quote(master_source)}" # Ordenar por resolucion / ancho de banda ascendente variants.sort(key=lambda v: (v['resolution'], v['bandwidth'])) count = len(variants) index = count - 1 # max = mayor calidad if quality == 'low' and count > 1: index = count - 2 selected = variants[index]['URI'] if (selected.startswith('http://') or selected.startswith('https://')) and self.isHlsManifestUrl(selected): return selected return f"{self.HLS_MANIFEST}?src={urllib.parse.quote(selected)}" def isHlsManifestUrl(self, url: str) -> bool: try: headers = {'User-Agent': 'VLC/3.0.20 LibVLC/3.0.20'} with self.http_session.get(url, headers=headers, stream=True, timeout=(8, 15), verify=False) as resp: if resp.status_code != 200: return False chunk = resp.raw.read(200) text = chunk.decode('utf-8', errors='ignore') return text.startswith('#EXTM3U') except Exception: return False def getFinalURL(self, manifest_uri: str) -> str: headers = { 'connection': 'Keep-Alive', 'accept-encoding': 'deflate', 'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/137.0.0.0 Safari/537.36', 'x-app-id': '01da9165-fa13-4d08-8f2f-825876ba8ffb', 'x-country-code': 'es', 'x-language-code': 'es', 'x-platform': 'ZWM1MDEyNjMtMjY2Ny00NTI5LWIyZGQtYjA1NWQ3ZDMzZjg4', 'x-tenant-id': 'b4773745-cf27-4bf6-a0d2-75b1a79a457c' } try: resp = self.http_session.get(manifest_uri, headers=headers, timeout=10, verify=False) if resp.status_code == 200 and m3u8 is not None: parsed = m3u8.loads(resp.text, uri=manifest_uri) if parsed.is_variant and parsed.playlists: playlists = sorted( parsed.playlists, key=lambda p: (p.stream_info.resolution[0] * p.stream_info.resolution[1]) if (p.stream_info and p.stream_info.resolution) else (p.stream_info.bandwidth or 0) ) best = playlists[-1] return urllib.parse.urljoin(manifest_uri, best.uri) return manifest_uri except Exception: pass return '' def login(self) -> bool: post_data = {'email': self.username, 'password': self.password} max_attempts = 5 attempt = 0 backoff = 2 success = False result = None while not success: attempt += 1 result = self.apiCall('POST', self.LOGIN_URL, json.dumps(post_data), self.DEFAULT_HEADERS) if isinstance(result, dict) and result.get('success') is True: success = True break retryable = self.last_http_code in (403, 429, 500, 502, 503) if not retryable or attempt >= max_attempts: break time.sleep(backoff) backoff = min(backoff * 2, 30) if isinstance(result, dict) and result.get('success') is True: session_id = result.get('session_id', '') user_id = '' if isinstance(result.get('user'), dict) and '_id' in result['user']: user_id = str(result['user']['_id']) elif 'user_id' in result: user_id = str(result['user_id']) expires = None fcmanager_token = result.get('fcmanager_token', '') if isinstance(fcmanager_token, str) and fcmanager_token: parts = fcmanager_token.split('.') if len(parts) >= 2: try: # Rellenar padding base64 payload_b64 = parts[1] + '=' * (-len(parts[1]) % 4) payload = json.loads(base64.urlsafe_b64decode(payload_b64)) if isinstance(payload, dict) and 'exp' in payload: tz_madrid = ZoneInfo('Europe/Madrid') expires = datetime.fromtimestamp(payload['exp'], tz=tz_madrid).strftime('%Y-%m-%d %H:%M:%S %Z') except Exception: expires = None save_data = { 'access_token': result.get('access_token', ''), 'local_token': result.get('local_token', ''), 'session_id': session_id, 'user_id': user_id, 'email': self.username, 'fcmanager_token': fcmanager_token, 'expires': expires, } try: with open(self.login_file, 'w', encoding='utf-8') as f: json.dump(save_data, f, indent=2, ensure_ascii=False) except Exception as e: print(f"Error guardando login: {e}", file=sys.stderr) return True return False def getAccessToken(self) -> str: session = self.getSession() return session.get('access_token', '') def getSessions(self) -> List[Dict[str, Any]]: session = self.getSession() if not session: return [] session_id = session.get('session_id', '') if not session_id: return [] headers = dict(self.DEFAULT_HEADERS) headers['Accept'] = 'application/json' headers['x-user-id'] = self.getUserId(session) url = f"{self.SESSION_LIST}?session_id={urllib.parse.quote(session_id)}" result = self.apiCall('GET', url, headers=headers) if isinstance(result, dict) and isinstance(result.get('sessions'), list): return result['sessions'] return [] def getUserId(self, session: Dict[str, Any]) -> str: return str(session.get('user_id', '')) def revokeSession(self, session_id: str) -> bool: session = self.getSession() if not session: return False user_id = self.getUserId(session) if not session_id or not user_id: return False headers = dict(self.DEFAULT_HEADERS) headers['Accept'] = 'application/json' headers['x-user-id'] = user_id token = session.get('access_token', '') if token: headers['access-token'] = token headers['access_token'] = token body = json.dumps({'userId': user_id, 'session_id': session_id}) result = self.apiCall('POST', self.SESSION_REVOKE, post_data=body, headers=headers) return isinstance(result, dict) and result.get('success') is True def getSession(self) -> Dict[str, Any]: if self.session_checked is not None: return self.cached_session login_data = self.loadStoredLogin() if not self.use_remote_token and not self.isSessionValid(login_data): if self.login(): login_data = self.loadStoredLogin() self.session_checked = True self.cached_session = login_data return login_data def loadStoredLogin(self) -> Dict[str, Any]: if self.use_remote_token and self.remote_token_url: try: resp = self.http_session.get(self.remote_token_url, timeout=10) if resp.status_code == 200: data = resp.json() if isinstance(data, dict): return data except Exception: pass return {} if not self.use_remote_token and self.login_file.exists(): try: with open(self.login_file, 'r', encoding='utf-8') as f: data = json.load(f) if isinstance(data, dict): return data except Exception: pass return {} def isSessionValid(self, login_data: Dict[str, Any]) -> bool: session_id = login_data.get('session_id', '') access_token = login_data.get('access_token', '') user_id = login_data.get('user_id', '') if not session_id or not access_token or not user_id: return False headers = dict(self.DEFAULT_HEADERS) headers['Accept'] = 'application/json' headers['x-user-id'] = user_id headers['access-token'] = access_token headers['access_token'] = access_token url = f"{self.SESSION_CHECK}?session_id={urllib.parse.quote(session_id)}" result = self.apiCall('GET', url, headers=headers) return isinstance(result, dict) and result.get('valid') is True def getUserEmail(self) -> str: session = self.getSession() if session.get('email'): return session['email'] return self.username def getProductId(self) -> int: session = self.getSession() if not session: return 0 headers = self.buildAuthHeaders(session) result = self.apiCall('POST', self.PRODUCTS_ENDPOINT, post_data=json.dumps({'email': self.getUserEmail()}), headers=headers) if isinstance(result, dict) and isinstance(result.get('products'), list) and len(result['products']) > 0: first = result['products'][0] if 'product_id' in first: try: return int(first['product_id']) except (ValueError, TypeError): return 0 return 0 def getSchedulesForGroup(self, group: str, product_id: int, limit: int = 100) -> List[Dict[str, Any]]: session = self.getSession() if not session: return [] headers = self.buildAuthHeaders(session) schedules = [] offset = 0 has_more = True while has_more: body = json.dumps({ 'group': group, 'email': self.getUserEmail(), 'product_id': product_id, 'limit': limit, 'offset': offset, }) result = self.apiCall('POST', self.CONTENTS_ENDPOINT, post_data=body, headers=headers) if not isinstance(result, dict): break batch = result.get('schedules') if isinstance(result.get('schedules'), list) else [] schedules.extend(batch) has_more = result.get('has_more') is True if has_more: offset += len(batch) if len(batch) == 0: break return schedules def findScheduleById(self, schedule_id: int) -> Dict[str, Any]: product_id = self.getProductId() if product_id == 0: return {} for group in ('live', 'upcoming', 'past'): for schedule in self.getSchedulesForGroup(group, product_id): if schedule.get('schedule_id') is not None and int(schedule['schedule_id']) == schedule_id: return schedule return {} def buildAuthHeaders(self, session: Dict[str, Any]) -> Dict[str, str]: headers = dict(self.DEFAULT_HEADERS) token = session.get('access_token', '') user_id = session.get('user_id', '') if token: headers['access-token'] = token headers['access_token'] = token if user_id: headers['x-user-id'] = user_id headers['x-user-email'] = self.getUserEmail() return headers def apiCall(self, method: str, url: str, post_data: str = '', headers: Optional[Dict[str, str]] = None, allow_retry: bool = True) -> Any: result = self._apiCallRaw(method, url, post_data, headers) auth_failed = self.last_http_code in (401, 403) is_auth_endpoint = (self.LOGIN_URL in url) or (self.SESSION_CHECK in url) if allow_retry and auth_failed and not is_auth_endpoint and not self.use_remote_token: self.session_checked = None self.cached_session = {} if self.login(): new_headers = self.buildAuthHeaders(self.getSession()) result = self._apiCallRaw(method, url, post_data, new_headers) return result def _apiCallRaw(self, method: str, url: str, post_data: str = '', headers: Optional[Dict[str, str]] = None) -> Any: req_headers = dict(headers or self.DEFAULT_HEADERS) try: if method.upper() == 'POST': resp = self.http_session.post( url, data=post_data, headers=req_headers, timeout=30, verify=False ) else: resp = self.http_session.get( url, headers=req_headers, timeout=30, verify=False ) self.last_http_code = resp.status_code return resp.json() except Exception: try: self.last_http_code = resp.status_code # type: ignore except Exception: self.last_http_code = 0 return None def getMatchesFileName(self) -> str: return str(self.matches_today) def getMatchesToday(self, offsethours: int = 0) -> Dict[str, Any]: stored_date = '1970-01-01' matches: Dict[str, Any] = {'matches': {}} for tournament in self.SLUGS: matches['matches'][tournament['slug']] = [] tz_madrid = ZoneInfo('Europe/Madrid') now = datetime.now(tz_madrid) today_str = now.strftime('%Y-%m-%d') # Offset en horas desde hoy 00:00 (Madrid) today_midnight = datetime.combine(now.date(), datetime.min.time(), tzinfo=tz_madrid) window_start = today_midnight + timedelta(hours=offsethours) window_end = window_start + timedelta(days=1) window_start_ts = int(window_start.timestamp()) window_end_ts = int(window_end.timestamp()) asked_date = window_start.strftime('%Y-%m-%d') is_for_asked_date = False if self.matches_today.exists(): try: with open(self.matches_today, 'r', encoding='utf-8') as f: json_data = json.load(f) if isinstance(json_data, dict) and 'date' in json_data: stored_date = json_data['date'] if asked_date == stored_date: matches = json_data is_for_asked_date = True except Exception: is_for_asked_date = False if not is_for_asked_date: matches['date'] = asked_date product_id = self.getProductId() if product_id > 0: schedules = ( self.getSchedulesForGroup('live', product_id) + self.getSchedulesForGroup('upcoming', product_id) + self.getSchedulesForGroup('past', product_id) ) def filter_schedule(data: Dict[str, Any]) -> bool: start = (data.get('startDatetimeUtc') or '').strip() if not start: return False try: ts_start = int(datetime.fromisoformat(start.replace('Z', '+00:00')).timestamp()) return window_start_ts <= ts_start < window_end_ts except Exception: return False seen = set() grouped_by_slug: Dict[str, List[Dict[str, Any]]] = {} for sched in schedules: if 'schedule_id' not in sched: continue try: sid = int(sched['schedule_id']) except (ValueError, TypeError): continue if sid in seen: continue seen.add(sid) if not filter_schedule(sched): continue content_name = sched.get('_content_name', '') slug = self.slugFromContentName(content_name) start_utc = sched.get('startDatetimeUtc', '') try: ts_start = int(datetime.fromisoformat(start_utc.replace('Z', '+00:00')).timestamp()) except Exception: ts_start = 0 filtered_schedule = { 'name': self.buildMatchName(sched), 'id': sid, 'date_start': start_utc, 'date_end': sched.get('endDatetimeUtc', ''), 'tournament': content_name, '_ts': ts_start } if slug not in grouped_by_slug: grouped_by_slug[slug] = [] grouped_by_slug[slug].append(filtered_schedule) # Por cada torneo se ordena cronológicamente por inicio y se asigna canal 1, 2, 3... for slug, items in grouped_by_slug.items(): items.sort(key=lambda x: x['_ts']) chan = 1 for item in items: item.pop('_ts', None) item['canal'] = chan chan += 1 if slug in matches['matches']: matches['matches'][slug].append(item) else: matches['matches'][slug] = [item] if offsethours == 0: try: with open(self.matches_today, 'w', encoding='utf-8') as f: json.dump(matches, f, indent=4, ensure_ascii=False) except Exception as e: print(f"Error escribiendo {self.matches_today}: {e}", file=sys.stderr) return matches def buildMatchName(self, schedule: Dict[str, Any]) -> str: local = schedule.get('local', '') visitor = schedule.get('visitor', '') local_clean = self.cleanTeamName(local) visitor_clean = self.cleanTeamName(visitor) if local_clean and visitor_clean: return f"{local_clean} Vs {visitor_clean}" # Fallback usando el título title = (schedule.get('title') or '').strip() title = re.sub(r'^[0-9]+[ªº]?\s*RFEF\s*', '', title, flags=re.IGNORECASE) title = re.sub(r'\s*J\s*[0-9]+\s*$', '', title, flags=re.IGNORECASE) return title.strip() def cleanTeamName(self, name: str) -> str: name = (name or '').strip() name = re.sub(r'\s*[0-9]+[ªº]?\s*RFEF\s*$', '', name, flags=re.IGNORECASE) name = re.sub(r'\s*[0-9]+[ªº]?\s*FEDERACION\s*$', '', name, flags=re.IGNORECASE) return name.strip() def monthNameEs(self, timestamp: int) -> str: dt = datetime.fromtimestamp(timestamp, tz=ZoneInfo('Europe/Madrid')) month_idx = dt.month return self.MONTHS_ES[month_idx] if 1 <= month_idx <= 12 else '' def sendCalendar(self, offset: int = 0): matches = self.getMatchesToday(offset) schedule_date = matches.get('date', '') if not schedule_date: return tz_madrid = ZoneInfo('Europe/Madrid') try: dt_schedule = datetime.fromisoformat(schedule_date).replace(tzinfo=tz_madrid) except Exception: return schedule_ts = int(dt_schedule.timestamp()) weekday_num = dt_schedule.isoweekday() # 1 = lunes, 7 = domingo month_day = dt_schedule.day # Calcular el lunes de esa semana monday_dt = dt_schedule - timedelta(days=(weekday_num - 1)) monday_month = self.monthNameEs(int(monday_dt.timestamp())) monday_mday = monday_dt.day for sport, matchs in matches.get('matches', {}).items(): if not matchs: continue query_id = self.queryIdForSlug(sport) nombre_tour = '' chans_prefix = '' imagen = '' if query_id == 1: nombre_tour = 'Partidos 1º FEF' chans_prefix = 'EVENTOS Primera RFEF' imagen = self.messages_img_primera elif query_id == 2: nombre_tour = 'Partidos 2º FEF' chans_prefix = 'Segunda RFEF' imagen = self.messages_img_segunda elif query_id == 3: nombre_tour = 'Partidos 3º FEF' chans_prefix = 'Tercera REF' imagen = self.messages_img_tercera if not nombre_tour: continue text_send = ( f"📣📣️ {nombre_tour} 📣📣\n" f"📣📣 Semana del {monday_mday} de {monday_month} 📣📣\n\n" f"📍 🗓 Dia {month_day}\n" ) logo_sent = False for match in matchs: chan_num = match.get('canal', '') date_start = match.get('date_start', '') try: dt_start = datetime.fromisoformat(date_start.replace('Z', '+00:00')).astimezone(tz_madrid) match_hour = dt_start.strftime('%H:%M') except Exception: match_hour = '' match_name = match.get('name', '') sports_text = f"\n📺 {chans_prefix} {chan_num}\n⏰{match_hour}\n{match_name}\n" if (len(sports_text) + len(text_send)) >= 1024: if logo_sent: self.notifyTelegram(text_send) text_send = '' else: self.notifyTelegram(text_send, imagen) logo_sent = True text_send += sports_text if text_send: if not logo_sent: self.notifyTelegram(text_send, imagen) else: self.notifyTelegram(text_send) def sendCalendarText(self, offset: int = 0): matches = self.getMatchesToday(offset) schedule_date = matches.get('date', '') if not schedule_date: return tz_madrid = ZoneInfo('Europe/Madrid') try: dt_schedule = datetime.fromisoformat(schedule_date).replace(tzinfo=tz_madrid) except Exception: return month_day = dt_schedule.day for sport, matchs in matches.get('matches', {}).items(): if not matchs: continue query_id = self.queryIdForSlug(sport) nombre_tour = '' chans_prefix = '' if query_id == 1: nombre_tour = 'Partidos 1º FEF' chans_prefix = 'Primera RFEF' elif query_id == 2: nombre_tour = 'Partidos 2º FEF' chans_prefix = 'Segunda RFEF' elif query_id == 3: nombre_tour = 'Partidos 3º FEF' chans_prefix = 'Tercera RFEF' if not nombre_tour: continue text_send = f"{nombre_tour}\n📍 🗓 Dia {month_day}\n" for match in matchs: chan_num = match.get('canal', '') date_start = match.get('date_start', '') try: dt_start = datetime.fromisoformat(date_start.replace('Z', '+00:00')).astimezone(tz_madrid) match_hour = dt_start.strftime('%H:%M') except Exception: match_hour = '' match_name = match.get('name', '') sports_text = f"\n{chans_prefix} {chan_num}\n⏰{match_hour}\n{match_name}\n" if (len(sports_text) + len(text_send)) >= 1024: self.notifyTelegram(text_send) text_send = '' text_send += sports_text if text_send: self.notifyTelegram(text_send) def notifyTelegram(self, text: str, img_encabezado: Optional[str] = None) -> None: if not self.bot_token or not self.telegram_users: return url_base = f"https://api.telegram.org/bot{self.bot_token}" custom_ua = 'Mozilla/5.0 (QtEmbedded; U; Linux; C) AppleWebKit/533.3 (KHTML, like Gecko) MAG200 stbapp ver: 2 rev: 250 Safari/533.3' for chat_id in self.telegram_users: try: if img_encabezado and os.path.exists(img_encabezado): with open(img_encabezado, 'rb') as photo_file: self.http_session.post( f"{url_base}/sendPhoto", data={'chat_id': chat_id, 'caption': text}, files={'photo': photo_file}, headers={'User-Agent': custom_ua}, timeout=10, verify=False ) else: self.http_session.post( f"{url_base}/sendMessage", data={'chat_id': chat_id, 'text': text}, headers={'User-Agent': custom_ua}, timeout=10, verify=False ) except Exception as e: print(f"Error enviando mensaje a Telegram (chat {chat_id}): {e}", file=sys.stderr) # Aliases en snake_case para mayor flexibilidad FootballClub.get_event_by_channel_id = FootballClub.getEventByChannelId FootballClub.slug_for_competition = FootballClub.slugForCompetition FootballClub.get_channel_url = FootballClub.getChannelUrl FootballClub.get_event_url = FootballClub.getEventUrl FootballClub.get_quality_manifest_url = FootballClub.getQualityManifestUrl FootballClub.is_hls_manifest_url = FootballClub.isHlsManifestUrl FootballClub.get_final_url = FootballClub.getFinalURL FootballClub.get_access_token = FootballClub.getAccessToken FootballClub.get_sessions = FootballClub.getSessions FootballClub.get_user_id = FootballClub.getUserId FootballClub.revoke_session = FootballClub.revokeSession FootballClub.get_session = FootballClub.getSession FootballClub.load_stored_login = FootballClub.loadStoredLogin FootballClub.is_session_valid = FootballClub.isSessionValid FootballClub.get_user_email = FootballClub.getUserEmail FootballClub.get_product_id = FootballClub.getProductId FootballClub.get_schedules_for_group = FootballClub.getSchedulesForGroup FootballClub.find_schedule_by_id = FootballClub.findScheduleById FootballClub.build_auth_headers = FootballClub.buildAuthHeaders FootballClub.api_call = FootballClub.apiCall FootballClub.get_matches_file_name = FootballClub.getMatchesFileName FootballClub.get_matches_today = FootballClub.getMatchesToday FootballClub.slug_from_content_name = FootballClub.slugFromContentName FootballClub.build_match_name = FootballClub.buildMatchName FootballClub.clean_team_name = FootballClub.cleanTeamName FootballClub.send_calendar = FootballClub.sendCalendar FootballClub.send_calendar_text = FootballClub.sendCalendarText FootballClub.notify_telegram = FootballClub.notifyTelegram if __name__ == '__main__': # CLI de prueba equivalente rápida fc = FootballClub(remote=False) cmd = sys.argv[1] if len(sys.argv) > 1 else 'cal' if cmd == 'cal': offset = int(sys.argv[2]) if len(sys.argv) > 2 else 0 cal = fc.getMatchesToday(offset) print(f"Calendario de eventos (offset {offset} horas):") print(f"Fecha: {cal.get('date', '-')}") print("-" * 50) for cat, matches in cal.get('matches', {}).items(): if matches: print(f"[{cat}] ({len(matches)} partidos):") for m in matches: print(f" Canal {m.get('canal')}: {m.get('name')} | Inicio: {m.get('date_start')}") elif cmd == 'event': canal = int(sys.argv[2]) if len(sys.argv) > 2 else 1 comp = int(sys.argv[3]) if len(sys.argv) > 3 else 1 qual = sys.argv[4] if len(sys.argv) > 4 else 'max' info = fc.getChannelUrl(canal, comp, qual) print(json.dumps(info, indent=2)) elif cmd == 'token': print(fc.getAccessToken()) elif cmd == 'sessions': print(json.dumps(fc.getSessions(), indent=2)) else: print(f"Comando desconocido: {cmd}. Opciones: cal, event, token, sessions")