tuxgyver
51b60cda70
Recopier un jeton de trente-trois caractères sur un clavier tactile n'était pas un réglage, c'était une corvée. On ouvre désormais une fenêtre depuis le serveur, on appuie sur Appairer côté téléphone, et c'est fait. Le jeton n'est pas pour autant distribué : ouvrir la fenêtre l'exige déjà, elle dure deux minutes, se referme dès qu'un appareil s'est servi, et ne répond qu'à une adresse privée. Le champ de saisie reste là pour qui préfère le coller. Les cinquante fichiers de code portent en outre leur notice de licence, comme la GPL le recommande. Elle est en français, comme le reste ; le texte qui fait foi demeure celui de LICENSE, non traduit.
246 lines
10 KiB
Python
246 lines
10 KiB
Python
# Noxely — collecte côté serveur, écoute hors connexion côté mobile.
|
|
# Copyright © 2026 Johnny (tuxgyver)
|
|
#
|
|
# Ce programme est un logiciel libre : vous pouvez le redistribuer et le modifier
|
|
# selon les termes de la GNU General Public License, version 3 ou ultérieure,
|
|
# telle que publiée par la Free Software Foundation.
|
|
#
|
|
# Il est distribué dans l'espoir d'être utile, mais SANS AUCUNE GARANTIE.
|
|
# Voir le fichier LICENSE à la racine du dépôt pour le texte complet.
|
|
|
|
"""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
|