Multi-upload batch queue + Telegram notifications

Changes:

1. Frontend multi-upload:
   - File input now has 'multiple' attribute, drag-drop accepts multiple
   - File queue list with per-file artist/title preview + remove button
   - 'Pošlji vse' uploads sequentially (one at a time to avoid network saturation)
   - Each file gets same batch_id for Telegram batch summary
   - After upload, queue clears, jobs appear in right sidebar

2. Backend queue worker:
   - New _queue_worker() background thread processes 'queued' jobs sequentially
   - Only 1 job at a time to keep openclaw stable (avoid CPU/RAM thrash)
   - FIFO order by created_at
   - Auto-starts on app startup after job resume

3. Job submission flow change:
   - /api/process and /api/youtube no longer call background.add_task directly
   - Just mark status='queued', queue worker picks up
   - This means upload completes fast, processing happens in background
   - User can close browser, jobs continue

4. Telegram notifications (FOLX Alerts bot):
   - Per-job: 'Reel pripravljen: Lady Gaga - Abracadabra (29s, 30 MB)'
   - Per-job failed: 'Reel ni uspel: <name> + error message'
   - Batch summary: 'Batch končan: 10/10 reels pripravljeni' (only if >1 in batch)
   - Uses existing TELEGRAM_TOKEN + TELEGRAM_CHAT_ID env vars
   - app/telegram.py module with notify_job_done(), notify_job_failed(),
     notify_batch_complete()

5. batch_id field:
   - Added to Job model + StartJobIn pydantic
   - Saved during upload + process
   - Used to count batch progress and trigger summary notification

