Hans-R-D/encoder-pipeline

1from fastapi import FastAPI, UploadFile, HTTPException, status2from fastapi.responses import JSONResponse, FileResponse3from pathlib import Path4import uuid5import os6import hmac7import hashlib8from datetime import datetime9from rq import Queue10from redis import from_url11from ..config import EncodingConfig, RedisConfig12from video_encoder.worker.tasks import encode_video_task, init_job13import logging14 15logger = logging.getLogger(__name__)16 17q = Queue(connection=RedisConfig.get_connection(), default_timeout=3600)18 19app = FastAPI(title="Video Encoding Service")20 21@app.on_event("startup")22async def startup_event():23    # Initialize Redis connection on startup24    RedisConfig.get_connection()25    logger.info("Application initialization complete")26 27@app.get("/")28async def health_check():29    return {"status": "ok", "redis": "connected"}30 31@app.post("/upload")32async def upload_video(file: UploadFile):33    job_id = str(uuid.uuid4())34    temp_dir = os.path.join(EncodingConfig.TEMP_DIR, job_id)35    os.makedirs(temp_dir, exist_ok=True)36 37    input_path = os.path.join(temp_dir, file.filename)38    with open(input_path, "wb") as f:39        f.write(await file.read())40 41    logger.info(f"Enqueuing job {job_id} for file {file.filename}")42    init_job(job_id)43    q.enqueue(encode_video_task, job_id, input_path)44 45    return JSONResponse(46        content={"job_id": job_id},47        status_code=status.HTTP_202_ACCEPTED48    )49 50@app.get("/status/{job_id}")51async def get_status(job_id: str):52    redis = RedisConfig.get_connection()53    progress = redis.hget(f"job:{job_id}", "progress")54 55    if not progress:56        logger.warning(f"Job {job_id} not found")57        raise HTTPException(status_code=404, detail="Job not found")58 59    return {"job_id": job_id, "progress": float(progress)}60 61@app.get("/play/{job_id}")62async def play_video(job_id: str, token: str):63    expected_token = hmac.new(64        EncodingConfig.HMAC_SECRET.encode(),65        job_id.encode(),66        hashlib.sha25667    ).hexdigest()68 69    if not hmac.compare_digest(token, expected_token):70        logger.warning(f"Invalid token for job {job_id}")71        raise HTTPException(status_code=403, detail="Invalid token")72 73    master_playlist = os.path.join(EncodingConfig.TEMP_DIR, job_id, f"{job_id}_master.m3u8")74    if not os.path.exists(master_playlist):75        logger.warning(f"Playlist not found for job {job_id}")76        raise HTTPException(status_code=404, detail="Playlist not found")77 78    return FileResponse(79        master_playlist,80        media_type="application/vnd.apple.mpegurl"81    )82 83@app.get("/token/{job_id}")84async def generate_token(job_id: str):85    timestamp = datetime.now().isoformat()86    signature = hmac.new(87        EncodingConfig.HMAC_SECRET.encode(),88        f"{job_id}{timestamp}".encode(),89        hashlib.sha25690    ).hexdigest()91 92    logger.info(f"Generated token for job {job_id}")93    return {"token": f"{timestamp}:{signature}"}94