dionysis-media-video/app.py

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/<int:job_id>', 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'))
    if job['status'] == 'processing':
        flash('Cannot delete a job that is currently processing.', '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/<int:job_id>', 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/<int:job_id>')
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)