chore(training): correct f-strings; log param counts via utils

src/models/emotion_detection/hf_loader.py

@@ -1,17 +1,17 @@
 #!/usr/bin/env python3
 from __future__ import annotations
 
-from dataclasses import dataclass
-from typing import Dict, Optional
 import os
-import tempfile
-import tarfile
 import shutil
+import tarfile
+import tempfile
+from dataclasses import dataclass
+from typing import Dict, Optional
 
-import torch
 import requests
-from transformers import AutoConfig, AutoTokenizer, AutoModelForSequenceClassification
+import torch
 from huggingface_hub import snapshot_download
+from transformers import AutoConfig, AutoModelForSequenceClassification, AutoTokenizer
 
 
 @dataclass
@@ -71,20 +71,31 @@ def predict(self, text: str, threshold: float = 0.5) -> Dict:
             headers["Authorization"] = f"Bearer {self.token}"
         payload = {"inputs": text}
         try:
-            resp = requests.post(self.endpoint_url, json=payload, headers=headers, timeout=30)
+            resp = requests.post(
+                self.endpoint_url, json=payload, headers=headers, timeout=30
+            )
             resp.raise_for_status()
             data = resp.json()
             # data can be [[{"label":..., "score":...}, ...]] or {"error":...}
             if isinstance(data, list):
                 items = data[0] if data and isinstance(data[0], list) else data
             else:
                 items = []
-            emotions = {item.get("label", f"L{i}"): float(item.get("score", 0.0)) for i, item in enumerate(items)}
+            emotions = {
+                item.get("label", f"L{i}"): float(item.get("score", 0.0))
+                for i, item in enumerate(items)
+            }
             if emotions:
                 primary_label, confidence = max(emotions.items(), key=lambda kv: kv[1])
             else:
                 primary_label, confidence = "neutral", 1.0
-            intensity = "high" if confidence >= 0.75 else ("moderate" if confidence >= 0.4 else "low")
+
+            if confidence >= 0.75:
+                intensity = "high"
+            elif confidence >= 0.4:
+                intensity = "moderate"
+            else:
+                intensity = "low"
             return {
                 "emotions": emotions,
                 "primary_emotion": primary_label,
@@ -95,20 +106,30 @@ def predict(self, text: str, threshold: float = 0.5) -> Dict:
             return {}
 
 
-def _wrap_local_model(local_dir: str, token: Optional[str] = None, force_multi_label: Optional[bool] = None) -> HFEmotionDetector:
+def _wrap_local_model(
+    local_dir: str,
+    token: Optional[str] = None,
+    force_multi_label: Optional[bool] = None,
+) -> HFEmotionDetector:
     cfg = AutoConfig.from_pretrained(local_dir, token=token)
     tok = AutoTokenizer.from_pretrained(local_dir, token=token, use_fast=True)
     mdl = AutoModelForSequenceClassification.from_pretrained(local_dir, token=token)
-    id2label = getattr(cfg, "id2label", None) or {i: str(i) for i in range(cfg.num_labels)}
+    id2label = getattr(cfg, "id2label", None) or {
+        i: str(i) for i in range(cfg.num_labels)
+    }
     if force_multi_label is not None:
         multi_label = bool(force_multi_label)
     else:
         problem_type = getattr(cfg, "problem_type", None)
         multi_label = problem_type == "multi_label_classification"
-    return HFEmotionDetector(model=mdl, tokenizer=tok, id2label=id2label, multi_label=multi_label)
+    return HFEmotionDetector(
+        model=mdl, tokenizer=tok, id2label=id2label, multi_label=multi_label
+    )
 
 
-def load_hf_emotion_model(model_id: str, token: Optional[str] = None, force_multi_label: Optional[bool] = None) -> HFEmotionDetector:
+def load_hf_emotion_model(
+    model_id: str, token: Optional[str] = None, force_multi_label: Optional[bool] = None
+) -> HFEmotionDetector:
     return _wrap_local_model(model_id, token=token, force_multi_label=force_multi_label)
 
 
@@ -120,7 +141,9 @@ def load_emotion_model_multi_source(
     endpoint_url: Optional[str] = None,
     force_multi_label: Optional[bool] = None,
 ) -> object:
-    """Try multiple sources to load the emotion model. Returns an object with .predict(text, threshold).
+    """Try multiple sources to load the emotion model.
+
+    Returns an object with .predict(text, threshold).
 
     Priority:
     1) Explicit local_dir if provided and exists
