feat: analysis schemas XAI inference YouTube downloader ve analysis job servisleri eklendi

app/routes/analyze_schemas.py

@@ -52,17 +52,24 @@ class AudioInfo(BaseModel):
 
 
 class ConfidenceBandModel(BaseModel):
+    """How far the probability sits from the decision threshold.
+
+    A qualitative tier only — there is no bootstrap behind it, so no
+    interval bounds are reported.
+    """
+
     tier: str
     labelTr: str
     labelEn: str
-    lowerBound: float
-    upperBound: float
+    margin: float = 0.0
 
 
 class ModelVoteModel(BaseModel):
     name: str
     probability: float
     vote: str
+    # False when the model file failed to load; probability is then a 0.5 placeholder.
+    available: bool = True
 
 
 class FeatureContributionModel(BaseModel):
@@ -131,6 +138,9 @@ class AnalysisResult(BaseModel):
     xai: Optional[XAIExplanation] = None
     towerScores: Dict[str, float] = Field(default_factory=dict)
     metaClassifier: Optional[MetaClassifierExplanation] = None
+    # Per-layer status: which model actually ran and how (e.g. CLAP embedding
+    # vs. its spectral heuristic fallback), so the UI can label each score.
+    layers: Dict[str, Dict[str, Any]] = Field(default_factory=dict)
 
 
 class AnalyzeResponse(BaseModel):

app/services/analysis_jobs.py

