import io
import json
import os
import time
import uuid
import base64
from datetime import datetime
from io import BytesIO
from tempfile import NamedTemporaryFile
from xmlrpc.client import Binary
import jwt
import threading
import tempfile
import numpy as np
import requests
import scipy
import soundfile as sf
import streamlit as st
import streamlit_vertical_slider as svs
from pydub import AudioSegment
from scipy.signal import butter, sosfilt
from gradio_client import Client as GradioClient
# Removed st.session_state alias - using st.session_state directly
# WordPress API base URL
WORDPRESS_API_BASE = "https://songlabai.com/wp-json/custom-api/v1"
from woocommerce import API
from wordpress_xmlrpc import Client
from wordpress_xmlrpc.compat import xmlrpc_client
from wordpress_xmlrpc.methods import media
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
# Enhanced CSS to block download buttons in audio players
st.markdown("""
""", unsafe_allow_html=True)
# Try to get API_URL from environment variables, if not found set to a default value
try:
API_URL = os.environ["API_URL"]
except KeyError:
st.error("API_URL environment variable is not set.")
st.stop()
# Try to get the Bearer token from environment variables, if not found set to a default value
try:
BEARER_TOKEN = os.environ["BEARER_TOKEN"]
except KeyError:
st.error("BEARER_TOKEN environment variable is not set.")
st.stop()
# =============================================================================
# HUGGING FACE SPACE CONFIGURATION for MusicGen
# =============================================================================
# Get HF Space token from environment variables or Streamlit secrets
try:
# Try Streamlit secrets first (for Streamlit Cloud deployment)
HF_SPACE_TOKEN = st.secrets.get("HF_SPACE_TOKEN", None)
except:
# Fall back to environment variable
HF_SPACE_TOKEN = os.environ.get("HF_SPACE_TOKEN", None)
# Your MusicGen Space URL
MUSICGEN_SPACE = "Healthydater/musicgen"
# MusicGen client will be initialized on-demand (lazy loading)
# This avoids timeout issues when the Space is sleeping at startup
musicgen_client = None
# =============================================================================
def get_musicgen_client_with_retry(max_retries=3, retry_delay=120):
"""
Initialize or get the MusicGen Space client with retry logic.
This handles cold starts by retrying with delays.
Args:
max_retries: Number of connection attempts
retry_delay: Seconds to wait between retries (default: 120s = 2 minutes)
Returns:
GradioClient instance or None if all retries failed
"""
global musicgen_client
# If client already initialized, return it
if musicgen_client is not None:
return musicgen_client
# Try to initialize with retries
for attempt in range(1, max_retries + 1):
try:
print(f"Connecting to model (attempt {attempt}/{max_retries})...")
# Gradio Client automatically uses HF_TOKEN environment variable if set
client = GradioClient(MUSICGEN_SPACE)
print(f"â Connected to model successfully")
musicgen_client = client
return client
except Exception as e:
error_msg = str(e)
print(f"Connection attempt {attempt} failed: {error_msg}")
if attempt < max_retries:
print(f"Waiting {retry_delay}s before retry...")
time.sleep(retry_delay)
else:
print(f"â All connection attempts failed")
return None
return None
page_bg_img = '''
'''
st.markdown(page_bg_img, unsafe_allow_html=True)
if "warmup_attempt_count" not in st.session_state:
st.session_state["warmup_attempt_count"] = 0
# Ensure session state keys are initialized
if "jwt_token" not in st.session_state:
st.session_state["jwt_token"] = None
if "session_id" not in st.session_state:
st.session_state["session_id"] = str(uuid.uuid4())
if 'login_needed' not in st.session_state:
st.session_state['login_needed'] = False
if 'generate_audio_params' not in st.session_state:
st.session_state['generate_audio_params'] = None
if 'audio_bytes' not in st.session_state:
st.session_state['audio_bytes'] = None
if 'audio_bytes_backup' not in st.session_state:
st.session_state['audio_bytes_backup'] = None
if 'playing_state' not in st.session_state:
st.session_state['playing_state'] = 'none' # 'none', 'generated', 'processed'
if 'audio_metadata' not in st.session_state:
st.session_state['audio_metadata'] = None
if 'audio_metadata_backup' not in st.session_state:
st.session_state['audio_metadata_backup'] = None
if 'login_refresh_needed' not in st.session_state:
st.session_state['login_refresh_needed'] = False
if 'ui_refresh_needed' not in st.session_state:
st.session_state['ui_refresh_needed'] = False
if 'audio_recovered' not in st.session_state:
st.session_state['audio_recovered'] = False
if 'promo_code_applied' not in st.session_state:
st.session_state['promo_code_applied'] = False
# Function to initialize session state variables for audio processing
def initialize_audio_session_state():
"""Initialize audio-related session state variables"""
if "vocal_audio" not in st.session_state:
st.session_state.vocal_audio = None
if "vocal_sample_rate" not in st.session_state:
st.session_state.vocal_sample_rate = None
if "audio" not in st.session_state:
st.session_state.audio = None
if "audio_pydub" not in st.session_state:
st.session_state.audio_pydub = None
if "audio_sample_rate" not in st.session_state:
st.session_state.audio_sample_rate = None
if "augmented_audio" not in st.session_state:
st.session_state.augmented_audio = None
if "augmented_audio_pydub" not in st.session_state:
st.session_state.augmented_audio_pydub = None
if "augmented_audio_sample_rate" not in st.session_state:
st.session_state.augmented_audio_sample_rate = None
# Initialize audio session state immediately
try:
initialize_audio_session_state()
except Exception as e:
# Fallback initialization
pass
# Magic Link Authentication Handler
def check_magic_link_auth():
"""Check for magic link authentication from URL parameters or browser storage"""
try:
# Check URL parameters first (direct from email click)
query_params = st.query_params
if 'magic_link_success' in query_params and 'jwt_token' in query_params:
jwt_token = query_params.get('jwt_token')
user_id = query_params.get('user_id')
action = query_params.get('action', 'login')
if jwt_token and user_id:
# Store the JWT token
st.session_state['jwt_token'] = jwt_token
st.session_state['session_id'] = str(uuid.uuid4())
st.session_state['magic_link_login'] = True
st.session_state['login_refresh_needed'] = True
# Show success message
success_msg = "Account created and logged in!" if action == 'register' else "Logged in successfully!"
st.success(f"â {success_msg} Welcome to SongLabAI!")
# Clear URL parameters
st.query_params.clear()
return True
elif 'magic_link_error' in query_params:
error = query_params.get('magic_link_error')
error_messages = {
'missing_token': 'Magic link is missing or invalid.',
'invalid_token': 'Magic link has expired or is invalid. Please request a new one.',
'user_exists': 'An account with this email already exists. Try logging in instead.',
'user_not_found': 'No account found with this email address.',
'user_creation_failed': 'Failed to create account. Please try again.',
'user_error': 'Authentication failed. Please try again.'
}
error_message = error_messages.get(error, 'Magic link authentication failed.')
st.error(f"â {error_message}")
# Clear URL parameters
st.query_params.clear()
return False
# Check browser storage (fallback method)
# This would require custom JavaScript component, so we'll skip for now
except Exception as e:
print(f"Error in magic link auth check: {e}")
return False
# Check for magic link authentication on page load
if not st.session_state.get("jwt_token"):
check_magic_link_auth()
def backup_audio_data():
"""Create backup of current audio data to prevent loss during reruns"""
if st.session_state.get('audio_bytes') is not None:
st.session_state['audio_bytes_backup'] = st.session_state['audio_bytes']
if st.session_state.get('audio_metadata') is not None:
st.session_state['audio_metadata_backup'] = st.session_state['audio_metadata'].copy()
def restore_audio_data():
"""Restore audio data from backup if main data is lost"""
if (st.session_state.get('audio_bytes') is None and
st.session_state.get('audio_bytes_backup') is not None):
st.session_state['audio_bytes'] = st.session_state['audio_bytes_backup']
if st.session_state.get('audio_metadata_backup') is not None:
st.session_state['audio_metadata'] = st.session_state['audio_metadata_backup'].copy()
# Set flag to show recovery notification
st.session_state['audio_recovered'] = True
return True
return False
def show_audio_recovery_notification():
"""Show notification when audio has been recovered"""
if st.session_state.get('audio_recovered', False):
st.info("đ Audio restored from backup after page refresh")
st.session_state['audio_recovered'] = False
def smart_rerun(force=False):
"""Smart rerun that avoids unnecessary reruns if audio is already available"""
# If we have audio data and this isn't a forced rerun, consider if rerun is necessary
if not force and st.session_state.get('audio_bytes') is not None:
# Only rerun if there's a specific flag indicating UI update is needed
if st.session_state.get('ui_refresh_needed', False):
st.session_state['ui_refresh_needed'] = False
backup_audio_data()
st.rerun()
else:
backup_audio_data()
st.rerun()
def secure_audio_player(audio_bytes, format="audio/wav"):
"""
Enhanced secure audio player that completely blocks downloads
Parameters:
audio_bytes: Audio data as bytes
format: Audio format mimetype
"""
if audio_bytes is None:
return
# Encode to base64
b64 = base64.b64encode(audio_bytes).decode()
# Custom HTML with additional JavaScript security
custom_html = f"""
"""
st.markdown(custom_html, unsafe_allow_html=True)
def get_api_headers():
return {
"Authorization": f"Bearer {BEARER_TOKEN}",
"Content-Type": "application/json",
}
def get_auth_headers():
jwt_token = st.session_state.get("jwt_token")
if jwt_token:
return {
"Authorization": f"Bearer {jwt_token}",
"Content-Type": "application/json",
"Cache-Control": "no-store",
}
else:
return {
"Content-Type": "application/json",
"Cache-Control": "no-store",
}
def set_saved_flag_before_rerun(track_id):
"""
This function sets the saved_by_user flag using a direct, synchronous request
that completes BEFORE Streamlit triggers a rerun
"""
import requests
import time
import random
# Prevent any string/None issues with track_id
if not track_id or not str(track_id).isdigit():
print(f"CRITICAL FIX: Invalid track_id: {track_id}")
return False
track_id = str(track_id).strip()
# Generate a cache busting parameter to prevent caching
cache_buster = f"nocache={int(time.time())}-{random.randint(1000, 9999)}"
# Make a SYNCHRONOUS request that will complete before the rerun
fix_url = f"https://songlabai.com/wp-admin/admin-ajax.php?action=songlab_emergency_flag_fix&track_id={track_id}&{cache_buster}"
print(f"CRITICAL FIX: Setting flag via {fix_url}")
try:
# Use a longer timeout to ensure the request has time to complete
response = requests.get(fix_url, timeout=30)
print(f"CRITICAL FIX: Response status: {response.status_code}")
print(f"CRITICAL FIX: Response content: {response.text[:200]}")
return response.status_code == 200
except Exception as e:
print(f"CRITICAL FIX ERROR: {str(e)}")
return False
def use_download_credit(jwt_token, user_id, duration, has_lyrics=False, metadata=None):
"""
Use a download credit or free sample based on duration using new audio API
"""
if not jwt_token:
return False
try:
url = f"{WORDPRESS_API_BASE}/audio-api/v1/use-audio-credit"
headers = {
'Authorization': f'Bearer {jwt_token}',
'Content-Type': 'application/json'
}
data = {
'duration': duration,
'has_lyrics': has_lyrics,
'metadata': metadata or {}
}
print(f"Using audio credit/sample for {duration}s (lyrics: {has_lyrics})")
response = requests.post(url, headers=headers, json=data, timeout=10)
if response.status_code == 200:
result = response.json()
print(f"Credit/sample used successfully: {result}")
# Clear subscription status cache so it refreshes next time
if "subscription_status_cache" in st.session_state:
del st.session_state["subscription_status_cache"]
del st.session_state["subscription_status_cache_time"]
print("Cleared subscription status cache after credit/sample use")
return True
else:
print(f"Failed to use credit/sample: HTTP {response.status_code}, Response: {response.text}")
return False
except Exception as e:
print(f"Error using download credit: {e}")
return False
def get_video_credits(jwt_token):
"""Get video credits from new video monetization system"""
if not jwt_token:
return None
try:
url = "https://songlabai.com/wp-json/video-api/v1/video-credits"
headers = {
'Authorization': f'Bearer {jwt_token}',
'Content-Type': 'application/json'
}
response = requests.get(url, headers=headers, timeout=10)
if response.status_code == 200:
return response.json()
else:
print(f"Failed to get video credits: HTTP {response.status_code}")
return None
except Exception as e:
print(f"Error getting video credits: {e}")
return None
def check_video_eligibility(jwt_token):
"""Check if user can generate videos"""
if not jwt_token:
return {'can_generate': False, 'reason': 'not_logged_in'}
try:
url = "https://songlabai.com/wp-json/video-api/v1/check-eligibility"
headers = {
'Authorization': f'Bearer {jwt_token}',
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, timeout=10)
if response.status_code == 200:
return response.json()
else:
return {'can_generate': False, 'reason': 'api_error'}
except Exception as e:
print(f"Error checking video eligibility: {e}")
return {'can_generate': False, 'reason': 'exception'}
def use_video_credit(jwt_token, generation_data):
"""Deduct video credit after successful generation"""
if not jwt_token:
return False
try:
url = "https://songlabai.com/wp-json/video-api/v1/use-video-credit"
headers = {
'Authorization': f'Bearer {jwt_token}',
'Content-Type': 'application/json'
}
response = requests.post(url, headers=headers, json=generation_data, timeout=10)
if response.status_code == 200:
# Clear video credit cache
if "video_credits_cache" in st.session_state:
del st.session_state["video_credits_cache"]
return True
else:
print(f"Failed to use video credit: HTTP {response.status_code}")
return False
except Exception as e:
print(f"Error using video credit: {e}")
return False
def get_current_subscription_status():
"""Get current user's credit-based status using new clean endpoints with caching"""
# Check if we have cached data that's still fresh (last 30 seconds)
cache_key = "subscription_status_cache"
cache_time_key = "subscription_status_cache_time"
current_time = time.time()
if (cache_key in st.session_state and
cache_time_key in st.session_state and
current_time - st.session_state[cache_time_key] < 30):
print("Using cached subscription status")
return st.session_state[cache_key]
if st.session_state.get("jwt_token"):
auth_headers = get_auth_headers()
credits_url = "https://songlabai.com/wp-json/audio-api/v1/audio-credits"
try:
print(f"Fetching credit status from: {credits_url}")
credits_response = requests.get(credits_url, headers=auth_headers)
if credits_response.status_code == 200:
credits_data = credits_response.json()
print(f"Credits response: {credits_data}")
print(f"DEBUG - Raw API response for audio credits: {credits_data}")
print(f"DEBUG - remaining_credits field: {credits_data.get('remaining_credits', 'FIELD_MISSING')}")
print(f"DEBUG - free_samples_remaining field: {credits_data.get('free_samples_remaining', 'FIELD_MISSING')}")
print(f"DEBUG - total_credits field: {credits_data.get('total_credits', 'FIELD_MISSING')}")
# NEW AUDIO CREDIT SYSTEM - Streamlit app is PAID ONLY (no free samples)
# IMPORTANT: API returns values as strings, convert to int
remaining_credits = int(credits_data.get('remaining_credits', 0))
total_credits = int(credits_data.get('total_credits', 0))
print(f"DEBUG - Parsed values -> remaining: {remaining_credits}, total: {total_credits}")
result = {
# Legacy compatibility fields
'subscription_plan_id': 'credits' if total_credits > 0 else 'free',
'is_free_tier': total_credits == 0,
'has_exceeded_download_limit': False,
'has_exceeded_generation_limit': False,
# NEW AUDIO CREDIT SYSTEM FIELDS (Streamlit = paid only)
'is_credit_based': True,
'unlimited_generation': True,
'remaining_credits': remaining_credits,
'free_samples_remaining': 0, # Streamlit app has no free samples
'total_credits': total_credits,
'used_credits': int(credits_data.get('used_credits', 0)),
'free_samples_used': 0 # Streamlit app has no free samples
}
# Cache the result
st.session_state[cache_key] = result
st.session_state[cache_time_key] = current_time
print(f"Cached subscription status - credits: {remaining_credits} (Streamlit = paid only, no free samples)")
print(f"DEBUG - Final result object: {result}")
return result
else:
print(f"Error fetching credits: HTTP {credits_response.status_code}")
print(f"Response: {credits_response.text}")
# Check for special admin accounts as fallback
try:
decoded_token = jwt.decode(
st.session_state["jwt_token"],
options={"verify_signature": False}
)
user_id = decoded_token.get('data', {}).get('user', {}).get('id')
# Special case for admin accounts - give them unlimited access
if user_id is not None and str(user_id) in ['206', '1']:
print(f"Admin account detected (ID {user_id}) - granting unlimited access")
return {
'subscription_plan_id': 'admin',
'is_free_tier': False,
'has_exceeded_download_limit': False,
'has_exceeded_generation_limit': False,
'is_credit_based': True,
'unlimited_generation': True,
'is_admin': True,
'remaining_credits': 999,
'free_samples_remaining': 999,
'total_credits': 999,
'used_credits': 0,
'free_samples_used': 0
}
except Exception as e:
print(f"Error checking for special admin account: {str(e)}")
except Exception as e:
print(f"Exception in get_current_subscription_status: {e}")
import traceback
print(traceback.format_exc())
# Default to free tier if not logged in or error occurs
return {
'subscription_plan_id': 'free',
'is_free_tier': True,
'has_exceeded_download_limit': False,
'has_exceeded_generation_limit': False,
'is_credit_based': True,
'unlimited_generation': True,
'remaining_credits': 0,
'free_samples_remaining': 0,
'total_credits': 0,
'used_credits': 0,
'free_samples_used': 0
}
def get_usage_counts():
"""Get current user's usage counts"""
if st.session_state.get("jwt_token"):
auth_headers = get_auth_headers()
usage_url = "https://songlabai.com/wp-json/custom-api/v1/usage-counts"
try:
print(f"Fetching usage counts from: {usage_url}")
usage_response = requests.get(usage_url, headers=auth_headers)
if usage_response.status_code == 200:
usage_data = usage_response.json()
print(f"Usage counts response: {usage_data}")
return {
'generation_count': usage_data.get('generation_count', 0),
'download_count': usage_data.get('download_count', 0),
'total_generation_count': usage_data.get('total_generation_count', 0),
'total_download_count': usage_data.get('total_download_count', 0)
}
else:
print(f"Error fetching usage counts: HTTP {usage_response.status_code}")
print(f"Response: {usage_response.text}")
except Exception as e:
print(f"Exception in get_usage_counts: {e}")
# Default values if not logged in or error occurs
return {
'generation_count': 0,
'download_count': 0,
'total_generation_count': 0,
'total_download_count': 0
}
def show_sidebar():
"""Display sidebar with login/user info and subscription details"""
with st.sidebar:
st.image("https://songlabai.com/wp-content/uploads/2024/03/songlabAI-logo-white-1024x267.png", width=200)
# Display login section if not logged in
if not st.session_state.get("jwt_token"):
st.markdown("### đĩ SongLabAI Login")
# Initialize session state for magic link UI
if 'magic_link_mode' not in st.session_state:
st.session_state['magic_link_mode'] = False
if 'magic_link_sent' not in st.session_state:
st.session_state['magic_link_sent'] = False
# Toggle between traditional login and magic link
col1, col2 = st.columns(2)
with col1:
if st.button("đ Email Login", use_container_width=True):
st.session_state['magic_link_mode'] = True
st.session_state['magic_link_sent'] = False
with col2:
if st.button("đ¤ Username Login", use_container_width=True):
st.session_state['magic_link_mode'] = False
st.session_state['magic_link_sent'] = False
if st.session_state.get('magic_link_mode', False):
# MAGIC LINK AUTHENTICATION
st.markdown("#### ⨠Magic Link Login")
st.markdown("Enter your email to receive a secure login link")
if not st.session_state.get('magic_link_sent', False):
with st.form("magic_link_form"):
email = st.text_input("Email Address", key="magic_email")
action = st.selectbox("Action", ["Login", "Register"], key="magic_action")
submit_magic = st.form_submit_button("Send Magic Link")
if submit_magic and email:
if not email or '@' not in email:
st.error("Please enter a valid email address")
else:
with st.spinner("Sending magic link..."):
try:
magic_url = "https://songlabai.com/wp-json/custom-api/v1/magic-link/request"
data = {
"email": email,
"action": action.lower()
}
response = requests.post(magic_url, json=data)
if response.status_code == 200:
result = response.json()
st.success(f"â {result.get('message', 'Magic link sent!')}")
st.session_state['magic_link_sent'] = True
# Store email in a different key to avoid widget conflict
st.session_state['stored_magic_email'] = email
else:
error_data = response.json() if response.headers.get('content-type', '').startswith('application/json') else {}
error_message = error_data.get('message', f'HTTP {response.status_code} error')
st.error(f"â {error_message}")
except Exception as e:
st.error(f"â Error sending magic link: {str(e)}")
else:
st.info(f"đ Magic link sent to {st.session_state.get('stored_magic_email', 'your email')}")
st.markdown("Check your email and click the link to log in automatically.")
if st.button("Send Another Link"):
st.session_state['magic_link_sent'] = False
st.rerun()
else:
# TRADITIONAL USERNAME/PASSWORD LOGIN
st.markdown("#### đ¤ Traditional Login")
with st.form("sidebar_login_form"):
username = st.text_input("Username", key="sidebar_username")
password = st.text_input("Password", type="password", key="sidebar_password")
submit_login = st.form_submit_button("Log In")
if submit_login and username and password:
with st.spinner("Logging in..."):
login_url = "https://songlabai.com/wp-json/jwt-auth/v1/token"
data = {
"username": username,
"password": password
}
response = requests.post(login_url, data=data)
if response.status_code == 200 and 'token' in response.json():
result = response.json()
st.session_state["jwt_token"] = result["token"]
st.session_state["session_id"] = str(uuid.uuid4())
# Set flag to refresh generate button
st.session_state["login_refresh_needed"] = True
st.success("â Logged in successfully!")
backup_audio_data()
st.rerun()
else:
st.error("â Invalid username or password")
st.markdown("Don't have an account? Use **Email Login** above to register!")
# If logged in, display user info and logout option
else:
try:
# Display user information
decoded_token = jwt.decode(
st.session_state["jwt_token"],
options={"verify_signature": False}
)
user_info = decoded_token.get('data', {}).get('user', {})
user_id = user_info.get('id')
# Get username (user_login) instead of display name
username = user_info.get('user_login') or user_info.get('user_nicename') or user_info.get('display_name') or 'User'
st.markdown(f"### đ Welcome, {username}!")
# Get subscription status and usage information
subscription_status = get_current_subscription_status()
usage_counts = get_usage_counts()
# Special handling for admin accounts
is_special_admin = user_id is not None and str(user_id) == '206'
is_super_admin = user_id is not None and str(user_id) == '1'
# Create credit system display
st.markdown("### đŗ Your Credits")
# Special admin display
if is_special_admin:
st.success("**Account Type:** Special Admin")
st.markdown("#### đĩ Generation Status")
st.markdown("- đĩ **Track Generation:** â (Unlimited)")
st.markdown("- đ¤ **Track with Vocals:** â (Unlimited)")
st.markdown("- đĨ **Downloads:** All formats available")
elif is_super_admin:
st.success("**Account Type:** Super Admin")
st.markdown("#### đĩ Generation Status")
st.markdown("- đĩ **Track Generation:** â (Unlimited)")
st.markdown("- đ¤ **Track with Vocals:** â (Unlimited)")
st.markdown("- đĨ **Downloads:** All formats available")
else:
# Get credit balances from the new system (Streamlit = paid only)
remaining_credits = subscription_status.get('remaining_credits', 0)
# Display what user can do across all tabs
st.markdown("#### đĩ What You Can Do")
st.markdown("- đĩ **Audio Generation:** Credits required")
st.markdown("- đŦ **Video Generation:** Credits required")
st.markdown("- đĨ **Downloads:** Included with credit purchases")
st.markdown("- đĄ **Free Samples:** Available in Gradio app only")
# Display audio credit balances
st.markdown("---")
st.markdown("#### đ° Audio Generation Credits")
if remaining_credits > 0:
st.markdown(f"- đĩ **Audio Credits:** {remaining_credits} (any duration)")
else:
st.markdown("- đĩ **Audio Credits:** 0")
st.info("đĄ Purchase credits below to generate and download your tracks!")
# Show video credits ONLY when on Commercial Ad tab
if st.session_state.get('current_tab') == 'commercial_ad':
st.markdown("---")
st.markdown("#### đŦ Video Generation Credits")
try:
video_credits = get_video_credits(st.session_state["jwt_token"])
if video_credits:
package_tier = video_credits.get('package_tier', 'free')
remaining = int(video_credits.get('remaining_credits', 0))
free_remaining = int(video_credits.get('free_generations_remaining', 0))
if package_tier == 'free':
st.markdown(f"**Package:** Free Tier")
st.markdown(f"**Free Presets Remaining:** {free_remaining}/2")
# Buy credits button for free tier
st.link_button(
"đ Buy Video Credits",
"https://songlabai.com/video-pricing/",
type="primary",
use_container_width=True
)
if free_remaining == 0:
st.warning("đ Buy credits to generate more videos!")
else:
package_names = {
'basic': 'Basic (5s)',
'advanced': 'Advanced (10s)',
'expert': 'Expert (15s)',
'custom_4': 'Custom (4 videos)',
'custom_12': 'Custom (12 videos)',
'custom_20': 'Custom (20 videos)'
}
st.markdown(f"**Package:** {package_names.get(package_tier, package_tier.title())}")
st.markdown(f"**Total Credits:** {remaining}")
# Show credits breakdown by FPS tier
available_fps_tiers = video_credits.get('available_fps_tiers', {})
if available_fps_tiers:
st.markdown("**Credits by FPS Tier:**")
for fps_tier_key in sorted(available_fps_tiers.keys(), key=lambda x: int(x)):
fps_tier = int(fps_tier_key) # Convert string key to int
credit_count = available_fps_tiers[fps_tier_key]
if credit_count > 0:
if fps_tier == 8:
st.markdown(f" âĸ Basic (8fps): {credit_count}")
elif fps_tier == 16:
st.markdown(f" âĸ Standard (16fps): {credit_count}")
elif fps_tier == 24:
st.markdown(f" âĸ HD (24fps): {credit_count}")
if free_remaining > 0:
st.markdown(f"**Free Presets:** {free_remaining}/2 remaining")
# Buy more credits button for paid tier users
st.link_button(
"đ Buy More Video Credits",
"https://songlabai.com/video-pricing/",
type="secondary",
use_container_width=True
)
if remaining == 0 and free_remaining == 0:
st.warning("đ You've used all your credits! Purchase more to continue.")
# Video "How It Works" - only on video tab
st.markdown("---")
st.markdown("#### âšī¸ How Video Credits Work")
st.markdown("- đŦ 1 free preset video generation")
st.markdown("- đŗ Buy credits for custom videos & more generations")
st.markdown("- đī¸ **Each credit is tied to the FPS tier you purchased**")
st.markdown("- đ To use 24fps, buy credits with 24fps addon")
st.markdown("- âąī¸ Choose 5s, 10s, or 15s videos based on package")
st.markdown("- đą Credits never expire")
except Exception as e:
st.error(f"Error loading video credits: {str(e)}")
# Audio track purchase buttons - only show on Track or Track Vocals tabs
if st.session_state.get('current_tab') in ['track', 'track_vocals']:
st.markdown("---")
st.markdown("### đ Buy Audio Download Credits")
st.markdown("**Download your generated audio tracks in high quality!**")
col1, col2 = st.columns(2)
st.markdown("#### đ Audio Credit Packages")
st.info("đ **Start FREE!** Get 1 free sample on [Generate for Free](https://songlabai.com/generate-music-free/) page")
st.markdown("**Available Packages:**")
st.markdown("- đĩ **Music with Lyrics**: $7.50 (up to 3 min)")
st.markdown("- đŧ **Music without Lyrics**: $5.00 (up to 3 min)")
st.markdown("- đŦ **Bundle Packages**: Music + Video from $10.00")
if st.button("đ View All Pricing & Purchase", type="primary", key="view_pricing"):
st.info("đ Visit our pricing page to purchase credits")
st.markdown("### [View Full Pricing & Buy Credits](https://songlabai.com/credit-pricing/)")
else:
st.markdown("[View Full Pricing](https://songlabai.com/credit-pricing/)")
# Audio "How It Works" - only on audio tabs
st.markdown("---")
st.markdown("#### âšī¸ How Audio Credits Work")
st.markdown("- đ **1 free sample** (available on [Generate for Free](https://songlabai.com/generate-music-free/) page)")
st.markdown("- đŗ **Credits charged per generation**")
st.markdown("- âąī¸ **1 credit = any duration** up to 3 minutes")
st.markdown("- đĩ **With or without lyrics** - same credit cost")
st.markdown("- đą Credits never expire")
# Logout button
if st.button("Log Out", key="sidebar_logout"):
# Preserve audio data during logout
audio_backup = st.session_state.get('audio_bytes_backup')
metadata_backup = st.session_state.get('audio_metadata_backup')
st.session_state.clear()
if audio_backup is not None:
st.session_state['audio_bytes_backup'] = audio_backup
if metadata_backup is not None:
st.session_state['audio_metadata_backup'] = metadata_backup
st.success("Logged out successfully!")
st.rerun()
except Exception as e:
st.error(f"Error displaying user info: {str(e)}")
st.code(traceback.format_exc()) # Show stack trace for debugging
if st.button("Log Out", key="sidebar_error_logout"):
# Preserve audio data during logout
audio_backup = st.session_state.get('audio_bytes_backup')
metadata_backup = st.session_state.get('audio_metadata_backup')
st.session_state.clear()
if audio_backup is not None:
st.session_state['audio_bytes_backup'] = audio_backup
if metadata_backup is not None:
st.session_state['audio_metadata_backup'] = metadata_backup
st.rerun()
# Help and support section
st.markdown("### âšī¸ Need Help?")
with st.expander("Tips & FAQs"):
st.markdown("""
**Generation Tips:**
- Use detailed descriptions
- Experiment with different tempos
- Try various genres for unique sounds
**Common Issues:**
- Server warmup may take several minutes
- Free tier has limited features
- Some effects work better with certain genres
**Contact us:**
For technical issues, visit our [contact page](https://songlabai.com/contact_us/)
""")
def wp_create_nonce(action=''):
"""Generate a WordPress-like nonce for REST API requests"""
import hashlib
import time
# Simple nonce generation based on current timestamp
timestamp = int(time.time())
data = f"{action}|{timestamp}"
return hashlib.md5(data.encode()).hexdigest()
def unified_track_upload(audio_data, sample_rate=None, user_id=None, metadata=None, saved_by_user=False):
"""
Simplified upload function - just handles the upload without complex flag setting
Parameters:
audio_data: Either numpy array or bytes of audio data
sample_rate: Required if audio_data is a numpy array, ignored if audio_data is bytes
user_id: User ID for track attribution (extracted from JWT if not provided)
metadata: Optional dictionary with track metadata (genre, energy, tempo, etc.)
saved_by_user: Boolean flag to mark if the user explicitly saved this track
Returns:
tuple: (success, file_url, file_id, message)
"""
try:
# Convert audio to bytes if it's a numpy array
if isinstance(audio_data, np.ndarray):
if not sample_rate:
return False, None, None, "Sample rate is required when audio_data is a numpy array"
# Convert to WAV format
audio_bytes = BytesIO()
sf.write(audio_bytes, audio_data, samplerate=sample_rate, format='WAV')
audio_bytes = audio_bytes.getvalue()
else:
# Assume it's already bytes
audio_bytes = audio_data
# Get user info from JWT token - CRITICAL for correct user attribution
jwt_token = st.session_state.get("jwt_token")
if not jwt_token:
return False, None, None, "JWT token not found. User must be logged in."
# Enhanced user data extraction
try:
# Decode JWT token to get user data
decoded = jwt.decode(jwt_token, options={"verify_signature": False})
user_data = decoded.get('data', {}).get('user', {})
# If user_id not provided, extract from token
if not user_id:
user_id = user_data.get('id')
if not user_id:
return False, None, None, "User ID not found in JWT token"
# CRITICAL: Extract user email and name correctly
user_email = user_data.get('user_email', '')
user_name = user_data.get('display_name', '')
# Debug output for troubleshooting
print(f"User data from JWT: ID={user_id}, Name={user_name}, Email={user_email}")
# If any user data is missing, try to get more info
if not user_email or not user_name:
print("WARNING: Missing user information from JWT token!")
# Try to get more user info from WordPress
try:
# Make a request to get user details
user_info_url = "https://songlabai.com/wp-json/wp/v2/users/me"
user_info_headers = {
"Authorization": f"Bearer {jwt_token}",
"Content-Type": "application/json"
}
user_response = requests.get(user_info_url, headers=user_info_headers)
if user_response.status_code == 200:
user_info = user_response.json()
if not user_name:
user_name = user_info.get('name', '')
if not user_email:
user_email = user_info.get('email', '')
print(f"Retrieved user info from WP API: Name={user_name}, Email={user_email}")
except Exception as e:
print(f"Error fetching additional user info: {str(e)}")
# Final check - if still missing info, use fallbacks
if not user_name:
user_name = f"User {user_id}"
print(f"Using fallback user name: {user_name}")
if not user_email:
user_email = f"user{user_id}@example.com"
print(f"Using fallback user email: {user_email}")
except Exception as e:
print(f"Error decoding JWT: {str(e)}")
return False, None, None, f"Failed to extract user info from token: {str(e)}"
# Get subscription tier information
subscription_status = get_current_subscription_status()
if not subscription_status['is_free_tier']:
plan_id = subscription_status.get('subscription_plan_id')
# UPDATED Subscription tier names
if plan_id == "3004":
subscription_tier = "Free Tier"
elif plan_id == "3005":
subscription_tier = "$25/month"
elif plan_id == "3007":
subscription_tier = "$99/month"
elif plan_id == "3008":
subscription_tier = "$445/month"
elif plan_id == "3009":
subscription_tier = "$849/month"
elif plan_id == "3051":
subscription_tier = "$849 - Special Occasion Song"
else:
subscription_tier = f"Premium (Plan ID: {plan_id})"
else:
subscription_tier = "Free Tier"
# Generate a unique filename
filename = f"songlab_track_{datetime.now().timestamp()}.wav"
# Get auth headers with JWT token
auth_headers = {
"Authorization": f"Bearer {jwt_token}",
"Content-Type": "application/json",
"Cache-Control": "no-store",
}
# Remove Content-Type from headers for file upload
upload_headers = auth_headers.copy()
if 'Content-Type' in upload_headers:
del upload_headers['Content-Type']
# Initialize result variables
file_url = "#"
file_id = "N/A"
upload_success = False
# CRITICAL FIX: Ensure saved_by_user is properly formatted for both form data and JSON
saved_by_user_value = '1' if saved_by_user else '0' # String for form data
saved_by_user_int = 1 if saved_by_user else 0 # Integer for JSON payloads
# Log the actual value being used
print(f"CRITICAL DEBUG: Setting saved_by_user to {saved_by_user_value} (string) / {saved_by_user_int} (int)")
# 1. Try direct AJAX upload endpoint with explicit form data
try:
print("Trying direct AJAX upload endpoint")
ajax_url = "https://songlabai.com/wp-admin/admin-ajax.php?action=upload_audio_direct"
files = {
'audio_file': (filename, audio_bytes, 'audio/wav')
}
# Include ALL user information explicitly with corrected saved_by_user value
form_data = {
'user_id': str(user_id),
'user_email': user_email, # CRITICAL - explicit email
'user_name': user_name, # CRITICAL - explicit name
'subscription_tier': subscription_tier,
'saved_by_user': saved_by_user_value # CRITICAL FIX - proper string value
}
# Debug output for form data
print(f"AJAX upload form data: {form_data}")
# Add metadata to form data if available
if metadata:
for key, value in metadata.items():
if value is not None:
form_data[key] = str(value)
response = requests.post(
ajax_url,
headers=upload_headers,
files=files,
data=form_data,
timeout=60
)
print(f"AJAX upload response: HTTP {response.status_code}")
print(f"Response content: {response.text[:200]}...")
if response.status_code == 200:
try:
response_data = response.json()
if response_data.get('success'):
file_url = response_data.get('data', {}).get('file_url', '#')
file_id = response_data.get('data', {}).get('attachment_id', 'N/A')
notification_id = response_data.get('data', {}).get('notification_id', 'N/A')
print(f"AJAX upload success! File URL: {file_url}, ID: {file_id}, Notification ID: {notification_id}")
upload_success = True
# Return notification_id as file_id for tracking
file_id = notification_id
else:
print(f"AJAX upload reported failure: {response_data.get('message', 'Unknown error')}")
except ValueError as json_error:
print(f"Error parsing AJAX response JSON: {json_error}")
print(f"Response text: {response.text[:500]}...")
elif response.status_code == 413:
print("File too large for WordPress upload (413)")
elif response.status_code == 403:
print("WordPress upload authentication failed (403)")
elif response.status_code == 404:
print("WordPress upload endpoint not found (404)")
elif response.status_code >= 500:
print(f"WordPress server error ({response.status_code}): {response.text[:500]}...")
else:
print(f"AJAX upload failed with status {response.status_code}: {response.text[:500]}...")
except requests.exceptions.Timeout:
print("AJAX upload timed out after 60 seconds")
except requests.exceptions.ConnectionError:
print("AJAX upload connection failed - check network connectivity")
except Exception as e:
print(f"Unexpected error in AJAX upload: {str(e)}")
import traceback
print(traceback.format_exc())
# 2. If AJAX failed, try the REST API v2 endpoint
if not upload_success:
try:
print("Trying REST API v2 endpoint")
v2_url = "https://songlabai.com/wp-json/songlab/v2/upload-audio"
files = {
'file': (filename, audio_bytes, 'audio/wav')
}
# Include ALL user information explicitly with corrected saved_by_user value
form_data = {
'user_id': str(user_id),
'user_email': user_email, # CRITICAL - explicit email
'user_name': user_name, # CRITICAL - explicit name
'subscription_tier': subscription_tier,
'saved_by_user': saved_by_user_value # CRITICAL FIX - proper string value
}
# Debug output for form data
print(f"REST API v2 form data: {form_data}")
# Add metadata to form data if available
if metadata:
for key, value in metadata.items():
if value is not None:
form_data[key] = str(value)
response = requests.post(
v2_url,
headers=upload_headers,
files=files,
data=form_data
)
print(f"v2 API response: HTTP {response.status_code}")
if response.status_code == 200:
try:
response_data = response.json()
if response_data.get('success'):
file_url = response_data.get('file_url', '#')
file_id = response_data.get('file_id', 'N/A')
notification_id = response_data.get('notification_id', 'N/A')
print(f"v2 API success! File URL: {file_url}, ID: {file_id}, Notification ID: {notification_id}")
upload_success = True
# Return notification_id as file_id for tracking
file_id = notification_id
else:
print(f"v2 API error: {response_data.get('message', 'Unknown error')}")
except Exception as e:
print(f"Error parsing v2 API response: {str(e)}")
print(f"Response text: {response.text[:500]}...")
else:
print(f"v2 API failed with status {response.status_code}: {response.text[:500]}...")
except Exception as e:
print(f"v2 API exception: {str(e)}")
import traceback
print(traceback.format_exc())
# 3. If both upload methods failed, use notification-only method
if not upload_success:
try:
print("All direct upload methods failed, creating notification with placeholders")
notification_url = "https://songlabai.com/wp-json/songlab/v1/notify"
# CRITICAL FIX: For JSON payloads, use integer 1/0 instead of string '1'/'0'
notification_payload = {
"link": "#",
"product_code": "pending-upload-" + datetime.now().strftime('%Y%m%d%H%M%S'),
"user_id": user_id,
"user_email": user_email, # CRITICAL - explicit email
"user_name": user_name, # CRITICAL - explicit name
"subscription_tier": subscription_tier,
"saved_by_user": saved_by_user_int # Send as integer for JSON payload
}
# Debug output for notification payload
print(f"Notification payload: {notification_payload}")
# Add metadata to notification payload if available
if metadata:
for key, value in metadata.items():
if value is not None:
notification_payload[key] = value
notify_response = requests.post(
notification_url,
headers=auth_headers,
json=notification_payload
)
print(f"Placeholder notification response: HTTP {notify_response.status_code}")
print(f"Response text: {notify_response.text}")
if notify_response.status_code == 200:
try:
notify_data = notify_response.json()
if notify_data.get('success'):
notification_id = notify_data.get('data', {}).get('id', 'N/A')
print(f"Placeholder notification created successfully with ID: {notification_id}")
# We'll consider this a partial success
upload_success = True
return upload_success, file_url, notification_id, "Track was saved with notification only. It may have limited functionality."
except Exception as e:
print(f"Error parsing notification response: {str(e)}")
except Exception as e:
print(f"Placeholder notification exception: {str(e)}")
import traceback
print(traceback.format_exc())
if upload_success:
return True, file_url, file_id, "Track uploaded successfully"
else:
return False, None, None, "All upload methods failed"
except Exception as e:
import traceback
print(f"Critical error in unified_track_upload: {str(e)}")
print(traceback.format_exc())
return False, None, None, f"Critical error: {str(e)}"
def fix_saved_by_user_flag(track_id):
"""
Bulletproof direct fix for saved_by_user flag - attempts multiple approaches
Parameters:
track_id: The ID of the track to update
Returns:
bool: True if successful, False otherwise
"""
import requests
import time
import random
# Prevent any string/None issues with track_id
if not track_id or not str(track_id).isdigit():
print(f"DIRECT FIX: Invalid track_id: {track_id}")
return False
track_id = str(track_id).strip()
success = False
# Anti-cache parameter
cache_buster = f"nocache={int(time.time())}-{random.randint(1000, 9999)}"
# ATTEMPT 1: Direct GET request (simplest approach)
try:
fix_url = f"https://songlabai.com/wp-admin/admin-ajax.php?action=songlab_emergency_flag_fix&track_id={track_id}&{cache_buster}"
print(f"DIRECT FIX ATTEMPT 1: Using emergency endpoint with GET for track {track_id}")
response = requests.get(fix_url, timeout=10)
print(f"DIRECT FIX: Response status: {response.status_code}")
if response.text:
print(f"DIRECT FIX: Response text: {response.text[:200]}...")
if response.status_code == 200:
try:
data = response.json()
if data.get('success'):
print(f"DIRECT FIX: Successfully updated track {track_id} with GET request")
return True
else:
print(f"DIRECT FIX: Error in response: {data}")
except Exception as e:
print(f"DIRECT FIX: Failed to parse JSON response: {str(e)}")
print(f"Full response text: {response.text}")
except Exception as e:
print(f"DIRECT FIX ATTEMPT 1 failed: {str(e)}")
# ATTEMPT 2: POST request
if not success:
try:
fix_url = f"https://songlabai.com/wp-admin/admin-ajax.php"
payload = {
'action': 'songlab_emergency_flag_fix',
'track_id': track_id
}
print(f"DIRECT FIX ATTEMPT 2: Using emergency endpoint with POST for track {track_id}")
response = requests.post(fix_url, data=payload, timeout=10)
print(f"DIRECT FIX: POST Response status: {response.status_code}")
if response.text:
print(f"DIRECT FIX: POST Response text: {response.text[:200]}...")
if response.status_code == 200:
try:
data = response.json()
if data.get('success'):
print(f"DIRECT FIX: Successfully updated track {track_id} with POST request")
return True
else:
print(f"DIRECT FIX: Error in response: {data}")
except Exception as e:
print(f"DIRECT FIX: Failed to parse POST JSON response: {str(e)}")
print(f"Full POST response text: {response.text}")
except Exception as e:
print(f"DIRECT FIX ATTEMPT 2 failed: {str(e)}")
# ATTEMPT 3: Alternative endpoint
if not success:
try:
# Try the direct_saved_flag_fix AJAX endpoint
fix_url = f"https://songlabai.com/wp-admin/admin-ajax.php"
payload = {
'action': 'direct_saved_flag_fix',
'track_id': track_id
}
print(f"DIRECT FIX ATTEMPT 3: Using alternative endpoint for track {track_id}")
response = requests.post(fix_url, data=payload, timeout=10)
print(f"DIRECT FIX: Alternative Response status: {response.status_code}")
if response.text:
print(f"DIRECT FIX: Alternative Response text: {response.text[:200]}...")
if response.status_code == 200:
try:
data = response.json()
if data.get('success'):
print(f"DIRECT FIX: Successfully updated track {track_id} with alternative endpoint")
return True
except:
# Even if we can't parse the JSON, if we got a 200 response, it might have worked
if "success" in response.text.lower():
print(f"DIRECT FIX: Likely succeeded with alternative endpoint based on response text")
return True
except Exception as e:
print(f"DIRECT FIX ATTEMPT 3 failed: {str(e)}")
# ATTEMPT 4: REST API call
if not success:
try:
# Try the force_update_saved_status REST API endpoint
rest_url = f"https://songlabai.com/wp-json/songlab/v1/force-update-saved-status"
headers = {
'Content-Type': 'application/json'
}
payload = {
'track_id': int(track_id),
'saved_by_user': 1
}
print(f"DIRECT FIX ATTEMPT 4: Using REST API endpoint for track {track_id}")
response = requests.post(rest_url, json=payload, headers=headers, timeout=10)
print(f"DIRECT FIX: REST API Response status: {response.status_code}")
if response.text:
print(f"DIRECT FIX: REST API Response text: {response.text[:200]}...")
if response.status_code == 200:
try:
data = response.json()
if data.get('success'):
print(f"DIRECT FIX: Successfully updated track {track_id} with REST API")
return True
except:
pass
except Exception as e:
print(f"DIRECT FIX ATTEMPT 4 failed: {str(e)}")
# All attempts failed
print(f"DIRECT FIX: All attempts failed for track {track_id}")
return False
def save_track_with_robust_flag_setting(audio_data, sample_rate, metadata=None):
"""
Save a track with robust flag handling - used for both generated and processed audio
Parameters:
audio_data: Either numpy array or bytes of audio data
sample_rate: Sample rate of the audio
metadata: Optional track metadata dictionary
Returns:
tuple: (success, message)
"""
import time
try:
with st.spinner("đž Saving track to your dashboard..."):
# Step 1: Upload the track normally first
success, file_url, track_id, message = unified_track_upload(
audio_data=audio_data,
sample_rate=sample_rate,
metadata=metadata,
saved_by_user=True # Try to set the flag directly first
)
if not success or not track_id:
return False, f"Failed to upload track: {message}"
# Log important information
print(f"Track upload success: URL={file_url}, ID={track_id}")
print(f"Now attempting to ensure saved_by_user flag is set...")
# Step 2: Wait a brief moment to ensure database writes complete
time.sleep(0.5)
# Step 3: Use our bulletproof fix function
fixed = fix_saved_by_user_flag(track_id)
if fixed:
print(f"Successfully set saved_by_user flag for track {track_id}")
return True, "Track saved successfully to your dashboard!"
else:
print(f"WARNING: Failed to set saved_by_user flag for track {track_id}")
# Step 4: If direct fix failed, try the most direct approach possible
try:
# Make a raw request to the best-working endpoint
import requests
# Try all possible endpoints with minimal parameters
endpoints = [
f"https://songlabai.com/wp-admin/admin-ajax.php?action=songlab_emergency_flag_fix&track_id={track_id}",
f"https://songlabai.com/wp-admin/admin-ajax.php?action=direct_saved_flag_fix&track_id={track_id}"
]
for i, url in enumerate(endpoints):
print(f"LAST RESORT FIX {i+1}: Trying {url}")
try:
response = requests.get(url, timeout=10)
print(f"Response status: {response.status_code}")
print(f"Response text: {response.text[:200]}")
if response.status_code == 200 and ("success" in response.text.lower()):
print(f"LAST RESORT FIX {i+1}: Appears successful!")
return True, "Track saved successfully after multiple attempts!"
except Exception as e:
print(f"LAST RESORT FIX {i+1} error: {str(e)}")
continue
except Exception as final_e:
print(f"ALL FLAG SETTING ATTEMPTS FAILED: {str(final_e)}")
# Return partial success - track was uploaded but flag might not be set
return True, "Track uploaded but flag setting may have failed. Please check your dashboard."
except Exception as e:
import traceback
print(f"Error in save_track_with_robust_flag_setting: {str(e)}")
print(traceback.format_exc())
return False, f"Error saving track: {str(e)}"
# Replace your existing save button code with this:
# In each of your button click handlers, replace:
# success, file_url, track_id, message = unified_track_upload(...)
# if success and track_id:
# fixed = fix_saved_by_user_flag(track_id)
# ...
#
# With:
# success, message = save_track_with_robust_flag_setting(audio_data, sample_rate, metadata)
# if success:
# st.success("â " + message)
# st.info("Your track will be reviewed by our team. If approved, it will be published and you'll earn 70% of all sales revenue.")
# else:
# st.error(f"â {message}")
# Function for admin-only notification
def save_on_generation(audio_array, sample_rate, metadata=None):
"""
Save a track to the admin dashboard immediately after generation with error handling
Parameters:
audio_array: Audio data as numpy array
sample_rate: Sample rate of the audio
metadata: Optional track metadata dictionary
Returns:
tuple: (file_url, file_id)
"""
print("Automatically saving generated track for admin review")
try:
# Use the unified upload function with saved_by_user=False
success, file_url, file_id, message = unified_track_upload(
audio_data=audio_array,
sample_rate=sample_rate,
metadata=metadata,
saved_by_user=False # Not saved by user, just for admin tracking
)
if success and file_id and file_id != "N/A":
print(f"Successfully saved track for admin: {file_url}, {file_id}")
return file_url, file_id
else:
print(f"Failed to save track for admin: {message}")
return "#", "N/A"
except Exception as e:
print(f"Exception during save_on_generation: {str(e)}")
# Check if it's a database connection error
if "database connection" in str(e).lower() or "Error establishing" in str(e):
print("Database connection error detected - track will still be playable")
return "#", "N/A"
def display_download_button(audio_bytes, is_premium_user=False, has_exceeded_limit=False, location="main"):
"""
Display download button based on subscription status
Parameters:
audio_bytes: Audio data as bytes
is_premium_user: Whether user has premium subscription
has_exceeded_limit: Whether user has exceeded download limits
location: Location identifier for unique key generation for buttons
"""
if audio_bytes is None:
return
# Show download button only for eligible premium users
if is_premium_user and not has_exceeded_limit:
st.download_button(
label="âŦī¸ Download Audio (Premium Feature)",
data=audio_bytes,
file_name=f"songlab_audio_{location}.wav",
mime="audio/wav",
on_click=update_download_count,
key=f"download_btn_{location}"
)
st.success("â Your premium subscription allows you to download this track!", icon="â ")
# For premium users who exceeded limits
elif is_premium_user:
st.warning("â ī¸ You've reached your download limit for this period. Your limits will reset automatically.", icon="â ī¸")
# For free users
else:
st.info("â **Premium Feature**: Downloading audio files requires a purchase. "
"[Buy Now](https://songlabai.com/orders/)", icon="âšī¸")
def save_track_fragment(audio_bytes, metadata, location="generated"):
"""
Save track to dashboard - removed fragment to prevent conflicts
"""
# Create a placeholder for status messages
status_placeholder = st.empty()
# Check if user is logged in
if not st.session_state.get("jwt_token"):
st.info("đ¤ **Log in** using the sidebar to save tracks to your dashboard")
return
# Add Save to Dashboard button
if st.button("đž Save to Dashboard", key=f"save_{location}_btn"):
# Check if we have a track ID from generation
if 'current_track_id' in st.session_state and st.session_state['current_track_id']:
track_id = st.session_state['current_track_id']
# Simply update the saved_by_user flag for the existing track
with st.spinner("đž Saving track to your dashboard..."):
print(f"Updating saved_by_user flag for existing track {track_id}")
# Call the fix_saved_by_user_flag function to set the flag to 1
success = fix_saved_by_user_flag(track_id)
if success:
status_placeholder.success("â Track saved to your dashboard successfully!")
status_placeholder.info("Your track will be reviewed by our team. If approved, it will be published and you'll earn 70% of all sales revenue.")
else:
status_placeholder.warning("â ī¸ Failed to save track to dashboard. Please try again.")
else:
# Fall back to the original upload method if no track ID is available
# (this should not normally happen, but added as a safety measure)
with st.spinner("đž Uploading track..."):
print("WARNING: No existing track ID found, creating new track entry...")
# Get audio array from audio bytes
with BytesIO(audio_bytes) as buf:
audio_array, sample_rate = sf.read(buf)
# Upload as a new track with saved_by_user=True
success, file_url, track_id, message = unified_track_upload(
audio_data=audio_array,
sample_rate=sample_rate,
metadata=metadata,
saved_by_user=True
)
if success and track_id:
# Store track_id in session state for reference
st.session_state[f'last_track_id_{location}'] = track_id
status_placeholder.success("â Track saved to your dashboard successfully!")
status_placeholder.info("Your track will be reviewed by our team. If approved, it will be published and you'll earn 70% of all sales revenue.")
else:
status_placeholder.error(f"â Failed to save track: {message}")
def generate_button_fragment(button_text, key, generate_func, *args):
"""
Generate button - removed fragment to prevent conflicts
"""
if st.button(button_text, key=key, type="primary"):
generate_func(*args)
def promo_code_fragment(promo_code, location="main"):
"""
Promo code handling - removed fragment to prevent conflicts
"""
# Check for valid promo code
if promo_code == "promosongtester" and not st.session_state.get('promo_code_applied', False):
if st.button("Apply Code", key=f"{location}_apply_promo"):
st.session_state['promo_code_applied'] = True
st.success("â Promo code applied! You now have access to full 3-minute duration.")
# Don't call smart_rerun here to avoid losing audio
# Show status if promo code is already applied
elif st.session_state.get('promo_code_applied', False):
st.success("â Promo code active")
# Add option to remove promo code
if st.button("Remove Code", key=f"{location}_remove_promo"):
st.session_state['promo_code_applied'] = False
# Don't call smart_rerun here to avoid losing audio
def display_audio_with_controls(audio_bytes, subscription_status, location="main"):
"""
Display audio player with download and save controls - removed fragment to prevent conflicts
"""
if audio_bytes is None:
return
# Display the audio player
st.markdown("### đĩ Generated Audio")
secure_audio_player(audio_bytes)
# Display columns for download and save buttons
col1, col2 = st.columns(2)
with col1:
# Show download button inline (not calling fragment from fragment)
if subscription_status and audio_bytes:
# Get user authentication info
jwt_token = subscription_status.get('jwt_token')
user_id = subscription_status.get('user_id')
# Check if user is admin
is_special_admin = user_id is not None and str(user_id) == '2'
is_super_admin = user_id is not None and str(user_id) == '1'
if is_special_admin or is_super_admin:
# Admins get unlimited downloads
if st.download_button(
label="âŦī¸ Download Audio (Admin)",
data=audio_bytes,
file_name=f"songlab_track_{location}.wav",
mime="audio/wav",
key=f"download_btn_{location}"
):
st.success("â Download started!", icon="â ")
else:
# Downloads available for credit purchasers only (Streamlit is paid-only)
remaining_credits = subscription_status.get('remaining_credits', 0)
if remaining_credits > 0 or subscription_status.get('total_credits', 0) > 0:
# User has purchased credits - allow download
if st.download_button(
label="âŦī¸ Download Audio",
data=audio_bytes,
file_name=f"songlab_track_{location}.wav",
mime="audio/wav",
key=f"download_btn_{location}"
):
st.success("â Download started!", icon="â ")
else:
# No credits purchased - block download
st.button("đ Download (Purchase Credits Required)", disabled=True, key=f"download_btn_{location}")
st.info("đĄ Download is included with credit purchases. Visit [Pricing Page](https://songlabai.com/credit-pricing/) to buy credits.")
with col2:
# User dashboard saving removed - only admin auto-save is active
pass # Info message removed
def download_button_no_rerun(audio_bytes, subscription_status, location="main"):
"""
Download button without causing reruns - removed fragment to prevent conflicts
"""
if audio_bytes is None:
return
# Get user authentication info
jwt_token = subscription_status.get('jwt_token')
user_id = subscription_status.get('user_id')
# Check if user is admin
is_special_admin = user_id is not None and str(user_id) == '2'
is_super_admin = user_id is not None and str(user_id) == '1'
if is_special_admin or is_super_admin:
# Admins get unlimited downloads
if st.download_button(
label="âŦī¸ Download Audio (Admin)",
data=audio_bytes,
file_name=f"songlab_track_{location}.wav",
mime="audio/wav",
key=f"download_btn_{location}"
):
st.success("â Download started!", icon="â ")
return
# Get credit balances
remaining_credits = subscription_status.get('remaining_credits', 0)
# Get audio metadata
metadata = st.session_state.get('audio_metadata', {})
has_lyrics = metadata.get('has_lyrics', False)
# Display download options (no nested columns since this is called inside a column)
st.markdown("#### đ¯ Download Options")
# 10-second download
st.markdown("**đĨ 10-Second Download**")
can_use_10s_credit = remaining_credits > 0
if can_use_10s_credit:
resource_label = "1 Credit"
if st.button(f"âŦī¸ Download (10s) - {resource_label}", key=f"download_10s_{location}", type="secondary"):
# Use credit/sample and download
success = use_download_credit(jwt_token, user_id, 10, has_lyrics, {'location': location, 'duration': '10s'})
if success:
# Create 10-second version of audio
try:
from pydub import AudioSegment
import io
# Load audio and trim to 10 seconds
audio = AudioSegment.from_wav(io.BytesIO(audio_bytes))
trimmed_audio = audio[:10000] # First 10 seconds (10000ms)
# Export trimmed audio
trimmed_bytes = trimmed_audio.export(format="wav").read()
# Create download
st.download_button(
label="âŦī¸ Click to Download 10s Version",
data=trimmed_bytes,
file_name=f"songlab_track_10s_{location}.wav",
mime="audio/wav",
key=f"download_10s_final_{location}"
)
st.success(f"â {resource_label} used! Credits: {remaining_credits}", icon="â ")
except Exception as e:
st.error(f"â Download failed: {str(e)}")
else:
st.error("â Failed to process download. Please try again.")
else:
st.button("đ No Credits", disabled=True, key=f"no_credit_10s_{location}")
st.caption("Purchase credits to download")
st.markdown("---")
# 20-second download
st.markdown("**đĨ 20-Second Download**")
can_use_20s_credit = remaining_credits > 0
if can_use_20s_credit:
resource_label = "1 Credit"
if st.button(f"âŦī¸ Download (20s) - {resource_label}", key=f"download_20s_{location}", type="primary"):
# Use credit/sample and download
success = use_download_credit(jwt_token, user_id, 20, has_lyrics, {'location': location, 'duration': '20s'})
if success:
# Create 20-second version of audio
try:
from pydub import AudioSegment
import io
# Load audio and trim to 20 seconds
audio = AudioSegment.from_wav(io.BytesIO(audio_bytes))
trimmed_audio = audio[:20000] # First 20 seconds (20000ms)
# Export trimmed audio
trimmed_bytes = trimmed_audio.export(format="wav").read()
# Create download
st.download_button(
label="âŦī¸ Click to Download 20s Version",
data=trimmed_bytes,
file_name=f"songlab_track_20s_{location}.wav",
mime="audio/wav",
key=f"download_20s_final_{location}"
)
st.success(f"â {resource_label} used! Credits: {remaining_credits}", icon="â ")
except Exception as e:
st.error(f"â Download failed: {str(e)}")
else:
st.error("â Failed to process download. Please try again.")
else:
st.button("đ No Credits", disabled=True, key=f"no_credit_20s_{location}")
st.caption("Purchase credits to download")
# Show purchase links if no credits
if remaining_credits == 0:
st.markdown("---")
st.info("đĄ **Need Credits?** Purchase audio credits to download your tracks!", icon="âšī¸")
st.markdown("[đ Buy Audio Credits](https://songlabai.com/songlab-melody/)")
# Modified audio player display function using fragments
def display_audio_player(is_final=False):
"""
Centralized function to display the appropriate audio player using fragments
"""
# Get subscription status
subscription_status = get_current_subscription_status()
is_premium = not subscription_status['is_free_tier']
has_exceeded_limit = subscription_status['has_exceeded_download_limit']
if is_final and st.session_state.augmented_audio_pydub is not None:
# Display processed audio using fragment
audio_bytes = st.session_state.augmented_audio_pydub.export(format="wav").read()
# Update metadata for processed audio
metadata = st.session_state.get('audio_metadata', {}).copy()
metadata['processed'] = True
metadata['processing_date'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
st.session_state.audio_metadata = metadata
# Use fragment to display final processed audio
st.markdown("### đ§ Final Processed Audio")
display_audio_with_controls(audio_bytes, subscription_status, "final")
elif not is_final and st.session_state.get('audio_bytes') is not None:
# Display audio with controls using fragment
display_audio_with_controls(st.session_state.audio_bytes, subscription_status, "main")
else:
# No audio available - offer recovery option if backup exists
if st.session_state.get('audio_bytes_backup') is not None:
st.info("đĄ Audio backup available")
if st.button("đ Restore Audio from Backup", key="manual_recovery"):
if restore_audio_data():
st.success("â Audio restored successfully!")
# Removed st.rerun() to prevent audio loss
else:
st.error("â Failed to restore audio")
def convert_audio_segment_to_float_array(audio_pydub):
"""
Convert a pydub AudioSegment to a NumPy array of type float32.
Args:
audio_pydub (AudioSegment): The AudioSegment object to be converted.
Returns:
np.ndarray: A NumPy array containing the audio data as float32.
"""
# Get the raw audio data as a sequence of samples
samples = audio_pydub.get_array_of_samples()
# Convert the samples to a NumPy array and normalize to float32
audio_array = np.array(samples).astype(np.float32)
# Normalize the audio array to range between -1.0 and 1.0
max_val = 2**15 # Assuming 16-bit audio, modify this if using different bit depths
audio_array /= max_val
return audio_array
def time_post_request(api_url, headers=None, payload=None, timeout=None):
"""
Times the execution of a POST request.
Parameters:
- api_url (str): The URL to which the POST request is sent.
- headers (dict): The headers to include in the POST request.
- payload (dict): The payload to include in the POST request.
- timeout (int): The timeout value in seconds for the POST request (optional).
Returns:
- response (requests.Response): The response object returned by the POST request.
- execution_time (float): The time it took to execute the POST request.
"""
start_time = time.time()
response = requests.post(api_url, headers=headers, json=payload, timeout=timeout)
end_time = time.time()
execution_time = end_time - start_time
print(f"Execution time: {execution_time} seconds")
return response
def check_server_status():
"""
Check if the server is warm by making a minimal request with proper error handling
Returns:
tuple: (is_ready, status_code, error_message)
- is_ready: Boolean indicating if server is ready
- status_code: HTTP status code from server (or None if request fails)
- error_message: Error message if any (or None if successful)
"""
try:
# Use a minimal request to check server status
warmup_payload = {
"inputs": {
"prompt": "warmup",
"duration": 1 # Use minimal duration
}
}
print("Sending warmup check request...")
response = requests.post(
API_URL,
headers=get_api_headers(),
json=warmup_payload,
timeout=5 # Short timeout for quick check
)
print(f"Warmup check response status: {response.status_code}")
# Server is ready if status code is 200
if response.status_code == 200:
print("Server is ready!")
return True, 200, None
else:
# If we get a 503, the server is starting up
if response.status_code == 503:
error_msg = "Server is starting up"
else:
try:
error_data = response.json()
error_msg = error_data.get("error", f"Unexpected status: {response.status_code}")
except:
error_msg = f"Unexpected status: {response.status_code}"
print(f"Server not ready: {error_msg}")
return False, response.status_code, error_msg
except requests.exceptions.Timeout:
print("Warmup check timed out - server is likely still starting")
return False, None, "Request timed out"
except Exception as e:
print(f"Error checking server status: {str(e)}")
return False, None, str(e)
def wait_for_server_warmup(status_placeholder, progress_placeholder, subscription_info_placeholder, stage_placeholder):
"""
Handle server warmup with detailed progress stages and visual feedback
Returns:
bool: True if server is ready, False if timeout
"""
# Define warmup stages with time estimations
stages = [
{"time": 0, "message": "đ Initializing server resources..."},
{"time": 60, "message": "âī¸ Loading the AI model into memory..."},
{"time": 180, "message": "đ§ Optimizing model parameters..."},
{"time": 300, "message": "đ Configuring generation settings..."},
{"time": 360, "message": "⨠Finalizing server preparation..."}
]
# NEW Credit system information to show during warmup
subscription_info = """
### đĩ **SongLabAI Credit System**
**đļ Generation:** Unlimited for all users (Always Free!)
- Generate as many tracks as you want
- Listen and preview all tracks for free
- No daily or monthly limits
**đŗ Download Credits** (Pay only when you want to download)
- **$60** - 10-Second Download Credit
- **$90** - 20-Second Download Credit
- Credits never expire
- High-quality WAV format
### đ **Server Warmup**
The AI model needs time to load when it's been inactive.
This is normal and happens to all users when the server restarts.
âąī¸ Please wait while we prepare your audio generation...
"""
# Initialization
start_time = time.time()
total_warmup_time = 420 # 7 minutes maximum wait time
current_stage = 0
last_check_time = 0
check_interval = 15 # Check every 15 seconds
# Initialize the UI elements
status_placeholder.warning("đ **Server is currently starting up**")
progress_placeholder.progress(0)
subscription_info_placeholder.markdown(subscription_info)
stage_placeholder.info(stages[0]["message"])
# Track if we've seen a 503 error (server starting)
server_starting = False
while time.time() - start_time < total_warmup_time:
# Calculate elapsed time and progress
elapsed = time.time() - start_time
remaining = total_warmup_time - elapsed
progress = min(elapsed / total_warmup_time, 0.99) # Cap at 99% until confirmed ready
# Update current stage based on elapsed time
while current_stage < len(stages) - 1 and elapsed > stages[current_stage + 1]["time"]:
current_stage += 1
# Format remaining time
minutes_remaining = int(remaining // 60)
seconds_remaining = int(remaining % 60)
# Update UI
progress_placeholder.progress(progress)
stage_message = (
f"{stages[current_stage]['message']}\n\n"
f"âŗ Estimated time remaining: {minutes_remaining}m {seconds_remaining}s"
)
# Check server status periodically
if elapsed - last_check_time >= check_interval:
last_check_time = elapsed
# Check if server is ready
is_ready, status_code, error_msg = check_server_status()
if is_ready:
# Server is ready!
status_placeholder.success("đ Server is ready! Proceeding to music generation...")
progress_placeholder.progress(1.0) # Show 100%
return True
# If we get a 503, the server is definitely starting up (not an error)
if status_code == 503:
server_starting = True
stage_placeholder.info(
f"{stage_message}\n\n"
"â Server is confirmed to be starting up. Please wait..."
)
else:
# Other error or still checking
if server_starting:
# We've seen a 503 before, so we know the server is starting
stage_placeholder.info(
f"{stage_message}\n\n"
"â Server is starting up. Please continue waiting..."
)
else:
# No confirmation yet that server is starting
stage_placeholder.info(
f"{stage_message}\n\n"
"âšī¸ Checking server status..."
)
# Small delay to prevent UI lag
time.sleep(0.1)
# Timeout reached - inform the user
status_placeholder.error(
"â Server initialization timed out.\n\n"
"The server might be experiencing high demand. Please try again in a few minutes."
)
return False
def update_download_count():
"""Update download count for the current user"""
auth_headers = get_auth_headers()
try:
update_url = "https://songlabai.com/wp-json/custom-api/v1/update-download-count"
requests.post(update_url, headers=auth_headers)
return True
except Exception as e:
print(f"Ã Error updating download count: {str(e)}")
return False
def load_and_play_generated_audio(response, genre=None, energy_level=None, tempo=None, description=None, duration=None, subscription_status=None):
"""
Load and display the generated audio with correct handling of nested array structure
and auto-save to admin dashboard
"""
try:
print(f"Processing response with status code: {response.status_code}")
# Parse the response JSON
try:
response_data = response.json()
print(f"Successfully parsed JSON response")
except Exception as json_err:
print(f"Error parsing JSON: {str(json_err)}")
print(f"First 200 chars of response: {response.text[:200]}")
raise ValueError(f"Invalid JSON response: {str(json_err)}")
# Extract the audio based on response format
if isinstance(response_data, dict) and "generated_audio" in response_data:
# Format from your endpoint: { "generated_audio": [[...]], "sample_rate": 32000 }
audio_data = response_data["generated_audio"]
sample_rate = response_data.get("sample_rate", 32000)
print(f"Found audio data structure type: {type(audio_data)}")
print(f"Audio data outer length: {len(audio_data)}")
# Handle the nested array structure - CRITICAL FIX
if isinstance(audio_data, list) and len(audio_data) == 1 and isinstance(audio_data[0], list):
# We have a nested array [[values...]], extract the inner array
print("Detected nested array structure, flattening")
audio_array = np.array(audio_data[0])
else:
# Direct array [values...]
audio_array = np.array(audio_data)
print(f"Final audio array shape: {audio_array.shape}")
elif isinstance(response_data, list) and len(response_data) > 0:
# Alternative format from older handler: [ { "generated_audio": [...], "sample_rate": 32000 } ]
first_item = response_data[0]
audio_data = first_item.get("generated_audio")
sample_rate = first_item.get("sample_rate", 32000)
if isinstance(audio_data, list) and len(audio_data) == 1 and isinstance(audio_data[0], list):
# Nested array in the list format too
audio_array = np.array(audio_data[0])
else:
audio_array = np.array(audio_data)
print(f"Extracted audio array from list format with shape: {audio_array.shape}")
else:
# Check for error response
if isinstance(response_data, dict) and "error" in response_data:
raise ValueError(f"Server error: {response_data['error']}")
else:
print(f"Unknown response structure: {type(response_data)}")
raise ValueError("Couldn't find audio data in the response")
if audio_array.size == 0:
raise ValueError("Audio array is empty")
print(f"Audio array statistics - min: {np.min(audio_array):.4f}, max: {np.max(audio_array):.4f}, mean: {np.mean(audio_array):.4f}")
# Normalize the audio to [-1.0, 1.0] if needed
max_val = np.max(np.abs(audio_array))
if max_val > 0:
audio_array = audio_array / max_val
print("Normalized audio array")
# Convert to int16 for WAV
int_audio = np.int16(audio_array * 32767)
# Write to a buffer
audio_buffer = BytesIO()
sf.write(audio_buffer, int_audio, sample_rate, format='WAV', subtype='PCM_16')
audio_buffer.seek(0)
print(f"Wrote audio to buffer with size: {len(audio_buffer.getvalue())} bytes")
# Process with pydub
audio_segment = AudioSegment.from_file(audio_buffer, format="wav")
print(f"Created AudioSegment with duration: {len(audio_segment)/1000:.2f}s")
# Store in session state
st.session_state.audio_pydub = audio_segment
st.session_state.audio_sample_rate = sample_rate
st.session_state.augmented_audio_pydub = audio_segment
st.session_state.audio_bytes = audio_buffer.getvalue()
st.session_state.playing_state = 'generated'
backup_audio_data()
# Store metadata
st.session_state.audio_metadata = {
'genre': genre,
'energy_level': energy_level,
'tempo': tempo,
'description': description,
'duration': duration,
'has_lyrics': False,
'generated_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
'sample_rate': sample_rate
}
# Auto-save to admin dashboard and get real track ID
print("Auto-saving track to admin dashboard...")
try:
file_url, file_id = save_on_generation(audio_array, sample_rate, st.session_state.audio_metadata)
if file_id and file_id != "N/A":
st.session_state['current_track_id'] = file_id
print(f"Successfully saved to admin dashboard with track ID: {file_id}")
else:
# Fallback to UUID if save fails
import uuid
temp_track_id = str(uuid.uuid4())
st.session_state['current_track_id'] = temp_track_id
print(f"Failed to save to admin dashboard, using temporary ID: {temp_track_id}")
except Exception as e:
print(f"Error saving to admin dashboard: {str(e)}")
# Fallback to UUID if save fails
import uuid
temp_track_id = str(uuid.uuid4())
st.session_state['current_track_id'] = temp_track_id
print(f"Exception occurred, using temporary ID: {temp_track_id}")
# Schedule auto-save for background (prevents rerun)
st.session_state['auto_save_audio_data'] = audio_array
st.session_state['auto_save_sample_rate'] = sample_rate
st.session_state['auto_save_metadata'] = st.session_state.audio_metadata.copy()
# Audio display now handled by fragment - removed to prevent duplicate display
print("Successfully displayed audio player without rerun-causing HTTP calls")
return audio_array, sample_rate
except Exception as e:
import traceback
error_details = traceback.format_exc()
print(f"Error in load_and_play_generated_audio: {e}")
print(f"Error details: {error_details}")
st.error(f"Failed to process generated audio: {e}")
return None, None
def debug_api_response(response):
"""Debug API response content"""
try:
# Log response headers and status
print(f"Response status: {response.status_code}")
# Try to parse as JSON
try:
content = response.json() # Use .json() instead of manually parsing
print(f"Response structure: {type(content)}")
# Handle different response formats
if isinstance(content, dict):
# Direct dictionary response format
if "generated_audio" in content:
audio_data = content["generated_audio"]
sample_rate = content.get("sample_rate", 32000)
print(f"Dictionary response - Audio length: {len(audio_data) if audio_data else 'None'}, Sample rate: {sample_rate}")
elif "error" in content:
print(f"Error in response: {content['error']}")
else:
print(f"Available keys in response: {list(content.keys())}")
elif isinstance(content, list) and len(content) > 0:
# List format response (possibly from original handler)
first_item = content[0]
if "generated_audio" in first_item:
audio_data = first_item["generated_audio"]
sample_rate = first_item.get("sample_rate", 32000)
print(f"List response - Audio length: {len(audio_data) if audio_data else 'None'}, Sample rate: {sample_rate}")
else:
print(f"Available keys in first item: {list(first_item.keys())}")
else:
print(f"Unexpected response format: {type(content)}")
except Exception as e:
print(f"Error parsing JSON: {str(e)}")
# Log first 200 characters of response
print(f"Raw response content: {response.text[:200]}...")
except Exception as e:
print(f"Error debugging response: {str(e)}")
def generate_audio(genre, energy_level, tempo, description, duration):
"""
Generate audio with improved warmup handling and error reporting
"""
# Create placeholders for warmup UI
status_placeholder = st.empty()
progress_placeholder = st.empty()
subscription_info_placeholder = st.empty()
stage_placeholder = st.empty()
try:
# Verify user is logged in
if st.session_state.get("jwt_token") is None:
st.error("You must be logged in to generate audio.")
st.session_state['login_needed'] = True
st.session_state['generate_audio_params'] = {
'genre': genre,
'energy_level': energy_level,
'tempo': tempo,
'description': description,
'duration': duration
}
# No rerun needed - login prompt will show naturally
return None
# Check for special admin account (User ID 206)
try:
decoded_token = jwt.decode(
st.session_state["jwt_token"],
options={"verify_signature": False}
)
user_id = decoded_token.get('data', {}).get('user', {}).get('id')
# Special case for user ID 206 (songlablowadmin)
if user_id == 206:
print("Detected special admin account (ID 206) - enforcing 30-second limit")
if duration > 30:
st.warning("â ī¸ This account is limited to 30-second generations.")
duration = 30 # Force duration to 30 seconds for this special admin
# Special case for user ID 1 (admin_h2ibbgql - super admin with unlimited access)
elif user_id == 1:
print("Detected super admin account (ID 1) - unlimited duration allowed")
except Exception as e:
print(f"Error checking for special admin account: {str(e)}")
# Check subscription status
subscription_status = get_current_subscription_status()
# REMOVED: Generation limits - now unlimited for all users
# The new system only limits downloads, not generation
# Check for free tier duration restriction (skip for super admin user ID 1)
is_free_tier = subscription_status['is_free_tier']
if is_free_tier and not st.session_state.get('promo_code_applied', False) and user_id != 1:
if duration > 30:
st.warning("â ī¸ Free tier users are limited to 30-second generations.")
st.info("đĄ Upgrade to a premium plan to generate longer tracks!")
duration = 30 # Force duration to 30 seconds for free tier
# Check duration for personal tiers (3005, 3006) - they also have 30 second limits (skip for super admin user ID 1)
plan_id = subscription_status.get('subscription_plan_id')
if plan_id in ['3005', '3006'] and not st.session_state.get('promo_code_applied', False) and user_id != 1:
if duration > 30:
st.warning("â ī¸ Your subscription plan includes 30-second generation limit.")
st.info("đĄ Upgrade to a commercial plan for extended duration tracks!")
duration = 30 # Force duration to 30 seconds for personal tiers
# NOTE: Warmup check removed - Gradio Space client handles cold start automatically
# The first request to a sleeping Space will take longer (1-2 minutes) but will wake it up
# Clear any warmup UI placeholders
for placeholder in [status_placeholder, progress_placeholder, subscription_info_placeholder, stage_placeholder]:
placeholder.empty()
# Prepare generation request
formatted_prompt = f"Genre: {genre}, Energy Level: {energy_level}, Tempo: {tempo}, Description: {description}"
payload = {
"inputs": {
"prompt": formatted_prompt,
"duration": float(duration)
},
"parameters": {
"top_k": 250,
"temperature": 1.0,
"guidance_scale": 3.0
}
}
# Log what's being sent
print(f"Sending generation request to: {API_URL}")
print(f"Request payload: {json.dumps(payload)}")
# Make the generation request
# Note: First request after model sleeps may take several minutes to wake up and load
with st.spinner("đĩ Generating your music... This may take a few moments (first request may take several minutes to load the model)."):
try:
# Use appropriate timeout based on duration
timeout = max(120, duration * 4) # Longer timeout for longer generations
print(f"Using timeout: {timeout}s")
# Get or initialize client with retry logic
# This handles cold starts gracefully
client = get_musicgen_client_with_retry(max_retries=3)
if client is None:
st.error("â Service unavailable. Please try again later.")
return None
result = client.predict(
formatted_prompt,
float(duration),
1.0, # temperature
250, # top_k
0.0, # top_p
3.0, # cfg_coef
True, # use_sampling
18.0 # extend_stride
)
print(f"Got response from model")
print(f"DEBUG - Response type: {type(result)}")
print(f"DEBUG - Response value: {result if not isinstance(result, tuple) else f'tuple with {len(result)} elements'}")
if isinstance(result, tuple) and len(result) > 0:
print(f"DEBUG - First element type: {type(result[0])}")
print(f"DEBUG - First element value: {result[0] if not isinstance(result[0], (list, tuple, np.ndarray)) else f'{type(result[0]).__name__} of length {len(result[0])}'}")
if isinstance(result, tuple) and len(result) > 1:
print(f"DEBUG - Second element type: {type(result[1])}")
print(f"DEBUG - Second element value (first 200 chars): {str(result[1])[:200]}")
# Result is (audio_file_path, metadata_dict)
if isinstance(result, tuple) and len(result) == 2:
audio_file_path, metadata = result
# Check for error response
if audio_file_path is None or (isinstance(metadata, dict) and "error" in metadata):
error_msg = metadata.get("error", "Unknown error") if isinstance(metadata, dict) else "Generation failed"
st.error(f"â {error_msg}")
return None
# audio_file_path is a string path to the audio file
if isinstance(audio_file_path, str):
# Load the audio file
print(f"Loading audio from: {audio_file_path}")
audio_array, sample_rate = sf.read(audio_file_path)
print(f"Loaded audio: {len(audio_array)} samples at {sample_rate}Hz")
# Convert Space response to match old endpoint format
class MockResponse:
def __init__(self, audio_data, sample_rate):
self.status_code = 200
# Convert to list if it's a numpy array
if isinstance(audio_data, np.ndarray):
audio_list = audio_data.tolist()
else:
audio_list = audio_data
self._json_data = {
"generated_audio": audio_list,
"sample_rate": sample_rate
}
def json(self):
return self._json_data
response = MockResponse(audio_array, sample_rate)
# Success! Show success message
st.success("⨠Music generated successfully!")
# DEDUCT CREDIT/FREE SAMPLE AFTER SUCCESSFUL GENERATION
try:
jwt_token = st.session_state.get("jwt_token")
if jwt_token:
# Decode JWT to get user_id
decoded_token = jwt.decode(jwt_token, options={"verify_signature": False})
user_id = decoded_token.get('data', {}).get('user', {}).get('id')
if user_id and user_id not in [1, 206]: # Skip for admin accounts
print(f"Deducting credit for normal track generation: user={user_id}, duration={duration}, has_lyrics=False")
credit_used = use_download_credit(
jwt_token,
user_id,
duration,
has_lyrics=False,
metadata={'type': 'normal_generation', 'duration': f'{duration}s'}
)
if credit_used:
print("â Credit/free sample deducted successfully")
else:
print("â ī¸ Warning: Credit deduction failed (generation still successful)")
except Exception as e:
print(f"Error deducting credit: {e}")
# Don't fail the generation if credit deduction fails
# Process and play the audio
result = load_and_play_generated_audio(
response=response,
genre=genre,
energy_level=energy_level,
tempo=tempo,
description=description,
duration=duration,
subscription_status=subscription_status
)
# Only update count if audio was successfully generated
if result[0] is not None:
# Update generation count on WordPress - with better error handling
try:
auth_headers = get_auth_headers()
update_generation_url = "https://songlabai.com/wp-json/custom-api/v1/update-generation-count"
response_update = requests.post(update_generation_url, headers=auth_headers, timeout=5)
print(f"Generation count update response: {response_update.status_code}")
except Exception as e:
print(f"Failed to update generation count: {str(e)} - continuing anyway")
return response
else:
st.error(f"â Unexpected response format from service")
return None
else:
st.error(f"â Unexpected response format from service")
return None
except Exception as e:
import traceback
print(f"Request exception: {e}")
print(traceback.format_exc())
st.error(f"â Error connecting to the generation service: {str(e)}")
return None
except Exception as e:
# Handle unexpected errors
import traceback
print(f"Detailed error in generate_audio: {traceback.format_exc()}")
st.error(f"An unexpected error occurred: {e}")
return None
def update_generate_button(genre, energy_level, tempo, description, duration):
"""Create or update the generate button based on current state"""
generate_btn_col, login_prompt_col = st.columns([1, 2])
with generate_btn_col:
generate_button_enabled = True
tooltip = ""
# Determine if button should be enabled
if not st.session_state.get("jwt_token"):
generate_button_enabled = False
tooltip = "Please log in using the sidebar to generate music"
elif genre and energy_level and tempo:
if not description: # Check confirmed_description instead
generate_button_enabled = False
tooltip = "Please add and confirm a description"
else:
# REMOVED: Generation limit checks - now unlimited for all users
# Generation is always allowed, only downloads require credits
pass
else:
generate_button_enabled = False
tooltip = "Please fill in all required fields"
# Create the button (disabled if needed)
if generate_button_enabled:
generate_button_fragment("đĩ Generate Audio", "generate_audio_button", generate_audio, genre, energy_level, tempo, description, duration)
else:
# Create disabled button with tooltip
st.markdown(
f"""
""",
unsafe_allow_html=True
)
with login_prompt_col:
# Show login prompt if not logged in
if not st.session_state.get("jwt_token"):
st.info("đ Please log in using the sidebar to generate music")
# If we just logged in, clear the flag - no rerun needed
if st.session_state.get("login_refresh_needed"):
st.session_state["login_refresh_needed"] = False
backup_audio_data()
def generate_audio_lyrics(genre, energy_level, tempo, description, duration, lyrics):
"""
Generate audio with lyrics using the new handler endpoint
"""
# Create placeholders for warmup UI
status_placeholder = st.empty()
progress_placeholder = st.empty()
subscription_info_placeholder = st.empty()
stage_placeholder = st.empty()
try:
# Verify user is logged in
if st.session_state.get("jwt_token") is None:
st.error("You must be logged in to generate audio.")
st.session_state['login_needed'] = True
st.session_state['generate_audio_params'] = {
'genre': genre,
'energy_level': energy_level,
'tempo': tempo,
'description': description,
'duration': duration,
'lyrics': lyrics
}
# No rerun needed - login prompt will show naturally
return None
# Check for special admin account (User ID 206)
try:
decoded_token = jwt.decode(
st.session_state["jwt_token"],
options={"verify_signature": False}
)
user_id = decoded_token.get('data', {}).get('user', {}).get('id')
# Special case for user ID 206 (songlablowadmin)
if user_id == 206:
print("Detected special admin account (ID 206) - enforcing 30-second limit")
if duration > 30:
st.warning("â ī¸ This account is limited to 30-second generations.")
duration = 30 # Force duration to 30 seconds for this special admin
# Special case for user ID 1 (admin_h2ibbgql - super admin with unlimited access)
elif user_id == 1:
print("Detected super admin account (ID 1) - unlimited duration allowed")
except Exception as e:
print(f"Error checking for special admin account: {str(e)}")
# Check subscription status
subscription_status = get_current_subscription_status()
# REMOVED: Generation limits - now unlimited for all users
# The new system only limits downloads, not generation
status_placeholder.info("đĩ Preparing lyrics generation...")
# Clear any previous audio data to avoid confusion
if "audio_bytes" in st.session_state:
del st.session_state.audio_bytes
if "audio_pydub" in st.session_state:
del st.session_state.audio_pydub
# Initialize warmup attempt counter
if "warmup_attempt_count" not in st.session_state:
st.session_state["warmup_attempt_count"] = 0
for placeholder in [status_placeholder, progress_placeholder, subscription_info_placeholder, stage_placeholder]:
placeholder.empty()
# Prepare generation request for lyrics endpoint
formatted_prompt = f"Genre: {genre}, Energy Level: {energy_level}, Tempo: {tempo}, Description: {description}"
# You'll need to set your lyrics endpoint URL here
LYRICS_API_URL = os.environ.get("LYRICS_API_URL", "YOUR_LYRICS_ENDPOINT_URL_HERE")
payload = {
"inputs": {
"prompt": formatted_prompt,
"lyrics": lyrics
},
"parameters": {
"duration": float(duration),
"infer_step": 60, # Note: singular as expected by handler
"guidance_scale": 15,
"scheduler_type": "euler",
"cfg_type": "apg",
"omega_scale": 10,
"guidance_interval": 0.5,
"guidance_interval_decay": 0.0,
"min_guidance_scale": 3.0,
"use_erg_tag": True,
"use_erg_lyric": False,
"use_erg_diffusion": True,
"oss_steps": [],
"guidance_scale_text": 0.0,
"guidance_scale_lyric": 0.0
}
}
headers = {
"Authorization": f"Bearer {BEARER_TOKEN}",
"Content-Type": "application/json"
}
with st.spinner("đ¤ Generating track with vocals..."):
response = requests.post(LYRICS_API_URL, headers=headers, json=payload, timeout=180)
if response.status_code == 503:
# Handle service unavailable - API is scaling up
st.warning("đ The vocals generation service is starting up. This may take 1-2 minutes...")
st.info("đĄ The AI model is being loaded. Please wait and try again in a moment.")
# Show retry button
if st.button("đ Retry Generation", key="retry_vocals_generation"):
# Clear previous error and let user try again
pass
return
elif response.status_code == 200:
result = response.json()
if "audio" in result:
# The handler returns base64 encoded audio
load_and_play_generated_audio_lyrics(result, genre, energy_level, tempo, description, duration, lyrics, subscription_status)
else:
st.error("â No audio returned from lyrics generation")
else:
st.error(f"â Lyrics generation failed: {response.status_code} - {response.text}")
except Exception as e:
st.error(f"An unexpected error occurred during lyrics generation: {e}")
return None
def load_and_play_generated_audio_lyrics(response, genre=None, energy_level=None, tempo=None, description=None, duration=None, lyrics=None, subscription_status=None):
"""
Load and display the generated audio with lyrics with correct handling
"""
try:
# Decode base64 audio
audio_b64 = response.get("audio")
if not audio_b64:
st.error("â No audio data found in response")
return
audio_bytes = base64.b64decode(audio_b64)
# Store in session state
st.session_state.audio_bytes = audio_bytes
# Convert to AudioSegment for processing
audio_buffer = io.BytesIO(audio_bytes)
audio_segment = AudioSegment.from_wav(audio_buffer)
st.session_state.audio_pydub = audio_segment
st.session_state.audio_sample_rate = audio_segment.frame_rate
st.session_state.augmented_audio_pydub = audio_segment
st.session_state.augmented_audio_sample_rate = audio_segment.frame_rate
# Set metadata including lyrics
st.session_state.audio_metadata = {
'genre': genre,
'energy_level': energy_level,
'tempo': tempo,
'description': description,
'duration': duration,
'has_lyrics': True,
'lyrics': lyrics,
'generation_type': 'lyrics'
}
# Set playing state
st.session_state.playing_state = 'generated'
# Display success message
st.success("đ Track with vocals generated successfully!")
# DEDUCT CREDIT/FREE SAMPLE AFTER SUCCESSFUL GENERATION
try:
jwt_token = st.session_state.get("jwt_token")
if jwt_token:
# Decode JWT to get user_id
decoded_token = jwt.decode(jwt_token, options={"verify_signature": False})
user_id = decoded_token.get('data', {}).get('user', {}).get('id')
if user_id and user_id not in [1, 206]: # Skip for admin accounts
print(f"Deducting credit for lyrics generation: user={user_id}, duration={duration}, has_lyrics=True")
credit_used = use_download_credit(
jwt_token,
user_id,
duration,
has_lyrics=True,
metadata={'type': 'lyrics_generation', 'duration': f'{duration}s'}
)
if credit_used:
print("â Credit/free sample deducted successfully")
else:
print("â ī¸ Warning: Credit deduction failed (generation still successful)")
except Exception as e:
print(f"Error deducting credit: {e}")
# Don't fail the generation if credit deduction fails
# Auto-save lyrics track to admin dashboard and get real track ID
print("Auto-saving lyrics track to admin dashboard...")
try:
# Convert audio bytes to numpy array for save_on_generation
audio_buffer = io.BytesIO(audio_bytes)
with sf.SoundFile(audio_buffer) as f:
audio_array = f.read()
sample_rate = f.samplerate
file_url, file_id = save_on_generation(audio_array, sample_rate, st.session_state.audio_metadata)
if file_id and file_id != "N/A":
st.session_state['current_track_id'] = file_id
print(f"Successfully saved lyrics track to admin dashboard with ID: {file_id}")
else:
# Fallback to UUID if save fails
import uuid
temp_track_id = str(uuid.uuid4())
st.session_state['current_track_id'] = temp_track_id
print(f"Failed to save lyrics track to admin dashboard, using temporary ID: {temp_track_id}")
except Exception as e:
print(f"Error saving lyrics track to admin dashboard: {str(e)}")
# Fallback to UUID if save fails
import uuid
temp_track_id = str(uuid.uuid4())
st.session_state['current_track_id'] = temp_track_id
print(f"Exception occurred with lyrics track, using temporary ID: {temp_track_id}")
# Audio display now handled by fragment - removed to prevent duplicate display
# Backup audio data (no HTTP requests, safe)
backup_audio_data()
# Schedule auto-save for background execution (prevents rerun)
if 'auto_save_pending' not in st.session_state:
st.session_state['auto_save_pending'] = True
# Auto-save will happen in background without affecting UI
except Exception as e:
st.error(f"â Error processing lyrics audio: {str(e)}")
def update_generate_button_lyrics(genre, energy_level, tempo, description, duration, lyrics):
"""Create or update the generate button for lyrics generation based on current state"""
generate_btn_col, login_prompt_col = st.columns([1, 2])
with generate_btn_col:
generate_button_enabled = True
tooltip = ""
# Determine if button should be enabled
if not st.session_state.get("jwt_token"):
generate_button_enabled = False
tooltip = "Please log in using the sidebar to generate music"
elif genre and energy_level and tempo:
if not description: # Check confirmed description
generate_button_enabled = False
tooltip = "Please add and confirm a description"
elif not lyrics:
generate_button_enabled = False
tooltip = "Please add lyrics"
else:
# REMOVED: Generation limit checks - now unlimited for all users
# Generation is always allowed, only downloads require credits
pass
else:
generate_button_enabled = False
tooltip = "Please fill in all required fields including lyrics"
# Create the button (disabled if needed)
if generate_button_enabled:
generate_button_fragment("đ¤ Generate Track with Vocals", "generate_lyrics_button", generate_audio_lyrics, genre, energy_level, tempo, description, duration, lyrics)
else:
# Create disabled button with tooltip
st.markdown(
f"""
""",
unsafe_allow_html=True
)
with login_prompt_col:
# Show login prompt if not logged in
if not st.session_state.get("jwt_token"):
st.info("đ Please log in using the sidebar to generate tracks with vocals")
# Initialize current_tab if not set (defaults to track)
if 'current_tab' not in st.session_state:
st.session_state['current_tab'] = 'track'
# Call the sidebar function once at the beginning
show_sidebar()
# Restore audio data if it was lost during a rerun
restore_audio_data()
# Show recovery notification if audio was restored
show_audio_recovery_notification()
# Streamlit app title
st.title("Songlab AI")
# Add tabs for different generation modes
generation_tab1, generation_tab2, generation_tab3 = st.tabs(["đĩ Track", "đ¤ Track with Vocals", "đŦ Commercial Ad"])
with generation_tab1:
st.markdown("### Generate Track")
# Main UI components for normal music generation
genres = [
"Pop",
"Rock",
"Hip Hop",
"Jazz",
"Blues",
"Country",
"Classical",
"Electronic",
"Reggae",
"Folk",
"R&B",
"Metal",
"Punk",
"Indie",
"Dance",
"World",
"Gospel",
"Soul",
"Funk",
"Ambient",
"Techno",
"Disco",
"House",
"Trance",
"Dubstep",
]
# Update current_tab when this widget is interacted with
def set_track_tab():
st.session_state['current_tab'] = 'track'
genre = st.selectbox("Select Genre:", genres, key="normal_genre", on_change=set_track_tab)
energy_levels = ["Low", "Medium", "High"]
energy_level = st.radio("Energy Level:", energy_levels, horizontal=True, key="normal_energy", on_change=set_track_tab)
desc_col, button_col = st.columns([4, 1])
with desc_col:
description = st.text_input(
"Description:",
"",
key="normal_description_input",
placeholder="Enter a description of your desired music"
)
# Add a visual indicator below the text input if description exists but isn't confirmed
if description and not st.session_state.get("normal_confirmed_description") == description:
st.markdown(
"""
â ī¸Don't forget to click the "Confirm" button!
""",
unsafe_allow_html=True
)
with button_col:
# Add some vertical spacing to align with the text box
st.markdown("", unsafe_allow_html=True)
# Always show an active confirm button
confirm_clicked = st.button(
"Confirm",
key="normal_confirm_description",
type="primary" # Make the button more prominent
)
if confirm_clicked:
# This will trigger a rerun with the current description value
st.session_state["normal_confirmed_description"] = description
# Show a success message
st.session_state["normal_show_confirm_success"] = True
backup_audio_data()
# No rerun needed - success message will show on next interaction
# Add success message after confirmation
if st.session_state.get("normal_show_confirm_success", False):
# Display success message
st.success("â Description confirmed!")
# Reset the flag to avoid showing the success message after page reloads
st.session_state["normal_show_confirm_success"] = False
# Use the confirmed description for button enablement
confirmed_description = st.session_state.get("normal_confirmed_description", "")
tempo = st.slider("Tempo (in bpm):", min_value=40, max_value=100, value=60, step=5, key="normal_tempo")
# Get subscription status for duration limits
subscription_status = get_current_subscription_status()
is_free_tier = subscription_status['is_free_tier']
duration_col, promo_col = st.columns([2, 1])
with duration_col:
is_special_admin = False
try:
if st.session_state.get("jwt_token"):
decoded_token = jwt.decode(
st.session_state["jwt_token"],
options={"verify_signature": False}
)
user_id = decoded_token.get('data', {}).get('user', {}).get('id')
if user_id is not None and str(user_id) == '206':
is_special_admin = True
print("Special admin account detected - limiting slider to 30 seconds")
except Exception as e:
print(f"Error checking for special admin in duration slider: {str(e)}")
# Apply the appropriate slider based on user type
if is_special_admin or (is_free_tier and not st.session_state.get('promo_code_applied', False)):
duration = st.slider(
"Duration (in seconds):",
min_value=15,
max_value=30, # Restricted to 30 seconds for free tier and special admin
value=30,
step=1,
help="Free tier users are limited to 30-second generations. Upgrade to create longer tracks or use a promo code!",
key="normal_duration"
)
else:
duration = st.slider(
"Duration (in seconds):",
min_value=15,
max_value=140,
value=30,
step=1,
key="normal_duration"
)
with promo_col:
# Add the promo code text input
promo_code = st.text_input("Promo Code:", "",
help="Enter a promo code to unlock premium features",
key="normal_promo_code")
# Check for valid promo code - use fragment to avoid reruns
promo_code_fragment(promo_code, "normal")
# Fragment for normal music generation to prevent full page reruns
@st.fragment
def normal_music_generation_fragment():
# Generate button for normal music
update_generate_button(genre, energy_level, tempo, confirmed_description, duration)
# Display audio player if audio exists (smooth appearance without full page reload)
if st.session_state.get('audio_bytes') and st.session_state.get('playing_state') == 'generated':
subscription_status = get_current_subscription_status()
display_audio_with_controls(st.session_state.audio_bytes, subscription_status, "generated")
# Run the fragment
normal_music_generation_fragment()
with generation_tab2:
st.markdown("### Generate Track with Vocals")
# Ensure session state is initialized for vocals
initialize_audio_session_state()
# Initialize any additional session state variables that might be needed
if 'audio_metadata' not in st.session_state:
st.session_state.audio_metadata = {}
if 'playing_state' not in st.session_state:
st.session_state.playing_state = None
if 'audio_bytes' not in st.session_state:
st.session_state.audio_bytes = None
# UI components for lyrics-based music generation
genres_lyrics = [
"Pop",
"Rock",
"Hip Hop",
"Jazz",
"Blues",
"Country",
"Classical",
"Electronic",
"Reggae",
"Folk",
"R&B",
"Metal",
"Punk",
"Indie",
"Dance",
"World",
"Gospel",
"Soul",
"Funk",
"Ambient",
"Techno",
"Disco",
"House",
"Trance",
"Dubstep",
]
# Update current_tab when this widget is interacted with
def set_track_vocals_tab():
st.session_state['current_tab'] = 'track_vocals'
genre_lyrics = st.selectbox("Select Genre:", genres_lyrics, key="lyrics_genre", on_change=set_track_vocals_tab)
energy_levels_lyrics = ["Low", "Medium", "High"]
energy_level_lyrics = st.radio("Energy Level:", energy_levels_lyrics, horizontal=True, key="lyrics_energy", on_change=set_track_vocals_tab)
# Description and lyrics input
desc_col_lyrics, button_col_lyrics = st.columns([4, 1])
with desc_col_lyrics:
description_lyrics = st.text_input(
"Description:",
"",
key="lyrics_description_input",
placeholder="Enter a description of your desired music"
)
# Add a visual indicator below the text input if description exists but isn't confirmed
if description_lyrics and not st.session_state.get("lyrics_confirmed_description") == description_lyrics:
st.markdown(
"""
â ī¸Don't forget to click the "Confirm" button!
""",
unsafe_allow_html=True
)
with button_col_lyrics:
# Add some vertical spacing to align with the text box
st.markdown("", unsafe_allow_html=True)
# Always show an active confirm button
confirm_clicked_lyrics = st.button(
"Confirm",
key="lyrics_confirm_description",
type="primary" # Make the button more prominent
)
if confirm_clicked_lyrics:
# This will trigger a rerun with the current description value
st.session_state["lyrics_confirmed_description"] = description_lyrics
# Show a success message
st.session_state["lyrics_show_confirm_success"] = True
backup_audio_data()
# No rerun needed - success message will show on next interaction
# Add success message after confirmation
if st.session_state.get("lyrics_show_confirm_success", False):
# Display success message
st.success("â Description confirmed!")
# Reset the flag to avoid showing the success message after page reloads
st.session_state["lyrics_show_confirm_success"] = False
# Lyrics input with confirmation system
lyrics_col, lyrics_confirm_col = st.columns([4, 1])
with lyrics_col:
lyrics = st.text_area(
"Lyrics:",
"",
key="lyrics_input",
placeholder="Enter the lyrics for your song...",
height=150,
help="Enter the complete lyrics that you want to be sung in the generated music"
)
# Add a visual indicator below the lyrics input if lyrics exist but aren't confirmed
if lyrics and not st.session_state.get("lyrics_confirmed_lyrics") == lyrics:
st.markdown(
"""
â ī¸Don't forget to confirm your lyrics!
""",
unsafe_allow_html=True
)
with lyrics_confirm_col:
# Add some vertical spacing to align with the text area
st.markdown("", unsafe_allow_html=True)
# Always show an active confirm button for lyrics
confirm_lyrics_clicked = st.button(
"Confirm Lyrics",
key="lyrics_confirm_lyrics",
type="secondary"
)
if confirm_lyrics_clicked:
# This will trigger a rerun with the current lyrics value
st.session_state["lyrics_confirmed_lyrics"] = lyrics
# Show a success message
st.session_state["lyrics_show_lyrics_confirm_success"] = True
backup_audio_data()
# No rerun needed - success message will show on next interaction
# Add success message after lyrics confirmation
if st.session_state.get("lyrics_show_lyrics_confirm_success", False):
# Display success message
st.success("â Lyrics confirmed!")
# Reset the flag to avoid showing the success message after page reloads
st.session_state["lyrics_show_lyrics_confirm_success"] = False
# Use the confirmed description and lyrics for button enablement
confirmed_description_lyrics = st.session_state.get("lyrics_confirmed_description", "")
confirmed_lyrics = st.session_state.get("lyrics_confirmed_lyrics", "")
tempo_lyrics = st.slider("Tempo (in bpm):", min_value=40, max_value=100, value=60, step=5, key="lyrics_tempo")
# Duration and promo code for lyrics generation
duration_col_lyrics, promo_col_lyrics = st.columns([2, 1])
with duration_col_lyrics:
is_special_admin = False
try:
if st.session_state.get("jwt_token"):
decoded_token = jwt.decode(
st.session_state["jwt_token"],
options={"verify_signature": False}
)
user_id = decoded_token.get('data', {}).get('user', {}).get('id')
if user_id is not None and str(user_id) == '206':
is_special_admin = True
print("Special admin account detected - limiting slider to 30 seconds")
except Exception as e:
print(f"Error checking for special admin in duration slider: {str(e)}")
# Apply the appropriate slider based on user type
if is_special_admin or (is_free_tier and not st.session_state.get('promo_code_applied', False)):
duration_lyrics = st.slider(
"Duration (in seconds):",
min_value=15,
max_value=30, # Restricted to 30 seconds for free tier and special admin
value=30,
step=1,
help="Free tier users are limited to 30-second generations. Upgrade to create longer tracks or use a promo code!",
key="lyrics_duration"
)
else:
duration_lyrics = st.slider(
"Duration (in seconds):",
min_value=15,
max_value=140,
value=30,
step=1,
key="lyrics_duration"
)
with promo_col_lyrics:
# Add the promo code text input
promo_code_lyrics = st.text_input("Promo Code:", "",
help="Enter a promo code to unlock premium features",
key="lyrics_promo_code")
# Check for valid promo code - use fragment to avoid reruns
promo_code_fragment(promo_code_lyrics, "lyrics")
# Fragment for vocals generation to prevent full page reruns
@st.fragment
def vocals_generation_fragment():
# Generate button for lyrics music
update_generate_button_lyrics(genre_lyrics, energy_level_lyrics, tempo_lyrics, confirmed_description_lyrics, duration_lyrics, confirmed_lyrics)
# Display audio player if vocals audio exists (smooth appearance without full page reload)
if st.session_state.get('audio_bytes') and st.session_state.get('playing_state') == 'generated':
# Check if this is vocals audio by looking for lyrics in metadata
metadata = st.session_state.get('audio_metadata', {})
if metadata.get('generation_type') == 'lyrics':
subscription_status = get_current_subscription_status()
display_audio_with_controls(st.session_state.audio_bytes, subscription_status, "vocals")
# Run the fragment
vocals_generation_fragment()
def get_wordpress_beats(jwt_token=None):
"""
Get beats from WordPress using the custom plugin endpoint
Parameters:
jwt_token: JWT token for authentication if needed
Returns:
list: List of beat dictionaries with id, title, url, etc.
"""
# Use the custom plugin's REST API endpoint
api_url = "https://songlabai.com/wp-json/songlab/v1/beats"
headers = {}
if jwt_token:
headers["Authorization"] = f"Bearer {jwt_token}"
try:
response = requests.get(api_url, headers=headers, timeout=10)
print(f"WordPress API response status: {response.status_code}")
if response.status_code == 200:
try:
beats = response.json()
print(f"Found {len(beats)} beats in WordPress")
return beats
except ValueError as json_error:
print(f"Error parsing WordPress response JSON: {json_error}")
return []
elif response.status_code == 404:
print("WordPress beats endpoint not found - check plugin installation")
return []
elif response.status_code == 403:
print("WordPress authentication failed - invalid JWT token")
return []
else:
print(f"WordPress API error: {response.status_code} - {response.text}")
return []
except requests.exceptions.Timeout:
print("WordPress API request timed out")
return []
except requests.exceptions.ConnectionError:
print("WordPress API connection failed - check network connectivity")
return []
except Exception as e:
print(f"Unexpected error in get_wordpress_beats: {str(e)}")
return []
def download_wordpress_beat(beat_url):
"""
Download a beat from WordPress
Parameters:
beat_url: URL of the beat file
Returns:
BytesIO: Beat data as BytesIO object or None if failed
"""
try:
response = requests.get(beat_url, stream=True, timeout=30)
if response.status_code == 200:
# Check if content is actually audio
content_type = response.headers.get('content-type', '').lower()
if 'audio' not in content_type and 'octet-stream' not in content_type:
print(f"Invalid content type for beat: {content_type}")
return None
return BytesIO(response.content)
elif response.status_code == 404:
print(f"Beat file not found at URL: {beat_url}")
return None
elif response.status_code == 403:
print(f"Access denied to beat file: {beat_url}")
return None
else:
print(f"Error downloading beat: {response.status_code} - {response.text}")
return None
except requests.exceptions.Timeout:
print(f"Timeout downloading beat from: {beat_url}")
return None
except requests.exceptions.ConnectionError:
print(f"Connection error downloading beat from: {beat_url}")
return None
except Exception as e:
print(f"Unexpected error in download_wordpress_beat: {str(e)}")
return None
def load_wordpress_beat_for_processing(beat_url, beat_name):
"""
Download and load a WordPress beat for processing
Parameters:
beat_url: URL of the beat file
beat_name: Name of the beat
Returns:
tuple: (success, message)
"""
try:
print(f"Downloading beat from: {beat_url}")
# Download the beat
beat_buffer = download_wordpress_beat(beat_url)
if not beat_buffer:
return False, "Failed to download beat from WordPress"
# Get file extension from beat URL
file_extension = beat_url.split('.')[-1].lower() if '.' in beat_url else "mp3"
# Validate file extension
valid_extensions = ['mp3', 'wav', 'flac', 'ogg', 'm4a']
if file_extension not in valid_extensions:
return False, f"Unsupported audio format: {file_extension}"
# Load into AudioSegment with error handling
try:
beat_segment = AudioSegment.from_file(beat_buffer, format=file_extension)
except Exception as audio_error:
return False, f"Failed to load audio file: {str(audio_error)}"
print(f"Loaded beat: {beat_name}, duration: {len(beat_segment)/1000:.2f}s")
# Convert to WAV if needed
if file_extension != 'wav':
wav_buffer = BytesIO()
beat_segment.export(wav_buffer, format='wav')
wav_buffer.seek(0)
beat_segment = AudioSegment.from_wav(wav_buffer)
# Store in session state
st.session_state.audio_pydub = beat_segment
st.session_state.audio_sample_rate = beat_segment.frame_rate
st.session_state.augmented_audio_pydub = beat_segment
st.session_state.augmented_audio_sample_rate = beat_segment.frame_rate
# Convert to bytes for player
wav_bytes = BytesIO()
beat_segment.export(wav_bytes, format='wav')
wav_bytes.seek(0)
st.session_state.audio_bytes = wav_bytes.getvalue()
backup_audio_data()
# Set playing state
st.session_state.playing_state = 'generated'
# Set metadata
st.session_state.audio_metadata = {
'genre': 'Beat',
'energy_level': 'Unknown',
'tempo': 'Unknown',
'description': f'Selected beat: {beat_name}',
'duration': len(beat_segment)/1000 # Convert ms to seconds
}
return True, "Beat loaded successfully"
except Exception as e:
import traceback
print(f"Error loading beat: {str(e)}")
print(traceback.format_exc())
return False, f"Error loading beat: {str(e)}"
# ============================================================================
# GENERATION TAB 3: COMMERCIAL AD GENERATION
# ============================================================================
with generation_tab3:
st.markdown("### đŦ Generate Commercial Ad / Video")
st.markdown("Create professional 10-second videos with optional jingles")
# Initialize session state for ad generation
if 'ad_jingle_bytes' not in st.session_state:
st.session_state.ad_jingle_bytes = None
if 'ad_video_path' not in st.session_state:
st.session_state.ad_video_path = None
if 'ad_final_video' not in st.session_state:
st.session_state.ad_final_video = None
if 'ad_uploaded_image' not in st.session_state:
st.session_state.ad_uploaded_image = None
# Generation Mode Selector
st.markdown("### đ¯ Select Generation Mode")
# Update current_tab when this widget is interacted with
def set_commercial_ad_tab():
st.session_state['current_tab'] = 'commercial_ad'
generation_mode = st.radio(
"Choose what you want to generate:",
["đŦ Video Only (Text-to-Video)", "đŧī¸ Video Only (Image-to-Video)", "đĩ Jingle + Video"],
key="ad_generation_mode",
horizontal=False,
on_change=set_commercial_ad_tab
)
# Instructions based on mode
with st.expander("âšī¸ How It Works"):
if generation_mode == "đŦ Video Only (Text-to-Video)":
st.markdown("""
**Generate a 10-second video from text description:**
1. **đ Describe your video** - Provide detailed prompt
2. **đŦ AI generates video** - Takes ~15 minutes
3. **âŦī¸ Download result** - Get your video file
**Tips:**
- Be specific about visuals, camera angles, and style
- Describe lighting, colors, and mood
- Generation takes ~15 minutes
""")
elif generation_mode == "đŧī¸ Video Only (Image-to-Video)":
st.markdown("""
**Animate a 10-second video from an image:**
1. **đŧī¸ Upload an image** - Your starting frame
2. **đ Describe motion** - How should it animate
3. **đŦ AI generates video** - Takes ~15 minutes
4. **âŦī¸ Download result** - Get your video file
**Tips:**
- Use high-quality images (PNG, JPG)
- Describe camera movements and transitions
- Generation takes ~15 minutes
""")
else: # Jingle + Video
st.markdown("""
**Create a complete 10-second ad with jingle and video:**
1. **đĩ Jingle Generation** - AI generates custom audio (10s)
2. **đŦ Video Generation** - Text-to-video OR image-to-video (10s)
3. **đĨ Combination** - Synchronized into final ad
**Tips:**
- Keep jingle and video themes aligned
- Total generation time: ~15-20 minutes
- Both jingle and video will be exactly 10 seconds
""")
st.markdown("---")
# Input Section - Changes based on mode
if generation_mode == "đŦ Video Only (Text-to-Video)":
st.markdown("### đ Video Settings")
ad_video_prompt = st.text_area(
"Video Description",
value=st.session_state.get('ad_video_prompt', ''),
placeholder="Sleek modern smartphone rotating on reflective black surface, studio lighting with blue gradient background, premium product photography, cinematic 4k",
height=150,
help="Describe the visual content, camera angles, lighting, and style",
key="ad_video_description"
)
ad_video_mode = "text_to_video"
ad_jingle_prompt = None
ad_jingle_lyrics = ""
elif generation_mode == "đŧī¸ Video Only (Image-to-Video)":
st.markdown("### đŧī¸ Upload Starting Image")
uploaded_image = st.file_uploader(
"Choose an image file (PNG, JPG, JPEG)",
type=["png", "jpg", "jpeg"],
key="ad_image_uploader",
help="This image will be the first frame of your video"
)
if uploaded_image:
st.image(uploaded_image, caption="Uploaded Image", use_column_width=True)
# Save to session state as base64
image_bytes = uploaded_image.read()
st.session_state.ad_uploaded_image = base64.b64encode(image_bytes).decode('utf-8')
st.markdown("### đ Animation Prompt")
ad_video_prompt = st.text_area(
"Describe the motion/animation",
value=st.session_state.get('ad_video_prompt', ''),
placeholder="Camera slowly zooms in while the product gently rotates, dramatic lighting shift from left to right",
height=120,
help="Describe camera movements, object motion, and transitions",
key="ad_video_description"
)
ad_video_mode = "image_to_video"
ad_jingle_prompt = None
ad_jingle_lyrics = ""
else: # Jingle + Video mode
st.markdown("### đĩ Jingle Settings")
ad_jingle_prompt = st.text_area(
"Jingle Description",
value=st.session_state.get('ad_jingle_prompt', ''),
placeholder="Upbeat pop jingle, catchy melody, modern production, energetic and positive",
height=100,
help="Describe the musical style, mood, and characteristics",
key="ad_jingle_description"
)
ad_jingle_lyrics = st.text_area(
"Jingle Lyrics or Slogan (Optional)",
value=st.session_state.get('ad_lyrics', ''),
placeholder="Enter your brand slogan, product name, tagline, or catchy phrases you want sung in the jingle (e.g., 'Fresh coffee, every morning', 'The future is here', 'Just do it')",
height=80,
help="Add lyrics, slogans, or brand messages to make your jingle memorable. Leave empty for an instrumental-only jingle.",
key="ad_jingle_lyrics_input"
)
st.markdown("### đŦ Video Settings")
# Choose text-to-video or image-to-video for the jingle+video mode
video_sub_mode = st.radio(
"Video generation method:",
["Text-to-Video", "Image-to-Video"],
key="jingle_video_submode",
horizontal=True
)
if video_sub_mode == "Image-to-Video":
uploaded_image = st.file_uploader(
"Upload starting image (PNG, JPG, JPEG)",
type=["png", "jpg", "jpeg"],
key="ad_image_uploader_jingle",
help="This image will be the first frame"
)
if uploaded_image:
col_img1, col_img2 = st.columns([1, 2])
with col_img1:
st.image(uploaded_image, caption="Uploaded Image", use_column_width=True)
image_bytes = uploaded_image.read()
st.session_state.ad_uploaded_image = base64.b64encode(image_bytes).decode('utf-8')
ad_video_mode = "image_to_video"
else:
ad_video_mode = "text_to_video"
ad_video_prompt = st.text_area(
"Video Description / Animation",
value=st.session_state.get('ad_video_prompt', ''),
placeholder="Sleek modern smartphone rotating on reflective surface, studio lighting, cinematic 4k" if video_sub_mode == "Text-to-Video" else "Camera zooms in while product rotates, dramatic lighting",
height=100,
help="Describe visuals (text-to-video) or motion/animation (image-to-video)",
key="ad_video_description"
)
st.markdown("---")
# Example Prompts Section
if generation_mode == "đĩ Jingle + Video":
st.markdown("### đĄ Example Commercial Ads")
EXAMPLE_COMMERCIAL_ADS = {
"Select an example...": {"jingle": "", "lyrics": "", "video": ""},
"Luxury Car Launch": {
"jingle": "cinematic orchestral, powerful strings, epic drums, sophisticated and premium, rising crescendo",
"lyrics": "Drive the future, feel the power, elegance in motion, every hour",
"video": "Sleek black luxury sedan driving through mountain roads at golden hour, dramatic low angle shots, cinematic camera movements, reflections on polished surface, mist rolling through valleys"
},
"Coffee Shop Morning": {
"jingle": "acoustic guitar, warm and cozy, jazzy piano, uplifting morning vibes, folk pop",
"lyrics": "Morning starts with you, fresh and true, every sip a new day, brewing your way",
"video": "Steaming coffee cup on rustic wooden table by cafe window, warm morning sunlight streaming through, barista pouring latte art, cozy atmosphere with bokeh background"
},
"Fitness App Energy": {
"jingle": "energetic electronic, powerful drums, motivational synths, upbeat tempo, pumping bass",
"lyrics": "Push harder, go further, stronger every day, your journey starts today",
"video": "Dynamic shots of athletes training at sunrise, running on urban streets with motion blur, high-energy movements, sweat droplets in slow motion, city skyline background"
},
"Gourmet Restaurant": {
"jingle": "elegant classical piano, soft strings, refined and luxurious, smooth jazz elements",
"lyrics": "Taste perfection, culinary art, every bite a masterpiece, dining from the heart",
"video": "Extreme close-up of gourmet dishes being plated by chef, flames on grill with premium steaks, artistic food presentation, shallow depth of field, vibrant colors"
},
"Adventure Travel": {
"jingle": "epic rock, electric guitar riffs, powerful drums, adventurous and bold, inspiring melody",
"lyrics": "Explore the world, break free, adventure calls, wild and free",
"video": "Drone shot soaring over mountain peaks at sunset, hikers reaching summit with arms raised, coastal cliffs with crashing waves, cinematic aerial views"
},
"Tech Innovation": {
"jingle": "futuristic electronic, synthesizers, modern beats, innovative sound, digital atmosphere",
"lyrics": "Tomorrow is here, innovation clear, technology that transforms, future takes form",
"video": "Modern smartphone with holographic interface projections, sleek product rotating on reflective black surface, blue neon lighting, high-tech laboratory background"
},
"Summer Beach Party": {
"jingle": "tropical house, upbeat rhythm, steel drums, party vibes, summer energy, electronic dance",
"lyrics": "Sun and sand, party all night, summer vibes, feeling right",
"video": "Beach party at sunset with bonfire, people dancing on the sand, colorful drinks with ice, palm trees swaying, ocean waves in background, golden hour lighting"
},
"Luxury Perfume": {
"jingle": "sensual piano, soft orchestral strings, elegant and mysterious, slow tempo, sophisticated",
"lyrics": "Essence of elegance, timeless grace, luxury captured, in every trace",
"video": "Elegant perfume bottle on marble pedestal with dramatic spotlight, golden liquid swirling inside, soft focus background, roses falling in slow motion, luxury studio aesthetic"
}
}
example_selector = st.selectbox(
"Choose an example commercial:",
list(EXAMPLE_COMMERCIAL_ADS.keys()),
key="commercial_example_selector"
)
if example_selector != "Select an example...":
if st.button("đ Use This Example", key="use_commercial_example_btn"):
# Set session state using the widget keys to ensure immediate update
st.session_state['ad_jingle_description'] = EXAMPLE_COMMERCIAL_ADS[example_selector]["jingle"]
st.session_state['ad_jingle_lyrics_input'] = EXAMPLE_COMMERCIAL_ADS[example_selector]["lyrics"]
st.session_state['ad_video_description'] = EXAMPLE_COMMERCIAL_ADS[example_selector]["video"]
st.rerun()
st.markdown("---")
# ========================================================================
# VIDEO GENERATION PARAMETERS - User Controls
# ========================================================================
# Check if user is logged in
is_logged_in = st.session_state.get("jwt_token") is not None
jwt_token = st.session_state.get("jwt_token", "")
# Get video credits and eligibility
video_credits = None
eligibility = None
can_generate = False
is_admin = False
if is_logged_in:
try:
video_credits = get_video_credits(jwt_token)
eligibility = check_video_eligibility(jwt_token)
can_generate = eligibility.get('can_generate', False)
# Check if admin
decoded_token = jwt.decode(jwt_token, options={"verify_signature": False})
user_id = decoded_token.get('data', {}).get('user', {}).get('id')
is_admin = user_id in [1, 206]
except Exception as e:
st.error(f"Error checking video eligibility: {str(e)}")
# Determine if user can customize
can_customize = is_admin or (video_credits and int(video_credits.get('remaining_credits', 0)) > 0)
# Mode toggle
is_custom_mode = st.toggle(
"đ¨ Customize Settings" if can_customize else "đ Customize (Requires Credits)",
value=False,
disabled=not can_customize,
help="Use presets (free) or customize all parameters (requires credits)"
)
# Compact settings in expander
with st.expander("âī¸ Video Settings" if is_custom_mode else "âī¸ Video Settings (Preset Mode)", expanded=True):
# Determine max duration based on package tier
if video_credits and video_credits.get('package_tier') == 'expert':
max_duration = 15
allows_intro = True
elif video_credits and video_credits.get('package_tier') in ['advanced', 'custom_4', 'custom_12', 'custom_20']:
max_duration = 10
allows_intro = False
elif video_credits and video_credits.get('package_tier') == 'basic':
max_duration = 5
allows_intro = False
else:
max_duration = 3 # Free tier
allows_intro = False
# Display preset info if not customizing
if not is_custom_mode:
st.info("đ Preset: 3s video, 16 FPS (Standard), Balanced creativity")
# Initialize intro variables before they're used in UI
add_expert_intro = False
intro_prompt = ""
company_name = ""
# Expert intro option (must come before duration slider so checkbox updates variables)
if allows_intro and is_custom_mode:
with st.expander("đŦ Expert: Add 5-Second Intro (Optional)"):
add_expert_intro = st.checkbox("Enable 5s branded intro")
if add_expert_intro:
st.info("đš **How it works:** We'll generate a 5-second intro video with your logo/branding, add a voiceover saying your company name, then combine it with your 10-second main video for a professional 15-second commercial!")
intro_prompt = st.text_area(
"Intro Visual Description",
placeholder="Company logo centered on dark blue gradient background with soft spotlight",
height=60,
help="Describe what you want to see in your 5-second intro (logo, background, lighting, etc.)"
)
company_name = st.text_input("Company Name (for voiceover)",
placeholder="YourBrand Inc.",
help="We'll generate a professional voiceover saying this name")
st.caption("đĄ Intro generates at 8fps (perfect for static logos) with professional voiceover âĸ Saves compute cost while maintaining quality")
# Compact 2-column layout for all settings
col1, col2 = st.columns(2)
with col1:
# Show duration note if intro is enabled
if add_expert_intro:
st.info("âšī¸ With intro enabled: 5s intro + 10s main = 15s final video")
# Determine default and max values for duration slider
default_duration = 3
max_duration_value = max_duration if is_custom_mode else 3
if add_expert_intro:
max_duration_value = 10 # Limit to 10s when intro enabled
default_duration = min(3, 10)
duration_seconds = st.slider(
"đš Duration" + (" (Main Video)" if add_expert_intro else ""),
min_value=2,
max_value=max_duration_value,
value=default_duration,
step=1,
disabled=not is_custom_mode,
help="With intro: Max 10s (5s intro + 10s main = 15s total) | Without intro: Free: 3s | Basic: 5s | Advanced: 10s | Expert: 15s"
)
guidance_scale = st.slider(
"đ¨ Creativity",
min_value=1.0,
max_value=15.0,
value=7.0,
step=0.5,
disabled=not is_custom_mode,
help="Lower=creative, Higher=accurate (7-9 optimal)"
)
with col2:
# FPS selection - based on credit stacks (each credit tied to specific FPS tier)
quality_options = {}
fps_to_label = {} # Maps FPS value back to label for lookup
if is_admin:
# Admin gets all options without credit requirements
quality_options = {"Basic (8fps)": 8, "Standard (16fps)": 16, "HD (24fps)": 24}
fps_to_label = {8: "Basic (8fps)", 16: "Standard (16fps)", 24: "HD (24fps)"}
elif video_credits:
# Get available FPS tiers from credit stacks
available_fps_tiers = video_credits.get('available_fps_tiers', {})
# Build dropdown showing ALL FPS options user can use
# User can use any FPS <= their max fps_tier
# Example: fps_tier=24 allows 8, 16, 24fps (custom packages)
# fps_tier=16 allows 8, 16fps (bought 16fps addon)
# fps_tier=8 allows 8fps only (bought 8fps addon)
if available_fps_tiers:
# Find the maximum FPS tier the user has credits for
max_fps_tier = max([int(tier) for tier in available_fps_tiers.keys() if int(available_fps_tiers[tier]) > 0], default=0)
# Get total credits (sum across all stacks)
total_credits = sum([int(available_fps_tiers[tier]) for tier in available_fps_tiers.keys()])
# Show all FPS options up to their max tier
if max_fps_tier >= 8 and total_credits > 0:
label = f"Basic (8fps) - {total_credits} credit{'s' if total_credits != 1 else ''}"
quality_options[label] = 8
fps_to_label[8] = label
if max_fps_tier >= 16 and total_credits > 0:
label = f"Standard (16fps) - {total_credits} credit{'s' if total_credits != 1 else ''}"
quality_options[label] = 16
fps_to_label[16] = label
if max_fps_tier >= 24 and total_credits > 0:
label = f"HD (24fps) - {total_credits} credit{'s' if total_credits != 1 else ''}"
quality_options[label] = 24
fps_to_label[24] = label
if is_custom_mode:
if quality_options:
# Show only FPS tiers where user has credits
available_options = list(quality_options.keys())
default_index = min(1, len(available_options) - 1) # Default to Standard if available
selected_quality = st.selectbox(
"đī¸ Quality",
options=available_options,
index=default_index,
help="Each credit is tied to the FPS tier you purchased it with"
)
fps = quality_options[selected_quality]
else:
# No credits available - show disabled dropdown
st.selectbox(
"đī¸ Quality",
options=["No credits available"],
index=0,
disabled=True,
help="Purchase credits to unlock custom video generation"
)
fps = 16 # Fallback value
else:
fps = 16
st.selectbox(
"đī¸ Quality",
options=["Standard (16fps)"],
index=0,
disabled=True
)
video_guidance_scale = st.slider(
"⥠Motion",
min_value=1.0,
max_value=10.0,
value=5.0,
step=0.5,
disabled=not is_custom_mode,
help="Lower=smooth, Higher=dynamic"
)
# ========================================================================
# LOGIN AND ELIGIBILITY CHECKS
# ========================================================================
# Check 1: Must be logged in
if not is_logged_in:
st.error("â ī¸ **Please log in to generate videos**")
st.info("đ Use the sidebar to log in or sign up")
st.stop()
# Check 2: Check eligibility (credits or free generations)
if not is_admin and not can_generate:
st.error("â ī¸ **No video credits or free generations remaining**")
if eligibility:
st.warning(f"Credits: {eligibility.get('credits', 0)} | Free remaining: {eligibility.get('free_remaining', 0)}/2")
st.markdown("### đ Purchase Video Credits")
st.markdown("Choose a package to continue generating videos:")
pkg_col1, pkg_col2, pkg_col3 = st.columns(3)
with pkg_col1:
st.markdown("**Basic** - $17.50")
st.markdown("- 1 **Basic-Tier** Credit")
st.markdown("- Up to 5 seconds")
st.markdown("- Choose quality (FPS)")
with pkg_col2:
st.markdown("**Advanced** - $35.00")
st.markdown("- 1 **Advanced-Tier** Credit")
st.markdown("- Up to 10 seconds")
st.markdown("- Choose quality (FPS)")
with pkg_col3:
st.markdown("**Expert** - $42.00")
st.markdown("- 1 **Expert-Tier** Credit")
st.markdown("- Up to 15 seconds + intro")
st.markdown("- Choose quality (FPS)")
st.markdown("---")
# Single "View All Packages" button that links to WordPress pricing page
st.markdown("### đ Purchase Video Credits")
st.markdown("[**Click Here to View All Packages & Buy Credits â**](https://songlabai.com/video-pricing/)")
st.info("đĄ You'll be redirected to our secure pricing page where you can choose your package and complete payment via Stripe.")
st.stop()
# Check 3: Preset mode warning for free users
if not is_custom_mode and video_credits:
free_remaining = video_credits.get('free_generations_remaining', 0)
if free_remaining == 1:
st.info(f"đ Using your 1 free preset generation - Purchase credits to continue after this!")
elif free_remaining == 0:
st.error(f"đ You've used your 1 free generation - Purchase credits to continue")
# Generation Button - Label changes based on mode and eligibility
if generation_mode == "đĩ Jingle + Video":
button_label = "đ Generate Jingle + Video"
else:
button_label = "đ Generate Video"
# CRITICAL: Disable button completely if user has no eligibility (no credits and no free generations)
# This prevents any endpoint calls from users who have exhausted their free generations
button_enabled = is_admin or can_generate
if button_enabled:
generate_ad_button = st.button(button_label, type="primary", use_container_width=True, key="generate_ad_btn")
else:
# Show disabled button with clear message why it's disabled
st.button(
"đ No Credits or Free Generations Available",
type="primary",
use_container_width=True,
disabled=True,
key="generate_ad_btn_disabled",
help="Purchase video credits to generate videos"
)
generate_ad_button = False # Ensure the button click is always False
if generate_ad_button:
# CRITICAL SAFETY CHECK: Double-check eligibility before starting ANY generation
# This prevents endpoint calls even if button was somehow clicked
if not is_admin and not can_generate:
st.error("â **Generation blocked: No credits or free generations available**")
st.info("đĄ Purchase video credits to continue generating videos")
st.stop()
# Validation based on mode
validation_passed = True
if generation_mode == "đŧī¸ Video Only (Image-to-Video)" or (generation_mode == "đĩ Jingle + Video" and ad_video_mode == "image_to_video"):
if not st.session_state.ad_uploaded_image:
st.error("â Please upload an image first")
validation_passed = False
if not ad_video_prompt:
st.error("â Please provide a video description/prompt")
validation_passed = False
if generation_mode == "đĩ Jingle + Video" and not ad_jingle_prompt:
st.error("â Please provide a jingle description")
validation_passed = False
if validation_passed:
# Reset previous generation
st.session_state.ad_jingle_bytes = None
st.session_state.ad_video_path = None
st.session_state.ad_final_video = None
# Progress tracking
st.markdown("### đ Generation Progress")
# Show estimated time and send video warmup request for Jingle + Video mode
if generation_mode == "đĩ Jingle + Video":
# Calculate dynamic time estimate based on duration and FPS
# Reference: 10s @ 24fps = 15 minutes (240 frames)
# Formula: estimated_minutes = (duration * fps) / 16 frames/min
video_estimate_minutes = (duration_seconds * fps) / 16
jingle_estimate_minutes = 2 # Jingle typically takes 1-2 minutes
intro_estimate_minutes = 0
# Add intro time if enabled
if add_expert_intro and intro_prompt:
intro_estimate_minutes = 4 # 5s @ 8fps intro + voiceover generation
video_estimate_minutes = (10 * fps) / 16 # Main video capped at 10s
total_estimate_minutes = int(video_estimate_minutes + jingle_estimate_minutes + intro_estimate_minutes)
if add_expert_intro and intro_prompt:
st.info(f"âąī¸ Total estimated time: ~{total_estimate_minutes} minutes (Jingle: ~2 min, Intro: ~4 min, Main Video: ~{int(video_estimate_minutes)} min, Combining: ~1 min)")
else:
st.info(f"âąī¸ Total estimated time: ~{total_estimate_minutes} minutes (Jingle: ~2 min, Video: ~{int(video_estimate_minutes)} min)")
# Send warmup request to video endpoint to start initialization in background
VIDEO_ENDPOINT_URL = os.environ.get("VIDEO_ENDPOINT_URL", "")
VIDEO_HF_TOKEN = os.environ.get("VIDEO_HF_TOKEN", "")
if VIDEO_ENDPOINT_URL and VIDEO_HF_TOKEN:
import threading
def warmup_video_endpoint():
try:
# Send a minimal warmup request to trigger endpoint initialization
warmup_payload = {
"inputs": {
"prompt": "warmup",
"mode": "text_to_video",
"width": 640,
"height": 384,
"temp": 16,
"guidance_scale": 7.0,
"video_guidance_scale": 5.0,
"num_inference_steps": [5, 5, 5],
"video_num_inference_steps": [3, 3, 3],
"fps": 24
}
}
warmup_headers = {
"Authorization": f"Bearer {VIDEO_HF_TOKEN}",
"Content-Type": "application/json"
}
requests.post(VIDEO_ENDPOINT_URL, headers=warmup_headers, json=warmup_payload, timeout=10)
except:
pass # Ignore errors, this is just a warmup to wake the endpoint
# Start warmup in background thread - non-blocking
warmup_thread = threading.Thread(target=warmup_video_endpoint)
warmup_thread.daemon = True
warmup_thread.start()
# ========================================================================
# STEP 1: GENERATE JINGLE (only for Jingle + Video mode)
# ========================================================================
if generation_mode == "đĩ Jingle + Video":
with st.container():
st.markdown("#### Step 1/3: Generating Jingle đĩ")
# Get lyrics endpoint from environment
LYRICS_API_URL = os.environ.get("LYRICS_API_URL", "")
if not LYRICS_API_URL:
st.error("â Jingle endpoint not configured. Please set LYRICS_API_URL environment variable.")
st.stop()
else:
# Prepare jingle generation payload
# Match jingle duration to video duration
jingle_duration = float(duration_seconds)
jingle_payload = {
"inputs": {
"prompt": ad_jingle_prompt,
"lyrics": ad_jingle_lyrics
},
"parameters": {
"duration": jingle_duration, # Match video duration
"infer_step": 60,
"guidance_scale": 15,
"scheduler_type": "euler",
"cfg_type": "apg",
"omega_scale": 10,
"guidance_interval": 0.5,
"guidance_interval_decay": 0.0,
"min_guidance_scale": 3.0,
"use_erg_tag": True,
"use_erg_lyric": False,
"use_erg_diffusion": True,
"oss_steps": [],
"guidance_scale_text": 0.0,
"guidance_scale_lyric": 0.0
}
}
jingle_headers = {
"Authorization": f"Bearer {BEARER_TOKEN}",
"Content-Type": "application/json"
}
# Retry logic for cold starts
max_retries = 5
retry_delay = 60 # seconds between retries (cold start ~80s)
jingle_success = False
for attempt in range(max_retries):
try:
if attempt == 0:
spinner_msg = "đĩ Generating jingle (1-2 minutes)..."
else:
spinner_msg = f"đĩ Retrying jingle generation (attempt {attempt + 1}/{max_retries})..."
with st.spinner(spinner_msg):
jingle_response = requests.post(
LYRICS_API_URL,
headers=jingle_headers,
json=jingle_payload,
timeout=180 # 3 minute timeout
)
if jingle_response.status_code == 200:
jingle_result = jingle_response.json()
if "audio" in jingle_result:
# Decode base64 audio
jingle_bytes = base64.b64decode(jingle_result["audio"])
st.session_state.ad_jingle_bytes = jingle_bytes
# Preview jingle
st.success(f"â Jingle generated! ({jingle_result.get('duration', 10):.1f}s)")
st.audio(jingle_bytes, format="audio/wav")
jingle_success = True
break
else:
st.error(f"â Jingle generation failed: {jingle_result.get('error', 'Unknown error')}")
st.stop()
elif jingle_response.status_code == 503:
if attempt < max_retries - 1:
st.warning(f"âŗ Jingle service is warming up (cold start ~80s). Waiting {retry_delay}s before retry {attempt + 2}/{max_retries}...")
time.sleep(retry_delay)
else:
st.error(f"â Jingle endpoint error: HTTP 503 - Service failed to warm up after {max_retries} attempts")
st.info("đĄ Please wait a moment and try generating again.")
st.stop()
else:
st.error(f"â Jingle endpoint error: HTTP {jingle_response.status_code}")
st.stop()
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
st.warning(f"âŗ Request timed out. Retrying ({attempt + 2}/{max_retries})...")
time.sleep(retry_delay)
else:
st.error(f"â Jingle generation timed out after {max_retries} attempts")
st.stop()
except Exception as e:
if attempt < max_retries - 1:
st.warning(f"â ī¸ Error: {str(e)}. Retrying ({attempt + 2}/{max_retries})...")
time.sleep(retry_delay)
else:
st.error(f"â Jingle generation error: {str(e)}")
st.stop()
if not jingle_success:
st.error("â Failed to generate jingle after all retry attempts")
st.stop()
st.markdown("---")
# ========================================================================
# STEP 2: GENERATE VIDEO (runs for all modes, but step number varies)
# ========================================================================
# Determine if we should proceed with video generation
should_generate_video = True
if generation_mode == "đĩ Jingle + Video":
# Only generate video if jingle was successful
if not st.session_state.ad_jingle_bytes:
should_generate_video = False
step_label = "#### Step 2/3: Generating Video đŦ"
else:
# Video-only modes - show time estimate
step_label = "#### Generating Video đŦ"
# Calculate dynamic time estimate based on duration and FPS
# Reference: 10s @ 24fps = 15 minutes (240 frames)
# Formula: estimated_minutes = (duration * fps) / 16 frames/min
intro_estimate_minutes = 0
main_video_duration = duration_seconds
if add_expert_intro and intro_prompt:
intro_estimate_minutes = 4 # 5s @ 8fps intro + voiceover
main_video_duration = min(duration_seconds, 10) # Main video capped at 10s
video_estimate_minutes = int((main_video_duration * fps) / 16)
total_estimate = video_estimate_minutes + intro_estimate_minutes
if add_expert_intro and intro_prompt:
st.info(f"âąī¸ Estimated time: ~{total_estimate} minutes (Intro: ~4 min, Main Video: ~{video_estimate_minutes} min, Combining: ~1 min)")
else:
st.info(f"âąī¸ Estimated time: ~{video_estimate_minutes} minutes")
if should_generate_video:
with st.container():
st.markdown(step_label)
# Video endpoint configuration from environment variables
VIDEO_ENDPOINT_URL = os.environ.get("VIDEO_ENDPOINT_URL", "")
VIDEO_HF_TOKEN = os.environ.get("VIDEO_HF_TOKEN", "")
VIDEO_REPO_ID = os.environ.get("VIDEO_REPO_ID", "Phoenixak99/savedvideos")
HF_TOKEN_DOWNLOAD = os.environ.get("HF_TOKEN_DOWNLOAD", "")
# Check if required environment variables are set
if not VIDEO_ENDPOINT_URL:
st.error("â Video endpoint not configured. Please set VIDEO_ENDPOINT_URL environment variable.")
st.stop()
if not VIDEO_HF_TOKEN:
st.error("â Video endpoint token not configured. Please set VIDEO_HF_TOKEN environment variable.")
st.stop()
if not HF_TOKEN_DOWNLOAD:
st.error("â HuggingFace download token not configured. Please set HF_TOKEN_DOWNLOAD environment variable.")
st.stop()
# ========================================================================
# EXPERT INTRO GENERATION (if enabled)
# ========================================================================
intro_video_path = None
if add_expert_intro and intro_prompt:
st.markdown("##### đŦ Generating 5-Second Intro...")
# Adjust main video duration to 10s (intro adds +5s)
main_video_duration = min(duration_seconds, 10)
# Generate 5-second intro video at 8fps (low quality for logo)
intro_temp = int((5 * 8 * 31) / 240) # 5 seconds at 8fps
intro_payload = {
"inputs": {
"prompt": intro_prompt,
"mode": "text_to_video",
"width": 1280,
"height": 768,
"temp": intro_temp,
"guidance_scale": 7.0,
"video_guidance_scale": 5.0,
"num_inference_steps": [10, 10, 10], # Lower quality for speed
"video_num_inference_steps": [5, 5, 5],
"fps": 8 # Low FPS for static logo
}
}
intro_headers = {
"Authorization": f"Bearer {VIDEO_HF_TOKEN}",
"Content-Type": "application/json"
}
try:
from huggingface_hub import HfApi
api = HfApi()
# Get initial count
initial_files = api.list_repo_files(
repo_id=VIDEO_REPO_ID,
repo_type="dataset",
token=HF_TOKEN_DOWNLOAD
)
intro_initial_count = len([f for f in initial_files if f.endswith('.mp4')])
# Send intro generation request
st.info("đ¤ Sending intro video generation request...")
intro_response = requests.post(VIDEO_ENDPOINT_URL, headers=intro_headers, json=intro_payload, timeout=60)
if intro_response.status_code != 200:
st.error(f"â Intro generation failed: HTTP {intro_response.status_code}")
st.stop()
# Monitor for intro video completion
intro_start_time = time.time()
intro_max_wait = 300 # 5 minutes for intro
intro_progress = st.progress(0)
intro_status = st.empty()
intro_video_path = None
while True:
intro_elapsed = time.time() - intro_start_time
if intro_elapsed > intro_max_wait:
st.error("â Intro video generation timeout")
st.stop()
intro_progress.progress(min(intro_elapsed / (4 * 60), 0.95))
intro_status.text(f"âŗ Generating intro... {intro_elapsed/60:.1f} min elapsed")
# Check for new intro video
try:
files = api.list_repo_files(
repo_id=VIDEO_REPO_ID,
repo_type="dataset",
token=HF_TOKEN_DOWNLOAD
)
video_files = [f for f in files if f.endswith('.mp4')]
if len(video_files) > intro_initial_count:
# Found new video
new_videos = sorted([f for f in video_files if f not in initial_files],
key=lambda x: x, reverse=True)
if new_videos:
intro_video_file = new_videos[0]
video_url = f"https://huggingface.co/datasets/{VIDEO_REPO_ID}/resolve/main/{intro_video_file}"
st.info("đĨ Downloading intro video...")
download_headers = {"Authorization": f"Bearer {HF_TOKEN_DOWNLOAD}"}
intro_video_response = requests.get(video_url, headers=download_headers, timeout=120)
if intro_video_response.status_code == 200:
# Save intro video to temp file
intro_video_path = tempfile.mktemp(suffix="_intro.mp4")
with open(intro_video_path, "wb") as f:
f.write(intro_video_response.content)
st.success("â Intro video generated!")
intro_progress.progress(1.0)
break
except Exception as e:
pass
time.sleep(10)
# Generate voiceover if company name provided
if company_name and intro_video_path:
st.info("đ¤ Generating voiceover...")
try:
from gtts import gTTS
# Create TTS audio
tts = gTTS(text=company_name, lang='en', slow=False)
voiceover_path = tempfile.mktemp(suffix="_voice.mp3")
tts.save(voiceover_path)
# Combine intro video + voiceover using FFmpeg
st.info("đŦ Adding voiceover to intro...")
import subprocess
intro_with_voice_path = tempfile.mktemp(suffix="_intro_final.mp4")
ffmpeg_cmd = [
'ffmpeg', '-y',
'-i', intro_video_path,
'-i', voiceover_path,
'-c:v', 'copy',
'-c:a', 'aac',
'-b:a', '192k',
'-filter_complex', '[1:a]afade=t=in:st=0:d=0.5,afade=t=out:st=4.5:d=0.5[a]',
'-map', '0:v',
'-map', '[a]',
'-shortest',
intro_with_voice_path
]
result = subprocess.run(ffmpeg_cmd, capture_output=True, text=True)
if result.returncode == 0:
# Replace intro_video_path with version that has voiceover
os.remove(intro_video_path)
os.remove(voiceover_path)
intro_video_path = intro_with_voice_path
st.success("â Voiceover added to intro!")
else:
st.warning("â ī¸ Could not add voiceover, using intro without voice")
if os.path.exists(voiceover_path):
os.remove(voiceover_path)
if os.path.exists(intro_with_voice_path):
os.remove(intro_with_voice_path)
except ImportError:
st.warning("â ī¸ TTS not available, intro will have no voiceover")
except Exception as e:
st.warning(f"â ī¸ Voiceover generation failed: {str(e)}")
except Exception as e:
st.error(f"â Intro generation error: {str(e)}")
st.stop()
st.markdown("---")
st.markdown("##### đŦ Generating Main Video (10 seconds)...")
else:
# No intro, use full duration
main_video_duration = duration_seconds
# Prepare main video generation payload
# Convert duration to temp parameter
# Reference: 31 temp = 10s at 24fps (240 frames)
# Formula: temp = (duration_seconds * fps) / 7.74
temp_value = int((main_video_duration * fps * 31) / 240)
video_payload = {
"inputs": {
"prompt": ad_video_prompt,
"mode": ad_video_mode, # text_to_video or image_to_video
"width": 1280,
"height": 768,
"temp": temp_value, # User-selected duration
"guidance_scale": guidance_scale, # User-selected creativity
"video_guidance_scale": video_guidance_scale, # User-selected transition energy
"num_inference_steps": [20, 20, 20],
"video_num_inference_steps": [10, 10, 10],
"fps": fps # User-selected FPS (8, 16, or 24)
}
}
# Add image to payload if image-to-video mode
if ad_video_mode == "image_to_video" and st.session_state.ad_uploaded_image:
video_payload["inputs"]["image"] = st.session_state.ad_uploaded_image
video_headers = {
"Authorization": f"Bearer {VIDEO_HF_TOKEN}",
"Content-Type": "application/json"
}
try:
# Get initial video count from HF dataset
from huggingface_hub import HfApi
import time
api = HfApi()
initial_files = api.list_repo_files(
repo_id=VIDEO_REPO_ID,
repo_type="dataset",
token=HF_TOKEN_DOWNLOAD
)
initial_video_count = len([f for f in initial_files if f.endswith('.mp4')])
# Send video generation request with retry logic for cold starts
import threading
max_video_retries = 5
video_retry_delay = 60 # seconds between retries (cold start ~90s)
request_success = False
for video_attempt in range(max_video_retries):
try:
if video_attempt == 0:
st.info("đ¤ Sending video generation request...")
else:
st.info(f"đ¤ Retrying video request (attempt {video_attempt + 1}/{max_video_retries})...")
request_sent = threading.Event()
response_status = {"code": None, "error": None}
def send_video_request():
try:
resp = requests.post(VIDEO_ENDPOINT_URL, headers=video_headers, json=video_payload, timeout=60)
response_status["code"] = resp.status_code
if resp.status_code != 200:
response_status["error"] = resp.text
except Exception as e:
response_status["error"] = str(e)
finally:
request_sent.set()
thread = threading.Thread(target=send_video_request)
thread.start()
request_sent.wait(timeout=70)
# Check if we got a 503 (cold start)
if response_status["code"] == 503:
if video_attempt < max_video_retries - 1:
st.warning(f"âŗ Video service is warming up (cold start ~90s). Waiting {video_retry_delay}s before retry {video_attempt + 2}/{max_video_retries}...")
time.sleep(video_retry_delay)
continue
else:
st.error(f"â Video endpoint error: HTTP 503 - Service failed to warm up after {max_video_retries} attempts")
st.info("đĄ Please wait a moment and try generating again.")
st.stop()
elif response_status["code"] and response_status["code"] != 200:
st.error(f"â Video endpoint error: HTTP {response_status['code']}")
if response_status["error"]:
st.error(f"Details: {response_status['error']}")
st.stop()
# Request successful or at least sent
request_success = True
break
except Exception as e:
if video_attempt < max_video_retries - 1:
st.warning(f"â ī¸ Error sending request: {str(e)}. Retrying ({video_attempt + 2}/{max_video_retries})...")
time.sleep(video_retry_delay)
else:
st.error(f"â Failed to send video request: {str(e)}")
st.stop()
if not request_success:
st.error("â Failed to send video generation request after all attempts")
st.stop()
# Set start time for monitoring
start_time = time.time()
# Monitor for video completion
progress_bar = st.progress(0)
status_text = st.empty()
# Dynamic timeout based on video complexity
# Reference: 10s @ 24fps = 15 minutes, add 20% buffer
estimated_minutes = (duration_seconds * fps) / 16
max_wait_seconds = int((estimated_minutes + 3) * 60 * 1.2) # Add 3 min buffer + 20% safety margin
check_interval = 10
while True:
elapsed = time.time() - start_time
if elapsed > max_wait_seconds:
st.error(f"â Video generation timeout ({int(max_wait_seconds/60)} minutes)")
st.stop()
# Update progress (show up to 95% until complete)
progress = min(elapsed / (max_wait_seconds * 0.95), 0.95)
progress_bar.progress(progress)
status_text.text(f"âŗ Generating video... {elapsed/60:.1f} min elapsed (est: ~{int(estimated_minutes)} min)")
# Check for new video
try:
files = api.list_repo_files(
repo_id=VIDEO_REPO_ID,
repo_type="dataset",
token=HF_TOKEN_DOWNLOAD
)
video_files = [f for f in files if f.endswith('.mp4')]
if len(video_files) > initial_video_count:
# New video detected!
latest_video = sorted(video_files)[-1]
video_url = f"https://huggingface.co/datasets/{VIDEO_REPO_ID}/resolve/main/{latest_video}"
progress_bar.progress(1.0)
status_text.text(f"â Video generated in {elapsed/60:.1f} minutes!")
# Download video
st.info("đĨ Downloading video...")
download_headers = {"Authorization": f"Bearer {HF_TOKEN_DOWNLOAD}"}
video_response = requests.get(video_url, headers=download_headers, timeout=120)
if video_response.status_code == 200:
# Save to temp file
with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
f.write(video_response.content)
video_path = f.name
st.session_state.ad_video_path = video_path
st.success(f"â Video downloaded! ({len(video_response.content)/1024/1024:.1f} MB)")
# ========================================================================
# DEDUCT VIDEO CREDIT
# ========================================================================
if not is_admin:
try:
generation_data = {
'mode': ad_video_mode,
'duration': duration_seconds,
'fps': fps,
'guidance_scale': guidance_scale,
'video_guidance_scale': video_guidance_scale,
'is_preset': not is_custom_mode,
'is_expert_intro': add_expert_intro,
'prompt': ad_video_prompt,
'intro_prompt': intro_prompt if add_expert_intro else None
}
credit_used = use_video_credit(jwt_token, generation_data)
if credit_used:
if not is_custom_mode:
st.info("â Free generation used!")
else:
st.info("â 1 video credit deducted")
else:
st.warning("â ī¸ Failed to deduct credit, but video was generated")
except Exception as e:
st.warning(f"â ī¸ Error deducting credit: {str(e)}")
# ========================================================================
# COMBINE INTRO + MAIN VIDEO (if intro was generated)
# ========================================================================
if intro_video_path and os.path.exists(intro_video_path):
st.markdown("---")
st.markdown("##### đŦ Combining Intro + Main Video...")
try:
import subprocess
# Create concat list for FFmpeg
concat_list_path = tempfile.mktemp(suffix="_concat.txt")
with open(concat_list_path, 'w') as f:
f.write(f"file '{intro_video_path}'\n")
f.write(f"file '{video_path}'\n")
# Output path for combined video
combined_video_path = tempfile.mktemp(suffix="_combined.mp4")
# Concatenate intro + main with crossfade transition
ffmpeg_concat_cmd = [
'ffmpeg', '-y',
'-f', 'concat',
'-safe', '0',
'-i', concat_list_path,
'-c', 'copy',
combined_video_path
]
result = subprocess.run(ffmpeg_concat_cmd, capture_output=True, text=True)
if result.returncode == 0:
# Replace video_path with combined version
os.remove(video_path)
os.remove(intro_video_path)
os.remove(concat_list_path)
# Update session state with combined video
video_path = combined_video_path
st.session_state.ad_video_path = combined_video_path
st.success("â Intro + Main video combined! Final video: 15 seconds")
else:
st.warning(f"â ī¸ Could not combine videos: {result.stderr}")
st.info("đĄ Using main video without intro")
except Exception as e:
st.warning(f"â ī¸ Error combining videos: {str(e)}")
st.info("đĄ Using main video without intro")
# Preview video only for Jingle + Video mode (before combining with audio)
# For video-only modes, skip preview and show final result with download button
if generation_mode == "đĩ Jingle + Video":
# Show preview of combined video (with intro if applicable)
with open(video_path, 'rb') as f:
st.video(f.read())
break
else:
st.error(f"â Failed to download video: HTTP {video_response.status_code}")
st.stop()
except Exception as e:
st.warning(f"â ī¸ Error checking for video: {e}")
time.sleep(check_interval)
except Exception as e:
st.error(f"â Video generation error: {str(e)}")
st.stop()
st.markdown("---")
# ========================================================================
# STEP 3: COMBINE JINGLE + VIDEO
# ========================================================================
if st.session_state.ad_jingle_bytes and st.session_state.ad_video_path:
with st.container():
st.markdown("#### Step 3/3: Creating Final Ad đĨ")
try:
# Save jingle to temp file
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
f.write(st.session_state.ad_jingle_bytes)
jingle_temp_path = f.name
# Output path for final video
final_video_path = tempfile.mktemp(suffix=".mp4")
# Use ffmpeg to combine audio and video
import subprocess
ffmpeg_cmd = [
"ffmpeg",
"-i", st.session_state.ad_video_path, # Input video
"-i", jingle_temp_path, # Input audio
"-c:v", "copy", # Copy video codec
"-c:a", "aac", # Encode audio as AAC
"-b:a", "192k", # Audio bitrate
"-shortest", # Match shortest stream
"-y", # Overwrite output
final_video_path
]
with st.spinner("đŦ Merging jingle with video..."):
result = subprocess.run(ffmpeg_cmd, capture_output=True, text=True)
if result.returncode == 0 and os.path.exists(final_video_path):
# Read final video
with open(final_video_path, "rb") as f:
final_video_bytes = f.read()
st.session_state.ad_final_video = final_video_bytes
# ========================================================================
# UPLOAD FULL AD TO HUGGINGFACE DATASET
# ========================================================================
try:
from huggingface_hub import upload_file
import uuid
# Generate unique filename
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
unique_id = str(uuid.uuid4())[:8]
# Get user ID for filename
try:
decoded_token = jwt.decode(jwt_token, options={"verify_signature": False})
user_id = decoded_token.get('data', {}).get('user', {}).get('id', 'unknown')
except:
user_id = 'unknown'
filename = f"full_ad_{timestamp}_{user_id}_{unique_id}.mp4"
# Silently upload to cloud storage (background)
try:
temp_upload_path = tempfile.mktemp(suffix=".mp4")
with open(temp_upload_path, "wb") as f:
f.write(final_video_bytes)
upload_file(
path_or_fileobj=temp_upload_path,
path_in_repo=filename,
repo_id=VIDEO_REPO_ID,
repo_type="dataset",
token=HF_TOKEN_DOWNLOAD,
)
# Clean up temp upload file
try:
os.remove(temp_upload_path)
except:
pass
except:
# Silently fail - user doesn't need to know about storage
pass
except Exception as e:
# Don't show upload errors to user
pass
# Cleanup temp files
try:
os.remove(jingle_temp_path)
os.remove(st.session_state.ad_video_path)
os.remove(final_video_path)
except:
pass
st.success("â Ad created successfully!")
# Display final ad
st.markdown("### đ Your Commercial Ad is Ready!")
st.video(final_video_bytes)
# Download button
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
st.download_button(
label="âŦī¸ Download Ad (MP4)",
data=final_video_bytes,
file_name=f"commercial_ad_{timestamp}.mp4",
mime="video/mp4",
use_container_width=True,
key="download_ad_btn"
)
st.balloons()
else:
st.error(f"â ffmpeg error: {result.stderr}")
except Exception as e:
st.error(f"â Error combining jingle and video: {str(e)}")
# ========================================================================
# VIDEO-ONLY DOWNLOAD (for modes without jingle)
# ========================================================================
elif st.session_state.ad_video_path and generation_mode != "đĩ Jingle + Video":
with st.container():
st.markdown("### đ Your Video is Ready!")
# Read the video file
try:
with open(st.session_state.ad_video_path, "rb") as f:
video_bytes = f.read()
# Display video
st.video(video_bytes)
# Download button
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
mode_name = "text_to_video" if ad_video_mode == "text_to_video" else "image_to_video"
st.download_button(
label="âŦī¸ Download Video (MP4)",
data=video_bytes,
file_name=f"video_{mode_name}_{timestamp}.mp4",
mime="video/mp4",
use_container_width=True,
key="download_video_only_btn"
)
st.balloons()
# Cleanup
try:
os.remove(st.session_state.ad_video_path)
except:
pass
except Exception as e:
st.error(f"â Error reading video file: {str(e)}")
st.divider()
st.header("đĩ Load Previous Music")
st.markdown("""
Upload a previously downloaded music clip or select from available beats to continue post-processing,
add vocals, or apply audio effects.
""")
# Add tabs to separate uploaded music and beat selection
tab1, tab2 = st.tabs(["Upload Your Music", "Choose from Available Beats"])
with tab1:
# Original file uploader UI
uploaded_music = st.file_uploader(
"Upload previous music clip",
type=["wav", "mp3", "ogg", "flac"],
help="Upload music you previously generated and downloaded from Songlab AI",
key="user_music_uploader"
)
if uploaded_music:
# Show a preview player for the uploaded music
st.audio(uploaded_music, format=f"audio/{uploaded_music.name.split('.')[-1]}")
# Process button
if st.button("Load for Post-Processing", key="load_uploaded_music"):
try:
# Read the uploaded music into memory
with st.spinner("Loading audio..."):
# Convert the uploaded file to an AudioSegment
audio_data = uploaded_music.read()
# Create a temporary file-like object
audio_buffer = BytesIO(audio_data)
audio_buffer.seek(0)
# Load into AudioSegment based on file extension
file_extension = uploaded_music.name.split('.')[-1].lower()
audio_segment = AudioSegment.from_file(audio_buffer, format=file_extension)
# Convert to the same format used in your application
if file_extension != 'wav':
wav_buffer = BytesIO()
audio_segment.export(wav_buffer, format='wav')
wav_buffer.seek(0)
audio_segment = AudioSegment.from_wav(wav_buffer)
# Store in session state
st.session_state.audio_pydub = audio_segment
st.session_state.audio_sample_rate = audio_segment.frame_rate
st.session_state.augmented_audio_pydub = audio_segment
st.session_state.augmented_audio_sample_rate = audio_segment.frame_rate
# Convert to bytes for player
wav_bytes = BytesIO()
audio_segment.export(wav_bytes, format='wav')
wav_bytes.seek(0)
st.session_state.audio_bytes = wav_bytes.getvalue()
backup_audio_data()
# Set playing state
st.session_state.playing_state = 'generated'
# Clear any existing metadata
st.session_state.audio_metadata = {
'genre': 'Unknown (Uploaded)',
'energy_level': 'Unknown',
'tempo': 'Unknown',
'description': f'Uploaded file: {uploaded_music.name}',
'duration': len(audio_segment)/1000 # Convert ms to seconds
}
# Success message
st.success("đ Music loaded successfully! Scroll down to use post-processing tools.")
# Force refresh to update the UI with the loaded audio
backup_audio_data()
st.rerun()
except Exception as e:
st.error(f"â Error loading audio file: {str(e)}")
st.info("Please make sure the file is a valid audio file in one of the supported formats.")
with tab2:
# Show beat selection for all users
st.markdown("### đĩ Select a Beat")
# Fetch beats from WordPress
with st.spinner("Loading beats from WordPress..."):
jwt_token = st.session_state.get("jwt_token")
available_beats = get_wordpress_beats(jwt_token)
if not available_beats:
st.info("No beats are currently available. Please upload your own music or contact an admin to add beats.")
else:
# Create a dictionary for easy lookup
beats_dict = {beat['title']: beat for beat in available_beats}
selected_beat_title = st.selectbox(
"Choose a beat to master and add vocals to",
options=["-- Select a beat --"] + list(beats_dict.keys()),
key="user_beat_selection"
)
if selected_beat_title and selected_beat_title != "-- Select a beat --":
# Get the selected beat info
selected_beat = beats_dict[selected_beat_title]
# Display information about the beat
st.markdown(f"#### Beat Preview")
st.markdown(f"**Beat:** {selected_beat_title}")
# Show additional metadata if available
if selected_beat.get('genre') or selected_beat.get('bpm') or selected_beat.get('key'):
meta_items = []
if selected_beat.get('genre'):
meta_items.append(f"Genre: {selected_beat['genre']}")
if selected_beat.get('bpm'):
meta_items.append(f"BPM: {selected_beat['bpm']}")
if selected_beat.get('key'):
meta_items.append(f"Key: {selected_beat['key']}")
if selected_beat.get('duration'):
meta_items.append(f"Duration: {selected_beat['duration']}s")
if meta_items:
st.caption(" | ".join(meta_items))
# Get file extension from URL
file_extension = selected_beat['url'].split('.')[-1].lower() if '.' in selected_beat['url'] else "mp3"
# Download and preview the beat
try:
with st.spinner("Loading beat preview..."):
beat_buffer = download_wordpress_beat(selected_beat['url'])
if beat_buffer:
# Display audio player for preview
st.audio(beat_buffer, format=f"audio/{file_extension}")
# Add button to load the beat for processing
if st.button("Load this Beat for Processing", key="load_selected_beat"):
with st.spinner("Loading beat for processing..."):
success, message = load_wordpress_beat_for_processing(
selected_beat['url'],
selected_beat_title
)
if success:
st.success("đ Beat loaded successfully! Scroll down to use post-processing tools.")
backup_audio_data()
st.rerun()
else:
st.error(f"â {message}")
else:
st.error("Failed to load beat preview")
except Exception as e:
st.error(f"Error previewing beat: {str(e)}")
import traceback
print(traceback.format_exc())
# Helper info section
with st.expander("âšī¸ About uploading previous music and using beats"):
st.markdown("""
### Ways to use music in Songlab AI
You have two options for working with existing music:
1. **Upload Your Own Music**
- Upload music you previously generated and downloaded from Songlab AI
- Import your own audio files to process with our tools
2. **Use Available Beats**
- Select from pre-made beats provided by Songlab AI
- Perfect for adding vocals or creating remixes
### What you can do with the music
- **Add vocals** to instrumental tracks
- **Apply different effects** to experiment with various sounds
- **Master** the audio with professional-quality tools
- **Adjust pitch and volume** to perfect your sound
### Supported file formats
- **WAV** (.wav) - Highest quality, uncompressed
- **MP3** (.mp3) - Common compressed format
- **OGG** (.ogg) - Free, open container format
- **FLAC** (.flac) - Lossless compression format
""")
# This divider separates the upload section from post-processing options
st.divider()
# Post-processing options - only show if we have audio to process
if st.session_state.audio_pydub is not None:
st.header("Post-processing Options")
vocal_file = st.file_uploader(
"Upload Vocal File", type=["mp3", "wav", "ogg", "flac", "aac"]
)
if vocal_file:
st.session_state.vocal_audio = vocal_file.read()
# Mixing
mix_vocals = st.checkbox("Mix Vocals")
if mix_vocals and st.session_state.vocal_audio is not None:
with NamedTemporaryFile() as f:
f.write(st.session_state.vocal_audio)
st.session_state.vocal_audio = AudioSegment.from_file(f.name)
st.session_state.augmented_audio_pydub = st.session_state.augmented_audio_pydub.overlay(
st.session_state.vocal_audio, position=100
)
st.session_state.augmented_audio = convert_audio_segment_to_float_array(
st.session_state.augmented_audio_pydub
)
st.session_state.augmented_audio_sample_rate = st.session_state.augmented_audio_pydub.frame_rate
st.session_state.playing_state = 'processed'
elif not mix_vocals and st.session_state.vocal_audio is not None:
st.session_state.augmented_audio_pydub = st.session_state.audio_pydub
st.session_state.augmented_audio_sample_rate = st.session_state.audio_pydub.frame_rate
# Mastering
st.header("Mastering")
st.markdown("")
# Volume Balance, Compression Ratio, and Reverb Amount
vol_col, pitch_shift, buttons_col = st.columns([2, 2, 2.5])
with buttons_col:
with st.container(height=371, border=True):
st.markdown("")
apply_stereo = st.button("Apply Stereo Effect")
st.markdown("")
reverse = st.button("Apply Audio Reverse")
st.markdown("")
reset_post_processing = st.button("Undo All Post-processings")
st.markdown("")
with vol_col:
with st.container(border=True):
volume_balance = svs.vertical_slider(
"Volume Balance",
min_value=-10.0,
max_value=10.0,
default_value=0.0,
step=0.1,
slider_color="green",
track_color="lightgray",
thumb_color="red",
thumb_shape="pill",
)
vol_button = st.button("Apply Vol-Balance")
# Pitch shifting
with pitch_shift:
with st.container(border=True):
pitch_semitones = svs.vertical_slider(
label="Pitch (semitones)",
min_value=-12,
max_value=12,
default_value=0,
step=1,
slider_color="red",
track_color="lightgray",
thumb_color="red",
thumb_shape="pill",
)
pitch_shift_button = st.button("Apply Pitch Shift")
effect_applied = False
if st.session_state.augmented_audio_pydub is not None:
if vol_button:
st.session_state.augmented_audio_pydub = st.session_state.augmented_audio_pydub + volume_balance
effect_applied = True
st.session_state.playing_state = 'processed'
if apply_stereo:
st.session_state.augmented_audio_pydub = st.session_state.augmented_audio_pydub.pan(
-0.5
).overlay(st.session_state.augmented_audio_pydub.pan(0.5))
effect_applied = True
st.session_state.playing_state = 'processed'
if reverse:
st.session_state.augmented_audio_pydub = st.session_state.augmented_audio_pydub.reverse()
effect_applied = True
st.session_state.playing_state = 'processed'
if pitch_shift_button:
st.session_state.augmented_audio_pydub = st.session_state.augmented_audio_pydub._spawn(
st.session_state.augmented_audio_pydub.raw_data,
overrides={
"frame_rate": int(
st.session_state.augmented_audio_pydub.frame_rate
* (2 ** (pitch_semitones / 12.0))
)
},
)
effect_applied = True
st.session_state.playing_state = 'processed'
if reset_post_processing and st.session_state.audio_pydub is not None:
st.session_state.augmented_audio_pydub = st.session_state.audio_pydub
st.session_state.playing_state = 'generated'
effect_applied = True
# Display the final audio if processed
if effect_applied or st.session_state.playing_state == 'processed':
display_audio_player(is_final=True)
# Force rerun if any effect was applied to show the updated audio
if effect_applied:
st.session_state['ui_refresh_needed'] = True
smart_rerun(force=True)