feat: background analysis jobs servisi ile queueing progress tracking ve caching eklendi

app/services/analysis_jobs.py

@@ -63,6 +63,8 @@ class Progress:
         self._started: Dict[str, float] = {}
         self._took: Dict[str, float] = {}
         self.cancelled = False
+        # queued: waiting for the CPU; running: downloading or analysing.
+        self.phase = "queued"
         # Returns True when nobody is waiting for the result any more.
         self.gone: Callable[[], Awaitable[bool]] = _never
 
@@ -143,6 +145,8 @@ class CpuGate:
 
     @asynccontextmanager
     async def slot(self, ticket: str, progress: Optional[Progress] = None) -> AsyncIterator[None]:
+        if progress is not None:
+            progress.phase = "queued"
         event = asyncio.Event()
         self._events[ticket] = event
         self._waiting.append(ticket)
@@ -159,6 +163,7 @@ class CpuGate:
         started = time.monotonic()
         try:
             if progress is not None:
+                progress.phase = "running"
                 progress.check()
                 if await progress.gone():
                     raise JobCancelled("job_abandoned")
@@ -216,22 +221,26 @@ class Job:
     created: float = field(default_factory=time.time)
     updated: float = field(default_factory=time.time)
     last_seen: float = field(default_factory=time.monotonic)
-    status: str = "queued"          # queued | running | done | error
+    status: str = "active"          # active | done | error
     cached: bool = False
     response: Optional[Dict[str, Any]] = None
     task: Optional[asyncio.Task] = None
 
+    @property
+    def finished(self) -> bool:
+        return self.status in ("done", "error")
+
     def to_dict(self) -> Dict[str, Any]:
         data: Dict[str, Any] = {
             "jobId": self.id,
             "sourceType": self.source_type,
-            "status": self.status,
+            "status": self.status if self.finished else self.progress.phase,
             "steps": self.progress.snapshot(),
-            "elapsedSec": round((self.updated if self.status in ("done", "error") else time.time()) - self.created, 2),
+            "elapsedSec": round((self.updated if self.finished else time.time()) - self.created, 2),
             "response": self.response,
         }
         position = cpu_gate.position(self.id)
-        if self.status == "queued" and position is not None:
+        if not self.finished and position is not None:
             data["queue"] = {"position": position, "estimatedWaitSec": cpu_gate.estimated_wait(position)}
         if self.cached:
             data["cached"] = True
@@ -252,25 +261,25 @@ class JobStore:
 
     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 in ("done", "error") and j.updated < cutoff]:
+        for job_id in [j.id for j in self._jobs.values() if j.finished and j.updated < cutoff]:
             del self._jobs[job_id]
         if len(self._jobs) > MAX_JOBS:
-            finished = sorted((j for j in self._jobs.values() if j.status in ("done", "error")), key=lambda j: j.updated)
+            finished = sorted((j for j in self._jobs.values() if j.finished), key=lambda j: j.updated)
             for job in finished[: len(self._jobs) - MAX_JOBS]:
                 del self._jobs[job.id]
 
     def pending_count(self) -> int:
-        return sum(1 for j in self._jobs.values() if j.status in ("queued", "running"))
+        return sum(1 for j in self._jobs.values() if not j.finished)
 
     def running_count(self) -> int:
-        return sum(1 for j in self._jobs.values() if j.status == "running")
+        return sum(1 for j in self._jobs.values() if not j.finished and j.progress.phase == "running")
 
     def start(self, source_type: str, steps: List[str], work: Work, key: Optional[str] = None) -> Job:
         """Queue a job; the same file or link already in flight is shared, a recent one is served from cache."""
         self._prune()
         if key and key in self._inflight:
             shared = self._jobs.get(self._inflight[key])
-            if shared and shared.status in ("queued", "running"):
+            if shared and not shared.finished:
                 shared.last_seen = time.monotonic()
                 return shared
 
@@ -305,6 +314,11 @@ class JobStore:
                 job.progress.skip_pending()
                 job.response = {"result": None, "warnings": [], "errors": [stop.code]}
                 job.status = "error"
+            except asyncio.CancelledError:
+                # Cancelled while still waiting for the CPU.
+                job.progress.skip_pending()
+                job.response = {"result": None, "warnings": [], "errors": ["cancelled"]}
+                job.status = "error"
             except Exception:  # noqa: BLE001 — the job must always settle
                 job.progress.skip_pending()
                 job.response = {"result": None, "warnings": [], "errors": ["internal_error"]}
@@ -330,10 +344,10 @@ class JobStore:
         job = self._jobs.get(job_id)
         if job is None:
             return None
-        if job.status in ("queued", "running"):
+        if not job.finished:
             job.progress.cancelled = True
             # A job still waiting for the CPU never gets there.
-            if job.status == "queued" and job.task is not None:
+            if cpu_gate.position(job.id) is not None and job.task is not None:
                 job.task.cancel()
         return job
 

app/services/feature_extractor.py

@@ -84,6 +84,22 @@ class AudioMeta:
     channels: int
 
 
+def decode_clip(source: Union[Path, bytes, io.BytesIO]) -> bytes:
+    """The clip every layer analyses (first 60 s, mono, 22.05 kHz) as a float WAV.
+
+    Decoding an MP3 is the slowest part of reading it, and each layer used
+    to decode the upload on its own. They now read this instead: loading it
+    back at 22.05 kHz returns exactly the samples extract_features would
+    have produced from the original. Raises ValueError for short or silent audio.
+    """
+    import soundfile as sf
+
+    y, sr = _load_audio(source, _TARGET_SR)
+    buf = io.BytesIO()
+    sf.write(buf, y, sr, format="WAV", subtype="FLOAT")
+    return buf.getvalue()
+
+
 def extract_features(
     source: Union[Path, bytes, io.BytesIO],
     *,