import asyncio import os import schedule import time import logging import aiohttp # --- KONFIGURATION --- INPUT_FOLDER = "./m3u_playlists" OUTPUT_FILE = "comb_and_cleaned.m3u" LOG_FILE = "m3u_manager.log" TIMEOUT_SECONDS = 3 MAX_CONCURRENT_TASKS = 50 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() ] ) # Globale Variablen für den Minuten-Fortschritt processed_count = 0 total_count = 0 valid_count = 0 def parse_m3u(file_path): """Liest eine M3U-Datei und merkt sich den Ursprungspfad jedes 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: streams.append({"info": current_info, "url": line, "source_file": file_path}) current_info = None else: streams.append({ "info": f'#EXTINF:-1,Channel {len(streams)+1}', "url": line, "source_file": file_path }) return streams async def check_and_write_stream(session, semaphore, stream, file_handle, active_sources, lock): """Prüft einen Stream und schreibt ihn bei Erfolg SOFORT in die Datei.""" global processed_count, valid_count async with semaphore: is_valid = False try: # HEAD-Request async with session.head(stream["url"], timeout=TIMEOUT_SECONDS, allow_redirects=True) as response: if response.status == 200: is_valid = True # GET-Gegenprüfung bei Fehlschlag (manche Server blocken HEAD) 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 # Inkrementiere den Zähler für bearbeitete Streams processed_count += 1 if is_valid: valid_count += 1 # Datei-Zugriff und Set-Aktualisierung müssen thread/async-safe sein async with lock: active_sources.add(stream["source_file"]) file_handle.write(f"{stream['info']}\n") file_handle.write(f"{stream['url']}\n") file_handle.flush() # Erzwingt das sofortige Schreiben auf die Festplatte 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}%). Aktiv: {valid_count}") async def validate_and_stream_write_async(all_streams): """Steuert die parallele Überprüfung und das Live-Schreiben.""" global processed_count, total_count, valid_count processed_count = 0 valid_count = 0 total_count = len(all_streams) semaphore = asyncio.Semaphore(MAX_CONCURRENT_TASKS) lock = asyncio.Lock() active_sources = set() headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"} # Datei öffnen und Header schreiben with open(OUTPUT_FILE, "w", encoding="utf-8") as f: f.write("#EXTM3U\n") f.flush() # Fortschrittsmelder im Hintergrund starten reporter_task = asyncio.create_task(progress_reporter()) async with aiohttp.ClientSession(headers=headers) as session: tasks = [ check_and_write_stream(session, semaphore, stream, f, active_sources, lock) for stream in all_streams ] await asyncio.gather(*tasks) # Fortschrittsmelder beenden, wenn alle Streams durch sind reporter_task.cancel() return active_sources def job_merge_and_clean(): """Hauptprozess.""" 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. Bitte lege M3U-Dateien dort ab.") 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 all_streams = [] for file_path in found_files: all_streams.extend(parse_m3u(file_path)) # Duplikate entfernen unique_urls = set() deduplicated_streams = [] for s in all_streams: if s["url"] not in unique_urls: unique_urls.add(s["url"]) deduplicated_streams.append(s) if not deduplicated_streams: logging.warning("Keine Streams zur Überprüfung vorhanden.") return logging.info(f"Starte Echtzeit-Prüfung von {len(deduplicated_streams)} Streams...") start_time = time.time() # Führt die asynchrone Prüfung und das Live-Schreiben aus active_sources = asyncio.run(validate_and_stream_write_async(deduplicated_streams)) end_time = time.time() logging.info(f"Prüfung beendet in {end_time - start_time:.2f} Sekunden.") # Tote Originaldateien 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}") if valid_count > 0: logging.info(f"SUCCESS: '{OUTPUT_FILE}' ist einsatzbereit ({valid_count} Streams live).") else: if os.path.exists(OUTPUT_FILE): os.remove(OUTPUT_FILE) logging.warning(f"ALERT: Kein einziger Stream war erreichbar. '{OUTPUT_FILE}' wurde entfernt.") logging.info("M3U Smart-Update beendet.\n" + "-"*50) if __name__ == "__main__": # Sofortiger Start beim Ausführen job_merge_and_clean() # Intervall-Planung 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)