feat: backend API routes servisleri ve tests için commend data processing dreams ve remix features eklendi

app/routes/commend/router.py

@@ -119,7 +119,9 @@ class CommendHealthResponse(BaseModel):
 
 _COMMEND_API_KEY = os.getenv("COMMEND_API_KEY")
 _COMMEND_POSTING_ENABLED = os.getenv("COMMEND_ENABLE_POSTING", "false").lower() == "true"
-_COMMEND_REQUIRE_AUTH = os.getenv("COMMEND_REQUIRE_AUTH", "true").lower() != "false"
+# Drafts are public (rate limited per visitor); set COMMEND_REQUIRE_AUTH=true to lock
+# them behind COMMEND_API_KEY too. Posting always needs the key, see _require_posting_key.
+_COMMEND_REQUIRE_AUTH = os.getenv("COMMEND_REQUIRE_AUTH", "false").lower() == "true"
 
 # Simple in-memory rate limiter: IP -> list of timestamps
 _rate_limit_store: dict[str, list[float]] = defaultdict(list)
@@ -232,7 +234,7 @@ async def commend_health_check():
 
 
 @router.post("/video-details", response_model=VideoDetailsResponse, dependencies=[Depends(_rate_limit_only)])
-async def get_video_info(request: GenerateCommentRequest):
+def get_video_info(request: GenerateCommentRequest):
     """
     Fetch video details without generating a comment.
     Useful for preview before generation.
@@ -256,7 +258,7 @@ async def get_video_info(request: GenerateCommentRequest):
 
 
 @router.post("/generate", response_model=GenerateCommentResponse, dependencies=[Depends(_require_api_key)])
-async def generate_youtube_comment(request: GenerateCommentRequest):
+def generate_youtube_comment(request: GenerateCommentRequest):
     """
     Generate an AI-powered comment for a YouTube video.
     Optimized for minimal API calls and token usage.
@@ -317,7 +319,7 @@ async def generate_youtube_comment(request: GenerateCommentRequest):
 
 
 @router.post("/post", response_model=PostCommentResponse, dependencies=[Depends(_require_posting_key)])
-async def post_comment_to_youtube(request: PostCommentRequest):
+def post_comment_to_youtube(request: PostCommentRequest):
     """
     Post a comment to YouTube.
 

app/routes/data_processing.py

@@ -6,7 +6,7 @@ import time
 import re
 from collections import defaultdict
 from fastapi import APIRouter, File, UploadFile, Form, HTTPException, Request, Depends, status
-from fastapi.responses import StreamingResponse
+from fastapi.responses import Response
 from pydantic import ValidationError
 import json
 import logging
@@ -15,6 +15,7 @@ from app.schemas import AudioAugmentationOptions, AudioConvertOptions, DatasetEn
 from app.services.audio_processor import process_audio, convert_audio_format
 from app.services.dataset_organizer import analyze_for_organization
 from app.services.client_ip import client_ip as _client_ip
+from app.services.tool_gate import run_blocking
 
 router = APIRouter(prefix="/api/process", tags=["Data Processing"])
 logger = logging.getLogger(__name__)
@@ -111,12 +112,13 @@ async def process_audio_endpoint(
         mix_content = await _read_capped(mix_file, MAX_PAYLOAD_BYTES) if mix_file is not None else None
 
         # Process audio
-        processed_audio = process_audio(content, parsed_options, mix_with_bytes=mix_content)
+        processed_audio = await run_blocking(process_audio, content, parsed_options, mix_with_bytes=mix_content)
 
-        # Return as downloadable file
+        # Return as downloadable file. Sent as one body: iterating a BytesIO
+        # yields tiny "lines" at every 0x0A byte, which crawls over the proxy.
         output_filename = f"processed_{safe_filename}.wav"
-        return StreamingResponse(
-            processed_audio,
+        return Response(
+            content=processed_audio.getvalue(),
             media_type="audio/wav",
             headers={"Content-Disposition": f"attachment; filename={output_filename}"}
         )
@@ -186,12 +188,12 @@ async def convert_audio_endpoint(
             chunks.append(chunk)
         content = b"".join(chunks)
 
-        converted = convert_audio_format(content, parsed_options)
+        converted = await run_blocking(convert_audio_format, content, parsed_options)
 
         target = parsed_options.target_format
         output_filename = f"{safe_filename}.{target}"
-        return StreamingResponse(
-            converted,
+        return Response(
+            content=converted.getvalue(),
             media_type=_CONVERT_MEDIA_TYPES[target],
             headers={"Content-Disposition": f"attachment; filename={output_filename}"}
         )
@@ -234,7 +236,7 @@ async def organize_audio_endpoint(file: UploadFile = File(...)):
             chunks.append(chunk)
         content = b"".join(chunks)
 
-        metadata = analyze_for_organization(content)
+        metadata = await run_blocking(analyze_for_organization, content)
         return DatasetEntryMetadata(**metadata.to_dict())
 
     except ValueError as e:

app/routes/dreams.py

@@ -8,6 +8,7 @@ import time
 from collections import defaultdict
 
 from fastapi import APIRouter, Depends, HTTPException, Request, status
+from starlette.concurrency import run_in_threadpool
 
 from app.schemas import DreamAnalysisRequest, DreamAnalysisResponse
 from app.services.dream_analyzer import analyze_dream
@@ -54,13 +55,16 @@ async def analyze_dream_endpoint(payload: DreamAnalysisRequest) -> DreamAnalysis
     """
     logger.info(f"Dream analysis request ({len(payload.dream_text)} chars, lang={payload.language})")
 
