import asyncio import os import schedule import time import logging import json import aiohttp # --- KONFIGURATION --- INPUT_FOLDER = "./m3u_playlists" # Quellordner für deine M3U-Dateien OUTPUT_FILE = "comb_and_cleaned.m3u" # Die kombinierte Gesamt-Playlist OUTPUT_LIVE_ONLY = "comb_live_only.m3u" # Reine Live-TV-Playlist OUTPUT_VOD_ONLY = "comb_vod_only.m3u" # Reine Film/Serien-Playlist LOG_FILE = "m3u_manager.log" # Name des Protokolls RETRY_JSON = "failed_counts.json" # Speichert die Fehlversuche für den Löschschutz TIMEOUT_SECONDS = 5 # Maximale Wartezeit pro Stream (Sekunden) MAX_CONCURRENT_TASKS = 500 # Parallele Netzwerk-Verbindungen CHECK_INTERVAL_MINUTES = 1440 # Intervall für die automatische Wiederholung # --- LOGGING INITIALISIEREN --- 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 load_failed_counts(): """Lädt die Liste der Fehlversuche aus der JSON-Datei.""" if os.path.exists(RETRY_JSON): try: with open(RETRY_JSON, "r", encoding="utf-8") as f: return json.load(f) except Exception: return {} return {} def save_failed_counts(counts): """Speichert den aktuellen Stand der Fehlversuche permanent ab.""" try: with open(RETRY_JSON, "w", encoding="utf-8") as f: json.dump(counts, f, indent=4) except Exception as e: logging.error(f"Fehler beim Speichern der Retry-Daten: {e}") def parse_m3u(file_path): """Liest eine M3U-Datei Zeile für Zeile und extrahiert die Streams.""" 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 anhand von URL-Mustern, ob ein Stream 'Live' oder 'VOD' ist.""" 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 ein Link-Ziel. Nutzt erst HEAD, bei Bedarf GET.""" 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 im Minutentakt den aktuellen Prüfstatus im Protokoll 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} geprüft ({prozent:.1f}%). Online: {valid_count}") async def validate_streams_async(all_streams): """Bündelt und kontrolliert alle übergebenen Streams zeitgleich.""" global processed_count, total_count, valid_count processed_count, valid_count, total_count = 0, 0, 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(): """Zusammenführung, Netzwerkcheck, Lösch-Counter-Logik und Dateigenerierung.""" logging.info("M3U Smart-Update gestartet.") if not os.path.exists(INPUT_FOLDER): os.makedirs(INPUT_FOLDER) 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("Keine M3U-Dateien gefunden.") return raw_streams = [] for file_path in found_files: raw_streams.extend(parse_m3u(file_path)) if not raw_streams: 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)) logging.info(f"Prüfung beendet in {time.time() - start_time:.2f} Sekunden.") unique_urls, unique_titles, final_live, final_vod, active_sources = set(), set(), [], [], 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"]) final_live.append(stream) if is_live_stream(stream["url"]) else final_vod.append(stream) # --- RETRY- & LÖSCHLOGIK (3-Versuche-Löschschutz) --- failed_counts = load_failed_counts() for file_path in found_files: rel_path = os.path.relpath(file_path, INPUT_FOLDER) if file_path in active_sources: if rel_path in failed_counts: failed_counts[rel_path] = 0 else: current_fails = failed_counts.get(rel_path, 0) + 1 failed_counts[rel_path] = current_fails if current_fails >= 3: try: os.remove(file_path) logging.info(f"CRITICAL: Datei nach 3 Fehlversuchen gelöscht: {rel_path}") del failed_counts[rel_path] except Exception as e: logging.error(f"Fehler beim Löschen: {e}") else: logging.warning(f"RETRY-WARNUNG: {rel_path} offline. Fehlversuch {current_fails}/3.") all_rel_paths = [os.path.relpath(f, INPUT_FOLDER) for f in found_files] failed_counts = {k: v for k, v in failed_counts.items() if k in all_rel_paths} save_failed_counts(failed_counts) # --- DATEIEN SCHREIBEN --- total_saved = len(final_live) + len(final_vod) if total_saved > 0: with open(OUTPUT_FILE, "w", encoding="utf-8") as f_comb: f_comb.write("#EXTM3U\n") for s in final_live + final_vod: f_comb.write(f"{s['info']}\n{s['url']}\n") if final_live: with open(OUTPUT_LIVE_ONLY, "w", encoding="utf-8") as fl: fl.write("#EXTM3U\n") for s in final_live: fl.write(f"{s['info']}\n{s['url']}\n") elif os.path.exists(OUTPUT_LIVE_ONLY): os.remove(OUTPUT_LIVE_ONLY) if final_vod: with open(OUTPUT_VOD_ONLY, "w", encoding="utf-8") as fv: fv.write("#EXTM3U\n") for s in final_vod: fv.write(f"{s['info']}\n{s['url']}\n") elif os.path.exists(OUTPUT_VOD_ONLY): os.remove(OUTPUT_VOD_ONLY) logging.info(f"SUCCESS: Listen aktualisiert -> '{OUTPUT_FILE}' ({total_saved} Streams).") else: 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.") 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) while True: schedule.run_pending() time.sleep(1)