SumGuyV5/python_scripts
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

update to the process_uploads upload.py and new database table called data

Browse Source
This commit is contained in:
SumGuyV5 committed 2026-09-16 01:11:38 +00:00
1 parent afe6490d20
commit 2773200a94
17 files changed
+245 -70
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

package/youtube/__init__.py
-3
View File
@@ -2,14 +2,11 @@
# Expose components from individual modules using relative imports
from .find_tags import find_tags
from .hashtags import get_top_nouns
from .shorts import fit_to_9_16_letterbox
from .uploader import upload_video, CategoryId
# Explicitly define what is exposed when importing the subpackage
__all__ = [
"find_tags",
"get_top_nouns",
"fit_to_9_16_letterbox",
"upload_video",
]
package/youtube/hashtags.py
-104
View File
@@ -1,104 +0,0 @@
#!/usr/bin/env python3
import re
import json
import nltk
import string
from collections import Counter
from nltk.corpus import stopwords
from nltk.tokenize import word_tokenize
# Download necessary NLTK data modules
nltk.download('punkt', quiet=True)
nltk.download('stopwords', quiet=True)
nltk.download('punkt_tab', quiet=True)
nltk.download('averaged_perceptron_tagger', quiet=True) # Required for POS tagging
nltk.download('averaged_perceptron_tagger_eng', quiet=True)
def extract_text_from_srt(file_path: str) -> str:
"""
Extracts the text from the srt file.
Args:
file_path (str): is the path to the str file.
Returns:
returns (str): returns the cleaned up text.
"""
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 extract_text_from_json(file_path: str) -> str:
"""
Extracts the text from the json file.
Args:
file_path (str): is the path to the json file.
Returns:
returns (str): returns the cleaned up text.
"""
with open(file_path, "r", encoding="utf-8") as file:
data = json.load(file)
messages = []
for comment in data.get("comments", []):
message_text = comment.get("message", {}).get("body", "")
messages.append(message_text)
# Return a single merged string of all chat text
return " ".join(messages)
def get_top_nouns(file_path: str, top_n: int = 10) -> list:
"""
Gets the top number of nouns based on past (top_n) arg.
Args:
file_path (str): is the path to the text file.
top_n (int): is the number of nouns to return.
Returns:
returns (list): returns a list of nouns.
"""
# 1. Extract raw text
if file_path.endswith(".srt"):
raw_text = extract_text_from_srt(file_path)
else:
raw_text = extract_text_from_json(file_path)
# 2. Basic cleanup (Keep original case for proper noun accuracy)
# Strip basic punctuation but leave words intact
clean_text = raw_text.translate(str.maketrans('', '', string.punctuation))
# 3. Tokenize words
words = word_tokenize(clean_text)
# 4. Part-of-Speech Tagging
tagged_words = nltk.pos_tag(words)
# 5. Filter for Nouns (NN = Singular Noun, NNP = Proper Noun, NNS = Plural Noun)
stop_words = set(stopwords.words('english'))
nouns = []
for word, tag in tagged_words:
word_lower = word.lower()
# Filter out short fragments and standard stopwords
if tag in ['NN', 'NNP', 'NNS'] and len(word_lower) > 2 and word_lower not in stop_words:
nouns.append(word_lower)
# 6. Count frequencies
noun_counts = Counter(nouns)
return [noun for noun, count in noun_counts.most_common(top_n)]
if __name__ == "__main__":
# Example usage
# Replace 'your_video.srt' with the path to your file
results = get_top_nouns('download/videos/2813112936/2813112936.srt', top_n=10)
print("Trending Hashtags SRT:", results)
results = get_top_nouns("download/videos/2813112936/2813112936_chat.json", top_n=10)
print("Trending Hashtags JSON:", results)
package/youtube/shorts.py
-142
View File
@@ -1,142 +0,0 @@
#!/usr/bin/env python3
import os
import json
import package.utils as utils
import tempfile
from pathlib import Path
from moviepy import VideoFileClip, ColorClip, CompositeVideoClip
from moviepy.video.VideoClip import TextClip
from pathlib import Path
import numpy as np
from PIL import Image, ImageFilter
def get_video_info(input_path: Path) -> tuple:
"""Uses ffprobe to instantly read input video dimensions and frame rate."""
cmd = f"ffprobe -v error -select_streams v:0 -show_entries stream=width,height,r_frame_rate -of json {input_path}"
# Run command and capture output (assumes linux.run_command prints or you use subprocess)
import subprocess
result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
#linux.run_command(cmd)
try:
data = json.loads(result.stdout)
stream = data['streams'][0]
w = int(stream['width'])
h = int(stream['height'])
# Convert fractional FPS string (e.g. "60/1" or "30000/1001") to float
fps_parts = stream['r_frame_rate'].split('/')
fps = float(fps_parts[0]) / float(fps_parts[1]) if len(fps_parts) > 1 else float(fps_parts[0])
return w, h, fps
except Exception:
return 1920, 1080, 60.0 # Safe defaults if probe fails
def fit_to_9_16_letterbox(input_path: str, top_text: str = "TOP TEXT", bottom_text: str = "BOTTOM TEXT", use_blur: bool = True, force: bool = False):
threads = "8"
input_file = Path(input_path)
output_suffix = "gaussian_9_16" if use_blur else "black_9_16"
output_path = input_file.parent / f"{input_file.stem}_{output_suffix}{input_file.suffix}"
if os.path.exists(output_path):
if not force:
print(f"❌ Error: Short video already exists: {output_path}")
return
else:
os.remove(output_path)
# 1. Probe input metadata instantly
orig_w, orig_h, fps = get_video_info(input_file)
canvas_w = 1080
canvas_h = 1920
print("✍️ Generating text overlay graphics via MoviePy...")
# Render static images for text instead of running a video context
title_clip = TextClip(
text=top_text, font_size=55, color="white", font="DejaVuSans-Bold",
text_align="center", size=(canvas_w - 100, 300), method="caption"
)
bottom_clip = TextClip(
text=bottom_text, font_size=55, color="white", font="DejaVuSans-Bold",
text_align="center", size=(canvas_w - 100, 300), method="caption"
)
# Save text layers to temporary PNGs
top_png = tempfile.NamedTemporaryFile(suffix=".png", delete=False).name
bottom_png = tempfile.NamedTemporaryFile(suffix=".png", delete=False).name
title_clip.save_frame(top_png)
bottom_clip.save_frame(bottom_png)
title_clip.close()
bottom_clip.close()
print("🎬 Dispatching compilation workload to FFmpeg filtergraph...")
# 2. Build the complex FFmpeg filtergraph
# [0:v] is the raw input video stream
filter_complex = []
if use_blur:
# Scale height to 1920, crop center 1080x1920, apply fast boxblur (power of 3 approximates Gaussian)
filter_complex.append(
f"[0:v]scale=-1:{canvas_h},crop={canvas_w}:{canvas_h}:(iw-{canvas_w})/2:0,boxblur=luma_radius=35:luma_power=3[bg];"
)
else:
# Generate a pure black background canvas matching video frame specs
filter_complex.append(
f"color=c=black:s={canvas_w}x{canvas_h}:r={fps}[bg];"
)
# Scale the foreground video to a clean 1080 width, keeping aspect ratio
filter_complex.append(
f"[0:v]scale={canvas_w}:-1[fg];"
)
# Layer composition chain:
# Overlay 1: Put scaled foreground onto background (centered vertically)
filter_complex.append(
f"[bg][fg]overlay=0:(H-h)/2[tmp1];"
)
# Overlay 2: Drop top text asset onto position Y=180
filter_complex.append(
f"[tmp1][1:v]overlay=(W-w)/2:180[tmp2];"
)
# Overlay 3: Drop bottom text asset onto position Y=1430
filter_complex.append(
f"[tmp2][2:v]overlay=(W-w)/2:1430[finalv]"
)
filter_graph = "".join(filter_complex)
# 3. Execute the native assembly command
# -map_chapters -1 -sn: Strips unnecessary metadata chunks instantly
# -c:a copy: Safely pulls original digital audio directly without decompression cycles
# -threads 0: Forces FFmpeg to auto-consume all available processing cores
ffmpeg_cmd = (
f'ffmpeg -y -v error -i "{input_file}" -i "{top_png}" -i "{bottom_png}" '
f'-filter_complex "{filter_graph}" '
f'-map "[finalv]" -map 0:a? -c:v libx264 -crf 18 -preset slow -pix_fmt yuv420p '
f'-c:a copy -map_chapters -1 -sn -threads {threads} "{output_path}"'
)
try:
utils.run_command(ffmpeg_cmd)
print(f"🎉 High-speed processing complete! Video saved to: {output_path}")
finally:
# Clean up temporary PNG picture files safely
for path in (top_png, bottom_png):
if os.path.exists(path):
os.unlink(path)
if __name__ == "__main__":
import time
start_time = time.perf_counter()
creator_name = "greenskiesbluegrass"
top_txt = "TeamPGP Live on Twitch...\n Every Friday and Sunday @7:30 ET.\n twitch.tv/teampgp"
bottom_txt = f"Clipped By: {creator_name}."
fit_to_9_16_letterbox("download/clips/AbnegateAgitatedGrassPJSalt/AbnegateAgitatedGrassPJSalt.mp4", top_txt, bottom_txt, True, True)
end_time = time.perf_counter()
execution_time = end_time - start_time
print(f"The function took {execution_time:.6f} seconds to complete.")
package/youtube/uploader.py
+10 -8
View File
@@ -2,7 +2,7 @@
import os
import argparse
from enum import Enum
from datetime import datetime
from datetime import datetime, timezone
from googleapiclient.discovery import build
from googleapiclient.http import MediaFileUpload
@@ -29,7 +29,7 @@ class CategoryId(Enum):
SCIENCE_AND_TECHNOLOGY = "28"
NONPROFITS_AND_ACTIVISM = "29"
def upload_video(file_path: str, title: str, category: CategoryId, description: str = "", privacyStatus: str = 'private', tags: list = None, release_time: datetime = None) -> bool:
def upload_video(file_path: str, title: str, category: CategoryId, description: str = "", privacyStatus: str = 'private', tags: list = None, release_time: datetime = None) -> str | None:
"""
Upload a video to YouTube
@@ -51,11 +51,12 @@ def upload_video(file_path: str, title: str, category: CategoryId, description:
# Check if file exists
if not os.path.exists(file_path):
print(f"❌ Error: File not found: {file_path}")
return False
return None
# Load credentials
if not config.CREDENTIALS:
return False
print("❌ Error: YouTube API credentials not configured.")
return None
# Create YouTube API client
youtube = build('youtube', 'v3', credentials=config.CREDENTIALS)
@@ -71,7 +72,8 @@ def upload_video(file_path: str, title: str, category: CategoryId, description:
# If release_time is passed, YouTube forces privacyStatus to 'private'
status_body['privacyStatus'] = 'private'
status_body['publishAt'] = release_time.strftime('%Y-%m-%dT%H:%M:%S.000Z')
#status_body['publishAt'] = release_time.strftime('%Y-%m-%dT%H:%M:%S.000Z')
status_body['publishAt'] = release_time.isoformat()
print(f"Configuring timed release for: {status_body['publishAt']}")
else:
# Standard immediate upload
@@ -121,14 +123,14 @@ def upload_video(file_path: str, title: str, category: CategoryId, description:
else:
print(f"Current Privacy Status: {response['status']['privacyStatus']}")
return True
return response['id']
except HttpError as e:
print(f"An HTTP error occurred: {str(e)}")
return False
return None
except Exception as e:
print(f"An error occurred: {str(e)}")
return False
return None
def main():
parser = argparse.ArgumentParser(description='Upload a video to YouTube')