#!/usr/bin/env python # -*- coding: utf-8 -*- # pylint: disable=W0613, C0116 # type: ignore[union-attr] from pymongo import MongoClient from aiohttp import web import socketio import asyncio import os, glob import time import json import sys import uuid import logging from aiohttp_session import setup, get_session, new_session from aiohttp_session.cookie_storage import EncryptedCookieStorage from tornado.template import Loader from bson.json_util import dumps from urllib.parse import unquote import aiohttp_cors import aiofiles import urllib.parse import requests from PIL import Image import io, re import xml.etree.ElementTree as ET from bs4 import BeautifulSoup import pathlib from datetime import datetime from bson import ObjectId import urllib.request import ipaddress logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[logging.StreamHandler(sys.stdout)], ) logger = logging.getLogger(__name__) # Connexion Database MONGO_URI = os.getenv("MONGO_URI", "mongodb://root:example@mongodb:27017/?authSource=admin") def connect_mongo(): for attempt in range(1, 6): try: client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000, connectTimeoutMS=5000) client.admin.command("ping") logger.info("Connected to MongoDB") return client except Exception as exc: # retry if Mongo is not ready yet logger.warning("Mongo connection failed (attempt %s/5): %s", attempt, exc) time.sleep(5) raise RuntimeError("MongoDB is unreachable after retries") client = connect_mongo() # Connect to the database and collection db_iptv = client["db_iptv"] channels = db_iptv['channels'] channels.create_index("id_playlist") playlists = db_iptv['playlists'] logos = db_iptv['logos'] recents = db_iptv['recents'] settings_collection = db_iptv['settings'] devices_db = db_iptv['devices'] MAX_STREAMING_ERRORS_PER_DEVICE = 5 active_xtream_imports = {} sio_session_ips = {} ip_country_cache = {} # Configure Socket IO sio = socketio.AsyncServer(logger=True, engineio_logger=True, cors_allowed_origins='*') app = web.Application() sio.attach(app) # Variables path = os.path.dirname(os.path.realpath(__file__))+'/' UPLOAD_PLAYLIST_DIR = "playlists" DOWNLOAD_EPG_DIR = "epg" LOGO_DIR = "/home/logos" M3U_LOGO_SOURCE = '/home/playlists_download/tv_channels_2a7e1ba958_plus.m3u' URL_SERVER = 'https://iptv.mrk.ovh/' DEVICE_SNAPSHOT_DIR = pathlib.Path(__file__).parent / 'static' / 'device_snapshots' DEVICE_SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True) # Set up Tornado template loader loader = Loader('') # Generic functions def safe_join(base_dir, user_filename): """Resolve user_filename under base_dir, rejecting anything that would escape it (e.g. '../../etc/passwd') via path separators or symlinks.""" base_dir = os.path.realpath(base_dir) candidate = os.path.realpath(os.path.join(base_dir, user_filename or '')) if candidate != base_dir and not candidate.startswith(base_dir + os.sep): raise ValueError('invalid filename') return candidate def findInfo(tag, s): end = '"' if (s.find(tag)>-1): t = s[s.find(tag)+len(tag)+2:s.rfind(end)] return t[0:t.find(end)] else: return "" # Helper function to convert MongoDB documents to JSON-serializable format def serialize_doc(doc): doc["_id"] = str(doc["_id"]) # Convert ObjectId to string return doc def clean_name_channel(name_channel): new_name_channel = re.sub(r'\b\w{2}\b', '', name_channel) new_name_channel = re.sub(r'\s+', ' ', new_name_channel).strip() new_name_channel = new_name_channel.replace('- ','') new_name_channel = new_name_channel.replace('| ','') new_name_channel = new_name_channel.replace(' UHD','') new_name_channel = new_name_channel.replace(' FHD','') new_name_channel = new_name_channel.replace(' -','') new_name_channel = new_name_channel.replace(' /','') new_name_channel = new_name_channel.replace('','') pattern = r".{2}\|" new_name_channel = re.sub(pattern, "", new_name_channel) new_name_channel = re.sub(r"[\(\{\[].*?[\)\}\]]", "", new_name_channel) new_name_channel = new_name_channel.strip() return new_name_channel def get_playlist_collection(id_playlist): return db_iptv['channels_' + id_playlist] async def emit_analysis_progress(payload, sid_client=None): if not sid_client: return data = {'typeCommand': 'progress_analysis', 'payload': payload} await sio.emit('communication', data, room=sid_client, callback=messageReceived) def get_existing_playlist(id_playlist): return playlists.find_one({"id_playlist": id_playlist}) or {} def find_existing_xtream_playlist(server_url, username, password): return playlists.find_one({ "playlist_type": "xtream_codes", "xtream_server": normalize_xtream_server(server_url), "xtream_username": (username or '').strip(), "xtream_password": (password or '').strip(), }) or {} def build_playlist_document(id_playlist, title_playlist, date_end_playlist, selected_color, extra_fields=None): existing_playlist = get_existing_playlist(id_playlist) playlist_doc = { 'id_playlist': id_playlist, 'title_playlist': title_playlist, 'date_end_playlist': date_end_playlist, 'selectedColor': selected_color, 'activation': existing_playlist.get('activation', 1), } if extra_fields: playlist_doc.update(extra_fields) return playlist_doc def save_playlist_document(playlist_doc): playlists.update_one( {"id_playlist": playlist_doc["id_playlist"]}, {"$set": playlist_doc}, upsert=True ) def delete_playlist_resources(id_playlist): db_iptv.drop_collection("channels_" + id_playlist) result = playlists.delete_one({"id_playlist": id_playlist}) file_path = os.path.join(UPLOAD_PLAYLIST_DIR, id_playlist + '.m3u') if os.path.exists(file_path): os.remove(file_path) print(f"File {file_path} deleted.") return result.deleted_count def bulk_insert_channels(collection_channels, new_channels, num_channel): if not new_channels: return collection_channels.insert_many(new_channels) print("Save channels ", num_channel, flush=True) def normalize_xtream_server(server_url): server_url = (server_url or '').strip() if not server_url: return '' if not re.match(r'^https?://', server_url, re.IGNORECASE): server_url = 'http://' + server_url return server_url.rstrip('/') def build_xtream_playlist_url(server_url, username, password): base = normalize_xtream_server(server_url) if not base or not username or not password: return '' return f"{base}/get.php?username={username}&password={password}&type=m3u_plus&output=ts" def build_xtream_api_url(server_url, username, password, action, **extra_params): base = normalize_xtream_server(server_url) query = { 'username': username, 'password': password, 'action': action, } query.update(extra_params) return f"{base}/player_api.php?{urllib.parse.urlencode(query)}" XTREAM_REQUEST_HEADERS = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" } def fetch_xtream_json(server_url, username, password, action, **extra_params): url = build_xtream_api_url(server_url, username, password, action, **extra_params) response = requests.get(url, headers=XTREAM_REQUEST_HEADERS, timeout=60) response.raise_for_status() return response.json() def parse_xtream_category_map(items, id_key='category_id', name_key='category_name'): category_map = {} for item in items or []: category_id = str(item.get(id_key, '')).strip() if category_id: category_map[category_id] = item.get(name_key, '') or '' return category_map def make_xtream_live_channel(id_playlist, item, category_map, username, password, server_url): stream_id = item.get('stream_id') if not stream_id: return None container_extension = item.get('container_extension') or 'ts' return { "id_playlist": id_playlist, "id_channel": str(uuid.uuid4()), "Name": item.get('name') or f"Live {stream_id}", "Language": item.get('stream_type') or 'live', "Topic": category_map.get(str(item.get('category_id', '')), ''), "urlLogo": item.get('stream_icon', '') or '', "url": f"{normalize_xtream_server(server_url)}/live/{username}/{password}/{stream_id}.{container_extension}", "media_type": "live", "stream_type": item.get('stream_type', 'live'), "stream_id": stream_id, "category_id": item.get('category_id'), "epg_channel_id": item.get('epg_channel_id', ''), } def make_xtream_vod_channel(id_playlist, item, category_map, username, password, server_url): stream_id = item.get('stream_id') if not stream_id: return None container_extension = item.get('container_extension') or 'mp4' return { "id_playlist": id_playlist, "id_channel": str(uuid.uuid4()), "Name": item.get('name') or f"Movie {stream_id}", "Language": item.get('stream_type') or 'movie', "Topic": category_map.get(str(item.get('category_id', '')), ''), "urlLogo": item.get('stream_icon', '') or item.get('cover', '') or '', "url": f"{normalize_xtream_server(server_url)}/movie/{username}/{password}/{stream_id}.{container_extension}", "media_type": "movie", "stream_type": item.get('stream_type', 'movie'), "stream_id": stream_id, "category_id": item.get('category_id'), "rating": item.get('rating', ''), } def flatten_xtream_episodes(episodes_payload): episodes = [] if isinstance(episodes_payload, dict): for season_episodes in episodes_payload.values(): if isinstance(season_episodes, list): episodes.extend(season_episodes) elif isinstance(episodes_payload, list): episodes.extend(episodes_payload) return episodes def make_xtream_series_channels(id_playlist, series_item, series_info, category_map, username, password, server_url): topic = category_map.get(str(series_item.get('category_id', '')), '') series_name = series_item.get('name') or f"Series {series_item.get('series_id')}" series_logo = series_item.get('cover') or series_item.get('cover_big') or series_item.get('stream_icon') or '' channels_to_insert = [] for episode in flatten_xtream_episodes(series_info.get('episodes')): episode_id = episode.get('id') or episode.get('episode_id') if not episode_id: continue container_extension = episode.get('container_extension') or 'mp4' episode_title = episode.get('title') or episode.get('name') or f"Episode {episode_id}" episode_num = episode.get('episode_num') season_num = episode.get('season') episode_label = episode_title try: if season_num is not None and episode_num is not None: episode_label = f"{series_name} - S{int(season_num):02d}E{int(episode_num):02d} - {episode_title}" else: episode_label = f"{series_name} - {episode_title}" except (TypeError, ValueError): episode_label = f"{series_name} - {episode_title}" channels_to_insert.append({ "id_playlist": id_playlist, "id_channel": str(uuid.uuid4()), "Name": episode_label, "Language": "series", "Topic": topic, "urlLogo": episode.get('info', {}).get('movie_image', '') or series_logo, "url": f"{normalize_xtream_server(server_url)}/series/{username}/{password}/{episode_id}.{container_extension}", "media_type": "series", "stream_type": "series", "series_id": series_item.get('series_id'), "episode_id": episode_id, "category_id": series_item.get('category_id'), "series_name": series_name, "season": season_num, "episode_num": episode_num, }) if channels_to_insert: return channels_to_insert return [{ "id_playlist": id_playlist, "id_channel": str(uuid.uuid4()), "Name": series_name, "Language": "series", "Topic": topic, "urlLogo": series_logo, "url": "", "media_type": "series", "stream_type": "series", "series_id": series_item.get('series_id'), "category_id": series_item.get('category_id'), "series_name": series_name, }] def save_image_from_url(url, save_path): try: # Send a GET request to the URL response = requests.get(url, timeout=10) response.raise_for_status() # Raise an error for unsuccessful status codes # Write the content of the response to a file with open(save_path, 'wb') as file: file.write(response.content) print(f"Image saved to {save_path}") except requests.exceptions.RequestException as e: print(f"Error downloading the image: {e}") def load_epg_root(): """ Load EPG XML either from the URL stored in settings or from a local fallback file. Returns an ElementTree root or None if nothing can be loaded. """ xml_content = None # Try remote URL first (if configured) doc = settings_collection.find_one({'key': 'epg_url'}) epg_url = doc.get('value') if doc else None if epg_url: try: resp = requests.get(epg_url, timeout=10) if resp.status_code == 200 and resp.content: xml_content = resp.content except requests.RequestException as exc: logger.warning("Unable to fetch remote EPG (%s): %s", epg_url, exc) # Fallback to local file if xml_content is None: local_path = os.path.join(DOWNLOAD_EPG_DIR, 'epg.xml') if os.path.exists(local_path): with open(local_path, 'rb') as f: xml_content = f.read() if not xml_content: return None try: return ET.fromstring(xml_content) except ET.ParseError as exc: logger.error("Failed to parse EPG XML: %s", exc) return None # AIOHTTP functions # INDEX async def index(request): template = loader.load('static/index.html') return web.Response(text=template.generate().decode('utf-8'), content_type='text/html') # Access to playlist file from the web async def playlist_download(request): data = dict(request.query) try: file_path = safe_join('/home/playlists_download', data['filename']) except ValueError: raise web.HTTPBadRequest(text='Invalid filename') if os.path.exists(file_path): return web.FileResponse(file_path) else: raise web.HTTPNotFound(text='File not found') async def init_db(request): db_iptv.drop_collection("channels") db_iptv.drop_collection("playlists") return web.json_response({'status': 'ok'}) async def analyse_m3u_file(request): data = await request.post() id_playlist = data['id_playlist'] title_playlist = data['title'] url_playlist = data['url'] date_end_playlist = data['date_end'] filename = data['filename'] sidClient = data['sidClient'] selectedColor = data['selectedColor'] playlist_comment = data.get('comment', '').strip() new_path_filename = "./" + UPLOAD_PLAYLIST_DIR + '/' + id_playlist + '.m3u' old_path_filename = "./" + UPLOAD_PLAYLIST_DIR + '/' +filename os.rename(old_path_filename,new_path_filename) collection_channels = get_playlist_collection(id_playlist) collection_channels.drop() new_playlist = build_playlist_document( id_playlist, title_playlist, date_end_playlist, selectedColor, { 'playlist_type': 'm3u', 'url_playlist': url_playlist, 'comment': playlist_comment, 'xtream_server': '', 'xtream_username': '', 'xtream_password': '', } ) save_playlist_document(new_playlist) # Process the m3u file new_channels =[] f = open(new_path_filename, "r", encoding="utf8") Lines = f.readlines() print('ouverture fichier', flush=True) channel = {} separator = '--------' topic = '' num_channel = 0 TypeLangChannel = ['EN','FR','US','CA', 'UK' ] total_lines = len(Lines) number_line = 0 for line in Lines: number_line += 1 if separator in line: topic = line[10:] topic = topic.replace(separator,'') # PROCESS URL LINE if len(channel)>0 and line[0:4]=="http": channel["url"] = line.strip('\n') new_channels.append(channel) #print(channel, flush=True) if num_channel % 1000 == 0: collection_channels.insert_many(new_channels) new_channels = [] print("Save channels ", num_channel, flush=True) print('%d / %d' % (number_line, total_lines), flush=True) await emit_analysis_progress({'percentage': number_line / total_lines}, sidClient) # PROCESS CHANNEL LINE if line[0:4] == "#EXT": channel = {} # ID CHANNEL id_channel = str(uuid.uuid4()) # DEFINE LANGUAGE LanguageChannel = next((lang for lang in TypeLangChannel if lang in line), '') # deternine url logo url_logo = findInfo('tvg-logo', line) channel_exist = False # case separator is | name_channel = line[line.rfind(',')+1:].strip('\n') if len(name_channel)>0: channel = {"id_playlist":id_playlist, "id_channel":id_channel, "Name":name_channel, "Language":LanguageChannel, "Topic":topic, "urlLogo" : url_logo} num_channel += 1 channel_exist = True # case separator is , if channel_exist == False: name_channel = line[line.rfind('|')+2:].strip('\n') if len(name_channel) > 0: channel = {"id_playlist":id_playlist, "id_channel":id_channel, "Name":name_channel, "Language":LanguageChannel, "Topic":topic, "urlLogo" : url_logo} channel_exist = True num_channel += 1 bulk_insert_channels(collection_channels, new_channels, num_channel) print("Save final channels ", num_channel, flush=True) number_channels = collection_channels.count_documents({"id_playlist":id_playlist}) playlists.update_many({'id_playlist':id_playlist}, { "$set": { "num_channels": number_channels } } ) result = {'num_channel': number_channels} return web.json_response(result) async def import_xtream_codes(request): data = await request.post() id_playlist = data.get('id_playlist', '').strip() title_playlist = data['title'].strip() date_end_playlist = data['date_end'] selectedColor = data.get('selectedColor', '') playlist_comment = data.get('comment', '').strip() sidClient = data.get('sidClient', '') xtream_server = normalize_xtream_server(data.get('xtream_server', '')) xtream_username = data.get('xtream_username', '').strip() xtream_password = data.get('xtream_password', '').strip() if not xtream_server or not xtream_username or not xtream_password: return web.json_response({'error': 'Xtream server, username and password are required.'}, status=400) if not id_playlist: existing_playlist = find_existing_xtream_playlist(xtream_server, xtream_username, xtream_password) if existing_playlist: id_playlist = existing_playlist['id_playlist'] logger.info("Reusing existing Xtream playlist id=%s for %s", id_playlist, xtream_server) else: id_playlist = str(uuid.uuid4()) import_key = sidClient.strip() or id_playlist if active_xtream_imports.get(import_key): logger.warning("Xtream import already running for key=%s", import_key) running_import = active_xtream_imports.get(import_key) or {} return web.json_response({ 'status': 'already_running', 'message': 'An Xtream import is already running for this client. Please wait for it to finish.', 'id_playlist': running_import.get('id_playlist', id_playlist), }, status=202) active_xtream_imports[import_key] = { 'id_playlist': id_playlist, 'started_at': time.time(), } collection_channels = get_playlist_collection(id_playlist) collection_channels.drop() playlist_doc = build_playlist_document( id_playlist, title_playlist, date_end_playlist, selectedColor, { 'playlist_type': 'xtream_codes', 'url_playlist': build_xtream_playlist_url(xtream_server, xtream_username, xtream_password), 'comment': playlist_comment, 'xtream_server': xtream_server, 'xtream_username': xtream_username, 'xtream_password': xtream_password, } ) save_playlist_document(playlist_doc) asyncio.create_task(run_xtream_import( id_playlist, import_key, collection_channels, xtream_server, xtream_username, xtream_password, sidClient )) return web.json_response({'status': 'started', 'id_playlist': id_playlist}) async def run_xtream_import(id_playlist, import_key, collection_channels, xtream_server, xtream_username, xtream_password, sidClient): # Runs in the background: the HTTP request already returned, so imports that take # minutes no longer hold a live connection open (which browsers/networks can silently drop). # Completion is reported to the client over the existing progress_analysis socketio channel. try: await emit_analysis_progress({'percentage': 0.02}, sidClient) loop = asyncio.get_running_loop() async def fetch(action, **extra_params): return await loop.run_in_executor( None, lambda: fetch_xtream_json(xtream_server, xtream_username, xtream_password, action, **extra_params) ) live_categories = parse_xtream_category_map(await fetch('get_live_categories')) vod_categories = parse_xtream_category_map(await fetch('get_vod_categories')) series_categories = parse_xtream_category_map(await fetch('get_series_categories')) live_streams = await fetch('get_live_streams') vod_streams = await fetch('get_vod_streams') series_list = await fetch('get_series') total_steps = max(1, len(live_streams) + len(vod_streams) + len(series_list)) processed_steps = 0 pending_channels = [] inserted_channels = 0 for item in live_streams: channel = make_xtream_live_channel(id_playlist, item, live_categories, xtream_username, xtream_password, xtream_server) if channel: pending_channels.append(channel) inserted_channels += 1 processed_steps += 1 if len(pending_channels) >= 1000: bulk_insert_channels(collection_channels, pending_channels, inserted_channels) pending_channels = [] if processed_steps % 250 == 0: await emit_analysis_progress({'percentage': processed_steps / total_steps}, sidClient) for item in vod_streams: channel = make_xtream_vod_channel(id_playlist, item, vod_categories, xtream_username, xtream_password, xtream_server) if channel: pending_channels.append(channel) inserted_channels += 1 processed_steps += 1 if len(pending_channels) >= 1000: bulk_insert_channels(collection_channels, pending_channels, inserted_channels) pending_channels = [] if processed_steps % 250 == 0: await emit_analysis_progress({'percentage': processed_steps / total_steps}, sidClient) for item in series_list: series_channels = make_xtream_series_channels( id_playlist, item, {}, series_categories, xtream_username, xtream_password, xtream_server ) pending_channels.extend(series_channels) inserted_channels += len(series_channels) processed_steps += 1 if len(pending_channels) >= 1000: bulk_insert_channels(collection_channels, pending_channels, inserted_channels) pending_channels = [] await emit_analysis_progress({'percentage': processed_steps / total_steps}, sidClient) bulk_insert_channels(collection_channels, pending_channels, inserted_channels) number_channels = collection_channels.count_documents({"id_playlist": id_playlist}) playlists.update_many({'id_playlist': id_playlist}, {"$set": {"num_channels": number_channels}}) await emit_analysis_progress({ 'percentage': 1, 'completed': True, 'num_channel': number_channels, 'id_playlist': id_playlist, }, sidClient) except requests.RequestException as exc: logger.exception("Xtream import failed") await emit_analysis_progress({ 'percentage': 1, 'completed': True, 'error': f'Xtream request failed: {exc}', 'id_playlist': id_playlist, }, sidClient) except Exception as exc: logger.exception("Xtream import failed") await emit_analysis_progress({ 'percentage': 1, 'completed': True, 'error': f'Xtream import failed: {exc}', 'id_playlist': id_playlist, }, sidClient) finally: active_xtream_imports.pop(import_key, None) async def handle_upload(request): reader = await request.multipart() id_playlist = None filename = None filepath = None # Handle both form field and file in one pass async for part in reader: if part.name == "id_playlist_to_action": id_playlist_data = await part.text() if len(id_playlist_data) == 0: print('No playlist to update', flush=True) id_playlist = str(uuid.uuid4()) else: id_playlist = id_playlist_data print('playlist to update :', id_playlist, flush=True) elif part.name == "file": filename = part.filename if not filename: return web.Response(text="No file provided.", status=400) filepath = os.path.join(UPLOAD_PLAYLIST_DIR, filename) with open(filepath, "wb") as f: while chunk := await part.read_chunk(): f.write(chunk) if filepath: return web.json_response({'status': 'ok', 'id_playlist': id_playlist}) return web.Response(text="No file field found in the request.", status=400) async def upload_snapshot(request): reader = await request.multipart() id_device = None file_bytes = bytearray() async for part in reader: if part.name == 'id_device': id_device = (await part.text()).strip() elif part.name == 'file': while True: chunk = await part.read_chunk() if not chunk: break file_bytes.extend(chunk) if not id_device or not file_bytes: return web.json_response({'status': 'error', 'message': 'missing id_device or file'}, status=400) try: destination = safe_join(str(DEVICE_SNAPSHOT_DIR), f'{id_device}.jpg') except ValueError: return web.json_response({'status': 'error', 'message': 'invalid id_device'}, status=400) with open(destination, 'wb') as output_file: output_file.write(file_bytes) snapshot_relpath = f'/static/device_snapshots/{id_device}.jpg' devices_db.update_one( {'id_device': id_device}, {'$set': {'snapshot_url': snapshot_relpath, 'snapshot_updated_at': time.time()}}, upsert=True ) return web.json_response({'status': 'ok', 'snapshot_url': snapshot_relpath}) # COUNT CHANNELS async def count_channels(request): data = dict(request.query) id_playlist = data['id_playlist'] print('id_playlist : ', id_playlist, flush=True) collection_channels = db_iptv['channels_'+id_playlist] number_channels = collection_channels.count_documents({"id_playlist":id_playlist}) print('number_channels : ', number_channels, flush=True) return web.json_response({'number_channels': number_channels}) # DELETE PLAYLIST async def delete_playlist(request): query_params = request.query id_playlist = query_params.get("id_playlist") nb_playlist_deleted = delete_playlist_resources(id_playlist) return web.json_response({ 'nb_playlist_deleted': nb_playlist_deleted }) async def cleanup_xtream_duplicates(request): duplicate_groups = {} for playlist in playlists.find({"playlist_type": "xtream_codes"}): xtream_server = normalize_xtream_server(playlist.get("xtream_server", "")) xtream_username = (playlist.get("xtream_username", "") or "").strip() xtream_password = (playlist.get("xtream_password", "") or "").strip() if not xtream_server or not xtream_username or not xtream_password: continue duplicate_key = (xtream_server, xtream_username, xtream_password) duplicate_groups.setdefault(duplicate_key, []).append(playlist) deleted_ids = [] kept_ids = [] for playlists_group in duplicate_groups.values(): if len(playlists_group) <= 1: continue playlists_group.sort( key=lambda playlist: ( int(playlist.get("num_channels", 0) or 0), int(playlist.get("activation", 0) or 0), str(playlist.get("_id", "")), ), reverse=True, ) kept_playlist = playlists_group[0] kept_ids.append(kept_playlist["id_playlist"]) for duplicate_playlist in playlists_group[1:]: duplicate_id = duplicate_playlist["id_playlist"] if duplicate_id == kept_playlist["id_playlist"]: continue if delete_playlist_resources(duplicate_id): deleted_ids.append(duplicate_id) return web.json_response({ "status": "ok", "deleted_count": len(deleted_ids), "deleted_ids": deleted_ids, "kept_ids": kept_ids, }) async def update_playlist_comment(request): data = await request.post() id_playlist = (data.get('id_playlist') or '').strip() comment = (data.get('comment') or '').strip() if not id_playlist: return web.json_response({'error': 'id_playlist is required'}, status=400) result = playlists.update_one( {'id_playlist': id_playlist}, {'$set': {'comment': comment}} ) if result.matched_count == 0: return web.json_response({'error': 'playlist not found'}, status=404) return web.json_response({'status': 'ok', 'id_playlist': id_playlist, 'comment': comment}) # Get List Playlists async def get_list_playlists(request): active_count = playlists.count_documents({}) #for p in playlists.find(): #id_playlist = p['id_playlist'] #number_channels = channels.count_documents({"id_playlist":id_playlist}) #playlists.update_many({'id_playlist':id_playlist}, { "$set": { "num_channels": number_channels } } ) #print("Document found:", p, flush=True) #print("Number of channels:", number_channels, flush=True) print(str(active_count), flush=True) if active_count == 0: return web.json_response({'status':'no data'}) else: return web.json_response(dumps(playlists.find())) async def get_source(request): with open('static/IPTV.py') as f: return web.Response(text=f.read(), content_type='text/plain') # List Channels def _search_channels_sync(channel_to_search, current_page): MAX_RECORDS = 30 escaped_keyword = re.escape(channel_to_search) # Channel names are typically prefixed with a country/format tag (e.g. # "AR - BEIN SPORTS 1 FHD"), so match at word boundaries rather than the # start of the string to rank those above unrelated names that merely # contain the keyword mid-word. prefix_pattern = re.compile(r'\b' + escaped_keyword, re.IGNORECASE) contains_pattern = re.compile(escaped_keyword, re.IGNORECASE) active_playlists = list(playlists.find({"activation": 1})) # Two relevance tiers, most relevant first: names starting with the keyword, # then everything else that merely contains it. Counted per tier/playlist so # pagination stays a cheap count+skip/limit instead of pulling every match # into memory to sort it. tiers = [] for playlist in active_playlists: collection_channels = db_iptv['channels_' + playlist['id_playlist']] prefix_filter = {'id_playlist': playlist['id_playlist'], 'Name': prefix_pattern} count = collection_channels.count_documents(prefix_filter) if count: tiers.append((playlist, prefix_filter, count)) for playlist in active_playlists: collection_channels = db_iptv['channels_' + playlist['id_playlist']] contains_only_filter = { 'id_playlist': playlist['id_playlist'], '$and': [{'Name': contains_pattern}, {'Name': {'$not': prefix_pattern}}], } count = collection_channels.count_documents(contains_only_filter) if count: tiers.append((playlist, contains_only_filter, count)) total_records = sum(count for _, _, count in tiers) if total_records == 0: return {'num_total_page': 0, 'current_page': 1, 'data': []} num_total_page = max(1, (total_records + MAX_RECORDS - 1) // MAX_RECORDS) current_page = max(1, min(current_page, num_total_page)) begin = (current_page - 1) * MAX_RECORDS remaining = MAX_RECORDS # Second pass: fetch only the page slice serialized_data = [] offset = 0 for playlist, mongo_filter, count in tiers: if remaining == 0: break if offset + count <= begin: offset += count continue local_skip = max(0, begin - offset) fetch_limit = min(remaining, count - local_skip) collection_channels = db_iptv['channels_' + playlist['id_playlist']] cursor = collection_channels.find(mongo_filter).skip(local_skip).limit(fetch_limit) serialized_data.extend(serialize_doc(doc) for doc in cursor) remaining = MAX_RECORDS - len(serialized_data) offset += count return {'num_total_page': num_total_page, 'current_page': current_page, 'data': serialized_data} async def list_channel(request): data = await request.post() channel_to_search = unquote(data['keyword']) current_page = int(data['num_page']) logger.info("Search keyword='%s' page=%s", channel_to_search, current_page) # Searching can mean scanning a multi-hundred-thousand-channel collection # with an unindexed regex (several seconds); run it off the event loop so # it doesn't stall every other request/socket while it works. result = await asyncio.get_running_loop().run_in_executor( None, _search_channels_sync, channel_to_search, current_page ) return web.json_response(result) def get_image_urls(keyword, max_results=10): search_url = f"https://duckduckgo.com/?q={keyword}&iax=images&ia=images" headers = { "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36" } # Send the search request response = requests.get(search_url, headers=headers, timeout=10) soup = BeautifulSoup(response.text, 'html.parser') # Find image containers image_urls = [] for img in soup.find_all('img', {'class': 'tile--img__img'}): src = img.get('src') if src and src.startswith('http'): image_urls.append(src) if len(image_urls) >= max_results: break return image_urls async def get_logo(request): data = dict(request.query) id_channel = data['id_channel'] id_playlist = data['id_playlist'] try: logo_channel_file_path = safe_join(LOGO_DIR, id_channel + '.jpg') except ValueError: raise web.HTTPBadRequest(text='Invalid id_channel') collection_channels = db_iptv['channels_'+id_playlist] d = collection_channels.find_one({ 'id_channel':id_channel}) url_logo = d['urlLogo'] name_channel = d['Name'] cleaned_channel_name = clean_name_channel(name_channel) if not os.path.exists(logo_channel_file_path): print(d['Name']+' :::: ' + id_channel + ' :::: ' + url_logo) if len(url_logo) == 0: print('Recherche url pour : ' + cleaned_channel_name, flush=True) #words = cleaned_channel_name.split() first_two_words = " ".join(cleaned_channel_name.split()[:3]) regex_pattern = ".*" + ".*".join(first_two_words.split()) + ".*" #regex_pattern = ".*" + ".*".join(words) + ".*" l = logos.find_one({ 'logo_name': { '$regex': regex_pattern, '$options': "i" } }) if l : url_logo = l['logo_url'] print('url trouvée : ' + url_logo, flush=True) await asyncio.get_running_loop().run_in_executor(None, save_image_from_url, url_logo, logo_channel_file_path) query = { 'id_channel':id_channel} update = { "$set": { "urlLogo": url_logo } } collection_channels.update_one(query, update) else: print('url identifiee et sauvegarde du fichier : ' + url_logo, flush=True) if len(url_logo) > 0: await asyncio.get_running_loop().run_in_executor(None, save_image_from_url, url_logo, logo_channel_file_path) image_data = None if not os.path.exists(logo_channel_file_path): with open(LOGO_DIR+'/none.jpg', "rb") as file: image_data = file.read() else: with open(logo_channel_file_path, "rb") as file: image_data = file.read() print('image found '+logo_channel_file_path, flush=True) # Return the image as a response return web.Response(body=image_data, content_type="image/jpeg") async def list_logo(request): result = logos.find({}).limit(1000) search_string = "harry potter" words = search_string.split() regex_pattern = ".*" + ".*".join(words) + ".*" logo_data = logos.find({ 'logo_name': { '$regex': regex_pattern, '$options': "i" } }) html = '' for r in logo_data: html += '

