""" youtube_studio_manager.py - Autonomous YouTube Channel Operations Suite Supports: 1. Playwright Chrome (Stealth) with user profile 'mirsixabcf2' + Custom Extension 2. invisible_playwright (Patched Firefox C++ Engine) for extreme anti-bot protection """ import os import sys import time import json import random import argparse from pathlib import Path # Enable UTF-8 encoding on Windows console if hasattr(sys.stdout, 'reconfigure'): try: sys.stdout.reconfigure(encoding='utf-8') sys.stderr.reconfigure(encoding='utf-8') except Exception: pass PROJECT_DIR = Path(__file__).resolve().parent EXTENSION_DIR = PROJECT_DIR / "extension" METADATA_JSON = EXTENSION_DIR / "metadata.json" THUMBNAILS_4K_DIR = PROJECT_DIR / "content" / "thumbnails_4k" THUMBNAILS_DIR = PROJECT_DIR / "content" / "thumbnails" VIDEOS_4K_DIR = PROJECT_DIR / "content" / "videos_4k" PROFILES_DIR = PROJECT_DIR / "content" / "browser_profiles" PROFILES_DIR.mkdir(parents=True, exist_ok=True) CHROME_STUDIO_PROFILE = PROFILES_DIR / "chrome_mirsixabcf2" FIREFOX_STUDIO_PROFILE = PROFILES_DIR / "invisible_firefox" def load_metadata(): """Loads standardized metadata from extension/metadata.json.""" if METADATA_JSON.exists(): try: with open(METADATA_JSON, 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: print(f"[!] Warning reading metadata.json: {e}") # Fallback to main_automation if needed try: from main_automation import EPISODE_METADATA res = {} for k, v in EPISODE_METADATA.items(): res[str(k)] = v return res except Exception: return {} def human_delay(min_sec=0.8, max_sec=2.0): """Simulates realistic human interaction delay.""" time.sleep(random.uniform(min_sec, max_sec)) class YouTubeStudioManager: def __init__(self, mode="chrome_stealth"): """ mode: - 'chrome_stealth': Uses real Google Chrome with profile mirsixabcf2 + custom Extension. - 'invisible_playwright': Uses patched Firefox (feder-cr/invisible_playwright). """ self.mode = mode self.playwright = None self.browser = None self.context = None self.page = None self.ip_launcher = None def start(self, headless=False): print("=" * 70) print("🚀 KHỞI ĐỘNG HỆ THỐNG QUẢN TRỊ YOUTUBE STUDIO TỰ HÀNH") print(f" Động cơ: {self.mode.upper()}") print(f" Chế độ hiển thị: {'Headless (Chạy ngầm)' if headless else 'Visible (Cửa sổ)'}") print("=" * 70) if self.mode == "invisible_playwright": return self._start_invisible_playwright(headless) else: return self._start_chrome_stealth(headless) def _start_invisible_playwright(self, headless=False): try: from invisible_playwright import InvisiblePlaywright except ImportError: print("[-] Chưa cài đặt invisible_playwright. Đang cài đặt...") os.system(f"{sys.executable} -m pip install invisible-playwright") from invisible_playwright import InvisiblePlaywright FIREFOX_STUDIO_PROFILE.mkdir(parents=True, exist_ok=True) print(f"[1/3] Khởi chạy invisible_playwright (Firefox C++ Patched Engine)...") print(f" Profile: {FIREFOX_STUDIO_PROFILE}") self.ip_launcher = InvisiblePlaywright( profile_dir=FIREFOX_STUDIO_PROFILE, headless=headless, humanize=True ) self.browser = self.ip_launcher.__enter__() self.page = self.browser.new_page() self.page.set_viewport_size({"width": 1440, "height": 900}) print("✓ invisible_playwright khởi động thành công với bảo vệ vân tay C++!") return True def _start_chrome_stealth(self, headless=False): # Auto-clean any stale Chrome processes using this profile to avoid ProcessSingleton lock try: import psutil for proc in psutil.process_iter(['pid', 'name', 'cmdline']): try: if proc.info['name'] and 'chrome' in proc.info['name'].lower(): cmd = proc.info.get('cmdline') or [] if any(str(CHROME_STUDIO_PROFILE) in c for c in cmd): proc.kill() except (psutil.NoSuchProcess, psutil.AccessDenied): pass except Exception: pass from playwright.sync_api import sync_playwright self.playwright = sync_playwright().start() CHROME_STUDIO_PROFILE.mkdir(parents=True, exist_ok=True) print(f"[1/3] Khởi chạy Google Chrome Stealth với Profile 'mirsixabcf2'...") print(f" Profile Dir: {CHROME_STUDIO_PROFILE}") print(f" Extension: {EXTENSION_DIR}") args = [ "--profile-directory=Default", f"--load-extension={EXTENSION_DIR.resolve()}", f"--disable-extensions-except={EXTENSION_DIR.resolve()}", "--disable-blink-features=AutomationControlled", "--exclude-switches=enable-automation", "--no-first-run", "--no-default-browser-check", "--disable-infobars", "--disable-features=IsolateOrigins,site-per-process", "--lang=en-US", "--window-size=1440,900" ] self.context = self.playwright.chromium.launch_persistent_context( user_data_dir=str(CHROME_STUDIO_PROFILE.resolve()), channel="chrome", headless=headless, user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/133.0.0.0 Safari/537.36", args=args, ignore_default_args=["--enable-automation"] ) self.page = self.context.pages[0] if self.context.pages else self.context.new_page() self.page.set_viewport_size({"width": 1440, "height": 900}) # Inject stealth evasions into all frames self.page.add_init_script(""" Object.defineProperty(navigator, 'webdriver', { get: () => undefined }); window.chrome = { runtime: {}, loadTimes: function() {}, csi: function() {}, app: {} }; Object.defineProperty(navigator, 'plugins', { get: () => [1, 2, 3, 4, 5] }); Object.defineProperty(navigator, 'languages', { get: () => ['en-US', 'en', 'vi'] }); """) print("✓ Google Chrome Stealth khởi động thành công với Extension tự làm & chống nhận diện Bot!") return True def login_flow_native(self): """ Launches official Google Chrome DIRECTLY without any automation flags, DevTools, or Playwright hooks. This guarantees 100% bypass of Google's 'This browser or app may not be secure' error, because it is literally the official Google Chrome running normally. """ import subprocess chrome_path = r"C:\Program Files\Google\Chrome\Application\chrome.exe" if not os.path.exists(chrome_path): chrome_path = os.path.expandvars(r"%LOCALAPPDATA%\Google\Chrome\Application\chrome.exe") login_url = "https://accounts.google.com/ServiceLogin?service=youtube&continue=https%3A%2F%2Fstudio.youtube.com%2Fchannel%2FUClZnv08RDTD4vgLOw1jmOuA%2Fvideos" cmd = [ chrome_path, f"--user-data-dir={CHROME_STUDIO_PROFILE.resolve()}", "--profile-directory=Default", f"--load-extension={EXTENSION_DIR.resolve()}", "--no-first-run", "--no-default-browser-check", login_url ] print("\n" + "=" * 70) print("🔑 KHỞI CHẠY GOOGLE CHROME NGUYÊN BẢN (100% KHÔNG AUTOMATION FLAGS)") print("=" * 70) print("[*] Đang mở Google Chrome chính chủ để bạn đăng nhập an toàn 100%...") print("[*] Google sẽ KHÔNG THỂ phát hiện bot vì hoàn toàn không dùng Playwright/CDP!") print("[*] Vui lòng đăng nhập tài khoản mirsixabcf2@gmail.com trên cửa sổ vừa mở.") print("[*] Sau khi đăng nhập thành công và màn hình hiển thị YouTube Studio:") print(" -> Bạn chỉ cần nhấn phím [ENTER] trên cửa sổ này hoặc đóng trình duyệt.") print("=" * 70 + "\n") proc = subprocess.Popen(cmd) try: input("👉 Nhấn [ENTER] sau khi bạn đã đăng nhập thành công vào YouTube Studio...") except (EOFError, KeyboardInterrupt): pass print("\n✓ Đã ghi nhận phiên đăng nhập hợp lệ vào profile:") print(f" {CHROME_STUDIO_PROFILE}") print("✓ Toàn bộ Cookie, OAuth Token và App-Bound Encryption đã được lưu vĩnh viễn!") return True def login_flow(self, timeout_mins=5): """Interactive login flow for invisible_playwright or Playwright context.""" login_url = "https://accounts.google.com/ServiceLogin?service=youtube&continue=https%3A%2F%2Fstudio.youtube.com%2Fchannel%2FUClZnv08RDTD4vgLOw1jmOuA%2Fvideos" print("\n" + "=" * 70) print("🔑 THIẾT LẬP ĐĂNG NHẬP 1 LẦN (ONE-TIME AUTHENTICATION SETUP)") print("=" * 70) print(f"[*] Đang điều hướng đến: {login_url}") print("[*] Vui lòng đăng nhập tài khoản mirsixabcf2 trên cửa sổ trình duyệt...") print("[*] Hệ thống sẽ tự động phát hiện khi bạn vào YouTube Studio và lưu phiên vĩnh viễn.\n") self.page.goto(login_url) start_time = time.time() max_duration = timeout_mins * 60 while time.time() - start_time < max_duration: current_url = self.page.url.lower() if "studio.youtube.com" in current_url and "signin" not in current_url: print("\n" + "🎉" * 25) print("✓ PHÁT HIỆN ĐĂNG NHẬP THÀNH CÔNG VÀO YOUTUBE STUDIO!") print(f" URL Hiện Tại: {self.page.url}") print("✓ Phiên đăng nhập đã được ghi nhận vào Profile vĩnh viễn!") print("🎉" * 25) time.sleep(4) self.capture_status("downloads/studio_auth_success.png") return True time.sleep(2) print("\n[-] Đã quá thời gian chờ đăng nhập.") return False def goto_studio(self): """Navigates to YouTube Studio Video List and verifies authentication.""" studio_url = "https://studio.youtube.com/channel/UClZnv08RDTD4vgLOw1jmOuA/videos" print(f"\n[2/3] Điều hướng đến YouTube Studio: {studio_url}...") self.page.goto(studio_url, wait_until="domcontentloaded", timeout=45000) human_delay(3, 5) current_url = self.page.url if "accounts.google.com" in current_url or "signin" in current_url.lower(): print("\n[!] PHIÊN ĐĂNG NHẬP CHƯA CÓ TRONG PROFILE NÀY.") print(" Vui lòng chạy: python youtube_studio_manager.py --login") print(" để hoàn tất xác thực 1 lần duy nhất cho profile.") return False print(f"✓ Trang hiện tại: {self.page.title()} ({self.page.url})") return True def capture_status(self, out_path="downloads/studio_status.png"): """Captures a screenshot of the YouTube Studio dashboard.""" os.makedirs(os.path.dirname(out_path), exist_ok=True) self.page.screenshot(path=out_path) print(f"📸 Đã chụp ảnh màn hình Studio: {out_path}") return out_path def inspect_channel(self): """Reads recent video list and statuses from Studio DOM.""" print("\n[3/3] Quét danh sách video và tình trạng xử lý kênh...") try: # Let Polymer render self.page.wait_for_selector("#row-container, ytcp-video-row", timeout=15000) rows = self.page.locator("ytcp-video-row, #row-container").all() print(f"✓ Tìm thấy {len(rows)} video trên trang quản trị Studio.") for i, r in enumerate(rows[:5]): try: title_el = r.locator("#video-title, #entity-table-cell").first title_text = title_el.inner_text().strip() if title_el.is_visible() else "N/A" print(f" [{i+1}] {title_text[:60]}") except Exception: pass except Exception as e: print(f"[-] Không thể quét chi tiết bảng: {e}") self.capture_status("downloads/studio_status.png") def goto_video_edit(self, ep_key): """Navigates to the Studio Edit page for an episode.""" metadata = load_metadata() meta = metadata.get(str(ep_key)) if not meta or not meta.get("video_id"): print(f"[-] Không tìm thấy video_id cho mục: {ep_key}") return False video_id = meta["video_id"] edit_url = f"https://studio.youtube.com/video/{video_id}/edit" print(f"\n[*] Điều hướng đến trang chỉnh sửa video [{ep_key}] (ID: {video_id}): {edit_url}...") self.page.goto(edit_url, wait_until="domcontentloaded", timeout=45000) human_delay(3, 5) return True def auto_apply_extension_metadata(self, ep_key): """ Triggers the custom Chrome Extension's Auto-fill for a specific episode (or 'full'). """ ep_str = str(ep_key) print(f"\n[*] Kích hoạt Extension tự động điền Metadata Mục: {ep_str}...") # Deep query to find extension panel inside Shadow DOM or main DOM script_check = """ (targetKey) => { const panel = document.getElementById('hv-panel') || document.getElementById('hv-studio-assistant'); if (!panel) return { found: false, reason: 'panel_missing' }; const select = document.getElementById('hv-preset-select'); const autofillBtn = document.getElementById('hv-autofill-btn'); if (!select || !autofillBtn) return { found: false, reason: 'elements_missing' }; select.value = String(targetKey); select.dispatchEvent(new Event('change', { bubbles: true })); autofillBtn.click(); return { found: true }; } """ res = self.page.evaluate(script_check, ep_str) if res.get("found"): print(f"✓ Extension đã kích hoạt 1-Click Auto-Fill cho Tập {ep_str} thành công!") human_delay(3, 5) # Trigger thumbnail injection via extension self.page.evaluate(""" () => { const thumbBtn = document.getElementById('hv-upload-thumb-btn'); if (thumbBtn) thumbBtn.click(); } """) human_delay(2, 3) return True else: print(f"[-] Extension UI chưa sẵn sàng ({res.get('reason')}). Sử dụng Playwright trực tiếp...") return self.direct_fill_metadata(ep_str) def direct_fill_metadata(self, ep_key): """Fills metadata directly via Playwright Shadow DOM locators.""" ep_str = str(ep_key) metadata = load_metadata() meta = metadata.get(ep_str) if not meta: print(f"[-] Không tìm thấy metadata cho mục: {ep_str}") return False title = meta.get("title", "") desc = meta.get("description", "") tags = meta.get("tags", "") print(f" -> Tiêu đề: {title[:60]}...") print(f" -> Mô tả: {len(desc)} ký tự") print(f" -> Tags: {len(tags)} ký tự") # 1. Fill Title try: title_box = self.page.locator("#textbox[aria-label*='tiêu đề'], #textbox[aria-label*='title'], #title-textarea #textbox").first if title_box.is_visible(timeout=5000): title_box.click() self.page.keyboard.press("Control+A") self.page.keyboard.type(title, delay=15) print(" ✓ Đã điền tiêu đề thành công") except Exception as e: print(f" [-] Lỗi điền tiêu đề: {e}") # 2. Fill Description try: desc_box = self.page.locator("#description-textarea #textbox, #textbox[aria-label*='mô tả'], #textbox[aria-label*='description']").first if desc_box.is_visible(timeout=5000): desc_box.click() self.page.keyboard.press("Control+A") self.page.evaluate(""" (text) => { const el = document.querySelector('#description-textarea #textbox'); if (el) { el.innerText = text; el.dispatchEvent(new Event('input', { bubbles: true })); } } """, desc) print(" ✓ Đã điền mô tả kèm Chapters thành công") except Exception as e: print(f" [-] Lỗi điền mô tả: {e}") # 3. Upload Thumbnail thumb_path = meta.get("thumb_path") if not thumb_path or not os.path.exists(thumb_path): # Try 4K thumbnail if ep_str == "full": thumb_path = str(THUMBNAILS_4K_DIR / "full_movie_thumb_4k.jpg") else: thumb_path = str(THUMBNAILS_4K_DIR / f"ep{ep_str}_thumb_4k.jpg") if os.path.exists(thumb_path): try: thumb_input = self.page.locator("input#file-loader, input[type='file'][accept*='image']").first if thumb_input.count() > 0: thumb_input.set_input_files(thumb_path) print(f" ✓ Đã nạp Thumbnail 4K UHD: {Path(thumb_path).name}") except Exception as e: print(f" [-] Lỗi upload thumbnail: {e}") # 4. Save changes try: human_delay(3, 5) save_btn = self.page.locator("#save, ytcp-button#save-button, button:has-text('Lưu'), button:has-text('Save')").first if save_btn.is_visible(timeout=5000): if not save_btn.is_disabled(): save_btn.click() print(" ✓ Đã bấm nút [LƯU] thay đổi thành công!") human_delay(3, 5) else: print(" [*] Nút Lưu đang ở trạng thái không đổi (đã lưu từ trước).") except Exception as e: print(f" [-] Lỗi khi bấm Lưu: {e}") return True def upload_video(self, video_path, ep_key): """ Automates end-to-end video uploading: 1. Clicks Create -> Upload videos 2. Sets video file 3. Fills metadata (title, desc, tags, thumbnail) 4. Selects 'Not made for kids' 5. Saves video """ video_file = Path(video_path) if not video_file.exists(): print(f"[-] File video không tồn tại: {video_path}") return False file_size_mb = video_file.stat().st_size / (1024 * 1024) print(f"\n" + "=" * 70) print(f"📤 BẮT ĐẦU QUY TRÌNH UPLOAD VIDEO: {video_file.name} ({file_size_mb:.1f} MB)") print("=" * 70) if not self.goto_studio(): return False # 1. Click Create -> Upload try: create_btn = self.page.locator("#create-icon, button[aria-label*='Create'], button[aria-label*='Tạo']").first create_btn.click() human_delay(1, 2) upload_item = self.page.locator("tp-yt-paper-item:has-text('Upload videos'), tp-yt-paper-item:has-text('Tải video lên')").first upload_item.click() human_delay(2, 3) # 2. Attach file file_input = self.page.locator("input[type='file']").first file_input.set_input_files(str(video_file.resolve())) print(f"✓ Đã đưa file {video_file.name} vào luồng tải lên Studio!") # 3. Wait for Details dialog self.page.wait_for_selector("#title-textarea, #textbox", timeout=60000) print("✓ Hộp thoại chi tiết video đã hiển thị.") human_delay(3, 5) # 4. Fill metadata self.auto_apply_extension_metadata(ep_key) # 5. Set 'Not made for kids' not_for_kids = self.page.locator("tp-yt-paper-radio-button[name='NOT_MADE_FOR_KIDS'], tp-yt-paper-radio-button[name='VIDEO_MADE_FOR_KIDS_NOT_MFK']").first if not_for_kids.is_visible(timeout=5000): not_for_kids.click() print("✓ Đã chọn: Không dành cho trẻ em (Not made for kids).") human_delay(2, 3) self.capture_status(f"downloads/upload_progress_{ep_key}.png") print("✓ Quy trình upload & điền chi tiết đã hoàn tất an toàn!") return True except Exception as e: print(f"[-] Lỗi trong quá trình upload: {e}") self.capture_status("downloads/upload_error.png") return False def close(self): """Closes browser session cleanly.""" print("\n[*] Đang đóng phiên Playwright...") try: if self.context: self.context.close() if self.playwright: self.playwright.stop() if hasattr(self, 'ip_launcher') and self.ip_launcher: self.ip_launcher.__exit__(None, None, None) print("✓ Đã giải phóng tài nguyên sạch sẽ.") except Exception as e: print(f"[-] Lỗi khi đóng: {e}") if __name__ == "__main__": parser = argparse.ArgumentParser(description="YouTube Studio Autonomous Management Suite") parser.add_argument("--mode", choices=["chrome_stealth", "invisible_playwright"], default="chrome_stealth", help="Browser engine (default: chrome_stealth)") parser.add_argument("--login", action="store_true", help="Launch interactive 1-time login to authenticate profile") parser.add_argument("--status", action="store_true", help="Check channel and recent videos status") parser.add_argument("--ep", type=str, default=None, help="Episode key ('1'..'8' or 'full') to autofill or edit") parser.add_argument("--upload", type=str, default=None, help="Upload specified episode ('1'..'8' or 'full')") parser.add_argument("--headless", action="store_true", help="Run headlessly without opening window") args = parser.parse_args() # Determine mode manager = YouTubeStudioManager(mode=args.mode) try: # If --login requested if args.login: if args.mode == "chrome_stealth": manager.login_flow_native() else: manager.start(headless=False) manager.login_flow() sys.exit(0) # Standard execution manager.start(headless=args.headless) if args.upload: ep = args.upload if ep == "full": v_path = VIDEOS_4K_DIR / "Forgotten_Battles_1785_Full_4K.mp4" else: v_path = VIDEOS_4K_DIR / f"ep{ep}_4k.mp4" manager.upload_video(str(v_path), ep) elif args.status: if manager.goto_studio(): manager.inspect_channel() elif args.ep: if manager.goto_video_edit(args.ep): manager.auto_apply_extension_metadata(args.ep) manager.capture_status(f"downloads/studio_ep_{args.ep}.png") else: # Default action: check status if manager.goto_studio(): manager.inspect_channel() finally: manager.close()