@@ -132,32 +155,42 @@ def load_emotion_model_multi_source(
     # 1) Local directory
     if local_dir and os.path.isdir(local_dir):
         try:
-            return _wrap_local_model(local_dir, token=token, force_multi_label=force_multi_label)
+            return _wrap_local_model(
+                local_dir, token=token, force_multi_label=force_multi_label
+            )
         except Exception:
             pass
 
     # 2) HF Hub direct
     if model_id:
         try:
-            return load_hf_emotion_model(model_id, token=token, force_multi_label=force_multi_label)
+            return load_hf_emotion_model(
+                model_id, token=token, force_multi_label=force_multi_label
+            )
         except Exception:
             pass
 
     # 3) HF snapshot
     if model_id:
         try:
             cache_base = os.getenv("HF_HOME", "/var/tmp/hf-cache")
-            snap_dir = snapshot_download(repo_id=model_id, token=token, cache_dir=cache_base)
-            return _wrap_local_model(snap_dir, token=token, force_multi_label=force_multi_label)
+            snap_dir = snapshot_download(
+                repo_id=model_id, token=token, cache_dir=cache_base
+            )
+            return _wrap_local_model(
+                snap_dir, token=token, force_multi_label=force_multi_label
+            )
         except Exception:
             pass
 
     # 4) Archive URL
     if archive_url:
         try:
-            cache_dir = os.path.join(os.getenv("XDG_CACHE_HOME", "/var/tmp/hf-cache"), "model-archives")
+            cache_base = os.getenv("XDG_CACHE_HOME", "/var/tmp/hf-cache")
+            cache_dir = os.path.join(cache_base, "model-archives")
             os.makedirs(cache_dir, exist_ok=True)
-            archive_path = os.path.join(cache_dir, os.path.basename(archive_url.split("?")[0]))
+            archive_name = os.path.basename(archive_url.split("?")[0])
+            archive_path = os.path.join(cache_dir, archive_name)
             # Download if not exists
             if not os.path.exists(archive_path):
                 r = requests.get(archive_url, timeout=60)
@@ -171,17 +204,24 @@ def load_emotion_model_multi_source(
                     tar.extractall(path=extract_dir)
             elif archive_path.endswith(".zip"):
                 import zipfile
+
                 with zipfile.ZipFile(archive_path, "r") as zf:
                     zf.extractall(path=extract_dir)
             else:
                 # Unknown archive, try treating as directory
                 pass
             # Try load from extracted directory (assume single top-level)
-            candidates = [extract_dir] + [os.path.join(extract_dir, d) for d in os.listdir(extract_dir)]
+            candidates = [extract_dir] + [
+                os.path.join(extract_dir, d) for d in os.listdir(extract_dir)
+            ]
             for cand in candidates:
-                if os.path.isdir(cand) and os.path.exists(os.path.join(cand, "config.json")):
+                if os.path.isdir(cand) and os.path.exists(
+                    os.path.join(cand, "config.json")
+                ):
                     try:
-                        det = _wrap_local_model(cand, token=token, force_multi_label=force_multi_label)
+                        det = _wrap_local_model(
+                            cand, token=token, force_multi_label=force_multi_label
+                        )
                         return det
                     except Exception:
                         continue
@@ -198,4 +238,4 @@ def load_emotion_model_multi_source(
             pass
 
     # Exhausted all sources
-    raise RuntimeError("Could not load emotion model from any source")
+    raise RuntimeError("Could not load emotion model from any source")