-    result, error = analyze_dream(payload.dream_text, payload.language)
+    result, error = await run_in_threadpool(analyze_dream, payload.dream_text, payload.language)
 
     if error is not None:
         if "too short" in error.lower():
             raise HTTPException(status_code=400, detail={"code": "dream_too_short", "message": error})
-        if "api key" in error.lower():
+        if "not configured" in error.lower():
             raise HTTPException(status_code=503, detail={"code": "service_not_configured", "message": "Dream analysis is not configured."})
+        if "api key" in error.lower():
+            # A key is set but Gemini refuses it: the owner has to replace it.
+            raise HTTPException(status_code=503, detail={"code": "service_unavailable", "message": "Dream analysis is unavailable right now."})
         if "quota" in error.lower():
             raise HTTPException(status_code=503, detail={"code": "service_unavailable", "message": error})
         raise HTTPException(status_code=502, detail={"code": "analysis_failed", "message": error})

app/routes/remix.py

@@ -8,13 +8,14 @@ import time
 from collections import defaultdict
 
 from fastapi import APIRouter, Depends, File, Form, HTTPException, Request, UploadFile, status
-from fastapi.responses import StreamingResponse
+from fastapi.responses import Response
 from pydantic import ValidationError
 
 from app.schemas import MultitrackAnalysis, MultitrackOptions, RemixAnalysis, RemixOptions
 from app.services.multitrack_mixer import MAX_CHANNELS, MIN_CHANNELS, mix_tracks
 from app.services.remix_engine import build_remix
 from app.services.client_ip import client_ip as _client_ip
+from app.services.tool_gate import run_blocking
 
 router = APIRouter(prefix="/api/remix", tags=["Creator Studio"])
 logger = logging.getLogger(__name__)
