testtest5

app.py

@@ -1,143 +1,313 @@
+# freesound_preprocess_ui.py
+# -*- coding: utf-8 -*-
+
 import os
+import re
 import time
-import requests
-import pandas as pd
+import urllib.parse
+from typing import Any, Dict, Tuple, Optional, List
+
 import numpy as np
+import pandas as pd
+import requests
 import gradio as gr
-from sklearn.preprocessing import KBinsDiscretizer, StandardScaler
-from sklearn.feature_extraction.text import HashingVectorizer
-from sklearn.preprocessing import OneHotEncoder
-
-
-# =========================
-# CONFIG
-# =========================
-API_TOKEN = "zE9NjEOgUMzH9K7mjiGBaPJiNwJLjSM53LevarRK"
-BASE_URL = "https://freesound.org/apiv2"
-TIMEOUT = (6, 20)
-
-SESSION = requests.Session()
-SESSION.headers.update({"Authorization": f"Token {API_TOKEN}"})
-
-
-# =========================
-# API FREESOUND
-# =========================
-def fetch_sound(sound_id: int):
-    url = f"{BASE_URL}/sounds/{sound_id}/"
-    params = {
-        "fields": (
-            "id,name,username,description,tags,created,"
-            "duration,num_downloads,avg_rating,"
-            "category,subcategory,license,type"
-        )
-    }
-
-    r = SESSION.get(url, params=params, timeout=TIMEOUT)
-    if r.status_code != 200:
-        raise RuntimeError(f"Erreur API {r.status_code}")
-    return r.json()
-
-
-# =========================
-# PREPROCESSING (ONLINE)
-# =========================
-def discretize_num_downloads(x):
-    if x < 100:
-        return "Low"
-    elif x < 1000:
-        return "Medium"
+
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.preprocessing import StandardScaler
+
+
+# ----------------------------
+# Robust network helpers
+# ----------------------------
+
+DEFAULT_TIMEOUT = 20
+
+def _session() -> requests.Session:
+    s = requests.Session()
+    s.headers.update({
+        "User-Agent": "Mozilla/5.0 (freesound-metadata-preprocess/1.0)",
+        "Accept": "application/json,text/plain,*/*",
+        "Connection": "keep-alive",
+    })
+    return s
+
+def fetch_json_with_retry(
+    url: str,
+    headers: Dict[str, str],
+    max_retries: int = 6,
+    base_sleep: float = 0.8,
+    timeout: int = DEFAULT_TIMEOUT,
+) -> Dict[str, Any]:
+    """
+    GET JSON robuste: gère 429 (rate limit), 5xx et déconnexions.
+    """
+    sess = _session()
+    last_err = None
+
+    for attempt in range(max_retries):
+        try:
+            resp = sess.get(url, headers=headers, timeout=timeout)
+
+            # rate limit
+            if resp.status_code == 429:
+                time.sleep(base_sleep * (2 ** attempt))
+                continue
+
+            # serveur instable
+            if resp.status_code >= 500:
+                time.sleep(base_sleep * (2 ** attempt))
+                continue
+
+            resp.raise_for_status()
+            return resp.json()
+
+        except Exception as e:
+            last_err = e
+            time.sleep(base_sleep * (2 ** attempt))
+
+    raise RuntimeError(f"Échec requête après {max_retries} essais. Dernière erreur: {last_err}")
+
+
+# ----------------------------
+# URL -> sound_id -> API endpoint
+# ----------------------------
+
+def sound_id_from_freesound_page(url: str) -> int:
+    """
+    Extrait l'ID depuis une URL FreeSound de page son:
+    https://freesound.org/people/.../sounds/<id>/
+    """
+    u = url.strip()
+    u = urllib.parse.unquote(u)
+
+    m = re.search(r"freesound\.org\/.*\/sounds\/(\d+)\/?", u)
+    if not m:
+        # si l'utilisateur colle juste l'ID (optionnel)
+        if re.fullmatch(r"\d+", u):
+            return int(u)
+        raise ValueError("URL non reconnue. Colle l’URL FreeSound du son (page), ex: .../sounds/844708/")
+    return int(m.group(1))
+
+def api_url_from_sound_id(sound_id: int) -> str:
+    return f"https://freesound.org/apiv2/sounds/{sound_id}/"
+
+
+# ----------------------------
+# Preprocessing helpers
+# ----------------------------
+
+def clean_tags(tags: Any) -> str:
+    """
+    Nettoie tags :
+    - support list ou str
+    - décode %3B etc
+    - split sur ; , espace
+    - lower
+    - supprime doublons
+    """
+    if tags is None:
+        return ""
+
+    if isinstance(tags, list):
+        raw = " ".join([str(t) for t in tags])
     else:
