import asyncio import os import schedule import time import logging import aiohttp # --- KONFIGURATION --- INPUT_FOLDER = "./m3u_playlists" OUTPUT_FILE = "comb_and_cleaned.m3u" OUTPUT_LIVE_ONLY = "comb_live_only.m3u" # Neue Datei für reine Live-Streams OUTPUT_VOD_ONLY = "comb_vod_only.m3u" # Neue Datei für reine VOD-Inhalte LOG_FILE = "m3u_manager.log" TIMEOUT_SECONDS = 4 MAX_CONCURRENT_TASKS = 500 CHECK_INTERVAL_MINUTES = 60 # --- LOGGING EINRICHTEN --- logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOG_FILE, encoding="utf-8"), logging.StreamHandler() ] ) processed_count = 0 total_count = 0 valid_count = 0 def parse_m3u(file_path): """Liest eine M3U-Datei und extrahiert alle Kanäle.""" streams = [] try: with open(file_path, "r", encoding="utf-8", errors="ignore") as f: lines = f.readlines() except Exception as e: logging.error(f"Fehler beim Lesen von {file_path}: {e}") return streams current_info = None for line in lines: line = line.strip() if not line: continue if line.startswith("#EXTINF:"): current_info = line elif line.startswith("#"): continue else: if current_info: title_match = current_info.split(",")[-1].strip() streams.append({ "info": current_info, "url": line, "source_file": file_path, "title": title_match if title_match else "Unbekannter Kanal" }) current_info = None else: streams.append({ "info": f'#EXTINF:-1,Channel {len(streams)+1}', "url": line, "source_file": file_path, "title": f'Channel {len(streams)+1}' }) return streams def is_live_stream(url): """Erkennt präzise Live-Streams anhand von URL-Mustern.""" url_lower = url.lower() if "/live/" in url_lower or "device=live" in url_lower: return True if url_lower.endswith(".m3u8") or ".m3u8?" in url_lower: return True if url_lower.endswith(".ts") or ".ts?" in url_lower: if "/movie/" in url_lower or "/series/" in url_lower: return False return True vod_extensions = (".mp4", ".mkv", ".avi", ".mov", "/movie/", "/movies/", "/series/", "/get.php") if any(vod_indicator in url_lower for vod_indicator in vod_extensions): return False return True async def check_single_stream(session, semaphore, stream): """Prüft asynchron, ob ein einzelner Stream erreichbar ist.""" global processed_count, valid_count async with semaphore: is_valid = False try: async with session.head(stream["url"], timeout=TIMEOUT_SECONDS, allow_redirects=True) as response: if response.status == 200: is_valid = True if not is_valid: async with session.get(stream["url"], timeout=TIMEOUT_SECONDS, allow_redirects=True) as get_resp: if get_resp.status == 200: is_valid = True except Exception: pass processed_count += 1 if is_valid: valid_count += 1 return stream return None async def progress_reporter(): """Gibt jede Minute den aktuellen Fortschritt im Log aus.""" while True: await asyncio.sleep(60) if total_count > 0 and processed_count < total_count: prozent = (processed_count / total_count) * 100 logging.info(f"[Fortschritt] {processed_count}/{total_count} Streams geprüft ({prozent:.1f}%). Online: {valid_count}") async def validate_streams_async(all_streams): """Prüft alle Streams parallel.""" global processed_count, total_count, valid_count processed_count = 0 valid_count = 0 total_count = len(all_streams) semaphore = asyncio.Semaphore(MAX_CONCURRENT_TASKS) headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"} reporter_task = asyncio.create_task(progress_reporter()) async with aiohttp.ClientSession(headers=headers) as session: tasks = [check_single_stream(session, semaphore, stream) for stream in all_streams] results = await asyncio.gather(*tasks) reporter_task.cancel() return [s for s in results if s is not None] def job_merge_and_clean(): """Hauptprozess zur Erstellung der drei getrennten Ausgabedateien.""" logging.info("M3U Smart-Update gestartet.") if not os.path.exists(INPUT_FOLDER): os.makedirs(INPUT_FOLDER) logging.info(f"Ordner '{INPUT_FOLDER}' wurde erstellt.") return found_files = [] for root, dirs, files in os.walk(INPUT_FOLDER): for file in files: if file.lower().endswith(('.m3u', '.m3u8')): found_files.append(os.path.join(root, file)) if not found_files: logging.warning(f"Keine M3U-Dateien in '{INPUT_FOLDER}' gefunden.") return raw_streams = [] for file_path in found_files: raw_streams.extend(parse_m3u(file_path)) if not raw_streams: logging.warning("Keine Streams in den Dateien gefunden.") return logging.info(f"Starte Prüfung von {len(raw_streams)} Streams...") start_time = time.time() online_streams = asyncio.run(validate_streams_async(raw_streams)) end_time = time.time() logging.info(f"Netzwerk-Prüfung beendet in {end_time - start_time:.2f} Sekunden.") unique_urls = set() unique_titles = set() final_live_streams = [] final_vod_streams = [] active_sources = set() for stream in online_streams: if stream["url"] not in unique_urls and stream["title"] not in unique_titles: unique_urls.add(stream["url"]) unique_titles.add(stream["title"]) active_sources.add(stream["source_file"]) if is_live_stream(stream["url"]): final_live_streams.append(stream) else: final_vod_streams.append(stream) # Quell-Dateien ohne aktive Streams löschen for file_path in found_files: if file_path not in active_sources: try: os.remove(file_path) rel_path = os.path.relpath(file_path, INPUT_FOLDER) logging.info(f"CRITICAL: Datei gelöscht (0 aktive Streams): {rel_path}") except Exception as e: logging.error(f"Fehler beim Löschen von {file_path}: {e}") total_saved = len(final_live_streams) + len(final_vod_streams) if total_saved > 0: # 1. Datei: Kombinierte Liste (Live oben, VOD unten) with open(OUTPUT_FILE, "w", encoding="utf-8") as f_comb: f_comb.write("#EXTM3U\n") for stream in final_live_streams: f_comb.write(f"{stream['info']}\n{stream['url']}\n") for stream in final_vod_streams: f_comb.write(f"{stream['info']}\n{stream['url']}\n") # 2. Datei: Reine Live-Streams if final_live_streams: with open(OUTPUT_LIVE_ONLY, "w", encoding="utf-8") as f_live: f_live.write("#EXTM3U\n") for stream in final_live_streams: f_live.write(f"{stream['info']}\n{stream['url']}\n") logging.info(f"SUCCESS: '{OUTPUT_LIVE_ONLY}' erstellt ({len(final_live_streams)} Live-Streams).") elif os.path.exists(OUTPUT_LIVE_ONLY): os.remove(OUTPUT_LIVE_ONLY) # 3. Datei: Reine VOD-Inhalte if final_vod_streams: with open(OUTPUT_VOD_ONLY, "w", encoding="utf-8") as f_vod: f_vod.write("#EXTM3U\n") for stream in final_vod_streams: f_vod.write(f"{stream['info']}\n{stream['url']}\n") logging.info(f"SUCCESS: '{OUTPUT_VOD_ONLY}' erstellt ({len(final_vod_streams)} VOD-Streams).") elif os.path.exists(OUTPUT_VOD_ONLY): os.remove(OUTPUT_VOD_ONLY) logging.info(f"SUCCESS: Gesamte Playlist aktualisiert -> '{OUTPUT_FILE}' ({total_saved} Streams gesamt).") else: # Säuberung aller Ausgabedateien bei Totalausfall for file in [OUTPUT_FILE, OUTPUT_LIVE_ONLY, OUTPUT_VOD_ONLY]: if os.path.exists(file): os.remove(file) logging.warning("ALERT: Keine funktionierenden Streams gefunden. Alle Ausgabedateien entfernt.") logging.info("M3U Smart-Update beendet.\n" + "-"*50) if __name__ == "__main__": job_merge_and_clean() schedule.every(CHECK_INTERVAL_MINUTES).minutes.do(job_merge_and_clean) logging.info(f"Scheduler aktiv. Nächster Scan in {CHECK_INTERVAL_MINUTES} Minuten.") while True: schedule.run_pending() time.sleep(1)