Sincronizando con radio-browser

Tengo el modelo. Tengo la base. Ahora toca rellenarla. Y aquí aparece la primera tensión interesante del proyecto: ¿cómo hacer la sincronización sin que dependa de un broker?

Management command, no Celery

Ya lo conté åen el post del stack: descarté Celery por YAGNI. La alternativa es un management command de Django que se invoca desde un scheduler externo. La interfaz queda:

docker compose exec backend python manage.py update_radios AR BR CL
# o, con el listado por defecto:
docker compose exec backend python manage.py update_radios

Sin worker, sin beat. Solo Django.

El comando, por bloques

Vive en streams/management/commands/update_radios.py. Tiene tres responsabilidades: resolver qué países, consultar radio-browser y persistir cada estación.

1. Argumentos y resolución de países

def add_arguments(self, parser):
    parser.add_argument(
        "country_codes",
        nargs="*",
        help="Códigos ISO-2 (ej.: AR BR CL). Si se omiten, se usa --from-file.",
    )
    parser.add_argument(
        "--from-file",
        type=str,
        default=None,
        help="Fichero de texto con un código por línea.",
    )
    parser.add_argument("--limit", type=int, default=DEFAULT_LIMIT)
    parser.add_argument("--hide-broken", action="store_true", default=True)

Si no paso códigos, lee un fichero por defecto incluido en el paquete (streams/management/commands/data/latam_iso_codes.txt) con los 20 países LATAM. Si paso códigos en la línea de comandos, gana ese listado. Si paso --from-file, gana el fichero apuntado.

2. Consulta async dentro de un command sync

El SDK radios es async. Los management commands de Django son sync. El puente es asyncio.run:

def handle(self, *args, **options):
    country_codes = self._resolve_country_codes(options)
    repo = StreamRepositoryORM()
    for code in country_codes:
        try:
            stations = asyncio.run(
                self._query_radios(
                    code,
                    limit=options["limit"],
                    hide_broken=options["hide_broken"],
                )
            )
        except Exception as exc:
            logger.exception("query failed for %s: %s", code, exc)
            continue
        self._persist_stations(repo, stations, code)


async def _query_radios(self, country_code, *, limit, hide_broken):
    async with RadioBrowser(user_agent=USER_AGENT) as client:
        return await client.stations(
            limit=limit,
            hide_broken=hide_broken,
            order=Order.CLICK_COUNT,
            reverse=True,
            filter_by=FilterBy.COUNTRY_CODE_EXACT,
            filter_term=country_code,
        )

Cada país se consulta con su propio async with (sesión HTTP nueva) y se procesa de forma síncrona antes de pasar al siguiente. Si la consulta de un país revienta, los demás siguen.

3. Persistencia con tolerancia a basura

La salud del comando depende de no abortar al primer registro raro:

def _persist_stations(self, repo, stations, country_code):
    ok, err = 0, 0
    for station in stations:
        try:
            repo.create_or_update_station(station)
            ok += 1
        except Exception as exc:
            err += 1
            logger.warning(
                "[%s] saltando estación inválida (uuid=%s): %s",
                country_code, getattr(station, "uuid", "?"), exc,
            )
    return ok, err

Una emisora con country_code=None o uuid="" se loggea como warning y la importación continúa. Solo un fallo de red entero corta el bucle de ese país, no del proceso.

El repositorio: idempotencia y saneamiento

El comando delega en StreamRepositoryORM.create_or_update_station. Aquí pasan dos cosas críticas:

@transaction.atomic
def create_or_update_station(self, data):
    uuid = _safe_str(getattr(data, "uuid", ""))
    country_code = _safe_str(getattr(data, "country_code", "")).upper()
    if not uuid or len(country_code) != 2:
        raise ValueError(
            f"station with invalid identity (uuid={uuid!r}, country_code={country_code!r})"
        )

    station_data = {
        "name": _safe_str(getattr(data, "name", "")),
        "country_code": country_code,
        "url": _safe_url(getattr(data, "url", "")),
        "homepage": _safe_url(getattr(data, "homepage", "")),
        # ... resto de campos saneados
    }

    station, _ = Station.objects.update_or_create(uuid=uuid, defaults=station_data)
    # ... seteo de M2M (tags, languages, language_codes)
    return station

_safe_url rechaza esquemas que no sean http/https:

def _safe_url(value):
    if not isinstance(value, str):
        return ""
    value = value.strip()
    if value.startswith("http://") or value.startswith("https://"):
        return value
    return ""

Esto previene que entren cadenas tipo javascript:alert(1) que radio-browser puede tener: recordemos que la base es comunitaria, cualquiera registra emisoras. En el post de la auditoría veremos por qué este saneo es crítico, no cosmético.

Idempotencia

update_or_create(uuid=...) es la clave: si la emisora existía, se actualiza; si no, se crea. La sincronización puede ejecutarse N veces sin duplicar nada, lo cual es exactamente lo que quiero de un cron diario.

Lo que no hace este comando

No borra estaciones que desaparecen de radio-browser. Si una emisora se da de baja allí, aquí queda con lastcheckok=False y la API la filtra. Más simple que borrarla.
No reintenta automáticamente en caso de fallo. Si quiero retry, lo gestiona el scheduler externo.
No emite métricas. Los logger.info con guardadas=X errores=Y son suficientes hoy.

Cómo se dispara en producción

En despliegue uso Dokploy y aprovecho su panel de Schedules, que ejecuta un comando cron contra un servicio del compose sin añadir contenedores extra:

Campo	Valor
Name	`update-radios-daily`
Cron Expression	`0 4 * * *` (04:00 UTC)
Service	`backend`
Command	`python manage.py update_radios`

Sin worker que mantener, sin beat al que vigilar. Esto funciona porque el comando ya es idempotente (update_or_create por UUID) y tolerante (un error en un país no aborta los demás). Si mañana cambio de Dokploy a algo distinto, basta con apuntar un cron del host al mismo docker compose exec backend python manage.py update_radios y problema resuelto.

El detalle completo de despliegue (Postgres con PostGIS, env vars, configuración del Schedule) vive en DEPLOY.md del repo.

Lista la sincronización. En el siguiente post miramos el otro extremo del backend: la API que consume la app móvil.

Backend Latam Radio #4