Spaces:
Running
Running
| import re | |
| import base64 | |
| import json | |
| import time | |
| import os | |
| import requests | |
| from utils import * | |
| from .Tts import TTS | |
| from llm_provider import generate_text | |
| from config import * | |
| from status import * | |
| from uuid import uuid4 | |
| from typing import List | |
| from termcolor import colored | |
| from datetime import datetime | |
| # Lazy imports for browser-dependent modules | |
| _browser_imports_done = False | |
| def _ensure_browser_imports(): | |
| global _browser_imports_done | |
| if _browser_imports_done: | |
| return | |
| global aai, webdriver, By, Service, Options, GeckoDriverManager | |
| global YOUTUBE_TEXTBOX_ID, YOUTUBE_MADE_FOR_KIDS_NAME, YOUTUBE_NOT_MADE_FOR_KIDS_NAME | |
| global YOUTUBE_NEXT_BUTTON_ID, YOUTUBE_RADIO_BUTTON_XPATH, YOUTUBE_DONE_BUTTON_ID | |
| global get_youtube_cache_path | |
| import assemblyai as aai | |
| import selenium_firefox # noqa: F401 | |
| from selenium import webdriver | |
| from selenium.webdriver.common.by import By | |
| from selenium.webdriver.firefox.service import Service | |
| from selenium.webdriver.firefox.options import Options | |
| from webdriver_manager.firefox import GeckoDriverManager | |
| import constants | |
| YOUTUBE_TEXTBOX_ID = constants.YOUTUBE_TEXTBOX_ID | |
| YOUTUBE_MADE_FOR_KIDS_NAME = constants.YOUTUBE_MADE_FOR_KIDS_NAME | |
| YOUTUBE_NOT_MADE_FOR_KIDS_NAME = constants.YOUTUBE_NOT_MADE_FOR_KIDS_NAME | |
| YOUTUBE_NEXT_BUTTON_ID = constants.YOUTUBE_NEXT_BUTTON_ID | |
| YOUTUBE_RADIO_BUTTON_XPATH = constants.YOUTUBE_RADIO_BUTTON_XPATH | |
| YOUTUBE_DONE_BUTTON_ID = constants.YOUTUBE_DONE_BUTTON_ID | |
| from cache import get_youtube_cache_path | |
| _browser_imports_done = True | |
| # MoviePy imports (always needed for video generation) | |
| from moviepy.editor import ( | |
| ImageClip, AudioFileClip, TextClip, CompositeVideoClip, | |
| CompositeAudioClip, concatenate_videoclips, afx, | |
| ) | |
| from moviepy.video.fx.all import crop | |
| from moviepy.config import change_settings | |
| from moviepy.video.tools.subtitles import SubtitlesClip | |
| # Set ImageMagick Path | |
| imgk = get_imagemagick_path() | |
| if imgk: | |
| change_settings({"IMAGEMAGICK_BINARY": imgk}) | |
| class YouTube: | |
| """ | |
| Class for YouTube Automation. | |
| Steps to create a YouTube Short: | |
| 1. Generate a topic [DONE] | |
| 2. Generate a script [DONE] | |
| 3. Generate metadata (Title, Description, Tags) [DONE] | |
| 4. Generate AI Image Prompts [DONE] | |
| 4. Generate Images based on generated Prompts [DONE] | |
| 5. Convert Text-to-Speech [DONE] | |
| 6. Show images each for n seconds, n: Duration of TTS / Amount of images [DONE] | |
| 7. Combine Concatenated Images with the Text-to-Speech [DONE] | |
| """ | |
| def __init__( | |
| self, | |
| account_uuid: str, | |
| account_nickname: str, | |
| fp_profile_path: str, | |
| niche: str, | |
| language: str, | |
| use_browser: bool = True, | |
| ) -> None: | |
| """ | |
| Constructor for YouTube Class. | |
| Args: | |
| account_uuid (str): The unique identifier for the YouTube account. | |
| account_nickname (str): The nickname for the YouTube account. | |
| fp_profile_path (str): Path to the firefox profile that is logged into the specificed YouTube Account. | |
| niche (str): The niche of the provided YouTube Channel. | |
| language (str): The language of the Automation. | |
| use_browser (bool): If False, skip Selenium initialization (for headless video generation). | |
| Returns: | |
| None | |
| """ | |
| self._account_uuid: str = account_uuid | |
| self._account_nickname: str = account_nickname | |
| self._fp_profile_path: str = fp_profile_path | |
| self._niche: str = niche | |
| self._language: str = language | |
| self._use_browser: bool = use_browser | |
| self.images = [] | |
| if not self._use_browser: | |
| self.browser = None | |
| return | |
| _ensure_browser_imports() | |
| # Initialize the Firefox profile | |
| self.options = Options() | |
| # Set headless state of browser | |
| if get_headless(): | |
| self.options.add_argument("--headless") | |
| if not os.path.isdir(self._fp_profile_path): | |
| raise ValueError( | |
| f"Firefox profile path does not exist or is not a directory: {self._fp_profile_path}" | |
| ) | |
| self.options.add_argument("-profile") | |
| self.options.add_argument(self._fp_profile_path) | |
| # Set the service | |
| self.service = Service(GeckoDriverManager().install()) | |
| # Initialize the browser | |
| self.browser = webdriver.Firefox( | |
| service=self.service, options=self.options | |
| ) | |
| def niche(self) -> str: | |
| """ | |
| Getter Method for the niche. | |
| Returns: | |
| niche (str): The niche | |
| """ | |
| return self._niche | |
| def language(self) -> str: | |
| """ | |
| Getter Method for the language to use. | |
| Returns: | |
| language (str): The language | |
| """ | |
| return self._language | |
| def generate_response(self, prompt: str, model_name: str = None) -> str: | |
| """ | |
| Generates an LLM Response based on a prompt and the user-provided model. | |
| Args: | |
| prompt (str): The prompt to use in the text generation. | |
| Returns: | |
| response (str): The generated AI Repsonse. | |
| """ | |
| return generate_text(prompt, model_name=model_name) | |
| def generate_topic(self) -> str: | |
| """ | |
| Generates a topic based on the YouTube Channel niche. | |
| Returns: | |
| topic (str): The generated topic. | |
| """ | |
| completion = self.generate_response( | |
| f"Please generate a specific video idea that takes about the following topic: {self.niche}. Make it exactly one sentence. Only return the topic, nothing else." | |
| ) | |
| if not completion: | |
| error("Failed to generate Topic.") | |
| self.subject = completion | |
| return completion | |
| def generate_script(self) -> str: | |
| """ | |
| Generate a script for a video, depending on the subject of the video, the number of paragraphs, and the AI model. | |
| Returns: | |
| script (str): The script of the video. | |
| """ | |
| sentence_length = get_script_sentence_length() | |
| prompt = f""" | |
| Generate a script for a video in {sentence_length} sentences, depending on the subject of the video. | |
| The script is to be returned as a string with the specified number of paragraphs. | |
| Here is an example of a string: | |
| "This is an example string." | |
| Do not under any circumstance reference this prompt in your response. | |
| Get straight to the point, don't start with unnecessary things like, "welcome to this video". | |
| Obviously, the script should be related to the subject of the video. | |
| YOU MUST NOT EXCEED THE {sentence_length} SENTENCES LIMIT. MAKE SURE THE {sentence_length} SENTENCES ARE SHORT. | |
| YOU MUST NOT INCLUDE ANY TYPE OF MARKDOWN OR FORMATTING IN THE SCRIPT, NEVER USE A TITLE. | |
| YOU MUST WRITE THE SCRIPT IN THE LANGUAGE SPECIFIED IN [LANGUAGE]. | |
| ONLY RETURN THE RAW CONTENT OF THE SCRIPT. DO NOT INCLUDE "VOICEOVER", "NARRATOR" OR SIMILAR INDICATORS OF WHAT SHOULD BE SPOKEN AT THE BEGINNING OF EACH PARAGRAPH OR LINE. YOU MUST NOT MENTION THE PROMPT, OR ANYTHING ABOUT THE SCRIPT ITSELF. ALSO, NEVER TALK ABOUT THE AMOUNT OF PARAGRAPHS OR LINES. JUST WRITE THE SCRIPT | |
| Subject: {self.subject} | |
| Language: {self.language} | |
| """ | |
| max_retries = 3 | |
| for attempt in range(max_retries): | |
| completion = self.generate_response(prompt) | |
| completion = re.sub(r"\*", "", completion) | |
| if not completion: | |
| error("The generated script is empty.") | |
| return | |
| if len(completion) <= 5000: | |
| self.script = completion | |
| return completion | |
| if get_verbose(): | |
| warning(f"Generated Script is too long (attempt {attempt + 1}/{max_retries}). Retrying...") | |
| self.script = completion | |
| return completion | |
| def generate_metadata(self) -> dict: | |
| """ | |
| Generates Video metadata for the to-be-uploaded YouTube Short (Title, Description). | |
| Returns: | |
| metadata (dict): The generated metadata. | |
| """ | |
| max_retries = 3 | |
| title = "" | |
| for attempt in range(max_retries): | |
| title = self.generate_response( | |
| f"Please generate a YouTube Video Title for the following subject, including hashtags: {self.subject}. Only return the title, nothing else. Limit the title under 100 characters." | |
| ) | |
| if len(title) <= 100: | |
| break | |
| if get_verbose(): | |
| warning(f"Generated Title is too long (attempt {attempt + 1}/{max_retries}). Retrying...") | |
| description = self.generate_response( | |
| f"Please generate a YouTube Video Description for the following script: {self.script}. Only return the description, nothing else." | |
| ) | |
| self.metadata = {"title": title, "description": description} | |
| return self.metadata | |
| def generate_prompts(self) -> List[str]: | |
| """ | |
| Generates AI Image Prompts based on the provided Video Script. | |
| Returns: | |
| image_prompts (List[str]): Generated List of image prompts. | |
| """ | |
| n_prompts = len(self.script) / 3 | |
| prompt = f""" | |
| Generate {n_prompts} Image Prompts for AI Image Generation, | |
| depending on the subject of a video. | |
| Subject: {self.subject} | |
| The image prompts are to be returned as | |
| a JSON-Array of strings. | |
| Each search term should consist of a full sentence, | |
| always add the main subject of the video. | |
| Be emotional and use interesting adjectives to make the | |
| Image Prompt as detailed as possible. | |
| YOU MUST ONLY RETURN THE JSON-ARRAY OF STRINGS. | |
| YOU MUST NOT RETURN ANYTHING ELSE. | |
| YOU MUST NOT RETURN THE SCRIPT. | |
| The search terms must be related to the subject of the video. | |
| Here is an example of a JSON-Array of strings: | |
| ["image prompt 1", "image prompt 2", "image prompt 3"] | |
| For context, here is the full text: | |
| {self.script} | |
| """ | |
| completion = ( | |
| str(self.generate_response(prompt)) | |
| .replace("```json", "") | |
| .replace("```", "") | |
| ) | |
| image_prompts = [] | |
| if "image_prompts" in completion: | |
| image_prompts = json.loads(completion)["image_prompts"] | |
| else: | |
| try: | |
| image_prompts = json.loads(completion) | |
| if get_verbose(): | |
| info(f" => Generated Image Prompts: {image_prompts}") | |
| except Exception: | |
| if get_verbose(): | |
| warning( | |
| "LLM returned an unformatted response. Attempting to clean..." | |
| ) | |
| # Get everything between [ and ], and turn it into a list | |
| r = re.compile(r"\[.*\]") | |
| image_prompts = r.findall(completion) | |
| if len(image_prompts) == 0: | |
| if get_verbose(): | |
| warning("Failed to generate Image Prompts.") | |
| image_prompts = [self.subject] | |
| if len(image_prompts) > n_prompts: | |
| image_prompts = image_prompts[: int(n_prompts)] | |
| self.image_prompts = image_prompts | |
| success(f"Generated {len(image_prompts)} Image Prompts.") | |
| return image_prompts | |
| def _persist_image(self, image_bytes: bytes, provider_label: str) -> str: | |
| """ | |
| Writes generated image bytes to a PNG file in .mp. | |
| Args: | |
| image_bytes (bytes): Image payload | |
| provider_label (str): Label for logging | |
| Returns: | |
| path (str): Absolute image path | |
| """ | |
| image_path = os.path.join(ROOT_DIR, ".mp", str(uuid4()) + ".png") | |
| with open(image_path, "wb") as image_file: | |
| image_file.write(image_bytes) | |
| if get_verbose(): | |
| info(f' => Wrote image from {provider_label} to "{image_path}"') | |
| self.images.append(image_path) | |
| return image_path | |
| def generate_image_nanobanana2(self, prompt: str) -> str: | |
| """ | |
| Generates an AI Image using Nano Banana 2 API (Gemini image API). | |
| Args: | |
| prompt (str): Prompt for image generation | |
| Returns: | |
| path (str): The path to the generated image. | |
| """ | |
| print(f"Generating Image using Nano Banana 2 API: {prompt}") | |
| api_key = get_nanobanana2_api_key() | |
| if not api_key: | |
| error("nanobanana2_api_key is not configured.") | |
| return None | |
| base_url = get_nanobanana2_api_base_url().rstrip("/") | |
| model = get_nanobanana2_model() | |
| aspect_ratio = get_nanobanana2_aspect_ratio() | |
| endpoint = f"{base_url}/models/{model}:generateContent" | |
| payload = { | |
| "contents": [{"parts": [{"text": prompt}]}], | |
| "generationConfig": { | |
| "responseModalities": ["IMAGE"], | |
| "imageConfig": {"aspectRatio": aspect_ratio}, | |
| }, | |
| } | |
| try: | |
| response = requests.post( | |
| endpoint, | |
| headers={"x-goog-api-key": api_key, "Content-Type": "application/json"}, | |
| json=payload, | |
| timeout=300, | |
| ) | |
| response.raise_for_status() | |
| body = response.json() | |
| candidates = body.get("candidates", []) | |
| for candidate in candidates: | |
| content = candidate.get("content", {}) | |
| for part in content.get("parts", []): | |
| inline_data = part.get("inlineData") or part.get("inline_data") | |
| if not inline_data: | |
| continue | |
| data = inline_data.get("data") | |
| mime_type = inline_data.get("mimeType") or inline_data.get("mime_type", "") | |
| if data and str(mime_type).startswith("image/"): | |
| image_bytes = base64.b64decode(data) | |
| return self._persist_image(image_bytes, "Nano Banana 2 API") | |
| if get_verbose(): | |
| warning(f"Nano Banana 2 did not return an image payload. Response: {body}") | |
| return None | |
| except Exception as e: | |
| if get_verbose(): | |
| warning(f"Failed to generate image with Nano Banana 2 API: {str(e)}") | |
| return None | |
| def generate_image(self, prompt: str) -> str: | |
| """ | |
| Generates an AI Image based on the given prompt using Nano Banana 2. | |
| Args: | |
| prompt (str): Reference for image generation | |
| Returns: | |
| path (str): The path to the generated image. | |
| """ | |
| return self.generate_image_nanobanana2(prompt) | |
| def generate_script_to_speech(self, tts_instance: TTS) -> str: | |
| """ | |
| Converts the generated script into Speech using KittenTTS and returns the path to the wav file. | |
| Args: | |
| tts_instance (tts): Instance of TTS Class. | |
| Returns: | |
| path_to_wav (str): Path to generated audio (WAV Format). | |
| """ | |
| path = os.path.join(ROOT_DIR, ".mp", str(uuid4()) + ".wav") | |
| # Clean script, remove every character that is not a word character, a space, a period, a question mark, or an exclamation mark. | |
| self.script = re.sub(r"[^\w\s.?!]", "", self.script) | |
| tts_instance.synthesize(self.script, path) | |
| self.tts_path = path | |
| if get_verbose(): | |
| info(f' => Wrote TTS to "{path}"') | |
| return path | |
| def add_video(self, video: dict) -> None: | |
| """ | |
| Adds a video to the cache. | |
| Args: | |
| video (dict): The video to add | |
| Returns: | |
| None | |
| """ | |
| _ensure_browser_imports() | |
| videos = self.get_videos() | |
| videos.append(video) | |
| cache = get_youtube_cache_path() | |
| with open(cache, "r") as file: | |
| previous_json = json.loads(file.read()) | |
| # Find our account | |
| accounts = previous_json["accounts"] | |
| for account in accounts: | |
| if account["id"] == self._account_uuid: | |
| account["videos"].append(video) | |
| # Commit changes | |
| with open(cache, "w") as f: | |
| f.write(json.dumps(previous_json)) | |
| def generate_subtitles(self, audio_path: str) -> str: | |
| """ | |
| Generates subtitles for the audio using the configured STT provider. | |
| Args: | |
| audio_path (str): The path to the audio file. | |
| Returns: | |
| path (str): The path to the generated SRT File. | |
| """ | |
| provider = str(get_stt_provider() or "local_whisper").lower() | |
| if provider == "local_whisper": | |
| return self.generate_subtitles_local_whisper(audio_path) | |
| if provider == "third_party_assemblyai": | |
| return self.generate_subtitles_assemblyai(audio_path) | |
| warning(f"Unknown stt_provider '{provider}'. Falling back to local_whisper.") | |
| return self.generate_subtitles_local_whisper(audio_path) | |
| def generate_subtitles_assemblyai(self, audio_path: str) -> str: | |
| """ | |
| Generates subtitles using AssemblyAI. | |
| Args: | |
| audio_path (str): Audio file path | |
| Returns: | |
| path (str): Path to SRT file | |
| """ | |
| aai.settings.api_key = get_assemblyai_api_key() | |
| config = aai.TranscriptionConfig() | |
| transcriber = aai.Transcriber(config=config) | |
| transcript = transcriber.transcribe(audio_path) | |
| subtitles = transcript.export_subtitles_srt() | |
| srt_path = os.path.join(ROOT_DIR, ".mp", str(uuid4()) + ".srt") | |
| with open(srt_path, "w") as file: | |
| file.write(subtitles) | |
| return srt_path | |
| def _format_srt_timestamp(self, seconds: float) -> str: | |
| """ | |
| Formats a timestamp in seconds to SRT format. | |
| Args: | |
| seconds (float): Seconds | |
| Returns: | |
| ts (str): HH:MM:SS,mmm | |
| """ | |
| total_millis = max(0, int(round(seconds * 1000))) | |
| hours = total_millis // 3600000 | |
| minutes = (total_millis % 3600000) // 60000 | |
| secs = (total_millis % 60000) // 1000 | |
| millis = total_millis % 1000 | |
| return f"{hours:02d}:{minutes:02d}:{secs:02d},{millis:03d}" | |
| def generate_subtitles_local_whisper(self, audio_path: str) -> str: | |
| """ | |
| Generates subtitles using local Whisper (faster-whisper). | |
| Args: | |
| audio_path (str): Audio file path | |
| Returns: | |
| path (str): Path to SRT file | |
| """ | |
| try: | |
| from faster_whisper import WhisperModel | |
| except ImportError: | |
| error( | |
| "Local STT selected but 'faster-whisper' is not installed. " | |
| "Install it or switch stt_provider to third_party_assemblyai." | |
| ) | |
| raise | |
| model = WhisperModel( | |
| get_whisper_model(), | |
| device=get_whisper_device(), | |
| compute_type=get_whisper_compute_type(), | |
| ) | |
| segments, _ = model.transcribe(audio_path, vad_filter=True) | |
| lines = [] | |
| for idx, segment in enumerate(segments, start=1): | |
| start = self._format_srt_timestamp(segment.start) | |
| end = self._format_srt_timestamp(segment.end) | |
| text = str(segment.text).strip() | |
| if not text: | |
| continue | |
| lines.append(str(idx)) | |
| lines.append(f"{start} --> {end}") | |
| lines.append(text) | |
| lines.append("") | |
| subtitles = "\n".join(lines) | |
| srt_path = os.path.join(ROOT_DIR, ".mp", str(uuid4()) + ".srt") | |
| with open(srt_path, "w", encoding="utf-8") as file: | |
| file.write(subtitles) | |
| return srt_path | |
| def combine(self) -> str: | |
| """ | |
| Combines everything into the final video. | |
| Returns: | |
| path (str): The path to the generated MP4 File. | |
| """ | |
| combined_image_path = os.path.join(ROOT_DIR, ".mp", str(uuid4()) + ".mp4") | |
| threads = get_threads() | |
| tts_clip = AudioFileClip(self.tts_path) | |
| max_duration = tts_clip.duration | |
| req_dur = max_duration / len(self.images) | |
| # Make a generator that returns a TextClip when called with consecutive | |
| generator = lambda txt: TextClip( | |
| txt, | |
| font=os.path.join(get_fonts_dir(), get_font()), | |
| fontsize=100, | |
| color="#FFFF00", | |
| stroke_color="black", | |
| stroke_width=5, | |
| size=(1080, 1920), | |
| method="caption", | |
| ) | |
| print(colored("[+] Combining images...", "blue")) | |
| clips = [] | |
| tot_dur = 0 | |
| # Add downloaded clips over and over until the duration of the audio (max_duration) has been reached | |
| while tot_dur < max_duration: | |
| for image_path in self.images: | |
| clip = ImageClip(image_path) | |
| clip.duration = req_dur | |
| clip = clip.set_fps(30) | |
| # Not all images are same size, | |
| # so we need to resize them | |
| if round((clip.w / clip.h), 4) < 0.5625: | |
| if get_verbose(): | |
| info(f" => Resizing Image: {image_path} to 1080x1920") | |
| clip = crop( | |
| clip, | |
| width=clip.w, | |
| height=round(clip.w / 0.5625), | |
| x_center=clip.w / 2, | |
| y_center=clip.h / 2, | |
| ) | |
| else: | |
| if get_verbose(): | |
| info(f" => Resizing Image: {image_path} to 1920x1080") | |
| clip = crop( | |
| clip, | |
| width=round(0.5625 * clip.h), | |
| height=clip.h, | |
| x_center=clip.w / 2, | |
| y_center=clip.h / 2, | |
| ) | |
| clip = clip.resize((1080, 1920)) | |
| # FX (Fade In) | |
| # clip = clip.fadein(2) | |
| clips.append(clip) | |
| tot_dur += clip.duration | |
| final_clip = concatenate_videoclips(clips) | |
| final_clip = final_clip.set_fps(30) | |
| random_song = choose_random_song() | |
| subtitles = None | |
| try: | |
| subtitles_path = self.generate_subtitles(self.tts_path) | |
| equalize_subtitles(subtitles_path, 10) | |
| subtitles = SubtitlesClip(subtitles_path, generator) | |
| subtitles.set_pos(("center", "center")) | |
| except Exception as e: | |
| warning(f"Failed to generate subtitles, continuing without subtitles: {e}") | |
| random_song_clip = AudioFileClip(random_song).set_fps(44100) | |
| # Turn down volume | |
| random_song_clip = random_song_clip.fx(afx.volumex, 0.1) | |
| comp_audio = CompositeAudioClip([tts_clip.set_fps(44100), random_song_clip]) | |
| final_clip = final_clip.set_audio(comp_audio) | |
| final_clip = final_clip.set_duration(tts_clip.duration) | |
| if subtitles is not None: | |
| final_clip = CompositeVideoClip([final_clip, subtitles]) | |
| final_clip.write_videofile(combined_image_path, threads=threads) | |
| success(f'Wrote Video to "{combined_image_path}"') | |
| return combined_image_path | |
| def generate_video(self, tts_instance: TTS) -> str: | |
| """ | |
| Generates a YouTube Short based on the provided niche and language. | |
| Args: | |
| tts_instance (TTS): Instance of TTS Class. | |
| Returns: | |
| path (str): The path to the generated MP4 File. | |
| """ | |
| # Generate the Topic | |
| self.generate_topic() | |
| # Generate the Script | |
| self.generate_script() | |
| # Generate the Metadata | |
| self.generate_metadata() | |
| # Generate the Image Prompts | |
| self.generate_prompts() | |
| # Generate the Images | |
| for prompt in self.image_prompts: | |
| self.generate_image(prompt) | |
| # Generate the TTS | |
| self.generate_script_to_speech(tts_instance) | |
| # Combine everything | |
| path = self.combine() | |
| if get_verbose(): | |
| info(f" => Generated Video: {path}") | |
| self.video_path = os.path.abspath(path) | |
| return path | |
| def _require_browser(self): | |
| if not self._use_browser or self.browser is None: | |
| raise RuntimeError( | |
| "Browser is not available. This method requires use_browser=True." | |
| ) | |
| def get_channel_id(self) -> str: | |
| """ | |
| Gets the Channel ID of the YouTube Account. | |
| Returns: | |
| channel_id (str): The Channel ID. | |
| """ | |
| self._require_browser() | |
| driver = self.browser | |
| driver.get("https://studio.youtube.com") | |
| time.sleep(2) | |
| channel_id = driver.current_url.split("/")[-1] | |
| self.channel_id = channel_id | |
| return channel_id | |
| def upload_video(self) -> bool: | |
| """ | |
| Uploads the video to YouTube. | |
| Returns: | |
| success (bool): Whether the upload was successful or not. | |
| """ | |
| self._require_browser() | |
| _ensure_browser_imports() | |
| try: | |
| self.get_channel_id() | |
| driver = self.browser | |
| verbose = get_verbose() | |
| # Go to youtube.com/upload | |
| driver.get("https://www.youtube.com/upload") | |
| # Set video file | |
| FILE_PICKER_TAG = "ytcp-uploads-file-picker" | |
| file_picker = driver.find_element(By.TAG_NAME, FILE_PICKER_TAG) | |
| INPUT_TAG = "input" | |
| file_input = file_picker.find_element(By.TAG_NAME, INPUT_TAG) | |
| file_input.send_keys(self.video_path) | |
| # Wait for upload to finish | |
| time.sleep(5) | |
| # Set title | |
| textboxes = driver.find_elements(By.ID, YOUTUBE_TEXTBOX_ID) | |
| title_el = textboxes[0] | |
| description_el = textboxes[-1] | |
| if verbose: | |
| info("\t=> Setting title...") | |
| title_el.click() | |
| time.sleep(1) | |
| title_el.clear() | |
| title_el.send_keys(self.metadata["title"]) | |
| if verbose: | |
| info("\t=> Setting description...") | |
| # Set description | |
| time.sleep(10) | |
| description_el.click() | |
| time.sleep(0.5) | |
| description_el.clear() | |
| description_el.send_keys(self.metadata["description"]) | |
| time.sleep(0.5) | |
| # Set `made for kids` option | |
| if verbose: | |
| info("\t=> Setting `made for kids` option...") | |
| is_for_kids_checkbox = driver.find_element( | |
| By.NAME, YOUTUBE_MADE_FOR_KIDS_NAME | |
| ) | |
| is_not_for_kids_checkbox = driver.find_element( | |
| By.NAME, YOUTUBE_NOT_MADE_FOR_KIDS_NAME | |
| ) | |
| if not get_is_for_kids(): | |
| is_not_for_kids_checkbox.click() | |
| else: | |
| is_for_kids_checkbox.click() | |
| time.sleep(0.5) | |
| # Click next | |
| if verbose: | |
| info("\t=> Clicking next...") | |
| next_button = driver.find_element(By.ID, YOUTUBE_NEXT_BUTTON_ID) | |
| next_button.click() | |
| # Click next again | |
| if verbose: | |
| info("\t=> Clicking next again...") | |
| next_button = driver.find_element(By.ID, YOUTUBE_NEXT_BUTTON_ID) | |
| next_button.click() | |
| # Wait for 2 seconds | |
| time.sleep(2) | |
| # Click next again | |
| if verbose: | |
| info("\t=> Clicking next again...") | |
| next_button = driver.find_element(By.ID, YOUTUBE_NEXT_BUTTON_ID) | |
| next_button.click() | |
| # Set as unlisted | |
| if verbose: | |
| info("\t=> Setting as unlisted...") | |
| radio_button = driver.find_elements(By.XPATH, YOUTUBE_RADIO_BUTTON_XPATH) | |
| radio_button[2].click() | |
| if verbose: | |
| info("\t=> Clicking done button...") | |
| # Click done button | |
| done_button = driver.find_element(By.ID, YOUTUBE_DONE_BUTTON_ID) | |
| done_button.click() | |
| # Wait for 2 seconds | |
| time.sleep(2) | |
| # Get latest video | |
| if verbose: | |
| info("\t=> Getting video URL...") | |
| # Get the latest uploaded video URL | |
| driver.get( | |
| f"https://studio.youtube.com/channel/{self.channel_id}/videos/short" | |
| ) | |
| time.sleep(2) | |
| videos = driver.find_elements(By.TAG_NAME, "ytcp-video-row") | |
| first_video = videos[0] | |
| anchor_tag = first_video.find_element(By.TAG_NAME, "a") | |
| href = anchor_tag.get_attribute("href") | |
| if verbose: | |
| info(f"\t=> Extracting video ID from URL: {href}") | |
| video_id = href.split("/")[-2] | |
| # Build URL | |
| url = build_url(video_id) | |
| self.uploaded_video_url = url | |
| if verbose: | |
| success(f" => Uploaded Video: {url}") | |
| # Add video to cache | |
| self.add_video( | |
| { | |
| "title": self.metadata["title"], | |
| "description": self.metadata["description"], | |
| "url": url, | |
| "date": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| } | |
| ) | |
| # Close the browser | |
| driver.quit() | |
| return True | |
| except Exception as e: | |
| error(f"Failed to upload video: {e}") | |
| self.browser.quit() | |
| return False | |
| def get_videos(self) -> List[dict]: | |
| """ | |
| Gets the uploaded videos from the YouTube Channel. | |
| Returns: | |
| videos (List[dict]): The uploaded videos. | |
| """ | |
| _ensure_browser_imports() | |
| if not os.path.exists(get_youtube_cache_path()): | |
| # Create the cache file | |
| with open(get_youtube_cache_path(), "w") as file: | |
| json.dump({"videos": []}, file, indent=4) | |
| return [] | |
| videos = [] | |
| # Read the cache file | |
| with open(get_youtube_cache_path(), "r") as file: | |
| previous_json = json.loads(file.read()) | |
| # Find our account | |
| accounts = previous_json["accounts"] | |
| for account in accounts: | |
| if account["id"] == self._account_uuid: | |
| videos = account["videos"] | |
| return videos | |