created new files for transcribing videos , find tranding hashtags
This commit is contained in:
+31
-18
@@ -17,7 +17,7 @@ class Database:
|
|||||||
if self.table == "vods":
|
if self.table == "vods":
|
||||||
self.columns = "id, date, title, gamename, downloaded, uploaded_yt, chat_upload_yt"
|
self.columns = "id, date, title, gamename, downloaded, uploaded_yt, chat_upload_yt"
|
||||||
elif self.table == "clips":
|
elif self.table == "clips":
|
||||||
self.columns = "slug, date, title, gamename, clip_by, view_count, downloaded, uploaded_yt"
|
self.columns = "slug, date, title, gamename, clip_by, view_count, downloaded, uploaded_yt, shorts_upload_yt"
|
||||||
|
|
||||||
if not file_exists:
|
if not file_exists:
|
||||||
self.create_database()
|
self.create_database()
|
||||||
@@ -65,36 +65,49 @@ class Database:
|
|||||||
self.CONN.commit()
|
self.CONN.commit()
|
||||||
self.CONN.close()
|
self.CONN.close()
|
||||||
|
|
||||||
def mark_as_uploaded(self, record_id: int):
|
def __mark_as(self, record_id: int, set_sql: str):
|
||||||
"""Flags a specific clip row record to uploaded (1)."""
|
|
||||||
|
|
||||||
id = "id"
|
id = "id"
|
||||||
|
|
||||||
if self.table == "clips":
|
if self.table == "clips":
|
||||||
id = "slug"
|
id = "slug"
|
||||||
|
|
||||||
self.CURSOR.execute(
|
self.CURSOR.execute(
|
||||||
f"UPDATE {self.table} SET uploaded_yt = 1 WHERE {id} = ?",
|
f"UPDATE {self.table} SET {set_sql} = 1 WHERE {id} = ?",
|
||||||
(record_id,)
|
(record_id,)
|
||||||
)
|
)
|
||||||
self.CONN.commit()
|
self.CONN.commit()
|
||||||
|
|
||||||
|
def mark_as_uploaded_shorts_chats(self, record_id: int):
|
||||||
|
"""Flags a specific row record to uploaded (1)."""
|
||||||
|
|
||||||
|
set_sql = "chat_upload_yt"
|
||||||
|
if self.table == "clips":
|
||||||
|
set_sql = "shorts_uploaded_yt"
|
||||||
|
|
||||||
|
self.__mark_as(record_id, set_sql)
|
||||||
|
|
||||||
|
def mark_as_uploaded(self, record_id: int):
|
||||||
|
"""Flags a specific row record to uploaded (1)."""
|
||||||
|
|
||||||
|
self.__mark_as(record_id, "uploaded_yt")
|
||||||
|
|
||||||
def mark_as_downloaded(self, record_id: int):
|
def mark_as_downloaded(self, record_id: int):
|
||||||
"""Updates the downloaded status to True (1) for a specific record ID."""
|
"""Updates the downloaded status to True (1) for a specific record ID."""
|
||||||
|
|
||||||
id = "id"
|
self.__mark_as(record_id, "download")
|
||||||
if self.table == "clips":
|
|
||||||
id = "slug"
|
|
||||||
|
|
||||||
# Updates the row matching the specific ID
|
def get_unuploaded_shorts_chats(self):
|
||||||
# TODO clips uses slug
|
"""Retrieves all clip rows remaining to be chat uploaded."""
|
||||||
self.CURSOR.execute(
|
|
||||||
f"UPDATE {self.table} SET downloaded = 1 WHERE {id} = ?",
|
where_sql = "chat_upload_yt"
|
||||||
(record_id,)
|
if self.table == "clips":
|
||||||
)
|
where_sql = "shorts_uploaded_yt"
|
||||||
self.CONN.commit()
|
|
||||||
|
self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 1 AND uploaded_yt = 1 AND {where_sql} = 0")
|
||||||
|
return self.CURSOR.fetchall()
|
||||||
|
|
||||||
def get_unuploaded(self):
|
def get_unuploaded(self):
|
||||||
"""Retrieves all clip rows remaining to be chat uploaded."""
|
"""Retrieves all clip rows remaining to be uploaded."""
|
||||||
self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 1 AND uploaded_yt = 0")
|
self.CURSOR.execute(f"SELECT {self.columns} FROM {self.table} WHERE downloaded = 1 AND uploaded_yt = 0")
|
||||||
return self.CURSOR.fetchall()
|
return self.CURSOR.fetchall()
|
||||||
|
|
||||||
@@ -128,7 +141,7 @@ class Database:
|
|||||||
clean_date = record_date_str
|
clean_date = record_date_str
|
||||||
|
|
||||||
self.CURSOR.execute(
|
self.CURSOR.execute(
|
||||||
f"INSERT OR IGNORE INTO {self.table} ({self.columns}) VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
f"INSERT OR IGNORE INTO {self.table} ({self.columns}) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
|
||||||
(slug, str(clean_date), title, gamename, clip_by, views, False, False),
|
(slug, str(clean_date), title, gamename, clip_by, views, False, False, False),
|
||||||
)
|
)
|
||||||
self.CONN.commit()
|
self.CONN.commit()
|
||||||
@@ -0,0 +1,147 @@
|
|||||||
|
#!/usr/bin/env python3
|
||||||
|
import json
|
||||||
|
from google.oauth2.credentials import Credentials
|
||||||
|
from google.auth.transport.requests import Request
|
||||||
|
from googleapiclient.discovery import build
|
||||||
|
from googleapiclient.errors import HttpError
|
||||||
|
|
||||||
|
def load_credentials():
|
||||||
|
"""Load credentials from secrets.json (Reused from your previous flow)"""
|
||||||
|
try:
|
||||||
|
with open('secrets.json', 'r') as f:
|
||||||
|
creds_data = json.load(f)
|
||||||
|
|
||||||
|
credentials = Credentials(
|
||||||
|
token=creds_data['token'],
|
||||||
|
refresh_token=creds_data['refresh_token'],
|
||||||
|
token_uri=creds_data['token_uri'],
|
||||||
|
client_id=creds_data['client_id'],
|
||||||
|
client_secret=creds_data['client_secret'],
|
||||||
|
scopes=creds_data['scopes']
|
||||||
|
)
|
||||||
|
|
||||||
|
if credentials.expired:
|
||||||
|
credentials.refresh(Request())
|
||||||
|
|
||||||
|
return credentials
|
||||||
|
except FileNotFoundError:
|
||||||
|
print("Error: secrets.json not found!")
|
||||||
|
return None
|
||||||
|
except Exception as e:
|
||||||
|
print(f"Error loading credentials: {str(e)}")
|
||||||
|
return None
|
||||||
|
|
||||||
|
def get_my_uploads_playlist_id(youtube):
|
||||||
|
"""Retrieves the system upload playlist ID for the authenticated user's channel."""
|
||||||
|
try:
|
||||||
|
# mine=True automatically references the authorized account
|
||||||
|
request = youtube.channels().list(part="contentDetails", mine=True)
|
||||||
|
response = request.execute()
|
||||||
|
|
||||||
|
if "items" in response and len(response["items"]) > 0:
|
||||||
|
return response["items"][0]["contentDetails"]["relatedPlaylists"]["uploads"]
|
||||||
|
else:
|
||||||
|
print("No channel found for these credentials.")
|
||||||
|
return None
|
||||||
|
except HttpError as e:
|
||||||
|
print(f"API Error retrieving channel details: {str(e)}")
|
||||||
|
return None
|
||||||
|
|
||||||
|
def scan_channel_videos_for_tag(youtube, uploads_playlist_id: str, target_tag: str):
|
||||||
|
"""
|
||||||
|
Loops through all channel video uploads and filters those possessing the target tag.
|
||||||
|
"""
|
||||||
|
target_tag_lower = target_tag.lower()
|
||||||
|
all_videos_count = 0
|
||||||
|
matched_videos = []
|
||||||
|
next_page_token = None
|
||||||
|
|
||||||
|
print("Beginning channel scan...")
|
||||||
|
|
||||||
|
while True:
|
||||||
|
try:
|
||||||
|
# Step A: Retrieve a batch of video IDs from the uploads playlist container
|
||||||
|
playlist_request = youtube.playlistItems().list(
|
||||||
|
part="snippet",
|
||||||
|
playlistId=uploads_playlist_id,
|
||||||
|
maxResults=50,
|
||||||
|
pageToken=next_page_token
|
||||||
|
)
|
||||||
|
playlist_response = playlist_request.execute()
|
||||||
|
|
||||||
|
video_ids_batch = [
|
||||||
|
item["snippet"]["resourceId"]["videoId"]
|
||||||
|
for item in playlist_response.get("items", [])
|
||||||
|
]
|
||||||
|
|
||||||
|
if not video_ids_batch:
|
||||||
|
break
|
||||||
|
|
||||||
|
all_videos_count += len(video_ids_batch)
|
||||||
|
|
||||||
|
# Step B: Pass batch to videos().list to extract metadata details (including tags)
|
||||||
|
video_request = youtube.videos().list(
|
||||||
|
part="snippet",
|
||||||
|
id=",".join(video_ids_batch)
|
||||||
|
)
|
||||||
|
video_response = video_request.execute()
|
||||||
|
|
||||||
|
for video in video_response.get("items", []):
|
||||||
|
title = video["snippet"]["title"]
|
||||||
|
video_id = video["id"]
|
||||||
|
# Tags are optional fields on YouTube; default to an empty list if absent
|
||||||
|
tags = video["snippet"].get("tags", [])
|
||||||
|
|
||||||
|
# Normalize tags to lowercase for clean matching evaluation
|
||||||
|
tags_lower = [tag.lower() for tag in tags]
|
||||||
|
|
||||||
|
if target_tag_lower in tags_lower:
|
||||||
|
matched_videos.append({
|
||||||
|
"id": video_id,
|
||||||
|
"title": title,
|
||||||
|
"tags": tags
|
||||||
|
})
|
||||||
|
print(f"🔍 Found Match: '{title}' (ID: {video_id})")
|
||||||
|
|
||||||
|
# Check if another page token exists, if not break the pagination loop
|
||||||
|
next_page_token = playlist_response.get("nextPageToken")
|
||||||
|
if not next_page_token:
|
||||||
|
break
|
||||||
|
|
||||||
|
except HttpError as e:
|
||||||
|
print(f"An error occurred while fetching video batches: {str(e)}")
|
||||||
|
break
|
||||||
|
|
||||||
|
# Summary reporting
|
||||||
|
print("\n" + "="*40)
|
||||||
|
print(f"Scan complete. Analyzed {all_videos_count} total videos.")
|
||||||
|
print(f"Found {len(matched_videos)} videos containing the '{target_tag}' tag.")
|
||||||
|
print("="*40)
|
||||||
|
|
||||||
|
return matched_videos
|
||||||
|
|
||||||
|
def main():
|
||||||
|
credentials = load_credentials()
|
||||||
|
if not credentials:
|
||||||
|
return
|
||||||
|
|
||||||
|
youtube = build('youtube', 'v3', credentials=credentials)
|
||||||
|
|
||||||
|
# 1. Fetch your dynamic uploads playlist pointer
|
||||||
|
uploads_id = get_my_uploads_playlist_id(youtube)
|
||||||
|
|
||||||
|
if uploads_id:
|
||||||
|
print(f"Target Uploads Playlist ID: {uploads_id}")
|
||||||
|
|
||||||
|
# 2. Run the iterative match parser targeting the 'clips' keyword tag
|
||||||
|
target_keyword = "clips"
|
||||||
|
results = scan_channel_videos_for_tag(youtube, uploads_id, target_keyword)
|
||||||
|
|
||||||
|
# 3. Print out a neat clean list of matches
|
||||||
|
if results:
|
||||||
|
print(f"\n--- List of matching videos for tag '{target_keyword}': ---")
|
||||||
|
for index, item in enumerate(results, start=1):
|
||||||
|
print(f"{index}. {item['title']} -> https://youtu.be{item['id']}")
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
@@ -0,0 +1,41 @@
|
|||||||
|
import os
|
||||||
|
import whisper
|
||||||
|
from moviepy.editor import VideoFileClip
|
||||||
|
from whisper.utils import get_writer
|
||||||
|
|
||||||
|
def extract_audio(video_path, audio_temp_path):
|
||||||
|
print("Extracting uncompressed WAV audio...")
|
||||||
|
video = VideoFileClip(video_path)
|
||||||
|
# Extract as WAV, strictly setting the sample rate to 16000Hz for Whisper
|
||||||
|
video.audio.write_audiofile(
|
||||||
|
audio_temp_path,
|
||||||
|
codec="pcm_s16le",
|
||||||
|
ffmpeg_params=["-ar", "16000", "-ac", "1"]
|
||||||
|
)
|
||||||
|
video.close()
|
||||||
|
|
||||||
|
def transcribe_to_srt(audio_path, output_directory, output_filename):
|
||||||
|
print("Loading Whisper model...")
|
||||||
|
model = whisper.load_model("base")
|
||||||
|
|
||||||
|
print("Transcribing audio...")
|
||||||
|
result = model.transcribe(audio_path)
|
||||||
|
|
||||||
|
print("Creating SRT file...")
|
||||||
|
srt_writer = get_writer("srt", output_directory)
|
||||||
|
srt_writer(result, output_filename, {})
|
||||||
|
|
||||||
|
print(f"SRT subtitle file saved in: {output_directory}")
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
video_path = "my_video.mp4"
|
||||||
|
audio_temp_path = "temp_audio.wav" # Changed extension to .wav
|
||||||
|
|
||||||
|
output_dir = os.getcwd()
|
||||||
|
output_prefix = "my_video_subtitles"
|
||||||
|
|
||||||
|
extract_audio(video_path, audio_temp_path)
|
||||||
|
transcribe_to_srt(audio_temp_path, output_dir, output_prefix)
|
||||||
|
|
||||||
|
if os.path.exists(audio_temp_path):
|
||||||
|
os.remove(audio_temp_path)
|
||||||
@@ -22,13 +22,46 @@ def run_linux_command(command: str):
|
|||||||
except subprocess.CalledProcessError as e:
|
except subprocess.CalledProcessError as e:
|
||||||
return {"success": False, "stdout": e.stdout, "stderr": e.stderr}
|
return {"success": False, "stdout": e.stdout, "stderr": e.stderr}
|
||||||
|
|
||||||
|
def transcribe(slug: str):
|
||||||
|
"""Transcribes the Video File."""
|
||||||
|
video_file = f"save/clips/{slug}/{slug}.mp4"
|
||||||
|
|
||||||
|
# Check if the video file exists
|
||||||
|
if not os.path.exists(video_file):
|
||||||
|
print(f"Error: File not found: {video_file}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
# no need to continue if srt transcribe file already exists
|
||||||
|
if os.path.exists(f"save/clips/{slug}/transcribe_{slug}.srt"):
|
||||||
|
print(f"video already transcribed:")
|
||||||
|
return True
|
||||||
|
|
||||||
|
import transcribe_video
|
||||||
|
transcribe_video.extract_audio(video_file, f"save/clips/{slug}/temp_{slug}_audio.wav")
|
||||||
|
transcribe_video.transcribe_to_srt(f"save/clips/{slug}/temp_{slug}_audio.wav", f"save/clips/{slug}/", f"transcribe_{slug}")
|
||||||
|
|
||||||
|
return True
|
||||||
|
|
||||||
|
def top_hashtags(slug: str):
|
||||||
|
file_srt = f"save/clips/{slug}/transcribe_{slug}.srt"
|
||||||
|
|
||||||
|
# if srt transcribe file not exists
|
||||||
|
if not os.path.exists(file_srt):
|
||||||
|
print(f"Transcribe file not found {file_srt}")
|
||||||
|
return []
|
||||||
|
|
||||||
|
import youtube_hashtags
|
||||||
|
|
||||||
|
return youtube_hashtags.get_top_hashtags(file_srt)
|
||||||
|
|
||||||
|
|
||||||
def download_clips():
|
def download_clips():
|
||||||
"""Loops over undownloaded metadata entries to write files down locally."""
|
"""Loops over undownloaded metadata entries to write files down locally."""
|
||||||
undownloaded_clips = DB.get_undownloaded()
|
undownloaded_clips = DB.get_undownloaded()
|
||||||
|
|
||||||
print("Downloading Clips...")
|
print("Downloading Clips...")
|
||||||
|
|
||||||
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in undownloaded_clips:
|
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt, uploaded_shorts_yt in undownloaded_clips:
|
||||||
print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}")
|
print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}")
|
||||||
|
|
||||||
# Uses standard clipdownload directive
|
# Uses standard clipdownload directive
|
||||||
@@ -44,23 +77,25 @@ def download_clips():
|
|||||||
|
|
||||||
def upload_clips():
|
def upload_clips():
|
||||||
"""Loops over the downloaded videos entries and uploaded them to youtube."""
|
"""Loops over the downloaded videos entries and uploaded them to youtube."""
|
||||||
print("Uploading Clips...")
|
print(f"Uploading Clips...")
|
||||||
|
|
||||||
unpuloaded_clips = DB.get_unuploaded()
|
unuploaded_clips = DB.get_unuploaded()
|
||||||
|
|
||||||
categoryId = CategoryId.GAMING
|
categoryId = CategoryId.GAMING
|
||||||
|
|
||||||
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in unpuloaded_clips:
|
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in unuploaded_clips:
|
||||||
print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}")
|
print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}")
|
||||||
|
|
||||||
file_path = f"save/clips/{slug}/{slug}.mp4"
|
file_path = f"save/clips/{slug}/{slug}.mp4"
|
||||||
title = title
|
title = title
|
||||||
description = f"Game: {game_name}, Cliped By: {clip_by}, on {record_date}, #Shorts #Clips #Twitch"
|
description = f"Game: {game_name}, Cliped By: {clip_by}, on {record_date}, #Shorts #Clips #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp"
|
||||||
categoryId = CategoryId.GAMING
|
categoryId = CategoryId.GAMING
|
||||||
privatcyStatus = 'private'
|
privatcyStatus = 'private'
|
||||||
tags = ['shorts', 'gaming', 'TeamPGP', f'{game_name}', f'{clip_by}', 'twitch_clips', 'clips']
|
tags = ['shorts', 'gaming', 'TeamPGP', f'{game_name}', f'{clip_by}', 'twitch_clips', 'clips', 'Level1Techs', 'twitch']
|
||||||
|
|
||||||
output = uploader.upload_video(file_path, title, categoryId, description, privatcyStatus, tags)
|
tags.extend(top_hashtags({slug}))
|
||||||
|
|
||||||
|
#output = uploader.upload_video(file_path, title, categoryId, description, privatcyStatus, tags)
|
||||||
|
|
||||||
if output is True:
|
if output is True:
|
||||||
print(f"Slug: {slug} | Was successfully uploaded.")
|
print(f"Slug: {slug} | Was successfully uploaded.")
|
||||||
|
|||||||
+78
-10
@@ -27,13 +27,46 @@ def run_linux_command(command: str):
|
|||||||
# Handles errors if the Linux command returns a non-zero exit code
|
# Handles errors if the Linux command returns a non-zero exit code
|
||||||
return {"success": False, "stdout": e.stdout, "stderr": e.stderr}
|
return {"success": False, "stdout": e.stdout, "stderr": e.stderr}
|
||||||
|
|
||||||
def download_vods():
|
def transcribe(slug: str):
|
||||||
|
"""Transcribes the Video File."""
|
||||||
|
video_file = f"save/clips/{slug}/{slug}.mp4"
|
||||||
|
|
||||||
|
# Check if the video file exists
|
||||||
|
if not os.path.exists(video_file):
|
||||||
|
print(f"Error: File not found: {video_file}")
|
||||||
|
return False
|
||||||
|
|
||||||
|
# no need to continue if srt transcribe file already exists
|
||||||
|
if os.path.exists(f"save/clips/{slug}/transcribe_{slug}.srt"):
|
||||||
|
print(f"video already transcribed:")
|
||||||
|
return True
|
||||||
|
|
||||||
|
import transcribe_video
|
||||||
|
transcribe_video.extract_audio(video_file, f"save/clips/{slug}/temp_{slug}_audio.wav")
|
||||||
|
transcribe_video.transcribe_to_srt(f"save/clips/{slug}/temp_{slug}_audio.wav", f"save/clips/{slug}/", f"transcribe_{slug}")
|
||||||
|
|
||||||
|
return True
|
||||||
|
|
||||||
|
def top_hashtags(slug: str):
|
||||||
|
file_srt = f"save/clips/{slug}/transcribe_{slug}.srt"
|
||||||
|
|
||||||
|
# if srt transcribe file not exists
|
||||||
|
if not os.path.exists(file_srt):
|
||||||
|
print(f"Transcribe file not found {file_srt}")
|
||||||
|
return []
|
||||||
|
|
||||||
|
import youtube_hashtags
|
||||||
|
|
||||||
|
return youtube_hashtags.get_top_hashtags(file_srt)
|
||||||
|
|
||||||
|
def download():
|
||||||
"""Find all undownload vods and download them."""
|
"""Find all undownload vods and download them."""
|
||||||
undownloaded_vods = DB.get_undownloaded()
|
undownloaded = DB.get_undownloaded()
|
||||||
|
|
||||||
print("Download VODs...")
|
print(f"Download {DB.table}...")
|
||||||
|
|
||||||
for record_id, record_date, title, game_name, downloaded, uploaded_yt, chat_upload_yt in undownloaded_vods:
|
if DB.table == "vods":
|
||||||
|
for record_id, record_date, title, game_name, downloaded, uploaded_yt, chat_upload_yt in undownloaded:
|
||||||
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title}")
|
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title}")
|
||||||
output = run_linux_command(f"TwitchDownloaderCLI videodownload --id {record_id} -o save/vods/{record_id}/{title}.mp4")
|
output = run_linux_command(f"TwitchDownloaderCLI videodownload --id {record_id} -o save/vods/{record_id}/{title}.mp4")
|
||||||
output2 = run_linux_command(f"TwitchDownloaderCLI chatdownload --id {record_id} -o save/vods/{record_id}/{title}_chat.json -E")
|
output2 = run_linux_command(f"TwitchDownloaderCLI chatdownload --id {record_id} -o save/vods/{record_id}/{title}_chat.json -E")
|
||||||
@@ -43,23 +76,38 @@ def download_vods():
|
|||||||
else:
|
else:
|
||||||
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title} | was Failed")
|
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title} | was Failed")
|
||||||
|
|
||||||
print("Finished Downloading VODs...")
|
elif DB.table == "clips":
|
||||||
|
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt, uploaded_shorts_yt in undownloaded:
|
||||||
|
print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}")
|
||||||
|
|
||||||
def upload_vods():
|
# Uses standard clipdownload directive
|
||||||
|
output = run_linux_command(f"TwitchDownloaderCLI clipdownload --id {slug} -o save/clips/{slug}/{title}.mp4")
|
||||||
|
|
||||||
|
if output["success"] is True:
|
||||||
|
print(f"Slug: {slug} | Was successfully downloaded.")
|
||||||
|
DB.mark_as_downloaded(slug)
|
||||||
|
else:
|
||||||
|
print(f"Slug: {slug} | Download Process failed. {output["stdout"]}. Error: {output["stderr"]}")
|
||||||
|
|
||||||
|
|
||||||
|
print(f"Finished Downloading {DB.table}...")
|
||||||
|
|
||||||
|
def upload():
|
||||||
"""Loops over the downloaded videos entries and uploaded them to youtube."""
|
"""Loops over the downloaded videos entries and uploaded them to youtube."""
|
||||||
print("Uploading Clips...")
|
print(f"Uploading {DB.table}...")
|
||||||
|
|
||||||
unuploaded_vods = DB.get_unuploaded()
|
unuploaded = DB.get_unuploaded()
|
||||||
|
|
||||||
|
if DB.table == "vods":
|
||||||
for record_id, record_date, title, game_name, downloaded, uploaded_yt, chat_upload_yt in unuploaded_vods:
|
for record_id, record_date, title, game_name, downloaded, uploaded_yt, chat_upload_yt in unuploaded_vods:
|
||||||
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title}")
|
print(f"ID: {record_id} | Date: {record_date} | Game: {game_name} | Title: {title}")
|
||||||
|
|
||||||
file_path = f"save/vods/{record_id}/{title}.mp4"
|
file_path = f"save/vods/{record_id}/{title}.mp4"
|
||||||
title = title
|
title = title
|
||||||
description = f"Game: {game_name}, on {record_date}, #Shorts"
|
description = f"Game: {game_name}, on {record_date}, #VODS #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp"
|
||||||
categoryId = CategoryId.GAMING
|
categoryId = CategoryId.GAMING
|
||||||
privatcyStatus = 'private'
|
privatcyStatus = 'private'
|
||||||
tags = ['gaming', 'TeamPGP', f'{game_name}']
|
tags = ['gaming', 'TeamPGP', f'{game_name}', 'twitch_vods', 'vods', 'Level1Techs', 'twitch']
|
||||||
|
|
||||||
output = uploader.upload_video(file_path, title, description, categoryId, privatcyStatus, tags)
|
output = uploader.upload_video(file_path, title, description, categoryId, privatcyStatus, tags)
|
||||||
|
|
||||||
@@ -68,6 +116,26 @@ def upload_vods():
|
|||||||
DB.mark_as_uploaded(record_id)
|
DB.mark_as_uploaded(record_id)
|
||||||
else:
|
else:
|
||||||
print(f"ID: {record_id} | Upload Process failed.")
|
print(f"ID: {record_id} | Upload Process failed.")
|
||||||
|
elif DB.table == "clips":
|
||||||
|
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in unuploaded_clips:
|
||||||
|
print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}")
|
||||||
|
|
||||||
|
file_path = f"save/clips/{slug}/{title}.mp4"
|
||||||
|
title = title
|
||||||
|
description = f"Game: {game_name}, Cliped By: {clip_by}, on {record_date}, #Shorts #Clips #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp"
|
||||||
|
categoryId = CategoryId.GAMING
|
||||||
|
privatcyStatus = 'private'
|
||||||
|
tags = ['shorts', 'gaming', 'TeamPGP', f'{game_name}', f'{clip_by}', 'twitch_clips', 'clips', 'Level1Techs', 'twitch']
|
||||||
|
|
||||||
|
tags.extend(top_hashtags({slug}))
|
||||||
|
|
||||||
|
#output = uploader.upload_video(file_path, title, categoryId, description, privatcyStatus, tags)
|
||||||
|
|
||||||
|
if output is True:
|
||||||
|
print(f"Slug: {slug} | Was successfully uploaded.")
|
||||||
|
DB.mark_as_uploaded(slug)
|
||||||
|
else:
|
||||||
|
print(f"Slug: {slug} | Upload Process failed.")
|
||||||
|
|
||||||
def create_chat_vods():
|
def create_chat_vods():
|
||||||
pass
|
pass
|
||||||
|
|||||||
+20
-5
@@ -160,7 +160,6 @@ def upload_video(file_path: str, title: str, category: CategoryId, description:
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
def main():
|
def main():
|
||||||
# FIXED: Reconstructed arguments to provide all required inputs for upload_video
|
|
||||||
parser = argparse.ArgumentParser(description='Upload a video to YouTube')
|
parser = argparse.ArgumentParser(description='Upload a video to YouTube')
|
||||||
parser.add_argument('--file', required=True, help='Path to the video file')
|
parser.add_argument('--file', required=True, help='Path to the video file')
|
||||||
parser.add_argument('--title', required=True, help='Title of the video')
|
parser.add_argument('--title', required=True, help='Title of the video')
|
||||||
@@ -168,19 +167,35 @@ def main():
|
|||||||
parser.add_argument('--description', default='', help='Video description text')
|
parser.add_argument('--description', default='', help='Video description text')
|
||||||
parser.add_argument('--privacy', default='private', choices=['public', 'private', 'unlisted'], help='Video privacy settings')
|
parser.add_argument('--privacy', default='private', choices=['public', 'private', 'unlisted'], help='Video privacy settings')
|
||||||
|
|
||||||
|
# ADDED: Feature parsing to easily pass tags from the CLI split by commas
|
||||||
|
parser.add_argument('--tags', default='', help='Comma-separated tags list (e.g. "python,coding,api")')
|
||||||
|
|
||||||
|
# ADDED: Option to provide a scheduled upload timestamp natively from CLI
|
||||||
|
parser.add_argument('--schedule', default=None, help='UTC Release date/time in ISO format: YYYY-MM-DDTHH:MM:SS (e.g. 2026-08-15T14:30:00)')
|
||||||
|
|
||||||
args = parser.parse_args()
|
args = parser.parse_args()
|
||||||
|
|
||||||
# Map the text input choice back to your CategoryId Enum object
|
|
||||||
chosen_category = CategoryId[args.category]
|
chosen_category = CategoryId[args.category]
|
||||||
|
parsed_tags = [t.strip() for t in args.tags.split(',')] if args.tags else []
|
||||||
|
|
||||||
|
# ADDED: Parse schedule string into datetime object dynamically
|
||||||
|
release_datetime = None
|
||||||
|
if args.schedule:
|
||||||
|
try:
|
||||||
|
# Assumes format matches CLI help instruction text
|
||||||
|
release_datetime = datetime.strptime(args.schedule, '%Y-%m-%dT%H:%M:%S').replace(tzinfo=timezone.utc)
|
||||||
|
except ValueError:
|
||||||
|
print("Error: Schedule date must format strictly as YYYY-MM-DDTHH:MM:SS")
|
||||||
|
return
|
||||||
|
|
||||||
# FIXED: Properly pass all required positional and optional parameters
|
|
||||||
upload_video(
|
upload_video(
|
||||||
file_path=args.file,
|
file_path=args.file,
|
||||||
title=args.title,
|
title=args.title,
|
||||||
category=chosen_category,
|
category=chosen_category,
|
||||||
description=args.description,
|
description=args.description,
|
||||||
privacyStatus=args.privacy
|
privacyStatus=args.privacy,
|
||||||
|
tags=parsed_tags,
|
||||||
|
release_time=release_datetime
|
||||||
)
|
)
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
main()
|
main()
|
||||||
@@ -0,0 +1,48 @@
|
|||||||
|
import re
|
||||||
|
import string
|
||||||
|
from collections import Counter
|
||||||
|
from nltk.corpus import stopwords
|
||||||
|
from nltk.tokenize import word_tokenize
|
||||||
|
import nltk
|
||||||
|
|
||||||
|
# Download necessary NLTK data
|
||||||
|
nltk.download('punkt', quiet=True)
|
||||||
|
nltk.download('stopwords', quiet=True)
|
||||||
|
|
||||||
|
def extract_text_from_srt(file_path):
|
||||||
|
with open(file_path, 'r', encoding='utf-8') as file:
|
||||||
|
content = file.read()
|
||||||
|
# Remove SRT timestamps and sequence numbers
|
||||||
|
clean_text = re.sub(r'\d+\n\d{2}:\d{2}:\d{2},\d{3} --> \d{2}:\d{2}:\d{2},\d{3}\n', '', content)
|
||||||
|
clean_text = re.sub(r'\d+', '', clean_text)
|
||||||
|
return clean_text
|
||||||
|
|
||||||
|
def get_top_hashtags(srt_file_path, top_n=10):
|
||||||
|
raw_text = extract_text_from_srt(srt_file_path)
|
||||||
|
|
||||||
|
# Lowercase and remove punctuation
|
||||||
|
raw_text = raw_text.lower()
|
||||||
|
raw_text = raw_text.translate(str.maketrans('', '', string.punctuation))
|
||||||
|
|
||||||
|
# Tokenize and remove stopwords
|
||||||
|
words = word_tokenize(raw_text)
|
||||||
|
stop_words = set(stopwords.words('english'))
|
||||||
|
|
||||||
|
# Filter for alphabetical words longer than 3 characters that aren't stop words
|
||||||
|
filtered_words = [
|
||||||
|
word for word in words
|
||||||
|
if word.isalpha() and word not in stop_words and len(word) > 3
|
||||||
|
]
|
||||||
|
|
||||||
|
# Get frequency and create hashtags
|
||||||
|
word_counts = Counter(filtered_words)
|
||||||
|
top_words = word_counts.most_common(top_n)
|
||||||
|
|
||||||
|
hashtags = [f"#{word[0]}" for word in top_words]
|
||||||
|
return hashtags
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
# Example usage
|
||||||
|
# Replace 'your_video.srt' with the path to your file
|
||||||
|
results = get_top_hashtags('your_video.srt', top_n=10)
|
||||||
|
print("Trending Hashtags:", results)
|
||||||
+32
-1
@@ -1,7 +1,12 @@
|
|||||||
import sys
|
import sys
|
||||||
from moviepy.editor import VideoFileClip
|
from moviepy.editor import VideoFileClip
|
||||||
|
|
||||||
def convert_to_short(input_path, output_path="youtube_short.mp4"):
|
def convert_to_short(input_path):
|
||||||
|
|
||||||
|
input_file = Path(input_path)
|
||||||
|
|
||||||
|
output_path = f"{input_file.stem}_shorts{input_file.suffix}"
|
||||||
|
|
||||||
# Load the video file
|
# Load the video file
|
||||||
clip = VideoFileClip(input_path)
|
clip = VideoFileClip(input_path)
|
||||||
|
|
||||||
@@ -39,5 +44,31 @@ def convert_to_short(input_path, output_path="youtube_short.mp4"):
|
|||||||
print(f"🎉 Success! Short saved as: {output_path}")
|
print(f"🎉 Success! Short saved as: {output_path}")
|
||||||
return True
|
return True
|
||||||
|
|
||||||
|
def upload_shorts():
|
||||||
|
"""Loops over the downloaded videos entries and uploaded them to youtube."""
|
||||||
|
print("Uploading Shorts...")
|
||||||
|
|
||||||
|
unuploaded_shorts = DB.get_unuploaded()
|
||||||
|
|
||||||
|
categoryId = CategoryId.GAMING
|
||||||
|
|
||||||
|
for slug, record_date, title, game_name, clip_by, views, downloaded, uploaded_yt in unuploaded_shorts:
|
||||||
|
print(f"Slug: {slug} | Date: {record_date} | Game: {game_name} | By: {clip_by} | Views: {views} | Title: {title}")
|
||||||
|
|
||||||
|
file_path = f"save/clips/{slug}/{slug}_shorts.mp4"
|
||||||
|
title = title
|
||||||
|
description = f"Game: {game_name}, Cliped By: {clip_by}, on {record_date}, #Shorts #Clips #Twitch Every Friday and Sunday @7:30 EST https://twitch.tv/teampgp"
|
||||||
|
categoryId = CategoryId.GAMING
|
||||||
|
privatcyStatus = 'private'
|
||||||
|
tags = ['shorts', 'gaming', 'TeamPGP', f'{game_name}', f'{clip_by}', 'twitch_clips', 'clips', 'Level1Techs', 'twitch']
|
||||||
|
|
||||||
|
output = uploader.upload_video(file_path, title, categoryId, description, privatcyStatus, tags)
|
||||||
|
|
||||||
|
if output is True:
|
||||||
|
print(f"Slug: {slug} | Was successfully uploaded.")
|
||||||
|
DB.mark_as_uploaded(slug)
|
||||||
|
else:
|
||||||
|
print(f"Slug: {slug} | Upload Process failed.")
|
||||||
|
|
||||||
if __name__ == "__main__":
|
if __name__ == "__main__":
|
||||||
convert_to_short()
|
convert_to_short()
|
||||||
Reference in New Issue
Block a user