User experience:
- Drag 20 videos at once
- Click 'Pošlji'
- Close browser, go grab coffee
- Telegram sends 'Reel pripravljen' for each
- After all done: 'Batch končan: 20/20 reels pripravljeni' summary
- Open app to download all
This commit is contained in:
2026-04-29 15:12:38 +00:00
parent 157e6b781e
commit 91cc03658d
4 changed files with 426 additions and 93 deletions
+132 -2
View File
@@ -511,14 +511,62 @@ def process_job(job_id):
output_path=str(output_path),
output_size_mb=round(output_path.stat().st_size / 1024 / 1024, 2),
)
# Telegram obvestilo
try:
from app.telegram import notify_job_done
final_job = load_job(job_id)
notify_job_done(final_job)
except Exception as e:
print(f"⚠️ TG notify_job_done failed: {e}", flush=True)
# Batch tracking — če je zadnji v batchu, pošlji summary
_try_finalize_batch(job_id)
else:
update_job(
job_id,
status="failed",
error="Output datoteka ne obstaja po obdelavi",
)
try:
from app.telegram import notify_job_failed
notify_job_failed(load_job(job_id), "Output datoteka ne obstaja")
except Exception:
pass
_try_finalize_batch(job_id)
except Exception as e:
update_job(job_id, status="failed", error=str(e))
try:
from app.telegram import notify_job_failed
notify_job_failed(load_job(job_id), str(e))
except Exception:
pass
_try_finalize_batch(job_id)
def _try_finalize_batch(job_id):
"""Če je job del batch-a, preveri če so vsi zaključili in pošlji summary."""
try:
job = load_job(job_id)
batch_id = job.get("batch_id")
if not batch_id:
return
# Preštej batch jobe
all_jobs = [load_job(jp.stem) for jp in JOBS_DIR.glob("*.json")]
batch_jobs = [j for j in all_jobs if j and j.get("batch_id") == batch_id]
unfinished = [j for j in batch_jobs if j.get("status") not in ("done", "failed")]
if unfinished:
return # še niso vsi končani
# Vsi so končani — pošlji summary
total = len(batch_jobs)
succeeded = len([j for j in batch_jobs if j.get("status") == "done"])
failed = total - succeeded
try:
from app.telegram import notify_batch_complete
notify_batch_complete(batch_id, total, succeeded, failed)
except Exception as e:
print(f"⚠️ TG batch summary failed: {e}", flush=True)
except Exception as e:
print(f"⚠️ _try_finalize_batch error: {e}", flush=True)
# ────────────────────────────────────────────────────────────────
@@ -528,6 +576,77 @@ app = FastAPI(title="Reels Clipper")
app.mount("/static", StaticFiles(directory=Path(__file__).parent.parent / "static"), name="static")
# ────────────────────────────────────────────────────────────────
# Queue worker — procesira queued jobe enega za drugim
# ────────────────────────────────────────────────────────────────
import threading
_queue_lock = threading.Lock()
_queue_running = {"current_job": None} # tracked v dict da je accessible iz threadov
def _queue_worker():
"""Background thread ki preverja queued jobe in jih obdeluje 1 po 1.
Teče forever, sleep 3s med iteracijami če ni dela.
"""
print("🚜 Queue worker zagnan", flush=True)
while True:
try:
# Že nekaj v obdelavi? Počakaj.
if _queue_running["current_job"]:
# Preverim ali je status še processing
cur = load_job(_queue_running["current_job"])
if cur and cur.get("status") in ("processing", "downloading"):
time.sleep(3)
continue
# Ni več v obdelavi → sprosti
_queue_running["current_job"] = None
# Najdi naslednjega "queued" joba (FIFO po created_at)
queued_jobs = []
for f in JOBS_DIR.glob("*.json"):
try:
j = json.loads(f.read_text())
if j.get("status") == "queued":
queued_jobs.append(j)
except Exception:
continue
if not queued_jobs:
time.sleep(3)
continue
queued_jobs.sort(key=lambda x: x.get("created_at", 0))
next_job = queued_jobs[0]
job_id = next_job["id"]
# Mark "processing" + zaženi
with _queue_lock:
_queue_running["current_job"] = job_id
print(f"🚜 Queue worker: obdelujem {job_id}", flush=True)
try:
process_job(job_id)
except Exception as e:
print(f"❌ Queue worker error pri {job_id}: {e}", flush=True)
update_job(job_id, status="failed", error=f"Queue worker: {e}")
finally:
with _queue_lock:
_queue_running["current_job"] = None
except Exception as e:
print(f"❌ Queue worker outer error: {e}", flush=True)
time.sleep(5)
# Zaženi worker v ozadju (samo enkrat)
_worker_thread = None
def _start_queue_worker():
global _worker_thread
if _worker_thread is None or not _worker_thread.is_alive():
_worker_thread = threading.Thread(target=_queue_worker, daemon=True)
_worker_thread.start()
@app.on_event("startup")
async def resume_or_cleanup_jobs():
"""Ob startu containerja: avto-resume processing jobs ali jih označi kot error.
@@ -590,6 +709,9 @@ async def resume_or_cleanup_jobs():
print(f" ⚠️ Napaka pri {f.name}: {e}")
print(f" ✅ Resumed: {resumed_count}, Error: {error_count}")
# Zaženi queue worker za queued jobe (multi-upload batch)
_start_queue_worker()
async def _resume_job_async(job_id):
@@ -651,6 +773,8 @@ class StartJobIn(BaseModel):
llm_model: Optional[str] = None # specifičen model (privzeto najboljši za provider)
# STT provider (Scribe je 18x hitreje + boljša multilingual accuracy)
whisper_provider: str = "auto" # auto / elevenlabs / local
# Batch tracking za multi-upload (Telegram summary)
batch_id: Optional[str] = None
# ────────────────────────────────────────────────────────────────
@@ -661,6 +785,7 @@ async def upload_video(
file: UploadFile = File(...),
artist: Optional[str] = Form(None),
title: Optional[str] = Form(None),
batch_id: Optional[str] = Form(None),
user: str = Depends(check_auth),
):
if not file.filename:
@@ -692,6 +817,9 @@ async def upload_video(
"updated_at": time.time(),
}
if batch_id:
job["batch_id"] = batch_id
# Artist + title — najprej user-provided, potem parse iz filename
if artist and title:
# User je vpisal ali potrdil
@@ -740,7 +868,7 @@ async def submit_youtube(
"quality": payload.quality,
}
save_job(job)
background.add_task(process_job, job_id)
# Queue worker bo pograbil
return job
@@ -775,12 +903,14 @@ async def start_processing(
llm_provider=payload.llm_provider,
llm_model=payload.llm_model,
whisper_provider=payload.whisper_provider,
batch_id=payload.batch_id,
current_step="V vrsti za obdelavo",
# Počisti pretekle napake (retry-friendly)
chorus_error=None,
interrupted_at=None,
)
background.add_task(process_job, payload.job_id)
# Queue worker (background thread) bo pograbil ta job — ne zaganjamo neposredno.
# To pomeni, da se ob več upload-ih obdelujejo zaporedno (1 hkrati = stabilno).
return load_job(payload.job_id)
+119
View File
@@ -0,0 +1,119 @@
"""
telegram.py — Telegram bot helper za reels-app.
Pošilja obvestila o končanih jobih, batch summary, napakah.
Credentials se preberejo iz env vars (TELEGRAM_TOKEN, TELEGRAM_CHAT_ID).
"""
import os
import urllib.request
import urllib.parse
import json
def send_message(text, parse_mode="Markdown", disable_notification=False):
"""Pošlji sporočilo na Telegram. Vrne True ob uspehu, False ob napaki.
text: ne sme biti daljši od 4096 znakov.
parse_mode: 'Markdown' ali 'HTML' ali None.
disable_notification: True za tiho obvestilo (brez zvonca).
"""
token = os.environ.get("TELEGRAM_TOKEN")
chat_id = os.environ.get("TELEGRAM_CHAT_ID")
if not token or not chat_id:
return False
# Telegram limit: 4096 chars
if len(text) > 4090:
text = text[:4087] + "..."
data_dict = {
"chat_id": chat_id,
"text": text,
}
if parse_mode:
data_dict["parse_mode"] = parse_mode
if disable_notification:
data_dict["disable_notification"] = "true"
data = urllib.parse.urlencode(data_dict).encode()
req = urllib.request.Request(
f"https://api.telegram.org/bot{token}/sendMessage",
data=data,
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
result = json.loads(resp.read().decode())
return result.get("ok", False)
except Exception as e:
print(f"⚠️ Telegram send failed: {e}", flush=True)
return False
def notify_job_done(job, base_url="https://reels.biba.live"):
"""Obvesti o končanem reelu."""
artist = job.get("parsed_artist", "")
title = job.get("parsed_title", "")
job_id = job.get("id", "")
if artist and title:
name = f"*{artist} — {title}*"
elif title:
name = f"*{title}*"
else:
name = f"`{job_id}`"
duration = job.get("duration", 0)
output_size_mb = job.get("output_size_mb") or round((job.get("output_size", 0) / 1024 / 1024), 1)
text = (
f"✅ Reel pripravljen\n\n"
f"{name}\n"
f"⏱ {duration:.0f}s · 📦 {output_size_mb} MB\n\n"
f"[Predogled & Download]({base_url})"
)
return send_message(text)
def notify_job_failed(job, error_msg=""):
"""Obvesti o neuspehu joba."""
artist = job.get("parsed_artist", "")
title = job.get("parsed_title", "")
fname = job.get("filename", "")
if artist and title:
name = f"*{artist} — {title}*"
elif fname:
name = f"`{fname}`"
else:
name = f"`{job.get('id', '?')}`"
err_short = (error_msg or "neznana napaka")[:200]
text = f"⚠️ Reel ni uspel\n\n{name}\n\n```\n{err_short}\n```"
return send_message(text)
def notify_batch_complete(batch_id, total, succeeded, failed, base_url="https://reels.biba.live"):
"""Obvesti o končanem batch-u (več jobov hkrati)."""
if total <= 1:
return False # ne pošiljaj batch summary za en sam job
if failed == 0:
emoji = "🎉"
head = f"Batch končan: vsi {total} reelov pripravljeni"
else:
emoji = "✅"
head = f"Batch končan: {succeeded}/{total} uspelih"
if failed > 0:
head += f", {failed} neuspelih"
text = f"{emoji} {head}\n\n[Vsi reels →]({base_url})"
return send_message(text)
def notify_batch_started(batch_id, total):
"""Obvesti o začetku batch-a (samo če > 1 job)."""
if total <= 1:
return False
text = f"🎬 Batch za {total} datotek dodan v vrsto…"
return send_message(text, disable_notification=True)