Modernized and simplified the app.

This wasn't a crazy rewrite or anything, I just updated it to the new
YouTube Transcript and OpenAI API's, as well as super simplifying the
code. On top of that, it now works single threaded, just using multiple
gunicorn threads for concurrency. It's a lot simplier and cleaner,
although not up to my current standards.
This commit is contained in:
foreverpyrite
2025-11-03 22:43:15 -06:00
parent c6b608f125
commit 1fd6711da0
12 changed files with 850 additions and 409 deletions

View File

@@ -1,90 +1,63 @@
import logging
import os
import uuid
from flask import Flask, render_template, Response, request, session
from main import yoink, process, user_streams, stream_lock
from flask import Flask, render_template, request, session
# Only used for the main.process function, but it feels right to have that over just "process"
import main
app = Flask(__name__, static_folder="website/static", template_folder="website")
app.secret_key = os.urandom(24) # Necessary for using sessions
# Configure logging
logging.basicConfig(
filename='./logs/app.log',
level=logging.DEBUG,
format='%(asctime)s %(levelname)s: %(message)s',
datefmt='%Y-%m-%d %H:%M:%S'
)
def create_session():
def create_session() -> str:
"""
Create a new session by generating a UUID and ensuring it does not collide
Create a new session by generating a UUID and ensuring it does not collide
with an existing session in the user_streams global dictionary.
Returns:
str: A unique session ID.
"""
session_id = str(uuid.uuid4())
# Even though collisions are unlikely, we check for safety.
try:
if user_streams[session_id]:
session_id = create_session()
except KeyError:
pass
return session_id
return uuid.uuid4().hex
@app.route('/')
@app.get("/")
def home():
"""
Render the home page and initialize a session.
Returns:
Response: The rendered home page with a unique session id.
"""
session_id = create_session()
session['id'] = session_id
session["id"] = session_id
logging.info(f"Home page accessed. Assigned initial session ID: {session_id}")
return render_template('index.html', session_id=session_id)
return render_template("index.html", session_id=session_id)
@app.route('/process_url', methods=['POST'])
@app.post("/process-url")
def process_url():
"""
Accept a YouTube URL (from a form submission), initialize the session if necessary,
Accept a YouTube URL (from a form submission), initialize the session if necessary,
and trigger the transcript retrieval and AI processing.
Returns:
Response: Text response indicating start or error message.
Response: The results of processing the url.
"""
session_id = session.get('id')
session_id = session.get("id")
if not session_id:
session_id = create_session()
session['id'] = session_id
session["id"] = session_id
logging.info(f"No existing session. Created new session ID: {session_id}")
url = request.form['url']
url = request.form["url"]
logging.info(f"Received URL for processing from session {session_id}: {url}")
success, msg, status_code = process(url, session_id)
if success:
logging.info(f"Processing started successfully for session {session_id}.")
return Response("Processing started. Check /stream_output for updates.", content_type='text/plain', status=200)
else:
logging.error(f"Processing failed for session {session_id}: {msg}")
return Response(msg, content_type='text/plain', status=status_code)
# Before I had process return stuff in the form of (success: bool, msg: str, status: int)
# I don't know why I was doing all that back then, it's not like it was a library or anything
# I planned on using on other projects...
return main.process(url, session_id)
@app.route('/stream_output')
def stream_output():
"""
Stream the AI processing output for the current session.
Returns:
Response: A streaming response with text/plain content.
"""
session_id = session.get('id')
if not session_id or session_id not in user_streams:
logging.warning(f"Stream requested without a valid session ID: {session_id}")
return Response("No active stream for this session.", content_type='text/plain', status=400)
logging.info(f"Streaming output requested for session {session_id}.")
return Response(yoink(session_id), content_type='text/plain', status=200)
if __name__ == '__main__':
if __name__ == "__main__":
logging.info("Starting Flask application.")
# Running with threaded=True to handle multiple requests concurrently.
app.run(debug=True, threaded=True)
app.run(debug=True, threaded=True)