@@ -0,0 +1,135 @@
+"""
+Background analysis jobs with real, per-step progress.
+
+POST /api/analyze/jobs starts a job and returns its id; the client polls
+GET /api/analyze/jobs/{id}. Each pipeline step reports when it starts and
+finishes, so the waiting screen shows what the server is actually doing
+instead of a simulated progress bar. A job also survives the client
+leaving: a reloaded page can resume polling the same id.
+
+State lives in process memory — the Space runs one replica, and a job
+only needs to outlive the request that started it, not a restart.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+import uuid
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable, Dict, List, Optional
+
+JOB_TTL_SEC = 15 * 60
+MAX_JOBS = 200
+
+STEP_PENDING = "pending"
+STEP_RUNNING = "running"
+STEP_DONE = "done"
+STEP_FAILED = "failed"
+STEP_SKIPPED = "skipped"
+
+FILE_STEPS = ["features", "vocals", "wav2vec2", "clap", "fst", "xai", "meta"]
+URL_STEPS = ["download", *FILE_STEPS]
+
+
+class Progress:
+    """Ordered step states for one analysis. A no-op sink is fine too."""
+
+    def __init__(self, steps: List[str]) -> None:
+        self._steps: Dict[str, str] = {s: STEP_PENDING for s in steps}
+        self._started: Dict[str, float] = {}
+        self._took: Dict[str, float] = {}
+
+    def set(self, step: str, state: str) -> None:
+        if step not in self._steps:
+            return
+        now = time.monotonic()
+        if state == STEP_RUNNING:
+            self._started.setdefault(step, now)
+        elif step in self._started and step not in self._took:
+            self._took[step] = round(now - self._started[step], 2)
+        self._steps[step] = state
+
+    def running(self, *steps: str) -> None:
+        for s in steps:
+            self.set(s, STEP_RUNNING)
+
+    def snapshot(self) -> List[Dict[str, Any]]:
+        out: List[Dict[str, Any]] = []
+        for step, state in self._steps.items():
+            item: Dict[str, Any] = {"id": step, "state": state}
+            if step in self._took:
+                item["seconds"] = self._took[step]
+            out.append(item)
+        return out
+
+
+@dataclass
+class Job:
+    id: str
+    source_type: str
+    progress: Progress
+    created: float = field(default_factory=time.time)
+    updated: float = field(default_factory=time.time)
+    status: str = "running"          # running | done | error
+    response: Optional[Dict[str, Any]] = None
+    task: Optional[asyncio.Task] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "jobId": self.id,
+            "sourceType": self.source_type,
+            "status": self.status,
+            "steps": self.progress.snapshot(),
+            "elapsedSec": round((self.updated if self.status != "running" else time.time()) - self.created, 2),
+            "response": self.response,
+        }
+
+
+class JobStore:
+    def __init__(self) -> None:
+        self._jobs: Dict[str, Job] = {}
+
+    def _prune(self) -> None:
+        cutoff = time.time() - JOB_TTL_SEC
+        for job_id in [j.id for j in self._jobs.values() if j.status != "running" and j.updated < cutoff]:
+            del self._jobs[job_id]
+        # Hard cap: drop the oldest finished jobs first.
+        if len(self._jobs) > MAX_JOBS:
+            finished = sorted((j for j in self._jobs.values() if j.status != "running"), key=lambda j: j.updated)
+            for job in finished[: len(self._jobs) - MAX_JOBS]:
+                del self._jobs[job.id]
+
+    def running_count(self) -> int:
+        return sum(1 for j in self._jobs.values() if j.status == "running")
+
+    def start(
+        self,
+        source_type: str,
+        steps: List[str],
+        work: Callable[[Progress], Awaitable[Dict[str, Any]]],
+    ) -> Job:
+        self._prune()
+        job = Job(id=uuid.uuid4().hex, source_type=source_type, progress=Progress(steps))
+        self._jobs[job.id] = job
+
+        async def runner() -> None:
+            try:
+                job.response = await work(job.progress)
+                job.status = "error" if job.response.get("errors") and not job.response.get("result") else "done"
+            except Exception:  # noqa: BLE001 — the job must always settle
+                job.response = {"result": None, "warnings": [], "errors": ["internal_error"]}
+                job.status = "error"
+            finally:
+                job.updated = time.time()
+
+        # Keep a reference: asyncio only holds weak references to tasks.
+        job.task = asyncio.create_task(runner())
+        return job
+
+    def get(self, job_id: str) -> Optional[Job]:
+        self._prune()
+        return self._jobs.get(job_id)
+
+
+job_store = JobStore()

app/services/inference_xai.py

@@ -369,12 +369,11 @@ class FeatureContribution:
 
 @dataclass
 class ConfidenceBand:
-    """Human-readable confidence tier."""
+    """Human-readable confidence tier (qualitative, no interval)."""
     tier: str                   # "uncertain" | "likely" | "strong" | "very_strong"
     label_tr: str
     label_en: str
-    lower_bound: float          # bootstrap CI lower
-    upper_bound: float          # bootstrap CI upper
+    margin: float               # 0 at the threshold, 1 at the far end of the verdict's side
 
 
 @dataclass
@@ -383,6 +382,7 @@ class ModelVote:
     name: str                   # XGBoost / LightGBM / ...
     probability: float
     vote: str                   # "ai" | "human"
+    available: bool = True      # False: model failed to load, probability is a placeholder
 
 
 @dataclass
@@ -639,21 +639,24 @@ class XAIInferenceService:
         return m
 
     def _confidence_band(self, prob: float) -> ConfidenceBand:
-        """Map probability to human-readable confidence tier + CI."""
-        # Distance from 0.5 (decision boundary) determines confidence
-        dist = abs(prob - 0.5)
-        # Rough bootstrap CI — +/- 0.05 for very confident, +/- 0.1 for uncertain
-        ci_width = 0.05 + (0.10 - 0.05) * (1.0 - min(dist * 2, 1.0))
-        lower = max(0.0, prob - ci_width)
-        upper = min(1.0, prob + ci_width)
-
-        if dist < 0.10:
+        """Map probability to a qualitative tier.
+
+        Measured from the decision threshold (not 0.5): the margin is how far
+        the probability sits into its side of the threshold, 0 at the
+        threshold and 1 at 0.0 / 1.0. With a 0.5 threshold the cut points
+        match the old distance-from-0.5 tiers (0.1 / 0.25 / 0.4).
+        """
+        t = min(max(self.threshold, 1e-6), 1 - 1e-6)
+        margin = (prob - t) / (1.0 - t) if prob >= t else (t - prob) / t
+        margin = max(0.0, min(1.0, margin))
+
+        if margin < 0.2:
             tier = "uncertain"
             label_tr, label_en = "Belirsiz", "Uncertain"
-        elif dist < 0.25:
+        elif margin < 0.5:
             tier = "likely"
             label_tr, label_en = "Muhtemelen", "Likely"
-        elif dist < 0.40:
+        elif margin < 0.8:
             tier = "strong"
             label_tr, label_en = "Güçlü İşaret", "Strong"
         else:
@@ -664,8 +667,7 @@ class XAIInferenceService:
             tier=tier,
             label_tr=label_tr,
             label_en=label_en,
-            lower_bound=round(lower, 3),
-            upper_bound=round(upper, 3),
+            margin=round(margin, 3),
         )
 
     def _load_ensemble_models(self) -> None:
@@ -699,8 +701,9 @@ class XAIInferenceService:
     def _build_votes(self, x_scaled: "np.ndarray") -> List[ModelVote]:
         """Run real inference on all loaded ensemble models.
 
