oceansweep's picture
?
ed28876
raw
history blame
17.1 kB
# Utils.py
#########################################
# General Utilities Library
# This library is used to hold random utilities used by various other libraries.
#
####
####################
# Function List
#
# 1. extract_text_from_segments(segments: List[Dict]) -> str
# 2. download_file(url, dest_path, expected_checksum=None, max_retries=3, delay=5)
# 3. verify_checksum(file_path, expected_checksum)
# 4. create_download_directory(title)
# 5. sanitize_filename(filename)
# 6. normalize_title(title)
# 7.
#
#
#
####################
# Import necessary libraries
import configparser
import hashlib
import json
import logging
from datetime import timedelta
from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
import requests
import time
from tqdm import tqdm
import os
import re
import unicodedata
from App_Function_Libraries.Video_DL_Ingestion_Lib import get_youtube
#######################################################################################################################
# Function Definitions
#
def extract_text_from_segments(segments):
logging.debug(f"Segments received: {segments}")
logging.debug(f"Type of segments: {type(segments)}")
def extract_text_recursive(data):
if isinstance(data, dict):
for key, value in data.items():
if key == 'Text':
return value
elif isinstance(value, (dict, list)):
result = extract_text_recursive(value)
if result:
return result
elif isinstance(data, list):
return ' '.join(filter(None, [extract_text_recursive(item) for item in data]))
return None
text = extract_text_recursive(segments)
if text:
return text.strip()
else:
logging.error(f"Unable to extract text from segments: {segments}")
return "Error: Unable to extract transcription"
def download_file(url, dest_path, expected_checksum=None, max_retries=3, delay=5):
temp_path = dest_path + '.tmp'
for attempt in range(max_retries):
try:
# Check if a partial download exists and get its size
resume_header = {}
if os.path.exists(temp_path):
resume_header = {'Range': f'bytes={os.path.getsize(temp_path)}-'}
response = requests.get(url, stream=True, headers=resume_header)
response.raise_for_status()
# Get the total file size from headers
total_size = int(response.headers.get('content-length', 0))
initial_pos = os.path.getsize(temp_path) if os.path.exists(temp_path) else 0
mode = 'ab' if 'Range' in response.headers else 'wb'
with open(temp_path, mode) as temp_file, tqdm(
total=total_size, unit='B', unit_scale=True, desc=dest_path, initial=initial_pos, ascii=True
) as pbar:
for chunk in response.iter_content(chunk_size=8192):
if chunk: # filter out keep-alive new chunks
temp_file.write(chunk)
pbar.update(len(chunk))
# Verify the checksum if provided
if expected_checksum:
if not verify_checksum(temp_path, expected_checksum):
os.remove(temp_path)
raise ValueError("Downloaded file's checksum does not match the expected checksum")
# Move the file to the final destination
os.rename(temp_path, dest_path)
print("Download complete and verified!")
return dest_path
except Exception as e:
print(f"Attempt {attempt + 1} failed: {e}")
if attempt < max_retries - 1:
print(f"Retrying in {delay} seconds...")
time.sleep(delay)
else:
print("Max retries reached. Download failed.")
raise
def verify_checksum(file_path, expected_checksum):
sha256_hash = hashlib.sha256()
with open(file_path, 'rb') as f:
for byte_block in iter(lambda: f.read(4096), b''):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest() == expected_checksum
def create_download_directory(title):
base_dir = "Results"
# Remove characters that are illegal in Windows filenames and normalize
safe_title = normalize_title(title)
logging.debug(f"{title} successfully normalized")
session_path = os.path.join(base_dir, safe_title)
if not os.path.exists(session_path):
os.makedirs(session_path, exist_ok=True)
logging.debug(f"Created directory for downloaded video: {session_path}")
else:
logging.debug(f"Directory already exists for downloaded video: {session_path}")
return session_path
def sanitize_filename(filename):
# Remove invalid characters and replace spaces with underscores
sanitized = re.sub(r'[<>:"/\\|?*]', '', filename)
sanitized = re.sub(r'\s+', ' ', sanitized).strip()
return sanitized
def normalize_title(title):
# Normalize the string to 'NFKD' form and encode to 'ascii' ignoring non-ascii characters
title = unicodedata.normalize('NFKD', title).encode('ascii', 'ignore').decode('ascii')
title = title.replace('/', '_').replace('\\', '_').replace(':', '_').replace('"', '').replace('*', '').replace('?',
'').replace(
'<', '').replace('>', '').replace('|', '')
return title
def clean_youtube_url(url):
parsed_url = urlparse(url)
query_params = parse_qs(parsed_url.query)
if 'list' in query_params:
query_params.pop('list')
cleaned_query = urlencode(query_params, doseq=True)
cleaned_url = urlunparse(parsed_url._replace(query=cleaned_query))
return cleaned_url
def extract_video_info(url):
info_dict = get_youtube(url)
title = info_dict.get('title', 'Untitled')
return info_dict, title
def clean_youtube_url(url):
parsed_url = urlparse(url)
query_params = parse_qs(parsed_url.query)
if 'list' in query_params:
query_params.pop('list')
cleaned_query = urlencode(query_params, doseq=True)
cleaned_url = urlunparse(parsed_url._replace(query=cleaned_query))
return cleaned_url
def extract_video_info(url):
info_dict = get_youtube(url)
title = info_dict.get('title', 'Untitled')
return info_dict, title
def import_data(file):
# Implement this function to import data from a file
pass
#######################
# Config loading
#
def load_comprehensive_config():
# Get the directory of the current script
current_dir = os.path.dirname(os.path.abspath(__file__))
# Go up one level to the project root directory
project_root = os.path.dirname(current_dir)
# Construct the path to the config file in the project root directory
config_path = os.path.join(project_root, 'config.txt')
# Create a ConfigParser object
config = configparser.ConfigParser()
# Read the configuration file
files_read = config.read(config_path)
if not files_read:
raise FileNotFoundError(f"Config file not found at {config_path}")
return config
def load_and_log_configs():
try:
config = load_comprehensive_config()
if config is None:
logging.error("Config is None, cannot proceed")
return None
# API Keys
anthropic_api_key = config.get('API', 'anthropic_api_key', fallback=None)
logging.debug(
f"Loaded Anthropic API Key: {anthropic_api_key[:5]}...{anthropic_api_key[-5:] if anthropic_api_key else None}")
cohere_api_key = config.get('API', 'cohere_api_key', fallback=None)
logging.debug(
f"Loaded Cohere API Key: {cohere_api_key[:5]}...{cohere_api_key[-5:] if cohere_api_key else None}")
groq_api_key = config.get('API', 'groq_api_key', fallback=None)
logging.debug(f"Loaded Groq API Key: {groq_api_key[:5]}...{groq_api_key[-5:] if groq_api_key else None}")
openai_api_key = config.get('API', 'openai_api_key', fallback=None)
logging.debug(
f"Loaded OpenAI API Key: {openai_api_key[:5]}...{openai_api_key[-5:] if openai_api_key else None}")
huggingface_api_key = config.get('API', 'huggingface_api_key', fallback=None)
logging.debug(
f"Loaded HuggingFace API Key: {huggingface_api_key[:5]}...{huggingface_api_key[-5:] if huggingface_api_key else None}")
openrouter_api_key = config.get('API', 'openrouter_api_key', fallback=None)
logging.debug(
f"Loaded OpenRouter API Key: {openrouter_api_key[:5]}...{openrouter_api_key[-5:] if openrouter_api_key else None}")
deepseek_api_key = config.get('API', 'deepseek_api_key', fallback=None)
logging.debug(
f"Loaded DeepSeek API Key: {deepseek_api_key[:5]}...{deepseek_api_key[-5:] if deepseek_api_key else None}")
# Models
anthropic_model = config.get('API', 'anthropic_model', fallback='claude-3-sonnet-20240229')
cohere_model = config.get('API', 'cohere_model', fallback='command-r-plus')
groq_model = config.get('API', 'groq_model', fallback='llama3-70b-8192')
openai_model = config.get('API', 'openai_model', fallback='gpt-4-turbo')
huggingface_model = config.get('API', 'huggingface_model', fallback='CohereForAI/c4ai-command-r-plus')
openrouter_model = config.get('API', 'openrouter_model', fallback='microsoft/wizardlm-2-8x22b')
deepseek_model = config.get('API', 'deepseek_model', fallback='deepseek-chat')
logging.debug(f"Loaded Anthropic Model: {anthropic_model}")
logging.debug(f"Loaded Cohere Model: {cohere_model}")
logging.debug(f"Loaded Groq Model: {groq_model}")
logging.debug(f"Loaded OpenAI Model: {openai_model}")
logging.debug(f"Loaded HuggingFace Model: {huggingface_model}")
logging.debug(f"Loaded OpenRouter Model: {openrouter_model}")
# Local-Models
kobold_api_IP = config.get('Local-API', 'kobold_api_IP', fallback='http://127.0.0.1:5000/api/v1/generate')
kobold_api_key = config.get('Local-API', 'kobold_api_key', fallback='')
llama_api_IP = config.get('Local-API', 'llama_api_IP', fallback='http://127.0.0.1:8080/v1/chat/completions')
llama_api_key = config.get('Local-API', 'llama_api_key', fallback='')
ooba_api_IP = config.get('Local-API', 'ooba_api_IP', fallback='http://127.0.0.1:5000/v1/chat/completions')
ooba_api_key = config.get('Local-API', 'ooba_api_key', fallback='')
tabby_api_IP = config.get('Local-API', 'tabby_api_IP', fallback='http://127.0.0.1:5000/api/v1/generate')
tabby_api_key = config.get('Local-API', 'tabby_api_key', fallback=None)
vllm_api_url = config.get('Local-API', 'vllm_api_IP', fallback='http://127.0.0.1:500/api/v1/chat/completions')
vllm_api_key = config.get('Local-API', 'vllm_api_key', fallback=None)
logging.debug(f"Loaded Kobold API IP: {kobold_api_IP}")
logging.debug(f"Loaded Llama API IP: {llama_api_IP}")
logging.debug(f"Loaded Ooba API IP: {ooba_api_IP}")
logging.debug(f"Loaded Tabby API IP: {tabby_api_IP}")
logging.debug(f"Loaded VLLM API URL: {vllm_api_url}")
# Retrieve output paths from the configuration file
output_path = config.get('Paths', 'output_path', fallback='results')
logging.debug(f"Output path set to: {output_path}")
# Retrieve processing choice from the configuration file
processing_choice = config.get('Processing', 'processing_choice', fallback='cpu')
logging.debug(f"Processing choice set to: {processing_choice}")
# Prompts - FIXME
prompt_path = config.get('Prompts', 'prompt_path', fallback='prompts.db')
return {
'api_keys': {
'anthropic': anthropic_api_key,
'cohere': cohere_api_key,
'groq': groq_api_key,
'openai': openai_api_key,
'huggingface': huggingface_api_key,
'openrouter': openrouter_api_key,
'deepseek': deepseek_api_key
},
'models': {
'anthropic': anthropic_model,
'cohere': cohere_model,
'groq': groq_model,
'openai': openai_model,
'huggingface': huggingface_model,
'openrouter': openrouter_model,
'deepseek': deepseek_model
},
'local_apis': {
'kobold': {'ip': kobold_api_IP, 'key': kobold_api_key},
'llama': {'ip': llama_api_IP, 'key': llama_api_key},
'ooba': {'ip': ooba_api_IP, 'key': ooba_api_key},
'tabby': {'ip': tabby_api_IP, 'key': tabby_api_key},
'vllm': {'ip': vllm_api_url, 'key': vllm_api_key}
},
'output_path': output_path,
'processing_choice': processing_choice
}
except Exception as e:
logging.error(f"Error loading config: {str(e)}")
return None
# Log file
# logging.basicConfig(filename='debug-runtime.log', encoding='utf-8', level=logging.DEBUG)
def format_metadata_as_text(metadata):
if not metadata:
return "No metadata available"
formatted_text = "Video Metadata:\n"
for key, value in metadata.items():
if value is not None:
if isinstance(value, list):
# Join list items with commas
formatted_value = ", ".join(str(item) for item in value)
elif key == 'upload_date' and len(str(value)) == 8:
# Format date as YYYY-MM-DD
formatted_value = f"{value[:4]}-{value[4:6]}-{value[6:]}"
elif key in ['view_count', 'like_count']:
# Format large numbers with commas
formatted_value = f"{value:,}"
elif key == 'duration':
# Convert seconds to HH:MM:SS format
hours, remainder = divmod(value, 3600)
minutes, seconds = divmod(remainder, 60)
formatted_value = f"{hours:02d}:{minutes:02d}:{seconds:02d}"
else:
formatted_value = str(value)
formatted_text += f"{key.capitalize()}: {formatted_value}\n"
return formatted_text.strip()
# # Example usage:
# example_metadata = {
# 'title': 'Sample Video Title',
# 'uploader': 'Channel Name',
# 'upload_date': '20230615',
# 'view_count': 1000000,
# 'like_count': 50000,
# 'duration': 3725, # 1 hour, 2 minutes, 5 seconds
# 'tags': ['tag1', 'tag2', 'tag3'],
# 'description': 'This is a sample video description.'
# }
#
# print(format_metadata_as_text(example_metadata))
def convert_to_seconds(time_str):
if not time_str:
return 0
# If it's already a number, assume it's in seconds
if time_str.isdigit():
return int(time_str)
# Parse time string in format HH:MM:SS, MM:SS, or SS
time_parts = time_str.split(':')
if len(time_parts) == 3:
return int(timedelta(hours=int(time_parts[0]),
minutes=int(time_parts[1]),
seconds=int(time_parts[2])).total_seconds())
elif len(time_parts) == 2:
return int(timedelta(minutes=int(time_parts[0]),
seconds=int(time_parts[1])).total_seconds())
elif len(time_parts) == 1:
return int(time_parts[0])
else:
raise ValueError(f"Invalid time format: {time_str}")
def save_to_file(video_urls, filename):
with open(filename, 'w') as file:
file.write('\n'.join(video_urls))
print(f"Video URLs saved to {filename}")
def save_segments_to_json(segments, file_name="transcription_segments.json"):
"""
Save transcription segments to a JSON file.
Parameters:
segments (list): List of transcription segments
file_name (str): Name of the JSON file to save (default: "transcription_segments.json")
Returns:
str: Path to the saved JSON file
"""
# Ensure the Results directory exists
os.makedirs("Results", exist_ok=True)
# Full path for the JSON file
json_file_path = os.path.join("Results", file_name)
# Save segments to JSON file
with open(json_file_path, 'w', encoding='utf-8') as json_file:
json.dump(segments, json_file, ensure_ascii=False, indent=4)
return json_file_path