' + r['logo_name'] + ' :::: ' + r['original_name'] + ' :::: ' + r['logo_url'] return web.Response(text=html,content_type='text/html') async def list_devices(request): return web.json_response(dumps(devices_db.find({}).limit(50))) async def clean_recents(request): recents.delete_many({}) return web.json_response({'status':'ok'}) async def get_recents(request): # Step 1: Get the IDs of the 50 most recent documents top_50 = list(recents.find({}, {"_id": 1}).sort("datetime", -1).limit(50)) top_50_ids = [doc["_id"] for doc in top_50] # Step 2: Delete all documents that are NOT in the top 50 result = recents.delete_many({ "_id": {"$nin": top_50_ids} }) return web.json_response(dumps(recents.find({}).sort("datetime", -1))) async def most_recents(request): pipeline = [ { "$sort": {"datetime": -1} }, { "$limit": 50 }, { "$group": { "_id": "$id_channel", "id_channel": {"$first": "$id_channel"}, "id_playlist": {"$first": "$id_playlist"}, "name": {"$first": "$name_channel"}, "url": {"$first": "$url_channel"}, "datetime": {"$first": "$datetime"}, "count": {"$sum": 1} } }, { "$sort": {"count": -1, "datetime": -1, "name": 1} } ] # Run the aggregation results = recents.aggregate(pipeline) return web.json_response(dumps(results)) async def activation_playlist(request): data = dict(request.query) id_playlist = data['id_playlist'] activation = int(data['activation']) print('activation : ', activation, flush=True) print(id_playlist,activation, flush=True ) playlists.update_many({'id_playlist':id_playlist}, { "$set": { "activation": activation } } ) return web.json_response({'id_playlist':id_playlist,'status':'ok'}) async def delete_device(request): data = dict(request.query) id_device = data.get('id_device') # Use .get() to avoid KeyError print(" ID device to delete : ",id_device,flush=True) if not id_device: return web.json_response({'error': 'id_device is required'}, status=400) result = devices_db.delete_many({"id_device": id_device}) return web.json_response({'deleted_count': result.deleted_count}) async def delete_recent(request): data = await request.post() oid = data.get('oid') if not oid: return web.json_response({'error': 'Missing oid'}, status=400) result = recents.delete_one({'_id': ObjectId(oid)}) if result.deleted_count == 1: return web.json_response({'status': 'success'}) async def get_epg_url(request): doc = settings_collection.find_one({'key': 'epg_url'}) epg_url = doc.get('value', '') if doc else '' return web.json_response({'epg_url': epg_url}) async def set_epg_url(request): data = await request.post() epg_url = data.get('epg_url', '').strip() if not epg_url: return web.json_response({'error': 'epg_url is required'}, status=400) settings_collection.update_one( {'key': 'epg_url'}, {'$set': {'value': epg_url, 'updated_at': datetime.utcnow()}}, upsert=True, ) return web.json_response({'status': 'ok', 'epg_url': epg_url}) async def get_gemini_key(request): doc = settings_collection.find_one({'key': 'gemini_key'}) key = doc.get('value', '') if doc else '' masked_key = (key[:4] + '...' + key[-4:]) if len(key) > 8 else ('****' if key else '') return web.json_response({'configured': bool(key), 'masked_key': masked_key}) async def set_gemini_key(request): data = await request.post() key = data.get('gemini_key', '').strip() if not key: return web.json_response({'error': 'gemini_key is required'}, status=400) settings_collection.update_one( {'key': 'gemini_key'}, {'$set': {'value': key, 'updated_at': datetime.utcnow()}}, upsert=True, ) return web.json_response({'status': 'ok'}) async def gemini_search(request): data = await request.post() query = data.get('query', '').strip() if not query: return web.json_response({'error': 'query is required'}, status=400) doc = settings_collection.find_one({'key': 'gemini_key'}) api_key = doc.get('value', '') if doc else '' if not api_key: return web.json_response({'error': 'Gemini API key not configured. Go to Settings > Gemini API Key.'}, status=400) url = f'https://generativelanguage.googleapis.com/v1beta/models/gemini-2.0-flash:generateContent?key={api_key}' payload = { 'contents': [{'parts': [{'text': query}]}], 'tools': [{'google_search': {}}] } try: resp = requests.post(url, json=payload, timeout=20) resp.raise_for_status() result = resp.json() answer = '' sources = [] candidates = result.get('candidates', []) if candidates: for part in candidates[0].get('content', {}).get('parts', []): if 'text' in part: answer += part['text'] grounding = candidates[0].get('groundingMetadata', {}) for chunk in grounding.get('groundingChunks', []): web_info = chunk.get('web', {}) if web_info.get('uri'): sources.append({'uri': web_info['uri'], 'title': web_info.get('title', '')}) return web.json_response({'answer': answer, 'sources': sources}) except requests.exceptions.HTTPError as e: try: err_msg = e.response.json().get('error', {}).get('message', str(e)) except Exception: err_msg = str(e) return web.json_response({'error': err_msg}, status=502) except Exception as e: return web.json_response({'error': str(e)}, status=502) async def channels_epg(request): root = load_epg_root() if root is None: return web.json_response({'error': 'No EPG data available'}, status=503) channels_list = [] for channel in root.findall('channel'): channel_id = channel.attrib.get('id', '') display_name = '' name_node = channel.find('display-name') if name_node is not None and name_node.text: display_name = name_node.text if display_name: channels_list.append({'id': channel_id, 'name': display_name}) return web.json_response({'channels': channels_list}) def parse_epg_datetime(raw_value: str) -> datetime: ts = raw_value[:14] return datetime.strptime(ts, "%Y%m%d%H%M%S") async def epg_programs(request): channel = request.query.get('channel') if not channel: return web.json_response({'error': 'channel is required'}, status=400) root = load_epg_root() if root is None: return web.json_response({'error': 'No EPG data available'}, status=503) programs = [] for programme in root.findall('programme'): if programme.attrib.get('channel') != channel: continue start_raw = programme.attrib.get('start', '') stop_raw = programme.attrib.get('stop', '') title_node = programme.find('title') desc_node = programme.find('desc') try: start_dt = parse_epg_datetime(start_raw) stop_dt = parse_epg_datetime(stop_raw) except Exception: continue programs.append({ 'title': title_node.text if title_node is not None else 'No title', 'desc': desc_node.text if desc_node is not None else '', 'start': start_dt.isoformat(), 'stop': stop_dt.isoformat() }) programs.sort(key=lambda p: p['start']) return web.json_response({'programs': programs}) async def download_epg(request): data = dict(request.query) url = data['url'] filename = data['filename'] try: filepath = safe_join(DOWNLOAD_EPG_DIR, filename) except ValueError: raise web.HTTPBadRequest(text='Invalid filename') print('download file :', url, flush=True) response = requests.get(url, timeout=10) with open(filepath, 'wb') as f: f.write(response.content) return web.json_response({'status':'ok'}) # Routage BASE_DIR = pathlib.Path(__file__).parent app.router.add_static('/static/', path=BASE_DIR / 'static', name='static') app.router.add_static('/logos/', LOGO_DIR) app.router.add_get('/', index) app.router.add_get('/playlists', playlist_download) app.router.add_get('/init_db', init_db) app.router.add_get('/get_list_playlists', get_list_playlists) app.router.add_post('/upload', handle_upload) app.router.add_post('/upload_snapshot', upload_snapshot) app.router.add_post('/analyse_m3u_file', analyse_m3u_file) app.router.add_post('/import_xtream_codes', import_xtream_codes) app.router.add_post('/update_playlist_comment', update_playlist_comment) app.router.add_get('/delete_playlist', delete_playlist) app.router.add_get('/cleanup_xtream_duplicates', cleanup_xtream_duplicates) app.router.add_post('/list_channel', list_channel) app.router.add_get('/get_logo', get_logo) app.router.add_get('/list_logo', list_logo) app.router.add_get('/list_devices', list_devices) app.router.add_get('/recents', get_recents) app.router.add_get('/clean_recents', clean_recents) app.router.add_get('/activation_playlist', activation_playlist) app.router.add_get('/delete_device', delete_device) app.router.add_get('/count_channels', count_channels) app.router.add_post('/delete_recent', delete_recent) app.router.add_get('/get_source', get_source) app.router.add_get('/most_recents', most_recents) app.router.add_get('/epg_url', get_epg_url) app.router.add_post('/epg_url', set_epg_url) app.router.add_get('/gemini_key', get_gemini_key) app.router.add_post('/gemini_key', set_gemini_key) app.router.add_post('/gemini_search', gemini_search) app.router.add_get('/channels_epg', channels_epg) app.router.add_get('/epg', epg_programs) def resolve_ip_country(ip_address): if not ip_address: return '' if ip_address in ip_country_cache: return ip_country_cache[ip_address] try: ip_obj = ipaddress.ip_address(ip_address) if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local: ip_country_cache[ip_address] = '' return '' except ValueError: return '' country = '' try: response = requests.get( f'http://ip-api.com/json/{ip_address}', params={'fields': 'status,country'}, timeout=5, ) data = response.json() if data.get('status') == 'success': country = data.get('country', '') or '' except requests.RequestException as exc: logger.warning("IP country lookup failed for %s: %s", ip_address, exc) ip_country_cache[ip_address] = country return country # SIO FUNCTIONS def messageReceived(): print('messageReceived') @sio.event def connect(sid, environ): # engineio's aiohttp integration hardcodes REMOTE_ADDR to 127.0.0.1; # the real peer address is only available via the underlying aiohttp request. # Behind the nginx reverse proxy, that peer is nginx itself, so prefer the # client IP nginx forwards via X-Real-IP (set in iptv.mrk.ovh.conf). aiohttp_request = environ.get('aiohttp.request') ip_address = '' if aiohttp_request: ip_address = aiohttp_request.headers.get('X-Real-IP') or aiohttp_request.remote or '' sio_session_ips[sid] = ip_address print("connect ", sid, ip_address, flush=True) @sio.event async def communication(fromSid, data): print("search request from ", fromSid, flush=True) toSidDevice = data['toSidDevice'] print("send signal to ", toSidDevice, flush=True) await sio.emit('communication', data, room=toSidDevice ,callback=messageReceived) @sio.event async def streaming_error(sid, data): IdDevice = data.get('IdDevice') message = data.get('message', '') if not IdDevice: return print('streaming_error from device :', IdDevice, message, flush=True) error_entry = {'message': str(message)[:500], 'timestamp': time.time()} devices_db.update_one( {'id_device': IdDevice}, {'$push': {'last_errors': {'$each': [error_entry], '$slice': -MAX_STREAMING_ERRORS_PER_DEVICE}}} ) @sio.event def disconnect(sid): sio_session_ips.pop(sid, None) print('disconnect ', sid, flush=True) @sio.event async def polling(sid, data): device_name = data['name'] IdDevice = data['IdDevice'] mediaRunning = data['mediaRunning'] device_last_datetime = time.time() data_running = data['data_running'] auto_snapshot_enabled = data.get('auto_snapshot_enabled', True) snapshot_interval_seconds = data.get('snapshot_interval_seconds', 10) print('polling from device :', IdDevice, flush=True) print('data_running :', data_running, flush=True) try: vpn_status = data['vpn_status'] except KeyError: vpn_status = 'unknown' ip_address = sio_session_ips.get(sid, '') if ip_address in ip_country_cache: ip_country = ip_country_cache[ip_address] else: ip_country = await asyncio.get_event_loop().run_in_executor(None, resolve_ip_country, ip_address) device_data = devices_db.find_one({ 'id_device':IdDevice}) if device_data: query = { 'id_device':IdDevice} update = { "$set": { "last_datetime": device_last_datetime ,"name": device_name, "sid":sid, "mediaRunning":mediaRunning, 'vpn_status':vpn_status, 'data_running':data_running, 'auto_snapshot_enabled': auto_snapshot_enabled, 'snapshot_interval_seconds': snapshot_interval_seconds, 'ip_address': ip_address, 'ip_country': ip_country} } devices_db.update_one(query, update) else: new_device = { 'id_device': IdDevice, 'name' : device_name, 'last_datetime' : device_last_datetime, 'sid':sid ,'mediaRunning' : mediaRunning, 'vpn_status':vpn_status, 'data_running':data_running, 'auto_snapshot_enabled': auto_snapshot_enabled, 'snapshot_interval_seconds': snapshot_interval_seconds, 'ip_address': ip_address, 'ip_country': ip_country} devices_db.insert_one(new_device) if data_running: new_record = { "id_device": IdDevice, "id_channel": data_running['id_channel'], "id_playlist": data_running['id_playlist'], "datetime": datetime.utcnow(), "name_channel": data_running['name_channel'], "url_channel":data_running['url_channel'], "progress": data_running['progress'], "percentage": data_running['percentage'] } latest_recent_record = recents.find_one( {"id_device": IdDevice}, sort=[("datetime", -1)] ) if latest_recent_record: same_channel = ( latest_recent_record.get("id_channel") == data_running['id_channel'] and latest_recent_record.get("id_playlist") == data_running['id_playlist'] ) if same_channel: recents.find_one_and_replace( {"_id": latest_recent_record["_id"]}, new_record, return_document=True ) else: recents.insert_one(new_record) else: recents.insert_one(new_record) # Main if __name__ == '__main__': logger.info('Starting server on port 5023') web.run_app(app, port=5023)