import os import secrets import threading import time from pathlib import Path from flask import ( Flask, render_template, request, redirect, url_for, flash, send_from_directory, jsonify, abort ) from werkzeug.utils import secure_filename from config import Config from models import JobStore from worker import process_job app = Flask(__name__) app.config.from_object(Config) app.config['MAX_CONTENT_LENGTH'] = Config.MAX_UPLOAD_SIZE _cfg = Config() store = JobStore(_cfg.DATABASE) _worker_thread = None _worker_stop = threading.Event() def _allowed_file(filename, extensions): return '.' in filename and filename.rsplit('.', 1)[1].lower() in extensions _IMAGE_EXTENSIONS = {'png', 'jpg', 'jpeg', 'gif', 'webp', 'bmp', 'tiff'} _AUDIO_EXTENSIONS = {'mp3', 'wav', 'flac', 'aac', 'm4a', 'ogg', 'wma'} def _ensure_dirs(): _cfg.UPLOAD_FOLDER.mkdir(parents=True, exist_ok=True) _cfg.OUTPUT_FOLDER.mkdir(parents=True, exist_ok=True) _ensure_dirs() def _worker_loop(): while not _worker_stop.is_set(): job = store.pop_next_waiting() if job is None: time.sleep(1) continue try: process_job(job['id'], store, _cfg) except Exception: # error_message already stored; keep worker alive pass def start_worker(): global _worker_thread if _worker_thread is None or not _worker_thread.is_alive(): _worker_stop.clear() _worker_thread = threading.Thread(target=_worker_loop, daemon=True) _worker_thread.start() start_worker() def _admin_required(): if _cfg.ADMIN_PASSWORD: if request.path.startswith('/admin') and request.path != url_for('admin_login'): if request.form.get('password') == _cfg.ADMIN_PASSWORD: return None if request.args.get('password') == _cfg.ADMIN_PASSWORD: return None return abort(401) return None @app.before_request def _before_request(): # If admin password is set, protect the whole app except health check. if _cfg.ADMIN_PASSWORD and request.path != '/health': if request.path == url_for('admin_login'): return None if request.form.get('password') == _cfg.ADMIN_PASSWORD: return None if request.args.get('password') == _cfg.ADMIN_PASSWORD: return None return abort(401) _admin_required() @app.route('/') def index(): jobs = store.list_jobs() return render_template('index.html', jobs=jobs, cfg=_cfg) @app.route('/upload', methods=['POST']) def upload(): title = request.form.get('title', '').strip() or 'Untitled' image = request.files.get('image') audio = request.files.get('audio') if not image or not image.filename: flash('Please select an image file.', 'error') return redirect(url_for('index')) if not audio or not audio.filename: flash('Please select an audio file.', 'error') return redirect(url_for('index')) image_ext = image.filename.rsplit('.', 1)[1].lower() audio_ext = audio.filename.rsplit('.', 1)[1].lower() if not _allowed_file(image.filename, _IMAGE_EXTENSIONS): flash(f'Unsupported image format: {image_ext}', 'error') return redirect(url_for('index')) if not _allowed_file(audio.filename, _AUDIO_EXTENSIONS): flash(f'Unsupported audio format: {audio_ext}', 'error') return redirect(url_for('index')) _ensure_dirs() safe_title = secure_filename(title).replace('.', '_') or 'untitled' job_prefix = f"{int(time.time())}-{secrets.token_hex(4)}" image_filename = f"{job_prefix}-img-{safe_title}.{image_ext}" audio_filename = f"{job_prefix}-aud-{safe_title}.{audio_ext}" image.save(_cfg.UPLOAD_FOLDER / image_filename) audio.save(_cfg.UPLOAD_FOLDER / audio_filename) job_id = store.create_job(title, image_filename, audio_filename) flash(f'Job #{job_id} added to the queue.', 'success') return redirect(url_for('index')) @app.route('/delete/', methods=['POST']) def delete_job(job_id): job = store.get_job(job_id) if job is None: flash('Job not found.', 'error') return redirect(url_for('index')) deleted = store.delete_job(job_id) if deleted: # Best-effort file cleanup try: (_cfg.UPLOAD_FOLDER / job['image_file']).unlink(missing_ok=True) (_cfg.UPLOAD_FOLDER / job['audio_file']).unlink(missing_ok=True) if job['output_file']: (_cfg.OUTPUT_FOLDER / job['output_file']).unlink(missing_ok=True) except Exception: pass flash(f'Job #{job_id} deleted.', 'success') else: flash('Job could not be deleted.', 'error') return redirect(url_for('index')) @app.route('/retry/', methods=['POST']) def retry_job(job_id): job = store.get_job(job_id) if job is None: flash('Job not found.', 'error') return redirect(url_for('index')) if job['status'] != 'error': flash('Only error jobs can be retried.', 'error') return redirect(url_for('index')) store.move_to_end(job_id) flash(f'Job #{job_id} moved to the end of the queue for retry.', 'success') return redirect(url_for('index')) @app.route('/download/') def download(job_id): job = store.get_job(job_id) if job is None or job['status'] != 'complete' or not job['output_file']: abort(404) return send_from_directory(_cfg.OUTPUT_FOLDER, job['output_file'], as_attachment=True) @app.route('/admin/login', methods=['GET', 'POST']) def admin_login(): if not _cfg.ADMIN_PASSWORD: return redirect(url_for('index')) if request.method == 'POST': if request.form.get('password') == _cfg.ADMIN_PASSWORD: return redirect(url_for('index', password=request.form.get('password'))) flash('Incorrect password.', 'error') return render_template('admin/login.html') @app.route('/health') def health(): return jsonify({ 'ok': True, 'version': _cfg.VERSION, 'queue': { row['status']: sum(1 for r in store.list_jobs() if r['status'] == row['status']) for row in store.list_jobs() } or {} }) if __name__ == '__main__': app.run(host='0.0.0.0', port=_cfg.INTERNAL_PORT, debug=True)