import requests import praw import json import cv2 import numpy as np import textwrap from gtts import gTTS from pydub import AudioSegment import subprocess import re import os import random import time import sys import uuid from googleapiclient.discovery import build from googleapiclient.errors import HttpError from googleapiclient.http import MediaFileUpload from oauth2client.client import flow_from_clientsecrets from oauth2client.file import Storage from oauth2client.tools import run_flow from google.auth.transport.requests import Request # Define the output folder path output_folder = 'output' # Constants SCOPES = ["https://www.googleapis.com/auth/youtube.upload"] CLIENT_SECRETS_FILE = "client_secrets.json" # Update with your client_secrets.json file path YOUTUBE_UPLOAD_SCOPE = "https://www.googleapis.com/auth/youtube.upload" DRIVE_SCOPE = "https://www.googleapis.com/auth/drive" YOUTUBE_API_SERVICE_NAME = "youtube" YOUTUBE_API_VERSION = "v3" MAX_RETRIES = 10 RETRIABLE_STATUS_CODES = [500, 502, 503, 504] ELEVENLABS_KEY = "55bfc10fb7eecae379f73e6740807101" # Check if the folder exists, if not, create it if not os.path.exists(output_folder): os.makedirs(output_folder) banned_words = ["fuck", "pussy", "ass", "porn", "gay", "dick", "cock", "kill", "fucking", "shit", "bitch", "bullshit", "asshole","douchebag", "bitch", "motherfucker", "nigga","cunt", "whore", "piss", "shoot", "bomb", "palestine", "israel" ] def contains_banned_word(text, banned_words): for word in banned_words: if word in text.lower(): return True return False def fetch_reddit_data(subreddit_name): # Reddit API Credentials client_id = 'TIacEazZS9FHWzDZ3T-3cA' client_secret = '6Urwdiqo_cC8Gt040K_rBhnR3r8CLg' user_agent = 'script by u/lakpriya1' # Initialize PRAW with your credentials reddit = praw.Reddit(client_id=client_id, client_secret=client_secret, user_agent=user_agent) subreddit = reddit.subreddit(subreddit_name) for _ in range(10): # Limit the number of attempts to 10 post = subreddit.random() # Check if the title contains a pattern resembling a URL if post and not re.search(r'\w+\.\w+', post.title) and not contains_banned_word(post.title, banned_words) and not len(post.title) < 50: post_data = {'title': post.title} with open('top_post.json', 'w') as outfile: json.dump(post_data, outfile, indent=4) print("Top post data saved to top_post.json") return # Exit after finding a suitable post print("No suitable post found without a URL-like string in the title.") def read_json(filename): print("Reading data from", filename) with open(filename, 'r') as file: data = json.load(file) return data def wrap_text(text, wrap_width): return textwrap.wrap(text, width=wrap_width) def resize_background_image(image_path, frame_width, frame_height): print("Resizing background image") image = cv2.imread(image_path) h, w = image.shape[:2] scale = max(frame_width / w, frame_height / h) new_w, new_h = int(w * scale), int(h * scale) resized_image = cv2.resize(image, (new_w, new_h)) # Cropping the resized image to fill the frame startx = new_w // 2 - (frame_width // 2) starty = new_h // 2 - (frame_height // 2) cropped_image = resized_image[starty:starty+frame_height, startx:startx+frame_width] return cropped_image def put_text_with_stroke(frame, text, position, font_scale, line_height, wrap_width, font_color=(255, 255, 255), stroke_color=(0, 0, 0)): font = cv2.FONT_HERSHEY_COMPLEX lines = wrap_text(text, wrap_width) # Calculate the total height of the text block total_text_height = line_height * len(lines) # Starting Y position to center text vertically start_y = (frame.shape[0] - total_text_height) // 2 for line in lines: text_size = cv2.getTextSize(line, font, font_scale, 1)[0] # Calculate x coordinate for center alignment text_x = (frame.shape[1] - text_size[0]) // 2 text_y = start_y + line_height # Draw text stroke (increase thickness for a bolder stroke) cv2.putText(frame, line, (text_x, text_y), font, font_scale, stroke_color, 8, cv2.LINE_AA) # Draw original text on top cv2.putText(frame, line, (text_x, text_y), font, font_scale, font_color, 2, cv2.LINE_AA) start_y += line_height def create_video_from_title(title, background_image, output_filename, audio_duration): print("Creating video from title") # Video properties fps = 24 frame_width, frame_height = 720, 1280 # 9:16 aspect ratio frame_count = audio_duration * fps # Logo images top_logo = load_logo('logo.png', frame_width, frame_height, 'top') bottom_logo = load_logo('sub.png', frame_width, frame_height, 'bottom') # OpenCV VideoWriter fourcc = cv2.VideoWriter_fourcc(*'mp4v') out = cv2.VideoWriter(output_filename, fourcc, fps, (frame_width, frame_height)) # Resize the background image background = resize_background_image(background_image, frame_width, frame_height) for i in range(int(np.floor(frame_count))): frame = background.copy() # Use the resized background image # Overlay logos frame = overlay_logo(frame, top_logo) frame = overlay_logo(frame, bottom_logo) # Add title to frame with text wrapping and highlight put_text_with_stroke(frame, title, (50, 500), 1, 50, 25, font_color=(255, 255, 255), stroke_color=(0, 0, 0)) # Adjust wrap_width and line_height as needed out.write(frame) # Write the frame to the video out.release() def fetch_random_nature_image(api_key): print("Fetching random nature image from Unsplash") url = f"https://api.unsplash.com/photos/random?query=nature&client_id={api_key}" response = requests.get(url) if response.status_code == 200: img_url = response.json()['urls']['regular'] img_data = requests.get(img_url).content with open('nature_background.jpg', 'wb') as handler: handler.write(img_data) return 'nature_background.jpg' else: print("Failed to fetch image from Unsplash") return None def text_to_speech(text, output_file): print("Converting text to speech") tts = gTTS(text=text, lang='en') tts.save(output_file) return output_file def get_audio_duration(audio_file): print("Getting audio duration") audio = AudioSegment.from_mp3(audio_file) return len(audio) / 1000.0 # Convert to seconds def combine_audio_video(video_file, audio_file, output_file, audio_delay_seconds=0.3): # Construct the full path for the output file output_file = os.path.join(output_folder, output_file) # Add a delay to the audio start cmd = f'ffmpeg -i "{video_file}" -itsoffset {audio_delay_seconds} -i "{audio_file}" -c:v copy -c:a aac -strict experimental -map 0:v:0 -map 1:a:0 "{output_file}"' subprocess.call(cmd, shell=True) print("Successfully made the video:", output_file) def load_logo(logo_path, frame_width, frame_height, position='top'): logo = cv2.imread(logo_path, cv2.IMREAD_UNCHANGED) # Load with alpha channel logo_height, logo_width = logo.shape[:2] # Scaling down the logo if it's too big scale_factor = min(1, frame_width / 3 / logo_width, frame_height / 10 / logo_height) new_size = (int(logo_width * scale_factor*1.3), int(logo_height * scale_factor*1.3)) logo = cv2.resize(logo, new_size, interpolation=cv2.INTER_AREA) # Positioning x_center = frame_width // 2 - logo.shape[1] // 2 if position == 'top': y_pos = 100 # 10 pixels from the top else: # 'bottom' y_pos = frame_height - logo.shape[0] - 100 # 10 pixels from the bottom return logo, (x_center, y_pos) def overlay_logo(frame, logo_info): logo, (x, y) = logo_info y1, y2 = y, y + logo.shape[0] x1, x2 = x, x + logo.shape[1] if logo.shape[2] == 4: # If the logo has an alpha channel alpha_logo = logo[:, :, 3] / 255.0 alpha_frame = 1.0 - alpha_logo for c in range(0, 3): frame[y1:y2, x1:x2, c] = (alpha_logo * logo[:, :, c] + alpha_frame * frame[y1:y2, x1:x2, c]) else: # If the logo does not have an alpha channel frame[y1:y2, x1:x2] = logo return frame def get_authenticated_service(): flow = flow_from_clientsecrets(CLIENT_SECRETS_FILE, scope=YOUTUBE_UPLOAD_SCOPE) storage = Storage(f"{sys.argv[0]}-oauth2.json") credentials = storage.get() if credentials is None or credentials.invalid: credentials = run_flow(flow, storage) return build(YOUTUBE_API_SERVICE_NAME, YOUTUBE_API_VERSION, credentials=credentials) def upload_video_to_drive(video_file, folder_id=None): """Uploads a video to Google Drive.""" # Check if the credentials are stored storage = Storage(f"{sys.argv[0]}-oauth2.json") credentials = storage.get() # If credentials are not available or are invalid, run the flow if not credentials or credentials.invalid: flow = flow_from_clientsecrets(CLIENT_SECRETS_FILE, scope=[DRIVE_SCOPE]) credentials = run_flow(flow, storage) service = build('drive', 'v3', credentials=credentials) file_metadata = { 'name': os.path.basename(video_file), 'mimeType': 'video/mp4' } if folder_id: file_metadata['parents'] = [folder_id] media = MediaFileUpload(video_file, mimetype='video/mp4', resumable=True) file = service.files().create(body=file_metadata, media_body=media, fields='id').execute() print('File ID: %s' % file.get('id')) def initialize_upload(youtube, options): tags = None if 'keywords' in options and options['keywords']: tags = options['keywords'].split(",") body = dict( snippet=dict( title=options['title'], description=options['description'], tags=tags, categoryId=options['category'] ), status=dict( privacyStatus=options['privacyStatus'] ) ) # Call the API's videos.insert method to create and upload the video. insert_request = youtube.videos().insert( part=",".join(body.keys()), body=body, # The chunksize parameter specifies the size of each chunk of data, in # bytes, that will be uploaded at a time. Set a higher value for # reliable connections as fewer chunks lead to faster uploads. Set a lower # value for better recovery on less reliable connections. # # Setting "chunksize" equal to -1 in the code below means that the entire # file will be uploaded in a single HTTP request. (If the upload fails, # it will still be retried where it left off.) This is usually a best # practice, but if you're using Python older than 2.6 or if you're # running on App Engine, you should set the chunksize to something like # 1024 * 1024 (1 megabyte). media_body=MediaFileUpload(options["file"], chunksize=-1, resumable=True) ) resumable_upload(insert_request) # This method implements an exponential backoff strategy to resume a # failed upload. def resumable_upload(insert_request): response = None error = None retry = 0 while response is None: try: print("Uploading file...") status, response = insert_request.next_chunk() if response is not None: if 'id' in response: print("Video id '%s' was successfully uploaded." % response['id']) else: exit("The upload failed with an unexpected response: %s" % response) except HttpError as e: if e.resp.status in RETRIABLE_STATUS_CODES: error = "A retriable HTTP error %d occurred:\n%s" % (e.resp.status, e.content) else: raise # except RETRIABLE_EXCEPTIONS as e: # error = "A retriable error occurred: %s" % e if error is not None: print(error) retry += 1 if retry > MAX_RETRIES: exit("No longer attempting to retry.") max_sleep = 2 ** retry sleep_seconds = random.random() * max_sleep print("Sleeping %f seconds and then retrying..." % sleep_seconds) time.sleep(sleep_seconds) def eleven_labs_text_to_speech(text, output_file): voice_ids = { "ndntWUKwYjgJGYkvF6at", "SVLJSgUbrKWfY8HvF2Xd", "sjdiTCylizqR74A3ssv4", } # randomly pick one of the voices voice_id = random.choice(list(voice_ids)) url = f"https://api.elevenlabs.io/v1/text-to-speech/{voice_id}" headers = { "Accept": "audio/mpeg", "Content-Type": "application/json", "xi-api-key": ELEVENLABS_KEY } data = { "text": text, "model_id": "eleven_monolingual_v1", "voice_settings": { "stability": 0.5, "similarity_boost": 0.5, "speed": 0.3, } } response = requests.post(url, json=data, headers=headers) if response.status_code == 200: with open(output_file, 'wb') as f: for chunk in response.iter_content(chunk_size=1024): f.write(chunk) print(f"Audio content written to {output_file}") else: print(f"Failed to synthesize speech: {response.content}") api_key = 'VhLwkCKi3iu5Pf37LXfz-Lp7hTW69EV8uw_hkLAPkiA' # Replace with your Unsplash API key background_image = fetch_random_nature_image(api_key) if background_image: # Example usage fetch_reddit_data('Glitch_in_the_Matrix') # Read data from JSON reddit_data = read_json('top_post.json') # Change filename if needed title = reddit_data.get('title') filename = "video_" + str(uuid.uuid4()) # Convert text to speech # voiceover_file = text_to_speech(title, 'voiceover.mp3') voiceover_file = eleven_labs_text_to_speech(title, 'voiceover.mp3') # Get audio duration audio_duration = get_audio_duration('voiceover.mp3') # Create and save the video create_video_from_title(title, background_image, "reddit_post_video_cv2.mp4", audio_duration) # Combine audio and video combine_audio_video('reddit_post_video_cv2.mp4', 'voiceover.mp3', filename + '.mp4') options = { 'file': 'output/'+ filename + '.mp4', 'title': "Amazing Facts Revealed: Unveiling the World's Hidden Wonders #shorts", 'description': "Welcome to our latest YouTube video, 'Amazing Facts Revealed: Unveiling the World's Hidden Wonders'! In this enthralling episode, we dive deep into the most astonishing and lesser-known facts about our world. From the mysteries of the deep sea to the enigmas of outer space, we cover it all. Get ready to be amazed by incredible scientific discoveries, historical secrets, and mind-blowing natural phenomena. Each fact is meticulously researched and presented with stunning visuals and engaging narration. Don't forget to like, share, and subscribe for more fascinating content. Stay curious and let's explore the wonders of our world together #shorts", 'category': "22", 'keywords': "facts, shorts, funny", 'privacyStatus': "private" } # try: # youtube = get_authenticated_service() # initialize_upload(youtube, options) # upload_video_to_drive('output/'+ filename + '.mp4','1t2lcYNLgz6FTeabzccY_06rvcnTGdQiR') # except HttpError as e: # print("An HTTP error %d occurred:\n%s" % (e.resp.status, e.content))