""" state.py - State management using HuggingFace Dataset as persistent storage. Since HF Spaces free tier has ephemeral storage (lost on restart), we use a private HF Dataset as our "database" to persist: - Processed video IDs (avoid duplicates) - YouTube OAuth tokens (for uploading) - Processing history and stats """ import json import os import tempfile from datetime import datetime from pathlib import Path from typing import Any, Optional from huggingface_hub import HfApi, hf_hub_download, upload_file class StateManager: """Manages persistent state via a HuggingFace Dataset.""" def __init__(self): self.token = os.getenv("HF_TOKEN", "") self.repo_id = os.getenv("HF_STATE_DATASET", "autodub-herspace/state") self.api = HfApi(token=self.token) self._state: dict = {} self._local_path = Path(tempfile.gettempdir()) / "autodub_state.json" # Default state structure self._defaults = { "processed_videos": {}, # video_id -> {title, date_processed, es_title} "youtube_tokens": {}, # OAuth tokens for YouTube upload "stats": { "total_processed": 0, "total_failed": 0, "last_check": None, "last_process": None, }, "queue": [], # Videos queued for processing "config": { "check_interval_minutes": 60, "target_channel": "HerYTStory", "language": "es", }, # Visual positioning for blur + subtitles (720x1280 canvas) # VLM-VERIFIED config (2026-07-03): # English captions detected at y=920-960 in scaled frame # Blur at y=895-1005 (full width 720, 110px tall) covers all captions # Spanish subs ON the blur (MarginV=288) replacing English captions in same position "visual_config": { # Blur region (full width to cover all caption positions) "blur_x": 0, # X offset (0 = full width) "blur_y": 895, # Y start (just above captions at y=920) "blur_w": 720, # Full width "blur_h": 110, # Covers y=895-1005 (captions + padding) "blur_strength": 20, # Boxblur strength # Spanish subtitle positioning — ON the blur (same position as English captions) "sub_font_size": 42, # ASS FontSize (Tahoma Bold) "sub_margin_v": 288, # ASS MarginV (subs on blur at y=895-1005) "sub_alignment": 2, # ASS Alignment (2=bottom-center) "sub_margin_l": 20, # ASS MarginL "sub_margin_r": 20, # ASS MarginR # Voice configuration "tts_engine": "edge-tts", # "edge-tts" (primary), "supertonic" or "gtts" "edge_tts_voice": "es-ES-XimenaNeural", # edge-tts voice "edge_tts_rate": "+0%", # natural speed "supertonic_voice": "M1", # fallback Supertonic voice }, } def _ensure_repo(self): """Create the dataset repo if it doesn't exist.""" try: self.api.repo_info(repo_id=self.repo_id, repo_type="dataset") except Exception: self.api.create_repo( repo_id=self.repo_id, repo_type="dataset", private=True, exist_ok=True, ) # Upload initial state initial = json.dumps(self._defaults, indent=2, ensure_ascii=False) with open(self._local_path, "w", encoding="utf-8") as f: f.write(initial) self.api.upload_file( path_or_fileobj=str(self._local_path), path_in_repo="state.json", repo_id=self.repo_id, repo_type="dataset", token=self.token, ) def load(self) -> dict: """Load state from HF Dataset. Returns the state dict. CRITICAL: NEVER overwrite YouTube tokens with empty values. If HF download returns empty tokens, preserve from local cache. """ # First, load local cache if it exists (this has our tokens!) local_state = None if self._local_path.exists(): try: with open(self._local_path, "r", encoding="utf-8") as f: local_state = json.load(f) except Exception: pass # Save local tokens BEFORE any download (in case download overwrites) local_tokens = {} if local_state and local_state.get("youtube_tokens", {}).get("refresh_token"): local_tokens = local_state["youtube_tokens"] print(f"[STATE] Local cache has YouTube tokens (preserving)") try: self._ensure_repo() downloaded = hf_hub_download( repo_id=self.repo_id, filename="state.json", repo_type="dataset", token=self.token, ) with open(downloaded, "r", encoding="utf-8") as f: loaded = json.load(f) # Merge with defaults (in case new fields were added) self._state = {**self._defaults, **loaded} # Ensure nested dicts are merged too for key in self._defaults: if isinstance(self._defaults[key], dict) and key in loaded: self._state[key] = {**self._defaults[key], **loaded[key]} # CRITICAL: If remote state has NO tokens but local cache DOES, # use local tokens (don't let empty remote overwrite our tokens!) remote_tokens = loaded.get("youtube_tokens", {}) if not remote_tokens.get("refresh_token") and local_tokens.get("refresh_token"): self._state["youtube_tokens"] = local_tokens print("[STATE] ✅ Preserved YouTube tokens from local cache (remote was empty)") elif remote_tokens.get("refresh_token"): print("[STATE] ✅ Loaded YouTube tokens from remote dataset") except Exception as e: print(f"[STATE] Warning: Could not load state from HF: {e}") print("[STATE] Using local cache (preserving YouTube tokens)") if local_state: self._state = {**self._defaults, **local_state} for key in self._defaults: if isinstance(self._defaults[key], dict) and key in local_state: self._state[key] = {**self._defaults[key], **local_state[key]} if local_tokens: self._state["youtube_tokens"] = local_tokens print("[STATE] ✅ Preserved YouTube tokens from local cache") else: self._state = dict(self._defaults) # Save local copy as cache self._save_local() return self._state def save(self): """Save current state to HF Dataset and local cache. CRITICAL: Before saving, check if current state is missing youtube_tokens. If so, try to load existing tokens from dataset first to avoid overwriting them. """ # CRITICAL FIX: If we don't have youtube_tokens but the dataset does, # preserve them before saving if not self._state.get("youtube_tokens", {}).get("refresh_token"): try: # Try to fetch existing state from dataset via proxy import subprocess as _subproc proxy_url = "https://hf-proxy.t70512145.workers.dev" target = f"https://huggingface.co/datasets/{self.repo_id}/raw/main/state.json" encoded = target.replace(":", "%3A").replace("/", "%2F") result = _subproc.run([ 'curl', '-s', '--max-time', '15', f'{proxy_url}/proxy?url={encoded}', '-H', f'Authorization: Bearer {self.token}' ], capture_output=True, text=True, timeout=20) if result.returncode == 0 and result.stdout.strip(): import json as _json remote_state = _json.loads(result.stdout) remote_tokens = remote_state.get("youtube_tokens", {}) if remote_tokens.get("refresh_token"): self._state["youtube_tokens"] = remote_tokens print("[STATE] Preserved YouTube tokens from remote dataset before save") except Exception as e: print(f"[STATE] Could not check remote tokens: {e}") self._save_local() try: self._ensure_repo() with open(self._local_path, "w", encoding="utf-8") as f: json.dump(self._state, f, indent=2, ensure_ascii=False) # Try direct upload first (works if not rate-limited) try: self.api.upload_file( path_or_fileobj=str(self._local_path), path_in_repo="state.json", repo_id=self.repo_id, repo_type="dataset", token=self.token, ) print("[STATE] State saved to HF Dataset (direct)") return except Exception as direct_err: print(f"[STATE] Direct upload failed: {direct_err}, trying proxy...") # Fallback: upload via CF Worker proxy using commit API self._upload_via_proxy("state.json") print("[STATE] State saved to HF Dataset (via proxy)") except Exception as e: print(f"[STATE] Warning: Could not save state to HF: {e}") print("[STATE] State saved locally only") def _upload_via_proxy(self, path_in_repo: str): """Upload a file to HF Dataset via CF Worker proxy (bypasses rate limit).""" import subprocess import base64 proxy_url = os.getenv("HF_PROXY_URL", "https://hf-proxy.t70512145.workers.dev") with open(self._local_path, "r", encoding="utf-8") as f: content = f.read() # URL-encode the target URL target_url = f"https://huggingface.co/api/datasets/{self.repo_id}/commit/main" encoded_target = target_url.replace(":", "%3A").replace("/", "%2F").replace("?", "%3F") commit_body = { "summary": "Save state (via proxy)", "files": [{"path": path_in_repo, "content": content}], "deletions": [] } import tempfile as _tempfile import json as _json with _tempfile.NamedTemporaryFile(mode='w', suffix='.json', delete=False) as tmp: _json.dump(commit_body, tmp) tmp_path = tmp.name try: result = subprocess.run([ 'curl', '-s', '--max-time', '60', '-X', 'POST', f'{proxy_url}/proxy?url={encoded_target}', '-H', f'Authorization: Bearer {self.token}', '-H', 'Content-Type: application/json', '-d', f'@{tmp_path}', '-w', '\nHTTP: %{http_code}' ], capture_output=True, text=True, timeout=90) if 'HTTP: 200' not in result.stdout and 'HTTP: 201' not in result.stdout: raise Exception(f"Proxy upload failed: {result.stdout[-200:]}") finally: os.unlink(tmp_path) def _save_local(self): """Save state to local file as cache.""" try: with open(self._local_path, "w", encoding="utf-8") as f: json.dump(self._state, f, indent=2, ensure_ascii=False) except Exception as e: print(f"[STATE] Warning: Could not save local state: {e}") # --- Convenience methods --- def is_video_processed(self, video_id: str) -> bool: """Check if a video has already been SUCCESSFULLY processed. Videos that failed are NOT counted as processed, so they can be retried by the monitor. """ return video_id in self._state.get("processed_videos", {}) def is_video_known(self, video_id: str) -> bool: """Check if a video has been seen before (processed OR failed).""" processed = video_id in self._state.get("processed_videos", {}) failed = video_id in self._state.get("failed_videos", {}) return processed or failed def mark_video_processed(self, video_id: str, title: str, es_title: str = ""): """Mark a video as processed.""" if "processed_videos" not in self._state: self._state["processed_videos"] = {} self._state["processed_videos"][video_id] = { "title": title, "es_title": es_title, "date_processed": datetime.now().isoformat(), } self._state["stats"]["total_processed"] = self._state["stats"].get("total_processed", 0) + 1 self._state["stats"]["last_process"] = datetime.now().isoformat() self.save() def mark_video_failed(self, video_id: str, error: str, title: str = ""): """Record a failed processing attempt. Stores the video ID and error so it can be retried later. Tracks retry count to avoid infinite loops. """ if "failed_videos" not in self._state: self._state["failed_videos"] = {} existing = self._state["failed_videos"].get(video_id, {}) retry_count = existing.get("retry_count", 0) + 1 self._state["failed_videos"][video_id] = { "title": title or existing.get("title", ""), "error": error, "retry_count": retry_count, "last_failed_at": datetime.now().isoformat(), "first_failed_at": existing.get("first_failed_at", datetime.now().isoformat()), } self._state["stats"]["total_failed"] = self._state["stats"].get("total_failed", 0) + 1 self.save() def is_video_permanently_failed(self, video_id: str, max_retries: int = 3) -> bool: """Check if a video has failed too many times to retry.""" failed = self._state.get("failed_videos", {}).get(video_id, {}) return failed.get("retry_count", 0) >= max_retries def get_retryable_failed_videos(self, max_retries: int = 3) -> list[dict]: """Get failed videos that can still be retried.""" result = [] for vid_id, info in self._state.get("failed_videos", {}).items(): if info.get("retry_count", 0) < max_retries: result.append({ "video_id": vid_id, "title": info.get("title", ""), "error": info.get("error", ""), "retry_count": info.get("retry_count", 0), }) return result def clear_failed_video(self, video_id: str): """Remove a video from the failed list (e.g. after successful retry).""" if "failed_videos" in self._state and video_id in self._state["failed_videos"]: del self._state["failed_videos"][video_id] self.save() def get_youtube_tokens(self) -> dict: """Get stored YouTube OAuth tokens.""" return self._state.get("youtube_tokens", {}) def save_youtube_tokens(self, tokens: dict): """Save YouTube OAuth tokens to in-memory state AND force upload to HF Dataset. CRITICAL: This method MUST reliably save tokens to the dataset. The regular save() method can fail silently, losing tokens on restart. This method uses a dedicated upload with retry logic. """ self._state["youtube_tokens"] = tokens self._save_local() # FORCE upload to dataset with retry max_retries = 3 for attempt in range(max_retries): try: import json as _json # Write to temp file tmp_path = self._local_path.parent / "state_tokens.json" with open(tmp_path, "w", encoding="utf-8") as f: _json.dump(self._state, f, indent=2, ensure_ascii=False) self.api.upload_file( path_or_fileobj=str(tmp_path), path_in_repo="state.json", repo_id=self.repo_id, repo_type="dataset", token=self.token, ) print(f"[STATE] YouTube tokens saved to HF Dataset (attempt {attempt+1}) ✓") # Verify by reading back from huggingface_hub import hf_hub_download verify_path = hf_hub_download( repo_id=self.repo_id, filename="state.json", repo_type="dataset", token=self.token ) with open(verify_path) as f: verify_state = _json.load(f) if verify_state.get("youtube_tokens", {}).get("refresh_token"): print("[STATE] Verified: tokens are in dataset ✓") return else: print(f"[STATE] Verification failed: tokens not in dataset (attempt {attempt+1})") except Exception as e: print(f"[STATE] Token save attempt {attempt+1} failed: {e}") import time time.sleep(2) print("[STATE] WARNING: Could not save tokens to dataset after 3 attempts!") print("[STATE] Tokens are in memory only - will be lost on restart") def update_last_check(self): """Update the last check timestamp.""" self._state["stats"]["last_check"] = datetime.now().isoformat() self.save() def get_stats(self) -> dict: """Get processing statistics.""" return self._state.get("stats", {}) def get_config(self) -> dict: """Get current configuration.""" return self._state.get("config", {}) def update_config(self, config: dict): """Update configuration.""" self._state["config"] = {**self._state.get("config", {}), **config} self.save() def add_to_queue(self, video_id: str, title: str, url: str): """Add a video to the processing queue.""" if "queue" not in self._state: self._state["queue"] = [] # Don't add duplicates for item in self._state["queue"]: if item.get("video_id") == video_id: return self._state["queue"].append({ "video_id": video_id, "title": title, "url": url, "added_at": datetime.now().isoformat(), "status": "pending", }) self.save() def get_next_in_queue(self) -> Optional[dict]: """Get the next pending video from the queue.""" for item in self._state.get("queue", []): if item.get("status") == "pending": return item return None def update_queue_item(self, video_id: str, status: str, error: str = ""): """Update a queue item's status.""" for item in self._state.get("queue", []): if item.get("video_id") == video_id: item["status"] = status if error: item["error"] = error item["updated_at"] = datetime.now().isoformat() break self.save() # --- Visual config (blur + subtitle positions) --- def get_visual_config(self) -> dict: """Get visual positioning config (blur + subtitles).""" return self._state.get("visual_config", self._defaults.get("visual_config", {})) def update_visual_config(self, config: dict): """Update visual positioning config. Merges with existing.""" current = self._state.get("visual_config", {}) current.update(config) self._state["visual_config"] = current self.save() print(f"[STATE] Visual config updated: {config}") # --- Cookie persistence --- def load_cookies_txt(self) -> Optional[str]: """Load cookies.txt from the HF state dataset. Returns the text content or None.""" try: downloaded = hf_hub_download( repo_id=self.repo_id, filename="cookies.txt", repo_type="dataset", token=self.token, ) with open(downloaded, "r", encoding="utf-8") as f: content = f.read() print(f"[STATE] Loaded cookies.txt from dataset ({len(content)} bytes)") return content except Exception as e: print(f"[STATE] No cookies.txt in dataset: {e}") return None def save_cookies_txt(self, cookies_text: str): """Save cookies.txt to the HF state dataset for persistence across restarts.""" try: tmp_path = self._local_path.parent / "cookies_upload.txt" with open(tmp_path, "w", encoding="utf-8") as f: f.write(cookies_text) self.api.upload_file( path_or_fileobj=str(tmp_path), path_in_repo="cookies.txt", repo_id=self.repo_id, repo_type="dataset", token=self.token, ) print("[STATE] Cookies saved to HF Dataset") except Exception as e: print(f"[STATE] Failed to save cookies: {e}")