-        return "High"
+        raw = str(tags)
+
+    raw = urllib.parse.unquote(raw)
+    raw = raw.replace(",", " ").replace(";", " ").replace("|", " ")
+    raw = re.sub(r"\s+", " ", raw).strip().lower()
+
+    toks = [t for t in raw.split(" ") if t]
+    toks = [t for t in toks if len(t) >= 2]
+
+    seen = set()
+    out = []
+    for t in toks:
+        if t not in seen:
+            seen.add(t)
+            out.append(t)
+    return " ".join(out)
+
+def clean_text(x: Any) -> str:
+    if x is None:
+        return ""
+    s = str(x)
+    s = urllib.parse.unquote(s)
+    s = s.lower()
+    s = re.sub(r"\s+", " ", s).strip()
+    return s
+
+def safe_num(x: Any) -> float:
+    try:
+        if x is None:
+            return 0.0
+        return float(x)
+    except Exception:
+        return 0.0
+
+def safe_len_list(x: Any) -> int:
+    if isinstance(x, list):
+        return len(x)
+    return 0
+
+
+# ----------------------------
+# Extract raw features (before)
+# ----------------------------
+
+RAW_COLUMNS = [
+    "id", "name", "username", "license", "created",
+    "description", "tags",
+    "duration", "samplerate", "bitrate", "bitdepth", "channels",
+    "filesize", "type",
+    "num_downloads", "num_ratings", "avg_rating",
+]
+
+def extract_raw_df(sound_json: Dict[str, Any]) -> pd.DataFrame:
+    row = {k: sound_json.get(k) for k in RAW_COLUMNS}
+
+    # certains champs peuvent être absents selon droits/endpoint
+    if "tags" not in row:
+        row["tags"] = sound_json.get("tags")
 
+    return pd.DataFrame([row])
 
-def discretize_avg_rating(x):
-    if x == 0 or pd.isna(x):
-        return "MissedInfo"
-    elif x < 2.5:
-        return "Low"
-    elif x < 3.8:
-        return "Medium"
-    else:
-        return "High"
 
+# ----------------------------
+# Build "after preprocessing" features
+# ----------------------------
+
+def build_after_features(raw_df: pd.DataFrame) -> Tuple[pd.DataFrame, pd.DataFrame]:
+    """
+    Retourne:
+    - after_readable_df : colonnes interprétables (nettoyées + dérivées)
+    - after_vector_df   : features vectorisées (TFIDF + numeric scaled) pour "voir" l’embedding
+    """
+    df = raw_df.copy()
 
-def preprocess_metadata(sound: dict):
-    out = {}
+    # Nettoyages
+    df["tags_clean"] = df["tags"].apply(clean_tags)
+    df["name_clean"] = df["name"].apply(clean_text)
+    df["desc_clean"] = df["description"].apply(clean_text)
 