-        Falls back to training-result approximation for any model
-        that failed to load or raises at inference time.
+        A model that failed to load or raises at inference time still gets
+        a row, marked unavailable with a neutral 0.5 placeholder — never an
+        invented probability.
         """
         votes: List[ModelVote] = []
         best_name = self.training_results.get("_best_model", "LightGBM")
@@ -708,23 +711,16 @@ class XAIInferenceService:
         all_names = list({**_ML_MODEL_FILES, **_DL_MODEL_FILES}.keys())
         for name in all_names:
             model = self.ensemble_models.get(name)
+            prob: Optional[float] = None
             if model is not None:
                 try:
                     prob = float(model.predict_proba(x_scaled)[0, 1])
                 except Exception as e:
                     logger.warning(f"Inference failed for {name}: {e}")
-                    model = None
-
-            if model is None:
-                # Fallback: approximate from training accuracy
-                data = self.training_results.get(name, {})
-                acc = data.get("accuracy", 0.5) if isinstance(data, dict) else 0.5
-                # Use best model's actual prob as anchor
-                best_data = self.training_results.get(best_name, {})
-                best_acc = best_data.get("accuracy", 0.8) if isinstance(best_data, dict) else 0.8
-                # Scale approximation relative to best model's training accuracy
-                ratio = acc / best_acc if best_acc > 0 else 1.0
-                prob = round(max(0.03, min(0.97, 0.5 + (x_scaled.flatten()[0] * 0.0 + 0.5 - 0.5) * ratio)), 3)
+
+            if prob is None:
+                votes.append(ModelVote(name=name, probability=0.5, vote="human", available=False))
+                continue
 
             threshold = self.threshold if name == best_name else 0.5
             votes.append(ModelVote(
@@ -733,7 +729,8 @@ class XAIInferenceService:
                 vote="ai" if prob >= threshold else "human",
             ))
 
-        return sorted(votes, key=lambda v: v.probability, reverse=True)
+        # Loaded models first, highest probability first; placeholders last.
+        return sorted(votes, key=lambda v: (v.available, v.probability), reverse=True)
 
     def to_dict(self, result: XAIResult) -> Dict[str, Any]:
         """Serialize XAIResult for JSON response."""
@@ -745,8 +742,7 @@ class XAIInferenceService:
                 "tier": result.confidence_band.tier,
                 "labelTr": result.confidence_band.label_tr,
                 "labelEn": result.confidence_band.label_en,
-                "lowerBound": result.confidence_band.lower_bound,
-                "upperBound": result.confidence_band.upper_bound,
+                "margin": result.confidence_band.margin,
             },
             "baseProbability": round(result.base_probability, 4),
             "modelVotes": [
@@ -754,6 +750,7 @@ class XAIInferenceService:
                     "name": v.name,
                     "probability": round(v.probability, 4),
                     "vote": v.vote,
+                    "available": v.available,
                 }
                 for v in result.model_votes
             ],

app/services/youtube_downloader.py

@@ -13,7 +13,7 @@ import re
 import tempfile
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import List, Optional
+from typing import List, Optional, Tuple
 
 import yt_dlp
 
@@ -170,9 +170,18 @@ class _YtDlpLogCollector:
 
 
 class YouTubeDownloader:
-    def __init__(self, output_dir: Path) -> None:
+    def __init__(
+        self,
+        output_dir: Path,
+        clip: Optional[Tuple[float, float]] = None,
+        max_filesize: Optional[int] = None,
+    ) -> None:
+        """clip=(start, end) seconds downloads only that section (needs ffmpeg);
+        max_filesize caps the fallback path that downloads without ffmpeg."""
         self.output_dir = output_dir
         self.output_dir.mkdir(parents=True, exist_ok=True)
+        self.clip = clip
+        self.max_filesize = max_filesize
 
     def download(self, url: str, video_id: str) -> DownloadResult:
         if not url or not url.strip():
@@ -257,7 +266,13 @@ class YouTubeDownloader:
         """Try downloading with different player client combinations."""
         failures: List[DownloadAttemptFailure] = []
 
+        if self.max_filesize:
+            options["max_filesize"] = self.max_filesize
+
         if with_postprocessor:
+            if self.clip:
+                options["download_ranges"] = yt_dlp.utils.download_range_func(None, [self.clip])
+                options["force_keyframes_at_cuts"] = True
             options["format"] = "bestaudio/best"
             options["postprocessors"] = [
                 {