import logging import threading from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any, Callable, Dict, List, Optional import requests from config import DEFAULT_VIDEA_CONCURRENCY, DEFAULT_VK_CONCURRENCY, MAX_VIDEA_CONVERTING from database import ( count_videa_converting_videos, count_videa_uploading_videos, get_account_by_id, get_accounts, get_all_videos_with_details, get_media_item_by_id, get_saved_series_list, get_series_episodes_from_db, get_video_with_details, save_media_item, save_video, update_video_progress, ) from converter import EmbedConverter from episode_parser import extract_season_episode from mozisarok_client import MozisarokClient from providers import get_provider from stream_uploader import UploadCancelledException logger = logging.getLogger("QueueManager") class UploadQueueManager: """Manages pre-assigned upload tasks, concurrency, and multi-step pipeline.""" def __init__( self, videa_concurrency: int = DEFAULT_VIDEA_CONCURRENCY, vk_concurrency: int = DEFAULT_VK_CONCURRENCY, max_videa_converting: int = MAX_VIDEA_CONVERTING, auto_recover: bool = True, ): self.videa_concurrency = videa_concurrency self.vk_concurrency = vk_concurrency self.max_videa_converting = max_videa_converting self._videa_upload_semaphore = threading.Semaphore(videa_concurrency) self._vk_upload_semaphore = threading.Semaphore(vk_concurrency) self._lock = threading.Lock() self.is_running = False self._stop_event = threading.Event() self.mozisarok = MozisarokClient() self.converter = EmbedConverter() # In-memory queue state self.queue: List[Dict[str, Any]] = [] self._cancel_events: Dict[int, threading.Event] = {} self._active_converting_task_ids: set = set() self._executor: Optional[ThreadPoolExecutor] = None self._listeners: List[Callable[[Dict[str, Any]], None]] = [] # Automatic crash and stop recovery from MySQL database if auto_recover: self.recover_from_database(run_in_background=True) @property def _videa_semaphore(self): """Backward-compatibility alias pointing to upload semaphore.""" return self._videa_upload_semaphore def set_videa_concurrency(self, concurrency: int): """Sets the maximum number of simultaneous Videa uploads.""" with self._lock: self.videa_concurrency = max(1, concurrency) self._videa_upload_semaphore = threading.Semaphore(self.videa_concurrency) logger.info(f"Videa upload concurrency set to {self.videa_concurrency}") def set_vk_concurrency(self, concurrency: int): """Sets the maximum number of simultaneous VK uploads.""" with self._lock: self.vk_concurrency = max(1, concurrency) self._vk_upload_semaphore = threading.Semaphore(self.vk_concurrency) logger.info(f"VK upload concurrency set to {self.vk_concurrency}") def get_videa_uploading_count(self) -> int: """Returns the number of Videa videos currently uploading or starting upload.""" with self._lock: return sum( 1 for t in self.queue if t.get("provider") == "videa" and t.get("status") in ["uploading", "waiting"] ) def get_vk_uploading_count(self) -> int: """Returns the number of VK videos currently uploading or starting upload.""" with self._lock: return sum( 1 for t in self.queue if t.get("provider") == "vk" and t.get("status") in ["uploading", "waiting"] ) def get_videa_converting_count(self) -> int: """Returns the number of Videa videos currently in conversion.""" with self._lock: mem_count = sum( 1 for t in self.queue if t.get("provider") == "videa" and t.get("status") in ["converting", "uploaded"] ) try: db_count = count_videa_converting_videos() except Exception: db_count = 0 return max(mem_count, db_count) def can_start_videa_upload(self) -> bool: """ Determines whether a new Videa upload may start: 1. Actively uploading count < videa_concurrency (ellenőrzés/konvertálás mellett mehet feltöltés) 2. Converting count < max_videa_converting (16 konvertálás alatt videónál nem lehet újat feltölteni) """ uploading = self.get_videa_uploading_count() converting = self.get_videa_converting_count() return (uploading < self.videa_concurrency) and (converting < self.max_videa_converting) def can_start_vk_upload(self) -> bool: """Determines whether a new VK upload may start (actively uploading < vk_concurrency).""" return self.get_vk_uploading_count() < self.vk_concurrency def add_listener(self, callback: Callable[[Dict[str, Any]], None]): self._listeners.append(callback) def _notify_listeners(self, item: Dict[str, Any]): for listener in self._listeners: try: listener(item) except Exception as e: logger.error(f"Error in queue listener callback: {e}") def add_task( self, media_data: Dict[str, Any], file_path: str, provider: str = "videa", account_id: Optional[int] = None, season: Optional[int] = None, episode: Optional[int] = None, language: str = "Magyar", subtitle: Optional[str] = None, quality: str = "1080p", anonymous: bool = False, ) -> int: """ Pre-assigns a video to a datasheet and adds it to the upload queue. Persists media_item and video in MySQL database with 'queued' status. """ path = Path(file_path) if not path.is_file(): raise FileNotFoundError(f"A fájl nem található a megadott útvonalon: {file_path}") file_size = path.stat().st_size file_name = path.name # 1. Save or update media item in MySQL tmdb_id = int(media_data["tmdb_id"] if "tmdb_id" in media_data else media_data["id"]) media_type = media_data.get("media_type", "movie") title = media_data.get("title") or media_data.get("name", "Ismeretlen cím") db_media_id = save_media_item( tmdb_id=tmdb_id, media_type=media_type, title=title, original_title=media_data.get("original_title") or media_data.get("original_name"), overview=media_data.get("overview"), release_date=media_data.get("release_date") or media_data.get("first_air_date"), poster_path=media_data.get("poster_path"), backdrop_path=media_data.get("backdrop_path"), vote_average=media_data.get("vote_average"), ) # Auto-extract season and episode from filename if TV series and not provided if media_type == "tv" and (season is None or episode is None): detected_s, detected_e = extract_season_episode(file_name) if season is None and detected_s is not None: season = detected_s if episode is None and detected_e is not None: episode = detected_e # 2. Save video entry in MySQL video_id = save_video( media_id=db_media_id, account_id=account_id, file_path=str(path.resolve()), file_name=file_name, file_size=file_size, provider=provider, season=season, episode=episode, language=language, subtitle=subtitle, quality=quality, status="queued", ) task_item = { "id": video_id, "media_id": db_media_id, "media_title": title, "media_type": media_type, "tmdb_id": tmdb_id, "poster_path": media_data.get("poster_path"), "file_path": str(path.resolve()), "file_name": file_name, "file_size": file_size, "provider": provider.lower(), "account_id": account_id, "season": season, "episode": episode, "language": language, "subtitle": subtitle, "quality": quality, "anonymous": anonymous, "status": "queued", "progress": 0.0, "error_message": None, "direct_url": None, "embed_url": None, "verification": None, } with self._lock: self.queue.append(task_item) self._cancel_events[video_id] = threading.Event() self._notify_listeners(task_item) logger.info(f"Added task #{video_id} ('{file_name}' -> '{title}') to upload queue.") return video_id def add_batch_tasks( self, media_data: Dict[str, Any], file_paths: List[str], provider: str = "videa", account_id: Optional[int] = None, language: str = "Magyar", subtitle: Optional[str] = None, quality: str = "1080p", anonymous: bool = False, ) -> List[int]: """ Adds multiple episodes or files to the upload queue in batch. Automatically extracts season and episode from each filename (e.g. s01e01, s1e1, s1e01). """ task_ids = [] for fp in file_paths: s, e = extract_season_episode(fp) tid = self.add_task( media_data=media_data, file_path=fp, provider=provider, account_id=account_id, season=s, episode=e, language=language, subtitle=subtitle, quality=quality, anonymous=anonymous, ) task_ids.append(tid) logger.info(f"Batch added {len(task_ids)} tasks to upload queue.") return task_ids def cancel_task(self, task_id: int) -> bool: """Signals cancellation for an ongoing or queued upload task.""" with self._lock: for task in self.queue: if task["id"] == task_id: cancel_evt = self._cancel_events.get(task_id) if cancel_evt: cancel_evt.set() old_status = task["status"] task["status"] = "cancelled" task["error_message"] = "Feltöltés megszakítva a felhasználó által." update_video_progress( task_id, status="cancelled", error_message="Feltöltés megszakítva a felhasználó által.", ) self._notify_listeners(task) logger.info(f"Task #{task_id} (previous status: {old_status}) cancelled by user.") return True return False def remove_task(self, task_id: int) -> bool: """ Removes a task from the queue. If it is currently uploading or running, signals immediate cancellation to abort network transfer and release resources. """ with self._lock: for i, task in enumerate(self.queue): if task["id"] == task_id: cancel_evt = self._cancel_events.pop(task_id, None) if cancel_evt: cancel_evt.set() if task["status"] in ["waiting", "uploading", "converting", "submitting", "verifying"]: update_video_progress( task_id, status="cancelled", error_message="Feltöltés megszakítva és törölve a sorból.", ) self.queue.pop(i) logger.info(f"Task #{task_id} removed/cancelled from queue.") return True return False def clear_completed(self): with self._lock: self.queue = [t for t in self.queue if t["status"] not in ["completed", "failed", "cancelled"]] def get_queue_status(self) -> List[Dict[str, Any]]: with self._lock: return list(self.queue) def _build_task_dict_from_db(self, v: Dict[str, Any]) -> Dict[str, Any]: """Reconstructs an in-memory task dictionary from a MySQL record.""" return { "id": v["id"], "media_id": v["media_id"], "media_title": v.get("media_title") or "Ismeretlen cím", "media_type": v.get("media_type") or "movie", "tmdb_id": v.get("tmdb_id") or 0, "file_path": v.get("file_path") or "", "file_name": v.get("file_name") or "", "file_size": v.get("file_size") or 0, "provider": v.get("provider") or "videa", "account_id": v.get("account_id"), "season": v.get("season"), "episode": v.get("episode"), "language": v.get("language") or "Magyar", "subtitle": v.get("subtitle"), "quality": v.get("quality") or "1080p", "status": v.get("status_code") or "queued", "progress": 0.0, "provider_video_id": v.get("provider_video_id"), "direct_url": v.get("direct_url"), "embed_url": v.get("embed_url"), "error_message": v.get("error_message"), "verification": None, } def recover_from_database(self, run_in_background: bool = True) -> Dict[str, Any]: """ Recovers system state from the MySQL database after a system crash, shutdown, or freeze: 1. 'feltöltés alatt' (uploading): - DO NOT automatically continue uploading (user request: 'viszont a feltöltés alatt álló videókat ne folytasa automatikusan'). - Resets MySQL status to 'várakozik', sets error_message='A feltöltés korábban félbeszakadt (rendszerleállás). Manuálisan elindítható.', - Loads into self.queue in paused state ('queued'). 2. 'várakozik' (queued): - Restores into self.queue in paused state ('queued'). 3. 'ellenőrzés alatt' (verifying): - User request: 'pl ellenőrzés alatt videót ellenőrize hogy fel van e töltve' - Checks if the video is uploaded and live on provider / Mozisarok. If reachable, finalizes with 'kész'. 4. 'beküldés alatt' (submitting): - Submits embed link to Mozisarok, verifies and marks completed. 5. 'konvertálás alatt' (converting) / 'feltöltve' (uploaded): - Checks provider transcoding: if finished, scrapes clean embed link, submits, verifies, and marks 'kész'. """ if run_in_background: t = threading.Thread(target=self._run_recovery, daemon=True) t.start() return {"status": "recovery_started_in_background"} else: return self._run_recovery() def _run_recovery(self) -> Dict[str, Any]: summary = { "uploading_paused": 0, "queued_restored": 0, "verifying_completed": 0, "verifying_failed": 0, "submitting_completed": 0, "converting_completed": 0, "converting_pending": 0, } try: all_videos = get_all_videos_with_details() except Exception as e: logger.error(f"Nem sikerült lekérni a videókat az adatbázisból a helyreállításhoz: {e}") return summary uncompleted = [ v for v in all_videos if v.get("status_code") not in ["completed", "cancelled"] and not v.get("completed_at") ] logger.info(f"Rendszerállapot helyreállítása az adatbázisból: {len(uncompleted)} befejezetlen feladat vizsgálata...") for v in uncompleted: vid_id = v["id"] status_code = v.get("status_code") or "queued" # 1. 'feltöltés alatt' -> NE folytassa automatikusan, állítsa várakozóra! if status_code == "uploading": msg = "A feltöltés korábban félbeszakadt (rendszerleállás). Manuálisan elindítható." update_video_progress( vid_id, status="várakozik", error_message=msg, ) task = self._build_task_dict_from_db(v) task["status"] = "queued" task["error_message"] = msg task["progress"] = 0.0 with self._lock: if not any(t["id"] == vid_id for t in self.queue): self.queue.append(task) self._notify_listeners(task) summary["uploading_paused"] += 1 logger.info(f"Félbeszakadt feltöltés #{vid_id} ({v.get('file_name')}) visszaállítva 'várakozik' állapotba (nem folytatódik automatikusan).") # 2. 'várakozik' -> Sorba visszatöltés, manuális indításra várva elif status_code == "queued": task = self._build_task_dict_from_db(v) task["status"] = "queued" task["progress"] = 0.0 with self._lock: if not any(t["id"] == vid_id for t in self.queue): self.queue.append(task) summary["queued_restored"] += 1 # 3. 'ellenőrzés alatt' -> Ellenőrizze hogy fel van-e töltve és elérhető-e a Mozisarok-on elif status_code == "verifying": embed_url = v.get("embed_url") direct_url = v.get("direct_url") provider_video_id = v.get("provider_video_id") verified_ok = False if embed_url: try: ver_res = self.mozisarok.verify_video( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=embed_url, season=v.get("season"), episode=v.get("episode"), ) if ver_res.get("url_reachable"): verified_ok = True if not v.get("mozisarok_submitted") or not ver_res.get("mozisarok_found"): sub_res = self.mozisarok.submit_link( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=embed_url, language=v.get("language") or "Magyar", subtitle=v.get("subtitle"), quality=v.get("quality") or "1080p", season=v.get("season"), episode=v.get("episode"), anonymous=bool(v.get("anonymous")), ) if sub_res.get("success"): update_video_progress(vid_id, mozisarok_submitted=1) update_video_progress( vid_id, status="completed", mozisarok_verified=1, completed=True, error_message=None, ) summary["verifying_completed"] += 1 logger.info(f"Ellenőrzés alatt álló videó #{vid_id} sikeresen ellenőrizve és lezárva (kész).") except Exception as e: logger.warning(f"Hiba a(z) #{vid_id} ellenőrzésekor: {e}") if not verified_ok: is_on_provider = False if direct_url: try: chk = requests.get(direct_url, timeout=10) if chk.status_code == 200: is_on_provider = True except Exception: pass if is_on_provider: try: scraper_embed = self.converter.convert_url(url=direct_url, vcode=provider_video_id) final_embed = scraper_embed or embed_url or direct_url sub_res = self.mozisarok.submit_link( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=final_embed, language=v.get("language") or "Magyar", subtitle=v.get("subtitle"), quality=v.get("quality") or "1080p", season=v.get("season"), episode=v.get("episode"), anonymous=bool(v.get("anonymous")), ) v_res = self.mozisarok.verify_video( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=final_embed, season=v.get("season"), episode=v.get("episode"), ) is_ok = bool(v_res.get("url_reachable")) update_video_progress( vid_id, embed_url=final_embed, mozisarok_submitted=1 if sub_res.get("success") else 0, mozisarok_verified=1 if is_ok else 0, status="completed" if is_ok else "hiba", completed=is_ok, error_message=None if is_ok else "A videó link ellenőrzése sikertelen.", ) if is_ok: summary["verifying_completed"] += 1 else: summary["verifying_failed"] += 1 except Exception as e: logger.error(f"Hiba a feltöltött videó (#{vid_id}) újrakonvertálásakor: {e}") summary["verifying_failed"] += 1 else: update_video_progress( vid_id, error_message="A feltöltött videó nem érhető el a szolgáltatónál.", ) summary["verifying_failed"] += 1 # 4. 'beküldés alatt' -> Embed link beküldése és ellenőrzése elif status_code == "submitting": embed_url = v.get("embed_url") if not embed_url and v.get("direct_url"): embed_url = self.converter.convert_url(url=v["direct_url"], vcode=v.get("provider_video_id")) or v["direct_url"] if embed_url: try: sub_res = self.mozisarok.submit_link( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=embed_url, language=v.get("language") or "Magyar", subtitle=v.get("subtitle"), quality=v.get("quality") or "1080p", season=v.get("season"), episode=v.get("episode"), anonymous=bool(v.get("anonymous")), ) ver_res = self.mozisarok.verify_video( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=embed_url, season=v.get("season"), episode=v.get("episode"), ) is_ok = bool(ver_res.get("url_reachable")) update_video_progress( vid_id, embed_url=embed_url, mozisarok_submitted=1 if sub_res.get("success") else 0, mozisarok_verified=1 if is_ok else 0, status="completed" if is_ok else "hiba", completed=is_ok, ) summary["submitting_completed"] += 1 except Exception as e: logger.error(f"Hiba a(z) #{vid_id} beküldése során: {e}") # 5. 'konvertálás alatt' / 'feltöltve' -> Biztonságos helyreállítás a sorba (NE zárja le korábban!) elif status_code in ["converting", "uploaded"]: provider_name = v.get("provider") or "videa" msg = "Várakozás a Videa átalakító sorára (Folytatásra vár)..." if provider_name == "videa" else "Feltöltve, beküldésre vár (Folytatásra vár)..." update_video_progress( vid_id, status="converting" if provider_name == "videa" else "submitting", error_message=msg, completed=False, ) task = self._build_task_dict_from_db(v) task["status"] = "converting" if provider_name == "videa" else "submitting" task["error_message"] = msg task["progress"] = 0.0 with self._lock: if not any(t["id"] == vid_id for t in self.queue): self.queue.append(task) summary["converting_pending"] += 1 logger.info(f"Félbeszakadt videó #{vid_id} ({v.get('file_name')}) visszatöltve a várakozási sorba: {task['status']}") self._notify_listeners(task) logger.info(f"Rendszerállapot helyreállítása befejeződött: {summary}") return summary def force_check_video(self, video_id: int) -> Dict[str, Any]: """ Manually forces verification and advancement for a video that may be stuck in 'konvertálás alatt', 'ellenőrzés alatt', or showing 0% progress. 1. Checks provider transcoding (e.g. Videa player URL) 2. Converts via Video Scraper to obtain clean embed link 3. Submits link to Mozisarok (if not yet submitted) 4. Verifies link reachability and marks video as completed ('kész') """ logger.info(f"Manuális kényszerített ellenőrzés indítása a(z) #{video_id} azonosítójú videóhoz...") v = get_video_with_details(video_id) if not v: return {"success": False, "error": f"A #{video_id} azonosítójú videó nem található az adatbázisban."} provider_name = v.get("provider") or "videa" provider_video_id = v.get("provider_video_id") direct_url = v.get("direct_url") embed_url = v.get("embed_url") # Extract vcode from urls if missing if not provider_video_id: for candidate in [direct_url, embed_url]: if candidate and "videa.hu" in candidate: if "player?v=" in candidate: provider_video_id = candidate.split("player?v=")[-1].split("&")[0] elif "-" in candidate.rstrip("/").split("/")[-1]: provider_video_id = candidate.rstrip("/").split("/")[-1].split("-")[-1] else: provider_video_id = candidate.rstrip("/").split("/")[-1] if provider_video_id: break if provider_name == "videa" and not provider_video_id: return { "success": False, "error": "A videóhoz nem tartozik Videa kód (vcode). A nyers feltöltés feltehetően félbeszakadt vagy nem fejeződött be.", } final_embed = None if provider_name == "videa" and provider_video_id: account = get_account_by_id(v["account_id"]) if v.get("account_id") else None if not account: active_accs = get_accounts(provider="videa", active_only=True) if active_accs: account = active_accs[0] prov = get_provider("videa") is_done = prov.check_conversion_status(account=account, vcode=provider_video_id) logger.info(f"Videa konvertálás állapot #{video_id} (vcode: {provider_video_id}): {is_done}") # Try converting via Video Scraper with player URL player_url = f"https://videa.hu/player?v={provider_video_id}" scraper_embed = self.converter.convert_url(url=player_url, vcode=provider_video_id, max_retries=3, retry_delay=2.0) if scraper_embed: final_embed = scraper_embed logger.info(f"Video Scraper sikeresen előállította az embed linket: {final_embed}") elif is_done: # Video is ready on Videa, fallback to standard player embed final_embed = player_url logger.info(f"Video Scraper nem adott választ, fallback embed használata: {final_embed}") else: return { "success": False, "status": "converting", "message": "A videó még konvertálás alatt áll a Videa szerverén. Próbáld újra pár perc múlva.", } elif provider_name == "vk": final_embed = embed_url or direct_url else: final_embed = embed_url or direct_url if not final_embed: return { "success": False, "error": "Nem sikerült érvényes embed linket generálni a videóhoz.", } # Submit to Mozisarok logger.info(f"Link beküldése Mozisarok-ra: {final_embed}") sub_res = self.mozisarok.submit_link( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=final_embed, language=v.get("language") or "Magyar", subtitle=v.get("subtitle"), quality=v.get("quality") or "1080p", season=v.get("season"), episode=v.get("episode"), anonymous=bool(v.get("anonymous")), ) # Verify on Mozisarok ver_res = self.mozisarok.verify_video( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=final_embed, season=v.get("season"), episode=v.get("episode"), ) is_reachable = bool(ver_res.get("url_reachable")) # Update MySQL update_video_progress( video_id, embed_url=final_embed, mozisarok_submitted=1 if sub_res.get("success") else 0, mozisarok_verified=1 if is_reachable else 0, status="completed" if is_reachable else "hiba", completed=is_reachable, error_message=None if is_reachable else sub_res.get("error", "Az embed link nem elérhető."), ) # Update in-memory queue task with self._lock: for t in self.queue: if t["id"] == video_id: t["status"] = "completed" if is_reachable else "failed" t["progress"] = 100.0 if is_reachable else t.get("progress", 0.0) t["embed_url"] = final_embed t["error_message"] = None if is_reachable else sub_res.get("error") self._notify_listeners(t) return { "success": is_reachable, "status": "completed" if is_reachable else "failed", "embed_url": final_embed, "mozisarok_submitted": bool(sub_res.get("success")), "mozisarok_verified": is_reachable, "message": f"Kényszerített ellenőrzés sikeres! A videó lezárva (kész). Embed link: {final_embed}" if is_reachable else f"Beküldési/ellenőrzési hiba: {sub_res.get('error')}", } def resubmit_video_to_mozisarok(self, video_id: int, force: bool = False) -> Dict[str, Any]: """ Selects a saved video (movie or TV series episode) from MySQL, checks whether its embed link is already on Mozisarok for this media item, and submits/re-submits it if not already present (or if force=True). """ logger.info(f"Embed újrabeküldés indítása a(z) #{video_id} azonosítójú videóhoz (force={force})...") v = get_video_with_details(video_id) if not v: return {"success": False, "error": f"A #{video_id} azonosítójú videó nem található az adatbázisban."} embed_url = v.get("embed_url") if not embed_url and v.get("direct_url"): scraper_embed = self.converter.convert_url(url=v["direct_url"], vcode=v.get("provider_video_id")) embed_url = scraper_embed or v.get("direct_url") if not embed_url: return { "success": False, "error": "A videóhoz nem található érvényes embed link. Előbb fejezd be a feltöltést!", } res = self.mozisarok.check_and_resubmit_link( media_type=v["media_type"], tmdb_id=v["tmdb_id"], url=embed_url, direct_url=v.get("direct_url"), provider_video_id=v.get("provider_video_id"), language=v.get("language") or "Magyar", subtitle=v.get("subtitle"), quality=v.get("quality") or "1080p", season=v.get("season"), episode=v.get("episode"), anonymous=bool(v.get("anonymous")), force=force, ) if res.get("success"): update_video_progress( video_id, embed_url=embed_url, mozisarok_submitted=1, mozisarok_verified=1 if res.get("verified") else 0, status="completed" if res.get("verified") else v.get("status"), completed=bool(res.get("verified")), ) with self._lock: for t in self.queue: if t["id"] == video_id: if res.get("verified"): t["status"] = "completed" t["progress"] = 100.0 t["embed_url"] = embed_url self._notify_listeners(t) return res def check_and_submit_series_batch( self, tmdb_id: int, season: Optional[int] = None, only_missing: bool = True, force: bool = False, language: Optional[str] = None, quality: Optional[str] = None, check_only: bool = False, episodes_data: Optional[List[Dict[str, Any]]] = None, ) -> Dict[str, Any]: """ Batch checks and submits episodes of a TV series to Mozisarok. - tmdb_id: TMDb ID of the TV series. - season: Optional season filter (None = all seasons). - only_missing: If True, checks Mozisarok first and SKIPS episodes that are already present. - force: If True, re-submits even if already present. - check_only: If True, only checks Mozisarok links for each episode without submitting anything. - episodes_data: Optional custom list of episodes (dict with season, episode, url/embed_url). If omitted, fetches episodes from MySQL database. """ logger.info( f"Kötegelt sorozat feldolgozás indítása (TMDb: {tmdb_id}, Évad: {season}, " f"only_missing={only_missing}, force={force}, check_only={check_only})..." ) if episodes_data: raw_episodes = episodes_data else: raw_episodes = get_series_episodes_from_db(tmdb_id, season) if not raw_episodes: return { "success": False, "error": f"Nem található mentett epizód az adatbázisban a #{tmdb_id} azonosítójú sorozathoz.", "total_episodes": 0, "already_present": 0, "submitted": 0, "failed": 0, "episodes": [], } # Deduplicate raw episodes by (season, episode), keeping the most complete record episodes_by_key = {} for ep in raw_episodes: s = ep.get("season") or 1 e = ep.get("episode") or 1 key = (int(s), int(e)) if key not in episodes_by_key: episodes_by_key[key] = ep else: existing = episodes_by_key[key] # Priority: verified > submitted > has embed_url > higher video ID if not existing.get("mozisarok_verified") and ep.get("mozisarok_verified"): episodes_by_key[key] = ep elif not existing.get("embed_url") and ep.get("embed_url"): episodes_by_key[key] = ep elif ep.get("id", 0) > existing.get("id", 0): episodes_by_key[key] = ep sorted_episodes = sorted( episodes_by_key.values(), key=lambda x: (int(x.get("season") or 1), int(x.get("episode") or 1)), ) results = [] already_present_count = 0 submitted_count = 0 failed_count = 0 for ep in sorted_episodes: ep_season = int(ep.get("season") or 1) ep_episode = int(ep.get("episode") or 1) ep_id = ep.get("id") ep_file = ep.get("file_name") or ep.get("file_path") or f"S{ep_season:02d}E{ep_episode:02d}" # Determine embed URL embed_url = ep.get("embed_url") or ep.get("url") if not embed_url and ep.get("direct_url"): scraper_embed = self.converter.convert_url(url=ep["direct_url"], vcode=ep.get("provider_video_id")) embed_url = scraper_embed or ep.get("direct_url") # 1. Check existing links on Mozisarok for this specific (season, episode) across all positions existing_links = self.mozisarok.get_links( media_type="tv", tmdb_id=tmdb_id, season=ep_season, episode=ep_episode, ) matched_link, matched_pos = self.mozisarok.find_matching_link( url=embed_url or "", links=existing_links, direct_url=ep.get("direct_url"), provider_video_id=ep.get("provider_video_id"), ) is_present_on_mozisarok = bool(matched_link) ep_res = { "video_id": ep_id, "season": ep_season, "episode": ep_episode, "file_name": ep_file, "embed_url": embed_url, "links_count": len(existing_links), "matched_position": matched_pos, "matched_link": matched_link, "mozisarok_found": is_present_on_mozisarok, "status": "pending", "message": "", } # If check_only mode: report status and skip submitting if check_only: if is_present_on_mozisarok: ep_res["status"] = "already_exists" ep_res["message"] = ( f"Már fent van Mozisarok-on a(z) {matched_pos}. helyen ({len(existing_links)} db aktív linkből)." ) already_present_count += 1 else: ep_res["status"] = "missing" if existing_links: ep_res["message"] = f"Hiányzik Mozisarokról ({len(existing_links)} db egyéb link található)." else: ep_res["message"] = "Hiányzik Mozisarokról (nincs még aktív link)." results.append(ep_res) continue # Submitting mode: # If already present and only_missing is True and not force: SKIP if is_present_on_mozisarok and only_missing and not force: logger.info( f"Sorozat S{ep_season:02d}E{ep_episode:02d} már elérhető Mozisarok-on a(z) {matched_pos}. helyen " f"({len(existing_links)} db link), kihagyva a duplikáció elkerülése végett." ) ep_res["status"] = "already_exists" ep_res["message"] = ( f"Már fent van Mozisarok-on a(z) {matched_pos}. helyen ({len(existing_links)} db linkből), " f"kihagyva a duplikáció elkerülése végett." ) already_present_count += 1 # Sync MySQL status if ep_id: update_video_progress( ep_id, mozisarok_submitted=1, mozisarok_verified=1, ) results.append(ep_res) continue # If embed_url is missing, we cannot submit to Mozisarok if not embed_url: ep_res["status"] = "failed" ep_res["message"] = "Hiányzó embed URL; a videó még nincs feltöltve/konvertálva." failed_count += 1 results.append(ep_res) continue # Submit missing (or forced) episode to Mozisarok sub_lang = language or ep.get("language") or "Magyar" sub_qual = quality or ep.get("quality") or "1080p" sub_sub = ep.get("subtitle") sub_anon = bool(ep.get("anonymous", False)) logger.info(f"Hiányzó epizód beküldése Mozisarok-ra: S{ep_season:02d}E{ep_episode:02d} -> {embed_url}") sub_res = self.mozisarok.submit_link( media_type="tv", tmdb_id=tmdb_id, url=embed_url, language=sub_lang, subtitle=sub_sub, quality=sub_qual, season=ep_season, episode=ep_episode, anonymous=sub_anon, ) if not sub_res.get("success"): err_msg = sub_res.get("error", "Ismeretlen beküldési hiba") logger.warning(f"Mozisarok beküldés sikertelen (S{ep_season:02d}E{ep_episode:02d}): {err_msg}") ep_res["status"] = "failed" ep_res["message"] = f"Beküldési hiba: {err_msg}" failed_count += 1 if ep_id: update_video_progress(ep_id, error_message=err_msg) else: # Post-verify ver_res = self.mozisarok.verify_video( media_type="tv", tmdb_id=tmdb_id, url=embed_url, season=ep_season, episode=ep_episode, ) is_ok = bool(ver_res.get("url_reachable")) ep_res["status"] = "submitted" ep_res["message"] = "Sikeresen beküldve és ellenőrizve!" if is_ok else "Beküldve, de az embed URL elérhetősége nem 200." ep_res["mozisarok_verified"] = is_ok submitted_count += 1 if ep_id: update_video_progress( ep_id, embed_url=embed_url, mozisarok_submitted=1, mozisarok_verified=1 if is_ok else 0, status="completed" if is_ok else "hiba", completed=is_ok, ) results.append(ep_res) logger.info( f"Sorozat kötegelt beküldés befejezve: összes={len(sorted_episodes)}, " f"már fent={already_present_count}, beküldve={submitted_count}, hiba={failed_count}" ) return { "success": True, "tmdb_id": tmdb_id, "season": season, "total_episodes": len(sorted_episodes), "already_present": already_present_count, "submitted": submitted_count, "failed": failed_count, "episodes": results, "message": ( f"Kötegelt ellenőrzés lefutott: {len(sorted_episodes)} részből " f"{already_present_count} már fent volt, {submitted_count} új rész beküldve, {failed_count} hiba." ), } def start_queue(self): """Manually triggers the upload queue processing in the background.""" with self._lock: if self.is_running: logger.warning("Upload queue is already running.") return self.is_running = True self._stop_event.clear() self._executor = ThreadPoolExecutor(max_workers=max(32, self.videa_concurrency + 25)) worker_thread = threading.Thread(target=self._process_queue_loop, daemon=True) worker_thread.start() logger.info( f"Manual start triggered: Upload queue started (videa_concurrency={self.videa_concurrency}, max_converting={self.max_videa_converting})." ) def stop_queue(self): """Stops the upload queue.""" with self._lock: self.is_running = False self._stop_event.set() if self._executor: self._executor.shutdown(wait=False) self._executor = None logger.info("Upload queue paused/stopped.") def _process_queue_loop(self): """ Main dispatcher loop: 1. Videa upload limit: at most videa_concurrency videos actively uploading. 2. Videa conversion limit: at most 20 videos converting. If 20 are converting, no new video upload can be started until one frees up (drops below 20). 3. Simultaneous execution: while videos are converting or verifying, uploads can run. 4. Resumes monitoring of recovered converting tasks in the background. """ while not self._stop_event.is_set(): next_task = None with self._lock: # 1. Check for any recovered converting tasks that need active monitoring for task in self.queue: tid = task["id"] if ( task.get("provider") == "videa" and task.get("status") == "converting" and tid not in self._active_converting_task_ids ): self._active_converting_task_ids.add(tid) if self._executor: self._executor.submit(self._resume_converting_task, task) # 2. Look for next queued task that satisfies provider limits for task in self.queue: if task["status"] == "queued": prov = task.get("provider", "videa").lower() if prov == "videa": uploading_videa = sum( 1 for t in self.queue if t.get("provider") == "videa" and t.get("status") in ["uploading", "waiting"] ) converting_videa = sum( 1 for t in self.queue if t.get("provider") == "videa" and t.get("status") in ["converting", "uploaded"] ) try: db_converting = count_videa_converting_videos() converting_videa = max(converting_videa, db_converting) except Exception: pass # Korlát: 20 konvertálás alatt videónál nem lehet újat feltölteni! if converting_videa >= self.max_videa_converting: task["error_message"] = ( f"Várakozás konvertálási helyre ({converting_videa}/{self.max_videa_converting})..." ) continue # Korlát: Egyszerre feltöltés alatt álló videók száma (videa_concurrency) if uploading_videa >= self.videa_concurrency: continue # Mindkét feltétel teljesült: elindítható a feltöltés! if task.get("error_message") and "Várakozás konvertálási helyre" in task["error_message"]: task["error_message"] = None task["status"] = "waiting" next_task = task break elif prov == "vk": # VK párhuzamosítás: Videa-tól függetlenül futhat a megadott vk_concurrency szerint uploading_vk = sum( 1 for t in self.queue if t.get("provider") == "vk" and t.get("status") in ["uploading", "waiting"] ) if uploading_vk >= self.vk_concurrency: continue task["status"] = "waiting" next_task = task break else: # Egyéb szolgáltató task["status"] = "waiting" next_task = task break if not next_task: # Check if any tasks are still running or waiting with self._lock: running = any( t["status"] in ["waiting", "uploading", "converting", "submitting", "verifying"] for t in self.queue ) has_queued = any(t["status"] == "queued" for t in self.queue) if not running and not has_queued: self.is_running = False break self._stop_event.wait(1.0) continue # Submit task to thread pool if self._executor: self._executor.submit(self._execute_single_task, next_task) def _execute_single_task(self, task: Dict[str, Any]): """Executes the multi-step lifecycle for a single video task.""" task_id = task["id"] provider_name = task["provider"] # Acquire upload concurrency semaphore strictly for the raw upload phase (Videa vagy VK) upload_sem = self._videa_upload_semaphore if provider_name == "videa" else (self._vk_upload_semaphore if provider_name == "vk" else None) if upload_sem: upload_sem.acquire() upload_sem_released = False try: logger.info(f"Starting processing for task #{task_id}: {task['file_name']}") task["status"] = "uploading" update_video_progress(task_id, status="uploading") self._notify_listeners(task) cancel_evt = self._cancel_events.get(task_id) if cancel_evt and cancel_evt.is_set(): raise UploadCancelledException("Feltöltés már az indítás előtt meg lett szakítva.") # Step 1: Resolve Account account = None if task.get("account_id"): account = get_account_by_id(task["account_id"]) if not account: active_accounts = get_accounts(provider=provider_name, active_only=True) if active_accounts: account = active_accounts[0] provider = get_provider(provider_name) # Progress callback helper def on_progress(pct: float): task["progress"] = round(pct, 1) self._notify_listeners(task) # Step 2: Upload video to chosen provider title = f"{task['media_title']}" if task["season"] is not None and task["episode"] is not None: title += f" S{task['season']:02d}E{task['episode']:02d}" upload_result = provider.upload( file_path=task["file_path"], title=title, description=f"{task['media_title']} - Mozisarok feltöltés", account=account, progress_callback=on_progress, cancel_event=cancel_evt, ) if cancel_evt and cancel_evt.is_set(): raise UploadCancelledException("Feltöltés megszakítva a felhasználó által.") task["direct_url"] = upload_result.direct_url task["provider_video_id"] = upload_result.provider_video_id embed_url = upload_result.embed_url task["status"] = "uploaded" task["progress"] = 100.0 update_video_progress( task_id, status="uploaded", provider_video_id=upload_result.provider_video_id, direct_url=upload_result.direct_url, ) self._notify_listeners(task) # ------------------------------------------------------------------ # FELTÖLTÉSI FÁZIS VÉGE: Szemafor azonnali felszabadítása! # A nyers feltöltés befejeződött, így a feltöltési szál felszabadul, # és a következő várakozó videó megkezdheti a feltöltést, miközben # ez a videó konvertálás vagy ellenőrzés alatt áll. # ------------------------------------------------------------------ if upload_sem and not upload_sem_released: upload_sem.release() upload_sem_released = True # Step 3: Videa: várja meg mire lekonvertálja a videa teljesen, és utána konvertálja le Video Scraper-rel if provider_name == "videa": task["status"] = "converting" task["progress"] = 0.0 task["error_message"] = "Várakozás a Videa átalakító sorára..." with self._lock: self._active_converting_task_ids.add(task_id) update_video_progress( task_id, status="converting", provider_video_id=upload_result.provider_video_id, direct_url=upload_result.direct_url, ) self._notify_listeners(task) def on_conversion_progress(pct: float, msg: str = ""): task["progress"] = round(pct, 1) if msg: task["error_message"] = msg self._notify_listeners(task) logger.info(f"Waiting for Videa conversion for task #{task_id} (vcode: {upload_result.provider_video_id})...") provider_session = upload_result.raw_data.get("session") if upload_result.raw_data else None provider.wait_for_conversion( account=account, vcode=upload_result.provider_video_id, progress_callback=on_conversion_progress, cancel_event=cancel_evt, session=provider_session, ) if cancel_evt and cancel_evt.is_set(): raise UploadCancelledException("Feltöltés megszakítva a felhasználó által.") logger.info(f"Videa conversion complete for task #{task_id}. Converting with Video Scraper...") task["error_message"] = "Konvertálás a Video Scraper-rel..." self._notify_listeners(task) scraper_embed = self.converter.convert_url( url=upload_result.direct_url, vcode=upload_result.provider_video_id, cancel_event=cancel_evt, ) if scraper_embed: embed_url = scraper_embed logger.info(f"Task #{task_id} converted embed link: {embed_url}") else: logger.warning(f"Task #{task_id} Video Scraper did not return embed, using default: {embed_url}") task["error_message"] = None with self._lock: self._active_converting_task_ids.discard(task_id) task["embed_url"] = embed_url task["status"] = "submitting" update_video_progress( task_id, status="submitting", provider_video_id=upload_result.provider_video_id, direct_url=upload_result.direct_url, embed_url=embed_url, ) self._notify_listeners(task) # Step 4: Submit embed URL to Mozisarok API logger.info(f"Submitting link to Mozisarok: {embed_url}") sub_res = self.mozisarok.submit_link( media_type=task["media_type"], tmdb_id=task["tmdb_id"], url=embed_url, language=task["language"], subtitle=task["subtitle"], quality=task["quality"], season=task["season"], episode=task["episode"], anonymous=task.get("anonymous", False), ) if not sub_res.get("success"): err_msg = sub_res.get("error", "Ismeretlen Mozisarok beküldési hiba") logger.error(f"Mozisarok link submission error: {err_msg}") task["error_message"] = f"Mozisarok beküldési hiba: {err_msg}" update_video_progress(task_id, mozisarok_submitted=0, error_message=err_msg) else: logger.info(f"Mozisarok submission successful for task #{task_id}") update_video_progress(task_id, mozisarok_submitted=1) # Step 5: Verify the video task["status"] = "verifying" self._notify_listeners(task) verification = self.mozisarok.verify_video( media_type=task["media_type"], tmdb_id=task["tmdb_id"], url=embed_url, season=task["season"], episode=task["episode"], ) task["verification"] = verification task["status"] = "completed" task["progress"] = 100.0 update_video_progress( task_id, status="completed", mozisarok_verified=1 if verification.get("url_reachable") else 0, completed=True, ) logger.info(f"Task #{task_id} completed successfully. Verification: {verification}") self._notify_listeners(task) except UploadCancelledException as ce: logger.info(f"Task #{task_id} upload cancelled: {ce}") task["status"] = "cancelled" task["error_message"] = "Feltöltés megszakítva a felhasználó által." update_video_progress(task_id, status="cancelled", error_message="Feltöltés megszakítva a felhasználó által.") self._notify_listeners(task) except Exception as e: logger.exception(f"Error processing task #{task_id}: {e}") task["status"] = "failed" task["error_message"] = str(e) update_video_progress(task_id, status="failed", error_message=str(e)) self._notify_listeners(task) finally: if upload_sem and not upload_sem_released: upload_sem.release() upload_sem_released = True with self._lock: self._active_converting_task_ids.discard(task_id) self._cancel_events.pop(task_id, None) def _resume_converting_task(self, task: Dict[str, Any]): """Resumes monitoring and finalization for a task that was already uploaded and is in conversion.""" task_id = task["id"] cancel_evt = self._cancel_events.get(task_id) provider_name = task.get("provider", "videa") provider_video_id = task.get("provider_video_id") direct_url = task.get("direct_url") try: logger.info(f"Resuming conversion monitoring for task #{task_id}: {task.get('file_name', '')}") account = None if task.get("account_id"): account = get_account_by_id(task["account_id"]) if not account: active_accounts = get_accounts(provider=provider_name, active_only=True) if active_accounts: account = active_accounts[0] provider = get_provider(provider_name) def on_conversion_progress(pct: float, msg: str = ""): task["progress"] = round(pct, 1) if msg: task["error_message"] = msg self._notify_listeners(task) # Wait for conversion if provider_name == "videa" and provider_video_id: provider.wait_for_conversion( account=account, vcode=provider_video_id, progress_callback=on_conversion_progress, cancel_event=cancel_evt, ) if cancel_evt and cancel_evt.is_set(): raise UploadCancelledException("Feltöltés megszakítva a felhasználó által.") task["error_message"] = "Konvertálás a Video Scraper-rel..." self._notify_listeners(task) scraper_embed = self.converter.convert_url( url=direct_url or f"https://videa.hu/player?v={provider_video_id}", vcode=provider_video_id, cancel_event=cancel_evt, ) embed_url = scraper_embed or task.get("embed_url") or direct_url else: embed_url = task.get("embed_url") or direct_url task["embed_url"] = embed_url task["status"] = "submitting" update_video_progress( task_id, status="submitting", embed_url=embed_url, ) self._notify_listeners(task) # Submit to Mozisarok sub_res = self.mozisarok.submit_link( media_type=task["media_type"], tmdb_id=task["tmdb_id"], url=embed_url, language=task["language"], subtitle=task["subtitle"], quality=task["quality"], season=task["season"], episode=task["episode"], anonymous=task.get("anonymous", False), ) task["status"] = "verifying" self._notify_listeners(task) verification = self.mozisarok.verify_video( media_type=task["media_type"], tmdb_id=task["tmdb_id"], url=embed_url, season=task["season"], episode=task["episode"], ) is_ok = bool(verification.get("url_reachable")) task["verification"] = verification task["status"] = "completed" if is_ok else "failed" task["progress"] = 100.0 if is_ok else task.get("progress", 0.0) update_video_progress( task_id, status="completed" if is_ok else "hiba", mozisarok_submitted=1 if sub_res.get("success") else 0, mozisarok_verified=1 if is_ok else 0, completed=is_ok, ) logger.info(f"Resumed task #{task_id} completed: {verification}") self._notify_listeners(task) except UploadCancelledException: task["status"] = "cancelled" update_video_progress(task_id, status="cancelled", error_message="Feltöltés megszakítva.") self._notify_listeners(task) except Exception as e: logger.error(f"Error resuming task #{task_id}: {e}") task["status"] = "failed" task["error_message"] = str(e) update_video_progress(task_id, status="failed", error_message=str(e)) self._notify_listeners(task) finally: with self._lock: self._active_converting_task_ids.discard(task_id)