-    # ---- Targets (debug) ----
-    out["num_downloads_class"] = discretize_num_downloads(sound["num_downloads"])
-    out["avg_rating_class"] = discretize_avg_rating(sound["avg_rating"])
+    # Features dérivées (lisibles)
+    df["num_tags"] = df["tags"].apply(safe_len_list)
+    df["name_len"] = df["name_clean"].apply(lambda s: len(s))
+    df["desc_len"] = df["desc_clean"].apply(lambda s: len(s))
+    df["text_all"] = (df["name_clean"].fillna("") + " " + df["desc_clean"].fillna("") + " " + df["tags_clean"].fillna("")).str.strip()
 
-    # ---- Numériques ----
-    out["duration_log"] = np.log1p(sound["duration"])
-    out["num_downloads_log"] = np.log1p(sound["num_downloads"])
+    # Numeric basic
+    numeric_cols = ["duration", "samplerate", "bitrate", "bitdepth", "channels", "filesize", "num_downloads", "num_ratings", "avg_rating",
+                    "num_tags", "name_len", "desc_len"]
+    for c in numeric_cols:
+        df[c] = df[c].apply(safe_num)
 
-    # ---- Created → age_days ----
-    created = pd.to_datetime(sound["created"], errors="coerce")
-    age_days = (pd.Timestamp.now() - created).days if pd.notna(created) else 0
-    out["age_days_log"] = np.log1p(age_days)
+    # 1) after_readable_df (ce que tu veux lire facilement)
+    after_readable_cols = [
+        "id", "type", "license", "created",
+        "name_clean", "tags_clean",
+        "duration", "samplerate", "channels", "filesize",
+        "num_downloads", "num_ratings", "avg_rating",
+        "num_tags", "name_len", "desc_len",
+    ]
+    after_readable_df = df[after_readable_cols].copy()
 
-    # ---- Username freq (proxy) ----
-    out["username_len"] = len(sound["username"]) if sound["username"] else 0
+    # 2) vectorisation texte (TF-IDF) + standardisation numeric
+    # Sur un seul son, TF-IDF marche quand même (tu verras les termes présents).
+    tfidf = TfidfVectorizer(max_features=60, ngram_range=(1, 2))
+    X_text = tfidf.fit_transform(df["text_all"].fillna(""))
 
-    # ---- Name ----
-    name = sound["name"].lower()
-    out["name_len"] = len(name)
+    # Numeric scaling
+    scaler = StandardScaler()
+    X_num = scaler.fit_transform(df[numeric_cols].to_numpy())
+
+    # Assemble en DataFrame pour affichage
+    text_feature_names = [f"tfidf:{t}" for t in tfidf.get_feature_names_out()]
+    X_text_dense = X_text.toarray()
 
-    hv = HashingVectorizer(n_features=8, alternate_sign=False)
-    name_vec = hv.transform([name]).toarray()[0]
-    for i, v in enumerate(name_vec):
-        out[f"name_vec_{i}"] = v
+    num_feature_names = [f"num:{c}" for c in numeric_cols]
 
-    # ---- Tags (simple multi-hot) ----
-    tags = sound["tags"][:5]  # limiter
-    for t in tags:
-        out[f"tag_{t}"] = 1
+    all_features = np.concatenate([X_num, X_text_dense], axis=1)
+    all_names = num_feature_names + text_feature_names
 
-    # ---- Catégories ----
-    for col in ["category", "subcategory", "license", "type"]:
-        val = sound.get(col) or "Unknown"
-        out[f"{col}_{val}"] = 1
+    after_vector_df = pd.DataFrame(all_features, columns=all_names)
 
-    return out
+    return after_readable_df, after_vector_df
 
 
-# =========================
-# PIPELINE GRADIO
-# =========================
-def run(sound_id):
-    if not str(sound_id).isdigit():
-        raise gr.Error("ID invalide")
+# ----------------------------
+# Main analysis function
+# ----------------------------
 
-    sound = fetch_sound(int(sound_id))
+def analyze(url: str, api_key: str) -> Tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
+    if not url or not url.strip():
+        raise ValueError("Colle l’URL du son FreeSound.")
 