@@ -101,15 +102,15 @@ async def remix_blend_endpoint(
         content_a = await _read_capped(track_a, MAX_PAYLOAD_BYTES)
         content_b = await _read_capped(track_b, MAX_PAYLOAD_BYTES)
 
-        result = build_remix(content_a, content_b, parsed_options)
+        result = await run_blocking(build_remix, content_a, content_b, parsed_options)
 
         # Serialize via the Pydantic model so the header uses the same
         # camelCase field names the frontend's RemixAnalysis type expects —
         # result.analysis itself is a plain snake_case dict.
         analysis_model = RemixAnalysis(**result.analysis)
         analysis_json = analysis_model.model_dump_json(by_alias=True)
-        return StreamingResponse(
-            result.audio,
+        return Response(
+            content=result.audio.getvalue(),
             media_type="audio/wav",
             headers={
                 "Content-Disposition": "attachment; filename=crowncode-remix.wav",
@@ -167,13 +168,13 @@ async def multitrack_mix_endpoint(
     try:
         contents = [await _read_capped(f, MAX_PAYLOAD_BYTES) for f in tracks]
 
-        result = mix_tracks(contents, parsed_options)
+        result = await run_blocking(mix_tracks, contents, parsed_options)
 
         # Same camelCase-via-Pydantic serialization as the /blend endpoint.
         analysis_model = MultitrackAnalysis(**result.analysis)
         analysis_json = analysis_model.model_dump_json(by_alias=True)
-        return StreamingResponse(
-            result.audio,
+        return Response(
+            content=result.audio.getvalue(),
             media_type="audio/wav",
             headers={
                 "Content-Disposition": "attachment; filename=crowncode-multitrack-mix.wav",

app/services/dream_analyzer.py

@@ -27,6 +27,10 @@ _client = None
 _MAX_DREAM_CHARS = 4000
 _MIN_DREAM_CHARS = 10
 
+# Tried in order; a model Google has retired falls through to the next one.
+_MODELS = [m for m in (os.getenv("DREAMS_GEMINI_MODEL"), "gemini-2.5-flash-lite", "gemini-2.5-flash") if m]
+_MODEL_GONE = ("not found", "no longer available", "deprecated", "is not supported", "404")
+
 
 def _get_client():
     global _client
@@ -35,7 +39,7 @@ def _get_client():
 
     api_key = os.getenv("DREAMS_GEMINI_API_KEY") or os.getenv("COMMEND_GEMINI_API_KEY")
     if not api_key:
-        raise ValueError("DREAMS_GEMINI_API_KEY (or COMMEND_GEMINI_API_KEY) required")
+        raise ValueError("Dream analysis is not configured: set DREAMS_GEMINI_API_KEY or COMMEND_GEMINI_API_KEY")
 
     try:
         _client = genai.Client(api_key=api_key)
@@ -97,10 +101,19 @@ Dream description:
 
 JSON:"""
 
-        response = client.models.generate_content(
-            model="gemini-2.5-flash-lite",
-            contents=prompt,
-        )
+        response = None
+        last_error: Optional[Exception] = None
+        for model in _MODELS:
+            try:
+                response = client.models.generate_content(model=model, contents=prompt)
+                break
+            except Exception as e:  # noqa: BLE001 - classified below
+                last_error = e
+                if not any(marker in str(e).lower() for marker in _MODEL_GONE):
+                    raise
+                logger.warning(f"Gemini model {model} unavailable, trying the next one: {e}")
+        if response is None:
+            raise last_error or RuntimeError("no Gemini model answered")
 
         raw = response.text.strip() if response.text else ""
         parsed = _extract_json(raw)
@@ -132,7 +145,10 @@ JSON:"""
         # from _get_client() reads "..._api_key required" (underscore), and
         # Gemini's own errors tend to read "API key not valid" (space), so
         # a plain substring check on either form alone misses the other.
+        if "not configured" in err:
+            return None, "Dream analysis is not configured"
         if "api key" in err or "api_key" in err or "apikey" in err:
+            logger.error("Gemini refused the configured API key; replace DREAMS_GEMINI_API_KEY / COMMEND_GEMINI_API_KEY")
             return None, "Invalid API key"
         if "quota" in err or "limit" in err or "resource" in err:
             return None, "API quota exceeded"

app/services/tool_gate.py

@@ -0,0 +1,62 @@
+"""Keeps the Creator Studio and ML Toolkit audio work off the event loop.
+
+Remixing, mixing, converting and tagging audio are CPU-bound librosa calls. Run
+inline in an async handler they freeze the whole server (health checks, AURIS
+polling, other downloads) until they finish. They go to a worker thread
+instead, and at most TOOLS_CPU_SLOTS of them run at once so two visitors'
+jobs don't starve each other and the AURIS analysis on the same CPU.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from contextlib import asynccontextmanager
+from typing import Any, AsyncIterator, Callable, Optional, TypeVar
+
+from fastapi import HTTPException, status
+from starlette.concurrency import run_in_threadpool
+
+T = TypeVar("T")
+
+TOOL_SLOTS = max(1, int(os.getenv("TOOLS_CPU_SLOTS", "1")))
+# Jobs allowed to wait for a slot; past that the visitor is told to come back.
+MAX_WAITING = max(1, int(os.getenv("TOOLS_MAX_WAITING", "6")))
+
+_semaphore: Optional[asyncio.Semaphore] = None
+_semaphore_loop: Optional[asyncio.AbstractEventLoop] = None
+_waiting = 0
+
+
+def waiting() -> int:
+    return _waiting
+
+
+@asynccontextmanager
+async def tool_slot() -> AsyncIterator[None]:
+    global _semaphore, _semaphore_loop, _waiting
+    loop = asyncio.get_running_loop()
+    if _semaphore is None or _semaphore_loop is not loop:
+        # One per event loop: a semaphore must not be shared across loops (tests start several).
+        _semaphore, _semaphore_loop, _waiting = asyncio.Semaphore(TOOL_SLOTS), loop, 0
+    if _waiting >= MAX_WAITING:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail={"code": "server_busy", "message": "Too many audio jobs are waiting. Try again in a minute."},
+            headers={"Retry-After": "30"},
+        )
+    _waiting += 1
+    try:
+        await _semaphore.acquire()
+    finally:
+        _waiting -= 1
+    try:
+        yield
+    finally:
+        _semaphore.release()
+
+
+async def run_blocking(fn: Callable[..., T], *args: Any, **kwargs: Any) -> T:
+    """Run a CPU-bound function in a worker thread, waiting for a free slot first."""
+    async with tool_slot():
+        return await run_in_threadpool(fn, *args, **kwargs)

tests/test_tool_gate.py

@@ -0,0 +1,182 @@
+"""Audio tools run off the event loop, one at a time, and the dream errors reach the page unchanged."""
+
+from __future__ import annotations
+
+import asyncio
+import io
+from collections import defaultdict
+import threading
+import time
+
+import numpy as np
+import pytest
+import soundfile as sf
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+
+from app.routes import dreams as dreams_module
+from app.routes.data_processing import router as process_router
+from app.routes.dreams import router as dreams_router
+from app.routes.remix import router as remix_router
+from app.services import tool_gate
+
+
+@pytest.fixture(autouse=True)
+def fresh_gate(monkeypatch):
+    monkeypatch.setattr(tool_gate, "_semaphore", None)
+    monkeypatch.setattr(tool_gate, "_semaphore_loop", None)
+    monkeypatch.setattr(tool_gate, "_waiting", 0)
+
+
+def wav(seconds: float = 2.0, freq: float = 220.0, sr: int = 22050) -> bytes:
+    t = np.arange(int(seconds * sr)) / sr
+    buf = io.BytesIO()
+    sf.write(buf, (0.3 * np.sin(2 * np.pi * freq * t)).astype(np.float32), sr, format="WAV")
+    return buf.getvalue()
+
+
+# ── the gate ─────────────────────────────────────────────────────────
+
+def test_work_runs_in_a_worker_thread_not_on_the_loop() -> None:
+    async def scenario():
+        loop_thread = threading.get_ident()
+        return loop_thread, await tool_gate.run_blocking(threading.get_ident)
+
+    loop_thread, worker_thread = asyncio.run(scenario())
+    assert worker_thread != loop_thread
+
+
+def test_the_loop_stays_responsive_while_a_job_runs() -> None:
+    async def scenario():
+        ticks = 0
+
+        async def heartbeat():
+            nonlocal ticks
+            while True:
+                await asyncio.sleep(0.02)
+                ticks += 1
+
+        beat = asyncio.create_task(heartbeat())
+        await tool_gate.run_blocking(time.sleep, 0.3)
+        beat.cancel()
+        return ticks
+
+    assert asyncio.run(scenario()) >= 5
+
+
+def test_one_job_at_a_time_by_default(monkeypatch) -> None:
+    monkeypatch.setattr(tool_gate, "TOOL_SLOTS", 1)
+    running = {"now": 0, "peak": 0}
+    lock = threading.Lock()
+
+    def job():
+        with lock:
+            running["now"] += 1
+            running["peak"] = max(running["peak"], running["now"])
+        time.sleep(0.1)
+        with lock:
+            running["now"] -= 1
+
+    async def scenario():
+        await asyncio.gather(*(tool_gate.run_blocking(job) for _ in range(3)))
+
+    asyncio.run(scenario())
+    assert running["peak"] == 1
+
+
+def test_a_full_queue_says_busy_instead_of_piling_up(monkeypatch) -> None:
+    monkeypatch.setattr(tool_gate, "MAX_WAITING", 2)
+
+    async def scenario():
+        # One job holds the slot and two wait for it; a fourth has nowhere to go.
+        jobs = [asyncio.create_task(tool_gate.run_blocking(time.sleep, 0.3)) for _ in range(3)]
+        await asyncio.sleep(0.05)
+        with pytest.raises(HTTPException) as busy:
+            await tool_gate.run_blocking(time.sleep, 0)
+        await asyncio.gather(*jobs)
+        return busy.value
+
+    err = asyncio.run(scenario())
+    assert err.status_code == 503
+    assert err.detail["code"] == "server_busy"
+    assert err.headers["Retry-After"] == "30"
+
+
+# ── routes ───────────────────────────────────────────────────────────
+
+@pytest.fixture
+def tools_client() -> TestClient:
+    app = FastAPI()
+    app.include_router(process_router)
+    app.include_router(remix_router)
+    return TestClient(app)
+
+
+def test_convert_returns_the_file_in_one_body(tools_client) -> None:
+    response = tools_client.post(
+        "/api/process/audio/convert",
+        files={"file": ("tone.wav", wav(), "audio/wav")},
+        data={"options": '{"targetFormat": "flac"}'},
+    )
+    assert response.status_code == 200
+    assert response.headers["content-type"] == "audio/flac"
+    assert int(response.headers["content-length"]) == len(response.content) > 1000
+
+
+def test_organize_reads_the_clip(tools_client) -> None:
+    response = tools_client.post("/api/process/audio/organize", files={"file": ("tone.wav", wav(), "audio/wav")})
+    assert response.status_code == 200
+    assert response.json()["durationSec"] == pytest.approx(2.0, abs=0.1)
+
+
+def test_blend_returns_audio_and_its_analysis_header(tools_client) -> None:
+    response = tools_client.post(
+        "/api/remix/blend",
+        files={"track_a": ("a.wav", wav(freq=220), "audio/wav"), "track_b": ("b.wav", wav(freq=330), "audio/wav")},
+        data={"options": "{}"},
+    )
+    assert response.status_code == 200
+    assert response.content[:4] == b"RIFF"
+    assert int(response.headers["content-length"]) == len(response.content)
+    assert "outputDurationSec" in response.headers["x-remix-analysis"]
+
+
+def test_multitrack_needs_two_tracks(tools_client) -> None:
+    response = tools_client.post("/api/remix/multitrack", files=[("tracks", ("a.wav", wav(), "audio/wav"))], data={"options": "{}"})
+    assert response.status_code == 400
+
+
+# ── dreams ───────────────────────────────────────────────────────────
+
+@pytest.fixture
+def dreams_client(monkeypatch) -> TestClient:
+    monkeypatch.setattr(dreams_module, "_rate_store", defaultdict(list))
+    app = FastAPI()
+    app.include_router(dreams_router)
+    return TestClient(app)
+
+
+def post_dream(client: TestClient):
+    return client.post("/api/dreams/analyze", json={"dreamText": "a dream that is long enough", "language": "English"})
+
+
+@pytest.mark.parametrize("error,status,code", [
+    ("Dream analysis is not configured", 503, "service_not_configured"),
+    ("Invalid API key", 503, "service_unavailable"),
+    ("API quota exceeded", 503, "service_unavailable"),
+    ("Analysis produced an unexpected response format.", 502, "analysis_failed"),
+    ("Dream description is too short to analyze.", 400, "dream_too_short"),
+])
+def test_dream_errors_map_to_distinct_codes(dreams_client, monkeypatch, error, status, code) -> None:
+    monkeypatch.setattr(dreams_module, "analyze_dream", lambda text, language: (None, error))
+    response = post_dream(dreams_client)
+    assert response.status_code == status
+    assert response.json()["detail"]["code"] == code
+
+
+def test_a_good_reading_comes_back(dreams_client, monkeypatch) -> None:
+    reading = {"emotions": ["fear"], "themes": ["falling"], "symbols": ["moon"], "interpretation": "A reading.", "lucidityIndicator": False}
+    monkeypatch.setattr(dreams_module, "analyze_dream", lambda text, language: (reading, None))
+    response = post_dream(dreams_client)
+    assert response.status_code == 200
+    assert response.json()["interpretation"] == "A reading."