import os from pathlib import Path from typing import Dict from dotenv import load_dotenv env_path = Path(__file__).parent / '.env' if env_path.exists(): load_dotenv(dotenv_path=env_path) class Config: APP_NAME = os.getenv('APP_NAME', 'Media Gateway') APP_DESCRIPTION = os.getenv('APP_DESCRIPTION', 'Streaming Media Service') APP_VERSION = os.getenv('APP_VERSION', '1.0.0') API_KEY = os.getenv('API_KEY', '') REQUIRED_REFERER = os.getenv('REQUIRED_REFERER', '') LIST_LIVES_CID = os.getenv('LIST_LIVES_CID', '') UPSTREAM_HOSTS = { 'live': os.getenv('UPSTREAM_HOST_LIVE', ''), 'vod': os.getenv('UPSTREAM_HOST_VOD', '') } CID_API_URL = os.getenv('CID_API_URL', '') LOGIN_API_URL = os.getenv('LOGIN_API_URL', '') LIST_API_URL = os.getenv('LIST_API_URL', '') EPG_API_URL = os.getenv('EPG_API_URL', '') @classmethod @property def LIST_API_URL_WITH_EPG(cls): if cls.LIST_API_URL: return cls.LIST_API_URL.replace('no_epg=1', 'no_epg=0') return '' LOGIN_PASSWORD = os.getenv('LOGIN_PASSWORD', '') LOGIN_APP_ID = os.getenv('LOGIN_APP_ID', '') LOGIN_DEVICE_ID = os.getenv('LOGIN_DEVICE_ID', '') ADMIN_USERNAME = os.getenv('ADMIN_USERNAME', '') ADMIN_PASSWORD_HASH = os.getenv('ADMIN_PASSWORD_HASH', '') USER_PASSWORD_EXPIRY_DAYS = int(os.getenv('USER_PASSWORD_EXPIRY_DAYS', '30')) MAX_USERS = int(os.getenv('MAX_USERS', '50')) REDIS_URL = os.getenv('REDIS_URL', '') REDIS_TOKEN = os.getenv('REDIS_TOKEN', '') CACHE_TTL: Dict[str, int] = { 'CID': int(os.getenv('CACHE_TTL_CID', '86400')), 'AUTH': int(os.getenv('CACHE_TTL_AUTH', '10800')), 'CHANNELS': int(os.getenv('CACHE_TTL_CHANNELS', '86400')), 'STREAM': int(os.getenv('CACHE_TTL_STREAM', '21600')), 'EPG_TODAY': int(os.getenv('CACHE_TTL_EPG_TODAY', '300')), 'EPG_OTHER': int(os.getenv('CACHE_TTL_EPG_OTHER', '86400')), 'EPG_MAX_DAYS': int(os.getenv('CACHE_TTL_EPG_MAX_DAYS', '30')), 'EPG_FULL': int(os.getenv('CACHE_TTL_EPG_FULL', '3600')), } TIMEOUT = int(os.getenv('HTTP_TIMEOUT', '60')) CONNECT_TIMEOUT = int(os.getenv('HTTP_CONNECT_TIMEOUT', '10')) MAX_STREAM_CACHE = int(os.getenv('MAX_STREAM_CACHE', '200')) MAX_EPG_CACHE = int(os.getenv('MAX_EPG_CACHE', '1000')) DEBUG = os.getenv('DEBUG', 'false').lower() == 'true' ENABLE_GZIP = os.getenv('ENABLE_GZIP', 'true').lower() == 'true' MAX_CONCURRENT_REQUESTS = int(os.getenv('MAX_CONCURRENT_REQUESTS', '100')) @classmethod def get_cid_url(cls) -> str: return cls.CID_API_URL.replace('{API_KEY}', cls.API_KEY) @classmethod def get_login_url(cls, cid: str) -> str: return (cls.LOGIN_API_URL .replace('{CID}', cid) .replace('{PASSWORD}', cls.LOGIN_PASSWORD) .replace('{APP_ID}', cls.LOGIN_APP_ID) .replace('{DEVICE_ID}', cls.LOGIN_DEVICE_ID)) @classmethod def get_list_url(cls, uid: str, with_epg: bool = False) -> str: if with_epg: url = cls.LIST_API_URL.replace('no_epg=1', 'no_epg=0') else: url = cls.LIST_API_URL url = url.replace('{UPSTREAM_HOST}', cls.UPSTREAM_HOSTS['live']) url = url.replace('{CID}', cls.LIST_LIVES_CID) url = url.replace('{UID}', uid) url = url.replace('{REFERER}', cls.REQUIRED_REFERER) return url @classmethod def get_epg_url(cls, uid: str, vid: str) -> str: url = cls.EPG_API_URL url = url.replace('{UPSTREAM_HOST}', cls.UPSTREAM_HOSTS['live']) url = url.replace('{CID}', cls.LIST_LIVES_CID) url = url.replace('{UID}', uid) url = url.replace('{VID}', vid) url = url.replace('{REFERER}', cls.REQUIRED_REFERER) return url @classmethod def validate(cls) -> tuple[bool, list[str]]: missing = [] required_vars = { 'API_KEY': cls.API_KEY, 'REQUIRED_REFERER': cls.REQUIRED_REFERER, 'LIST_LIVES_CID': cls.LIST_LIVES_CID, 'UPSTREAM_HOST_LIVE': cls.UPSTREAM_HOSTS['live'], 'UPSTREAM_HOST_VOD': cls.UPSTREAM_HOSTS['vod'], 'CID_API_URL': cls.CID_API_URL, 'LOGIN_API_URL': cls.LOGIN_API_URL, 'LIST_API_URL': cls.LIST_API_URL, 'EPG_API_URL': cls.EPG_API_URL, 'LOGIN_PASSWORD': cls.LOGIN_PASSWORD, 'LOGIN_APP_ID': cls.LOGIN_APP_ID, 'LOGIN_DEVICE_ID': cls.LOGIN_DEVICE_ID } for var_name, var_value in required_vars.items(): if not var_value: missing.append(var_name) return (len(missing) == 0, missing) @classmethod def get_masked_config(cls) -> dict: def mask(value: str, show_chars: int = 4) -> str: if not value or len(value) <= show_chars: return '***' return value[:show_chars] + '***' return { 'APP_NAME': cls.APP_NAME, 'APP_VERSION': cls.APP_VERSION, 'API_KEY': mask(cls.API_KEY), 'REQUIRED_REFERER': mask(cls.REQUIRED_REFERER, 10), 'LIST_LIVES_CID': mask(cls.LIST_LIVES_CID, 8), 'UPSTREAM_HOST_LIVE': cls.UPSTREAM_HOSTS['live'], 'UPSTREAM_HOST_VOD': cls.UPSTREAM_HOSTS['vod'], 'CID_API_URL': mask(cls.CID_API_URL, 15), 'LOGIN_API_URL': mask(cls.LOGIN_API_URL, 15), 'LOGIN_PASSWORD': '***', 'LOGIN_APP_ID': mask(cls.LOGIN_APP_ID, 8), 'LOGIN_DEVICE_ID': mask(cls.LOGIN_DEVICE_ID, 8), 'CACHE_TTL': cls.CACHE_TTL, 'TIMEOUT': cls.TIMEOUT, 'CONNECT_TIMEOUT': cls.CONNECT_TIMEOUT, 'MAX_STREAM_CACHE': cls.MAX_STREAM_CACHE, 'MAX_EPG_CACHE': cls.MAX_EPG_CACHE, 'DEBUG': cls.DEBUG }