-    # AVANT
-    before_df = pd.DataFrame.from_dict(sound, orient="index", columns=["value"])
+    api_key = (api_key or "").strip() or os.environ.get("FREESOUND_API_KEY", "").strip()
+    if not api_key:
+        raise ValueError("Il faut une clé FreeSound API. Mets-la dans le champ 'API key' ou dans FREESOUND_API_KEY.")
 
-    # APRÈS
-    processed = preprocess_metadata(sound)
-    after_df = pd.DataFrame.from_dict(processed, orient="index", columns=["value"])
+    sound_id = sound_id_from_freesound_page(url)
+    api_url = api_url_from_sound_id(sound_id)
 
-    return before_df, after_df
+    headers = {"Authorization": f"Token {api_key}"}
 
+    sound_json = fetch_json_with_retry(api_url, headers=headers)
 
-# =========================
-# UI
-# =========================
-with gr.Blocks(title="Metadata preprocessing FreeSound") as demo:
-    gr.Markdown("""
-    # 🎧 FreeSound – Prétraitement Metadata  
-    **Objectif :** visualiser les features **avant** et **après** preprocessing  
-    """)
+    before_df = extract_raw_df(sound_json)
 
-    sound_id = gr.Textbox(label="Sound ID", placeholder="ex: 123456")
-    btn = gr.Button("Analyser")
+    after_readable_df, after_vector_df = build_after_features(before_df)
+
+    # Bonus: afficher seulement les top features TF-IDF non-nulles
+    # (sur un seul sample, c'est plus clair)
+    nonzero = after_vector_df.loc[0]
+    top = nonzero[nonzero != 0].sort_values(key=lambda s: np.abs(s), ascending=False).head(30)
+    top_df = top.reset_index()
+    top_df.columns = ["feature", "value"]
+
+    return before_df, after_readable_df, top_df
+
+
+# ----------------------------
+# Gradio UI
+# ----------------------------
+
+with gr.Blocks(title="FreeSound - Prétraitement Metadata") as demo:
+    gr.Markdown("## 🎧 FreeSound – Prétraitement Metadata\n"
+                "Objectif : **visualiser les features AVANT et APRÈS preprocessing**.\n\n"
+                "- Entrée = **URL du son FreeSound** (page)\n"
+                "- Sorties = **tableau avant**, **tableau après**, **top features (vectorisées)**")
 
     with gr.Row():
-        before = gr.Dataframe(label="AVANT preprocessing (brut FreeSound)")
-        after = gr.Dataframe(label="APRÈS preprocessing (features modèle)")
+        url_in = gr.Textbox(
+            label="URL du son FreeSound",
+            placeholder="https://freesound.org/people/.../sounds/844708/",
+            value="",
+        )
+
+    api_in = gr.Textbox(
+        label="API key (Token) FreeSound (optionnel si FREESOUND_API_KEY est set)",
+        placeholder="Colle ta clé ici (Token ...)",
+        type="password",
+        value="",
+    )
+
+    btn = gr.Button("Analyser")
+
+    gr.Markdown("### Avant (raw metadata)")
+    before_out = gr.Dataframe(interactive=False, wrap=True)
+
+    gr.Markdown("### Après (nettoyé + features dérivées lisibles)")
+    after_out = gr.Dataframe(interactive=False, wrap=True)
+
+    gr.Markdown("### Top features après vectorisation (num + TF-IDF) — valeurs non nulles")
+    top_out = gr.Dataframe(interactive=False, wrap=True)
 
-    btn.click(run, sound_id, [before, after])
+    btn.click(
+        fn=analyze,
+        inputs=[url_in, api_in],
+        outputs=[before_out, after_out, top_out],
+    )
 
-demo.launch()
+if __name__ == "__main__":
+    demo.launch(server_name="0.0.0.0", server_port=7860)