tuxgyver
ed1d40826b
Le téléphone balayait quarante radios toutes les trois minutes, cherchait dans quatre catalogues et enregistrait des directs. Tout cela le tenait éveillé pour un travail qui n'a aucune raison de se faire sur batterie. Un serveur reprend la collecte : recherche, guet, enregistrement, file de téléchargement, piloté depuis un navigateur. Le mobile demande ce qui a changé depuis son curseur, télécharge et se rendort — toutes les six heures, en Wi-Fi non facturé. Vingt fichiers quittent l'application au passage. Le balayage devient continu plutôt qu'espacé de trois minutes : une sonde ne lit que le premier bloc de métadonnées avant de refermer, ce qui rend la permanence tenable quand garder les flux ouverts coûterait plusieurs mégabits par seconde. Un morceau guetté a été capté en moins de vingt-cinq secondes. La recherche interroge les quatre catalogues à la fois et fond leurs résultats, les plus légers d'abord. Quatre défauts corrigés en chemin, tous trouvés en essayant plutôt qu'en relisant : un jeton régénéré à chaque démarrage, qui aurait obligé à reconfigurer le mobile ; un 403 de Wikimedia dû à un agent trop générique, que la fusion des sources masquait ; un dossier de données par défaut inaccessible au compte qui lance le serveur ; et une lecture de métadonnées qui n'itérait rien, faute d'être asynchrone.
236 lines
9.6 KiB
Python
236 lines
9.6 KiB
Python
"""Recherche dans les catalogues libres de droits."""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import re
|
|
import urllib.parse
|
|
|
|
import httpx
|
|
|
|
# Wikimedia refuse (403) les agents génériques : sa politique demande un nom identifiable
|
|
# et un moyen de contact. Les autres services s'en accommodent aussi.
|
|
UA = {"User-Agent": "Noxely/1.0 (https://git.tuxtech.fr/tuxgyver/Noxely)"}
|
|
TIMEOUT = httpx.Timeout(30.0, connect=15.0)
|
|
|
|
AUDIO_EXT = (".mp3", ".m4a", ".ogg", ".opus", ".flac", ".wav")
|
|
VIDEO_EXT = (".mp4", ".m4v", ".webm", ".ogv")
|
|
|
|
CATALOGS = {
|
|
"archive": "Internet Archive",
|
|
"underground": "Underground (netlabels)",
|
|
"librivox": "LibriVox",
|
|
"wikimedia": "Wikimedia Commons",
|
|
}
|
|
|
|
|
|
async def search_all(query: str, language: str | None = None) -> list[dict]:
|
|
"""Interroge les quatre catalogues en parallèle et fond les résultats.
|
|
|
|
Chaque ligne porte son catalogue d'origine : c'est lui qu'il faudra pour en lister les
|
|
pistes. Un catalogue en panne n'empêche pas les autres de répondre — on préfère une liste
|
|
incomplète à une erreur globale.
|
|
"""
|
|
names = list(CATALOGS)
|
|
outcomes = await asyncio.gather(
|
|
*(search(name, query, language) for name in names), return_exceptions=True
|
|
)
|
|
merged: list[dict] = []
|
|
for name, outcome in zip(names, outcomes):
|
|
if isinstance(outcome, Exception):
|
|
continue
|
|
for item in outcome:
|
|
merged.append({**item, "catalog": name})
|
|
# Les résultats les plus légers d'abord : sur un téléphone, la taille prime.
|
|
merged.sort(key=lambda item: item["size"] or 1 << 62)
|
|
return merged
|
|
|
|
|
|
async def search(catalog: str, query: str, language: str | None = None) -> list[dict]:
|
|
if catalog == "librivox":
|
|
return await _librivox(query, language)
|
|
if catalog == "wikimedia":
|
|
return await _wikimedia(query)
|
|
collections = ["netlabels", "freemusicarchive"] if catalog == "underground" else []
|
|
return await _archive(query, language, collections)
|
|
|
|
|
|
async def tracks(catalog: str, item_id: str, title: str = "", kind: str = "Audio") -> list[dict]:
|
|
"""Les fichiers d'un résultat : un album en a plusieurs, un morceau un seul."""
|
|
if catalog == "librivox":
|
|
return await _librivox_tracks(item_id)
|
|
if catalog == "wikimedia":
|
|
return [{"title": title or item_id, "artist": "Wikimedia Commons", "url": item_id, "size": 0}]
|
|
return await _archive_tracks(item_id, title, kind)
|
|
|
|
|
|
# --- archive.org --------------------------------------------------------
|
|
|
|
async def _archive(query: str, language: str | None, collections: list[str]) -> list[dict]:
|
|
clauses = [f"({query})", "mediatype:(audio OR movies)"]
|
|
if language:
|
|
clauses.append(f"language:({language})")
|
|
if collections:
|
|
clauses.append("collection:(" + " OR ".join(collections) + ")")
|
|
# Les éléments restreints répondent 401 au téléchargement : écartés dès la recherche.
|
|
clauses.append("NOT access-restricted-item:true")
|
|
|
|
params = {
|
|
"q": " AND ".join(clauses),
|
|
"fl[]": ["identifier", "title", "creator", "mediatype", "item_size"],
|
|
"rows": "30", "page": "1", "output": "json",
|
|
}
|
|
async with httpx.AsyncClient(headers=UA, timeout=TIMEOUT) as client:
|
|
response = await client.get("https://archive.org/advancedsearch.php", params=params)
|
|
response.raise_for_status()
|
|
docs = response.json().get("response", {}).get("docs", [])
|
|
|
|
return [
|
|
{
|
|
"id": doc["identifier"],
|
|
"title": _first(doc.get("title")) or doc["identifier"],
|
|
"artist": _first(doc.get("creator")) or "Internet Archive",
|
|
"kind": "Vidéo" if doc.get("mediatype") == "movies" else "Audio",
|
|
"catalog": "archive",
|
|
"size": int(doc.get("item_size") or 0),
|
|
}
|
|
for doc in docs if doc.get("identifier")
|
|
]
|
|
|
|
|
|
async def _archive_tracks(identifier: str, title: str, kind: str) -> list[dict]:
|
|
async with httpx.AsyncClient(headers=UA, timeout=TIMEOUT, follow_redirects=True) as client:
|
|
response = await client.get(f"https://archive.org/metadata/{identifier}")
|
|
response.raise_for_status()
|
|
meta = response.json()
|
|
|
|
if str(meta.get("metadata", {}).get("access-restricted-item", "")).lower() == "true":
|
|
raise ValueError(f"« {title or identifier} » est en accès restreint")
|
|
|
|
wants_video = kind == "Vidéo"
|
|
extensions = VIDEO_EXT if wants_video else AUDIO_EXT
|
|
files = [f for f in meta.get("files", []) if f.get("name", "").lower().endswith(extensions)]
|
|
if not files:
|
|
raise ValueError("Aucun fichier lisible dans cet élément")
|
|
|
|
# Un élément expose l'original et ses dérivés : on garde le plus léger de chaque morceau,
|
|
# autant de pistes que d'œuvres et non que de formats.
|
|
groups: dict[str, dict] = {}
|
|
for entry in files:
|
|
stem = entry["name"].rsplit(".", 1)[0].lower()
|
|
size = int(entry.get("size") or 1 << 62)
|
|
if stem not in groups or size < int(groups[stem].get("size") or 1 << 62):
|
|
groups[stem] = entry
|
|
|
|
out = []
|
|
for entry in sorted(groups.values(), key=lambda f: f["name"]):
|
|
name = entry["name"]
|
|
out.append({
|
|
"title": entry.get("title") or name.rsplit(".", 1)[0],
|
|
"artist": title or identifier,
|
|
"url": f"https://archive.org/download/{identifier}/" + urllib.parse.quote(name),
|
|
"size": int(entry.get("size") or 0),
|
|
"duration_ms": int(float(entry.get("length") or 0) * 1000),
|
|
"artwork_url": f"https://archive.org/services/img/{identifier}",
|
|
})
|
|
return out
|
|
|
|
|
|
# --- LibriVox -----------------------------------------------------------
|
|
|
|
async def _librivox(query: str, language: str | None) -> list[dict]:
|
|
term = urllib.parse.quote(query.strip())
|
|
words = query.strip().split()
|
|
# `title` est une correspondance exacte sans « ^ », et `author` ne matche que le nom de
|
|
# famille : trois interrogations valent mieux qu'une qui échouerait.
|
|
criteria = [f"title=^{term}", f"author={term}"]
|
|
if len(words) > 1:
|
|
criteria.append("author=" + urllib.parse.quote(words[-1]))
|
|
|
|
seen: dict[str, dict] = {}
|
|
limit = 25 if not language else 100
|
|
async with httpx.AsyncClient(headers=UA, timeout=TIMEOUT) as client:
|
|
for criterion in criteria:
|
|
url = f"https://librivox.org/api/feed/audiobooks/?format=json&limit={limit}&{criterion}"
|
|
response = await client.get(url)
|
|
if response.status_code == 404: # « aucun résultat », pas une panne
|
|
continue
|
|
response.raise_for_status()
|
|
for book in response.json().get("books", []):
|
|
# Le paramètre `language` de l'API est accepté puis ignoré : filtrage ici.
|
|
if language and (book.get("language") or "").lower() != language.lower():
|
|
continue
|
|
feed = book.get("url_rss")
|
|
if feed and feed not in seen:
|
|
authors = ", ".join(
|
|
" ".join(filter(None, [a.get("first_name"), a.get("last_name")]))
|
|
for a in book.get("authors", [])
|
|
)
|
|
seen[feed] = {
|
|
"id": feed,
|
|
"title": book.get("title") or "Livre audio",
|
|
"artist": authors or "LibriVox",
|
|
"kind": "Audio",
|
|
"size": 0,
|
|
"catalog": "librivox",
|
|
}
|
|
return list(seen.values())
|
|
|
|
|
|
async def _librivox_tracks(feed_url: str) -> list[dict]:
|
|
async with httpx.AsyncClient(headers=UA, timeout=TIMEOUT, follow_redirects=True) as client:
|
|
response = await client.get(feed_url)
|
|
response.raise_for_status()
|
|
xml = response.text
|
|
|
|
title = re.search(r"<title>(.*?)</title>", xml, re.S)
|
|
book = (title.group(1).strip() if title else "Livre audio")
|
|
out = []
|
|
for item in re.findall(r"<item>(.*?)</item>", xml, re.S):
|
|
url = re.search(r'<enclosure[^>]*url="([^"]+)"', item)
|
|
name = re.search(r"<title>(.*?)</title>", item, re.S)
|
|
if url:
|
|
out.append({
|
|
"title": (name.group(1).strip() if name else "Épisode"),
|
|
"artist": book,
|
|
"url": url.group(1),
|
|
"size": 0,
|
|
})
|
|
return out
|
|
|
|
|
|
# --- Wikimedia Commons --------------------------------------------------
|
|
|
|
async def _wikimedia(query: str) -> list[dict]:
|
|
params = {
|
|
"action": "query", "format": "json", "generator": "search",
|
|
"gsrsearch": f"filetype:audio {query.strip()}", "gsrnamespace": "6", "gsrlimit": "30",
|
|
"prop": "imageinfo", "iiprop": "url|size|mime|extmetadata",
|
|
}
|
|
async with httpx.AsyncClient(headers=UA, timeout=TIMEOUT) as client:
|
|
response = await client.get("https://commons.wikimedia.org/w/api.php", params=params)
|
|
response.raise_for_status()
|
|
pages = response.json().get("query", {}).get("pages", {})
|
|
|
|
out = []
|
|
for page in pages.values():
|
|
info = (page.get("imageinfo") or [{}])[0]
|
|
url = info.get("url")
|
|
if not url or not url.split("?")[0].lower().endswith(AUDIO_EXT):
|
|
continue
|
|
licence = (info.get("extmetadata", {}).get("LicenseShortName", {}) or {}).get("value")
|
|
out.append({
|
|
"id": url,
|
|
"title": page.get("title", "").removeprefix("File:").rsplit(".", 1)[0],
|
|
"artist": licence or "Wikimedia Commons",
|
|
"kind": "Audio",
|
|
"size": int(info.get("size") or 0),
|
|
"catalog": "wikimedia",
|
|
})
|
|
return out
|
|
|
|
|
|
def _first(value):
|
|
if isinstance(value, list):
|
|
return value[0] if value else None
|
|
return value
|