hacklab/FLUJOS
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat: coconews, ingesta RSS multiidioma

Browse source 
Subsistema de noticias completo: cuatro workers independientes que se
coordinan solo a traves de MongoDB, su API en :3100 y las unidades systemd.

  ingestor.py    236 feeds en 5 idiomas, ciclo de 15 min
  translator.py  traduccion al espanol
  ner.py         entidades con spaCy
  embedder.py    vectores MiniLM de 384 dimensiones

El ingestor fija un timeout de socket global. feedparser.parse() no acepta
timeout propio y usa el del socket, que por defecto es infinito: un feed que
acepta la conexion y luego no responde bloqueaba el hilo para siempre, el
ThreadPoolExecutor no cerraba y el ciclo no volvia a terminar.

Las unidades systemd son plantillas con __INSTALL_DIR__, __USER__ y
__NODE_BIN__; instalar.sh las rellena y las deja en /etc/systemd/system.
This commit is contained in:
hacklab 2026-08-23 17:11:31 +02:00
parent 8625746ace
commit 6de1b6de02
17 changed files with 1341 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

17
coconews/api/package.json Normal file
View file

@ -0,0 +1,17 @@
{
"name": "api",
"version": "1.0.0",
"description": "",
"main": "server.js",
"scripts": {
"test": "echo \"Error: no test specified\" && exit 1",
"start": "node server.js"
},
"keywords": [],
"author": "",
"license": "ISC",
"dependencies": {
"express": "^5.2.1",
"mongodb": "^7.2.0"
}
}

142
coconews/api/server.js Normal file
View file

@ -0,0 +1,142 @@
const express = require('express');
const { MongoClient } = require('mongodb');
const MONGO_URL = process.env.MONGO_URL || 'mongodb://localhost:27017';
const DB_NAME = process.env.DB_NAME || 'FLUJOS_DATOS';
const PORT = process.env.COCONEWS_PORT || 3100;
const app = express();
app.use(express.json());
let db;
async function connect() {
const client = new MongoClient(MONGO_URL);
await client.connect();
db = client.db(DB_NAME);
console.log('MongoDB conectado');
}
app.use((req, res, next) => {
res.setHeader('Access-Control-Allow-Origin', '*');
next();
});
// GET /api/news?page=1&per_page=30&tema=...&q=...&pais=...&entidad=...
app.get('/api/news', async (req, res) => {
const page = Math.max(1, parseInt(req.query.page || '1'));
const perPage = Math.min(100, parseInt(req.query.per_page || '30'));
const tema = req.query.tema ? req.query.tema.trim() : null;
const pais = req.query.pais ? req.query.pais.trim() : null;
const q = req.query.q ? req.query.q.trim() : null;
const entidad = req.query.entidad ? req.query.entidad.trim() : null;
const filter = { traducido: true, source_type: 'rss' };
if (tema) filter.tema = tema;
if (pais) filter.subtema = { $regex: pais, $options: 'i' };
if (q) filter.texto = { $regex: q.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'), $options: 'i' };
if (entidad) filter['entidades.texto'] = { $regex: entidad.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'), $options: 'i' };
try {
const [total, news] = await Promise.all([
db.collection('noticias').countDocuments(filter),
db.collection('noticias')
.find(filter, { projection: { titulo_original: 1, texto: 1, url: 1, fecha: 1, imagen_url: 1, tema: 1, subtema: 1, fuente: 1, lang: 1, entidades: 1 } })
.sort({ fecha: -1 })
.skip((page - 1) * perPage)
.limit(perPage)
.toArray()
]);
res.json({ news, total, page, per_page: perPage, total_pages: Math.ceil(total / perPage) });
} catch (e) {
res.status(500).json({ error: e.message });
}
});
// GET /api/stats
app.get('/api/stats', async (req, res) => {
try {
const hoyStart = new Date(); hoyStart.setHours(0,0,0,0);
const [totalNews, traducidas, feeds, hoy, conNer, conEmb] = await Promise.all([
db.collection('noticias').countDocuments({ source_type: 'rss' }),
db.collection('noticias').countDocuments({ source_type: 'rss', traducido: true }),
db.collection('coconews_feeds').countDocuments({ activo: true }),
db.collection('noticias').countDocuments({ source_type: 'rss', fecha: { $gte: hoyStart } }),
db.collection('noticias').countDocuments({ source_type: 'rss', procesado_ner: true }),
db.collection('noticias').countDocuments({ source_type: 'rss', procesado_embedding: true }),
]);
const [porTema, porIdioma] = await Promise.all([
db.collection('noticias').aggregate([
{ $match: { source_type: 'rss', traducido: true } },
{ $group: { _id: '$tema', count: { $sum: 1 } } },
{ $sort: { count: -1 } },
]).toArray(),
db.collection('noticias').aggregate([
{ $match: { source_type: 'rss', traducido: true } },
{ $group: { _id: '$lang', count: { $sum: 1 } } },
{ $sort: { count: -1 } },
]).toArray(),
]);
res.json({ total_news: totalNews, traducidas, feeds_activos: feeds, hoy, con_ner: conNer, con_embedding: conEmb, por_tema: porTema, por_idioma: porIdioma });
} catch (e) {
res.status(500).json({ error: e.message });
}
});
// GET /api/temas
app.get('/api/temas', async (req, res) => {
try {
const temas = await db.collection('noticias').distinct('tema', { source_type: 'rss' });
res.json(temas.filter(Boolean).sort());
} catch (e) {
res.status(500).json({ error: e.message });
}
});
// GET /api/entidades?tipo=persona|organizacion|lugar&limit=20&tema=...
app.get('/api/entidades', async (req, res) => {
const tipo = req.query.tipo || null;
const tema = req.query.tema || null;
const limit = Math.min(100, parseInt(req.query.limit || '30'));
const filter = {};
if (tipo) filter.tipo = tipo;
if (tema) filter.temas = tema;
try {
const entidades = await db.collection('coconews_entidades')
.find(filter, { projection: { texto: 1, tipo: 1, count: 1, temas: 1 } })
.sort({ count: -1 })
.limit(limit)
.toArray();
res.json(entidades);
} catch (e) {
res.status(500).json({ error: e.message });
}
});
// GET /api/feeds
app.get('/api/feeds', async (req, res) => {
const page = Math.max(1, parseInt(req.query.page || '1'));
const perPage = Math.min(100, parseInt(req.query.per_page || '50'));
try {
const [total, feeds] = await Promise.all([
db.collection('coconews_feeds').countDocuments(),
db.collection('coconews_feeds')
.find({}, { projection: { nombre: 1, url: 1, tema: 1, subtema: 1, idioma: 1, activo: 1, fallos: 1, ultimo_ok: 1 } })
.sort({ nombre: 1 })
.skip((page - 1) * perPage)
.limit(perPage)
.toArray()
]);
res.json({ feeds, total, page, per_page: perPage, total_pages: Math.ceil(total / perPage) });
} catch (e) {
res.status(500).json({ error: e.message });
}
});
connect().then(() => {
app.listen(PORT, '127.0.0.1', () => console.log(`Coconews API escuchando en :${PORT}`));
}).catch(err => {
console.error('Error conectando MongoDB:', err);
process.exit(1);
});

19
coconews/systemd/coconews-api.service Normal file
View file

@ -0,0 +1,19 @@
[Unit]
Description=Coconews API (Node.js + MongoDB)
After=network.target mongod.service
[Service]
Type=simple
User=__USER__
WorkingDirectory=__INSTALL_DIR__/coconews/api
Environment=MONGO_URL=mongodb://localhost:27017
Environment=DB_NAME=FLUJOS_DATOS
Environment=COCONEWS_PORT=3100
ExecStart=__NODE_BIN__ --experimental-global-webcrypto server.js
Restart=on-failure
RestartSec=10
StandardOutput=append:__INSTALL_DIR__/coconews/logs/api.log
StandardError=append:__INSTALL_DIR__/coconews/logs/api.log
[Install]
WantedBy=multi-user.target

18
coconews/systemd/coconews-embedder.service Normal file
View file

@ -0,0 +1,18 @@
[Unit]
Description=Coconews Embedder Worker (MiniLM)
After=network.target mongod.service
[Service]
Type=simple
User=__USER__
WorkingDirectory=__INSTALL_DIR__/coconews/workers
Environment=MONGO_URL=mongodb://localhost:27017
Environment=DB_NAME=FLUJOS_DATOS
ExecStart=/usr/bin/python3 embedder.py
Restart=on-failure
RestartSec=30
StandardOutput=append:__INSTALL_DIR__/coconews/logs/embedder.log
StandardError=append:__INSTALL_DIR__/coconews/logs/embedder.log
[Install]
WantedBy=multi-user.target

18
coconews/systemd/coconews-ingestor.service Normal file
View file

@ -0,0 +1,18 @@
[Unit]
Description=Coconews Ingestor RSS
After=network.target mongod.service
[Service]
Type=simple
User=__USER__
WorkingDirectory=__INSTALL_DIR__/coconews/workers
Environment=MONGO_URL=mongodb://localhost:27017
Environment=DB_NAME=FLUJOS_DATOS
ExecStart=/usr/bin/python3 ingestor.py
Restart=on-failure
RestartSec=30
StandardOutput=append:__INSTALL_DIR__/coconews/logs/ingestor.log
StandardError=append:__INSTALL_DIR__/coconews/logs/ingestor.log
[Install]
WantedBy=multi-user.target

18
coconews/systemd/coconews-ner.service Normal file
View file

@ -0,0 +1,18 @@
[Unit]
Description=Coconews NER Worker (spaCy)
After=network.target mongod.service
[Service]
Type=simple
User=__USER__
WorkingDirectory=__INSTALL_DIR__/coconews/workers
Environment=MONGO_URL=mongodb://localhost:27017
Environment=DB_NAME=FLUJOS_DATOS
ExecStart=/usr/bin/python3 ner.py
Restart=on-failure
RestartSec=30
StandardOutput=append:__INSTALL_DIR__/coconews/logs/ner.log
StandardError=append:__INSTALL_DIR__/coconews/logs/ner.log
[Install]
WantedBy=multi-user.target

18
coconews/systemd/coconews-translator.service Normal file
View file

@ -0,0 +1,18 @@
[Unit]
Description=Coconews Translator
After=network.target mongod.service
[Service]
Type=simple
User=__USER__
WorkingDirectory=__INSTALL_DIR__/coconews/workers
Environment=MONGO_URL=mongodb://localhost:27017
Environment=DB_NAME=FLUJOS_DATOS
ExecStart=/usr/bin/python3 translator.py
Restart=on-failure
RestartSec=30
StandardOutput=append:__INSTALL_DIR__/coconews/logs/translator.log
StandardError=append:__INSTALL_DIR__/coconews/logs/translator.log
[Install]
WantedBy=multi-user.target

44
coconews/systemd/instalar.sh Executable file
View file

@ -0,0 +1,44 @@
#!/usr/bin/env bash
# Instala las unidades de coconews en systemd.
#
# Las unidades de este directorio son plantillas: llevan __INSTALL_DIR__,
# __USER__ y __NODE_BIN__ en lugar de rutas de una máquina concreta. Este script
# las rellena y las deja en /etc/systemd/system/.
#
# sudo ./instalar.sh # deduce todo del entorno
# sudo INSTALL_DIR=/srv/flujos ./instalar.sh
#
set -euo pipefail
AQUI="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
INSTALL_DIR="${INSTALL_DIR:-$(cd "$AQUI/../.." && pwd)}"
SERVICE_USER="${SERVICE_USER:-${SUDO_USER:-$(id -un)}}"
NODE_BIN="${NODE_BIN:-$(command -v node || echo /usr/bin/node)}"
if [ "$(id -u)" -ne 0 ]; then
echo "Hace falta root: sudo $0" >&2
exit 1
fi
echo " directorio : $INSTALL_DIR"
echo " usuario : $SERVICE_USER"
echo " node : $NODE_BIN"
echo
mkdir -p "$INSTALL_DIR/coconews/logs"
chown "$SERVICE_USER" "$INSTALL_DIR/coconews/logs"
for plantilla in "$AQUI"/*.service; do
destino="/etc/systemd/system/$(basename "$plantilla")"
sed -e "s|__INSTALL_DIR__|$INSTALL_DIR|g" \
-e "s|__USER__|$SERVICE_USER|g" \
-e "s|__NODE_BIN__|$NODE_BIN|g" \
"$plantilla" > "$destino"
echo " instalada: $destino"
done
systemctl daemon-reload
echo
echo "Listo. Para arrancarlas:"
echo " systemctl enable --now coconews-ingestor coconews-translator \\"
echo " coconews-ner coconews-embedder coconews-api"

51
coconews/workers/_resiliencia.py Normal file
View file

@ -0,0 +1,51 @@
#!/usr/bin/env python3
"""
_resiliencia.py — helper compartido de los workers coconews.
Problema que resuelve: hoy cada worker (ingestor/translator/ner/embedder) hace
operaciones Mongo SIN try/except dentro de su `while True`. Si mongod cae, el
proceso muere → systemd lo reinicia (Restart=on-failure) → recarga el modelo
(~4-9 s) → vuelve a morir = CRASH-LOOP que quemó CPU/RAM y llenó logs durante
~3 semanas (incidente 2026-05-18).
`bucle_resiliente` ejecuta el paso de trabajo en bucle capturando los errores de
Mongo y reintentando con backoff, en lugar de dejar morir el proceso. Así una
caída de mongod se traduce en "espero y reintento", no en crash-loop.
Uso en un worker:
from _resiliencia import bucle_resiliente
if __name__ == "__main__":
ensure_indexes() # si aplica (ingestor)
bucle_resiliente(process_batch, sleep_trabajo=2, sleep_vacio=60)
"""
import logging
import time
from pymongo.errors import AutoReconnect, PyMongoError, ServerSelectionTimeoutError
log = logging.getLogger("resiliencia")
def bucle_resiliente(paso, sleep_trabajo=2, sleep_vacio=60, sleep_error=30):
"""Ejecuta `paso()` en bucle infinito de forma tolerante a fallos.
`paso` debe devolver el nº de items procesados (0 = nada pendiente) o None.
- Si Mongo no está disponible → log + espera `sleep_error` y reintenta (NO muere).
- Tras un paso con trabajo → espera `sleep_trabajo`; si no había nada → `sleep_vacio`.
"""
while True:
try:
n = paso()
except (ServerSelectionTimeoutError, AutoReconnect) as e:
log.warning("Mongo no disponible (%s); reintento en %ss", type(e).__name__, sleep_error)
time.sleep(sleep_error)
continue
except PyMongoError as e:
log.error("Error de Mongo en el paso: %s; reintento en %ss", e, sleep_error)
time.sleep(sleep_error)
continue
except Exception as e: # noqa: BLE001 — un worker no debe morir por un doc malo
log.exception("Error inesperado en el paso: %s; reintento en %ss", e, sleep_error)
time.sleep(sleep_error)
continue
time.sleep(sleep_vacio if not n else sleep_trabajo)

212
coconews/workers/add_feeds.py Normal file
View file

@ -0,0 +1,212 @@
#!/usr/bin/env python3
"""
Añade feeds RSS internacionales a coconews_feeds en MongoDB.
Solo inserta los que no existen ya (por URL).
"""
import os
from pymongo import MongoClient
from pymongo.errors import DuplicateKeyError
MONGO_URL = os.getenv("MONGO_URL", "mongodb://localhost:27017")
DB_NAME = os.getenv("DB_NAME", "FLUJOS_DATOS")
# (nombre, url, tema, subtema, idioma)
FEEDS = [
# ── GUERRA GLOBAL / GEOPOLITICA / CONFLICTOS ─────────────────────────────
("BBC World News", "http://feeds.bbci.co.uk/news/world/rss.xml", "guerra global", "Internacional", "en"),
("Reuters World", "https://feeds.reuters.com/reuters/worldNews", "guerra global", "Internacional", "en"),
("Al Jazeera English", "https://www.aljazeera.com/xml/rss/all.xml", "guerra global", "Internacional", "en"),
("Deutsche Welle", "https://rss.dw.com/xml/rss-en-all", "guerra global", "Internacional", "en"),
("France 24 English", "https://www.france24.com/en/rss", "guerra global", "Internacional", "en"),
("The Guardian World", "https://www.theguardian.com/world/rss", "guerra global", "Internacional", "en"),
("AP News World", "https://apnews.com/rss", "guerra global", "Internacional", "en"),
("Foreign Policy", "https://foreignpolicy.com/feed/", "guerra global", "Geopolítica", "en"),
("Bellingcat", "https://www.bellingcat.com/feed/", "guerra global", "Conflictos", "en"),
("OCCRP", "https://www.occrp.org/en/feed", "guerra global", "Investigación", "en"),
("War on the Rocks", "https://warontherocks.com/feed/", "guerra global", "Defensa", "en"),
("Defense News", "https://www.defensenews.com/arc/outbound/rss/", "guerra global", "Defensa", "en"),
("El País Internacional", "https://feeds.elpais.com/mrss-s/pages/ep/site/elpais.com/section/internacional/portada", "guerra global", "España", "es"),
("El Mundo Internacional", "https://e00-elmundo.uecdn.es/elmundo/rss/internacional.xml", "guerra global", "España", "es"),
("France 24 Español", "https://www.france24.com/es/rss", "guerra global", "Internacional", "es"),
("RT en Español", "https://actualidad.rt.com/rss", "guerra global", "Internacional", "es"),
("euronews EN", "https://feeds.feedburner.com/euronews/en/news/", "guerra global", "Europa", "en"),
("euronews ES", "https://feeds.feedburner.com/euronews/es/news/", "guerra global", "Europa", "es"),
("The Intercept", "https://theintercept.com/feed/?lang=en", "guerra global", "Investigación", "en"),
("Middle East Eye", "https://www.middleeasteye.net/rss", "guerra global", "Oriente Medio", "en"),
("Antiwar.com", "https://www.antiwar.com/feed/", "guerra global", "Conflictos", "en"),
("Responsible Statecraft", "https://responsiblestatecraft.org/feed/", "guerra global", "Geopolítica", "en"),
("RUSI Publications", "https://rusi.org/publications/rss", "guerra global", "Defensa", "en"),
("Breaking Defense", "https://breakingdefense.com/feed/", "guerra global", "Defensa", "en"),
("The Diplomat", "https://thediplomat.com/feed/", "guerra global", "Asia", "en"),
("Jamestown Foundation", "https://jamestown.org/feed/", "guerra global", "Seguridad", "en"),
("Atlantic Council", "https://www.atlanticcouncil.org/feed/", "guerra global", "Geopolítica", "en"),
("CFR — World", "https://www.cfr.org/rss/cfr_world.xml", "guerra global", "Geopolítica", "en"),
("Crisis Group", "https://www.crisisgroup.org/rss.xml", "guerra global", "Conflictos", "en"),
("Relief Web", "https://reliefweb.int/headlines/rss.xml", "guerra global", "Humanitario", "en"),
("UNHCR News", "https://www.unhcr.org/rss/news.rss", "guerra global", "Humanitario", "en"),
("NATO News", "https://www.nato.int/cps/en/natohq/news.rss", "guerra global", "Defensa", "en"),
("South China Morning Post Intl","https://www.scmp.com/rss/91/feed", "guerra global", "Asia", "en"),
("Kyiv Independent", "https://kyivindependent.com/feed/", "guerra global", "Europa", "en"),
("The Moscow Times", "https://www.themoscowtimes.com/rss/news", "guerra global", "Rusia", "en"),
("Meduza English", "https://meduza.io/rss/en/all", "guerra global", "Rusia", "en"),
("Jerusalem Post", "https://www.jpost.com/Rss/RssFeedsHeadlines.aspx", "guerra global", "Oriente Medio", "en"),
("Haaretz English", "https://www.haaretz.com/cmlink/1.628752", "guerra global", "Oriente Medio", "en"),
("Anadolu Agency", "https://www.aa.com.tr/en/rss/default?cat=world", "guerra global", "Internacional", "en"),
("TRT World", "https://www.trtworld.com/rss", "guerra global", "Internacional", "en"),
("Global Voices", "https://globalvoices.org/feed/", "guerra global", "Sociedad", "en"),
("Arab News", "https://www.arabnews.com/rss.xml", "guerra global", "Oriente Medio", "en"),
("Dawn Pakistan", "https://www.dawn.com/feeds/home", "guerra global", "Asia", "en"),
("The Hindu", "https://www.thehindu.com/feeder/default.rss", "guerra global", "Asia", "en"),
("Al Monitor", "https://www.al-monitor.com/rss.xml", "guerra global", "Oriente Medio", "en"),
("Times of India Intl", "https://timesofindia.indiatimes.com/rssfeeds/-2128936835.cms", "guerra global", "Asia", "en"),
# ── INTELIGENCIA Y SEGURIDAD ──────────────────────────────────────────────
("Lawfare", "https://www.lawfaremedia.org/feed", "inteligencia y seguridad", "EEUU", "en"),
("Just Security", "https://www.justsecurity.org/feed/", "inteligencia y seguridad", "Internacional", "en"),
("The Cipher Brief", "https://www.thecipherbrief.com/feed", "inteligencia y seguridad", "Inteligencia", "en"),
("Krebs on Security", "https://krebsonsecurity.com/feed/", "inteligencia y seguridad", "Ciberseguridad", "en"),
("Security Week", "https://feeds.feedburner.com/Securityweek", "inteligencia y seguridad", "Ciberseguridad", "en"),
("The Hacker News", "https://feeds.feedburner.com/TheHackersNews", "inteligencia y seguridad", "Ciberseguridad", "en"),
("Dark Reading", "https://www.darkreading.com/rss.xml", "inteligencia y seguridad", "Ciberseguridad", "en"),
("Recorded Future", "https://www.recordedfuture.com/feed", "inteligencia y seguridad", "Ciberseguridad", "en"),
("Bleeping Computer", "https://www.bleepingcomputer.com/feed/", "inteligencia y seguridad", "Ciberseguridad", "en"),
("ProPublica", "https://www.propublica.org/feeds/propublica/main", "inteligencia y seguridad", "Investigación", "en"),
("RAND Corporation", "https://www.rand.org/feed.xml", "inteligencia y seguridad", "Análisis", "en"),
("Brookings", "https://www.brookings.edu/feed/", "inteligencia y seguridad", "Análisis", "en"),
("Wired Security", "https://www.wired.com/feed/category/security/latest/rss", "inteligencia y seguridad", "Ciberseguridad", "en"),
("Ars Technica Security", "https://feeds.arstechnica.com/arstechnica/security", "inteligencia y seguridad", "Ciberseguridad", "en"),
("IISS", "https://www.iiss.org/rss/publications/", "inteligencia y seguridad", "Defensa", "en"),
("Schneier on Security", "https://www.schneier.com/feed/atom/", "inteligencia y seguridad", "Ciberseguridad", "en"),
("Europol News", "https://www.europol.europa.eu/rss/news", "inteligencia y seguridad", "Europa", "en"),
("El Confidencial Seguridad", "https://www.elconfidencial.com/rss/tecnologia/ciberseguridad.xml","inteligencia y seguridad", "España", "es"),
("Infosecurity Magazine", "https://www.infosecurity-magazine.com/rss/news/", "inteligencia y seguridad", "Ciberseguridad", "en"),
# ── CAMBIO CLIMATICO / MEDIO AMBIENTE ─────────────────────────────────────
("Climate Home News", "https://www.climatechangenews.com/feed/", "cambio climático", "Internacional", "en"),
("Carbon Brief", "https://www.carbonbrief.org/feed/", "cambio climático", "Ciencia", "en"),
("Inside Climate News", "https://insideclimatenews.org/feed/", "cambio climático", "EEUU", "en"),
("Yale Environment 360", "https://e360.yale.edu/feed/", "cambio climático", "Ciencia", "en"),
("The Guardian Environment", "https://www.theguardian.com/environment/rss", "cambio climático", "Internacional", "en"),
("BBC Science & Environment", "http://feeds.bbci.co.uk/news/science_and_environment/rss.xml", "cambio climático", "Internacional", "en"),
("Grist", "https://grist.org/feed/", "cambio climático", "EEUU", "en"),
("Mongabay", "https://news.mongabay.com/feed/", "cambio climático", "Biodiversidad", "en"),
("Eco-Business", "https://www.eco-business.com/feed/", "cambio climático", "Asia", "en"),
("Environmental Health News", "https://www.ehn.org/feed/", "cambio climático", "Salud", "en"),
("Climate Wire — E&E", "https://www.eenews.net/rss/climatewire.rss", "cambio climático", "EEUU", "en"),
("NASA Climate", "https://climate.nasa.gov/news/rss.xml", "cambio climático", "Ciencia", "en"),
("UNEP News", "https://www.unep.org/newsroom/rss.xml", "cambio climático", "Internacional", "en"),
("El País Planeta Futuro", "https://feeds.elpais.com/mrss-s/pages/ep/site/elpais.com/section/planeta-futuro/portada", "cambio climático", "España", "es"),
("Greenpeace Blog", "https://www.greenpeace.org/international/blog/feed/", "cambio climático", "ONG", "en"),
("WWF News", "https://www.worldwildlife.org/press-releases.rss", "cambio climático", "Biodiversidad", "en"),
("DeSmog", "https://www.desmog.com/feed/", "cambio climático", "Investigación", "en"),
("Energy Monitor", "https://www.energymonitor.ai/feed/", "cambio climático", "Energía", "en"),
("Renewable Energy World", "https://www.renewableenergyworld.com/feed/", "cambio climático", "Energía", "en"),
("CleanTechnica", "https://cleantechnica.com/feed/", "cambio climático", "Tecnología", "en"),
# ── ECONOMÍA Y CORPORACIONES ──────────────────────────────────────────────
("Reuters Business", "https://feeds.reuters.com/reuters/businessNews", "economía y corporaciones", "Internacional", "en"),
("BBC Business", "http://feeds.bbci.co.uk/news/business/rss.xml", "economía y corporaciones", "Internacional", "en"),
("CNBC Economy", "https://www.cnbc.com/id/20910258/device/rss/rss.html", "economía y corporaciones", "EEUU", "en"),
("MarketWatch", "https://feeds.marketwatch.com/marketwatch/topstories/", "economía y corporaciones", "EEUU", "en"),
("IMF News", "https://www.imf.org/en/News/rss", "economía y corporaciones", "Internacional", "en"),
("World Bank", "https://feeds.worldbank.org/worldbank/all", "economía y corporaciones", "Internacional", "en"),
("Forbes", "https://www.forbes.com/feeds/entrepreneur.rss", "economía y corporaciones", "EEUU", "en"),
("The Economist", "https://www.economist.com/international/rss.xml", "economía y corporaciones", "Internacional", "en"),
("Financial Times (free)", "https://www.ft.com/rss/home/uk", "economía y corporaciones", "Internacional", "en"),
("El País Economía", "https://feeds.elpais.com/mrss-s/pages/ep/site/elpais.com/section/economia/portada", "economía y corporaciones", "España", "es"),
("Expansión", "https://e00-expansion.uecdn.es/rss/mercados.xml", "economía y corporaciones", "España", "es"),
("Harvard Business Review", "https://feeds.hbr.org/harvardbusiness/", "economía y corporaciones", "Gestión", "en"),
("Quartz", "https://qz.com/feed", "economía y corporaciones", "Internacional", "en"),
("Business Insider", "https://feeds.businessinsider.com/custom/all", "economía y corporaciones", "EEUU", "en"),
("TechCrunch", "https://techcrunch.com/feed/", "economía y corporaciones", "Tecnología", "en"),
("Wired", "https://www.wired.com/feed/rss", "economía y corporaciones", "Tecnología", "en"),
("MIT Technology Review", "https://www.technologyreview.com/feed/", "economía y corporaciones", "Tecnología", "en"),
("Ars Technica", "https://feeds.arstechnica.com/arstechnica/index", "economía y corporaciones", "Tecnología", "en"),
("The Verge", "https://www.theverge.com/rss/index.xml", "economía y corporaciones", "Tecnología", "en"),
("ZDNet", "https://www.zdnet.com/news/rss.xml", "economía y corporaciones", "Tecnología", "en"),
("OCDE Insights", "https://oecdinsights.org/feed/", "economía y corporaciones", "Internacional", "en"),
("WTO News", "https://www.wto.org/english/news_e/news_e.rss", "economía y corporaciones", "Internacional", "en"),
("Multinational Monitor", "https://multinationalmonitor.org/feed/", "economía y corporaciones", "Corporaciones", "en"),
("Global Trade Review", "https://www.gtreview.com/feed/", "economía y corporaciones", "Comercio", "en"),
# ── DEMOGRAFIA Y SOCIEDAD ─────────────────────────────────────────────────
("BBC Health", "http://feeds.bbci.co.uk/news/health/rss.xml", "demografía y sociedad", "Salud", "en"),
("Reuters Health", "https://feeds.reuters.com/reuters/healthNews", "demografía y sociedad", "Salud", "en"),
("WHO News", "https://www.who.int/rss-feeds/news-releases-en.xml", "demografía y sociedad", "Salud", "en"),
("Human Rights Watch", "https://www.hrw.org/feeds/all-reports", "demografía y sociedad", "Derechos Humanos","en"),
("Amnesty International", "https://www.amnesty.org/en/feed/", "demografía y sociedad", "Derechos Humanos","en"),
("The Guardian Society", "https://www.theguardian.com/society/rss", "demografía y sociedad", "Social", "en"),
("Vox", "https://www.vox.com/rss/index.xml", "demografía y sociedad", "EEUU", "en"),
("El Diario ES", "https://www.eldiario.es/rss/", "demografía y sociedad", "España", "es"),
("El Salto Diario", "https://www.elsaltodiario.com/rss", "demografía y sociedad", "España", "es"),
("Público ES", "https://www.publico.es/rss/", "demografía y sociedad", "España", "es"),
("ONU News ES", "https://news.un.org/feed/subscribe/es/news/all/rss.xml", "demografía y sociedad", "Internacional", "es"),
("ONU News EN", "https://news.un.org/feed/subscribe/en/news/all/rss.xml", "demografía y sociedad", "Internacional", "en"),
("IPS News", "https://ipsnews.net/feed/", "demografía y sociedad", "Internacional", "en"),
("Open Democracy", "https://www.opendemocracy.net/en/rss.xml", "demografía y sociedad", "Política", "en"),
("Pew Research", "https://www.pewresearch.org/feed/", "demografía y sociedad", "Investigación", "en"),
("Politico EU", "https://www.politico.eu/news/feed/", "demografía y sociedad", "Europa", "en"),
("Politico US", "https://www.politico.com/rss/politicopicks.xml", "demografía y sociedad", "EEUU", "en"),
("El País Sociedad", "https://feeds.elpais.com/mrss-s/pages/ep/site/elpais.com/section/sociedad/portada", "demografía y sociedad", "España", "es"),
("Migration Policy Institute", "https://www.migrationpolicy.org/rss.xml", "demografía y sociedad", "Migración", "en"),
("ACNUR Noticias", "https://www.acnur.org/noticias.rss", "demografía y sociedad", "Migración", "es"),
("Social Europe", "https://socialeurope.eu/feed", "demografía y sociedad", "Europa", "en"),
("Jacobin", "https://jacobin.com/feed/", "demografía y sociedad", "Política", "en"),
("New Statesman", "https://www.newstatesman.com/feeds/blogs-news.rss", "demografía y sociedad", "Reino Unido", "en"),
("The Atlantic", "https://www.theatlantic.com/feed/all/", "demografía y sociedad", "EEUU", "en"),
]
def main():
client = MongoClient(MONGO_URL, serverSelectionTimeoutMS=5000)
db = client[DB_NAME]
db.coconews_feeds.create_index([("url", 1)], unique=True)
inserted = skipped = 0
for nombre, url, tema, subtema, idioma in FEEDS:
doc = {
"nombre": nombre,
"url": url,
"tema": tema,
"subtema": subtema,
"idioma": idioma,
"activo": True,
"fallos": 0,
"ultimo_ok": None,
}
result = db.coconews_feeds.update_one(
{"url": url},
{"$setOnInsert": doc},
upsert=True,
)
if result.upserted_id:
inserted += 1
else:
skipped += 1
total = db.coconews_feeds.count_documents({})
activos = db.coconews_feeds.count_documents({"activo": True})
print(f"Insertados: {inserted} nuevos | Ya existían: {skipped}")
print(f"Total feeds en BD: {total} ({activos} activos)")
# Marcar feeds afganos inaccesibles como inactivos
bad_domains = ["8am.af", "arezo.tv", "arezotv.com", "bakhtarnews.af",
"barya.news", "chaprast.com", "hamshahri.af",
"jomhornews.com", "avapress.com"]
deactivated = 0
for domain in bad_domains:
r = db.coconews_feeds.update_many(
{"url": {"$regex": domain}, "activo": True},
{"$set": {"activo": False, "fallos": 10}}
)
deactivated += r.modified_count
if deactivated:
print(f"Desactivados {deactivated} feeds afganos inaccesibles.")
client.close()
if __name__ == "__main__":
main()

229
coconews/workers/backfill_embeddings.py Normal file
View file

@ -0,0 +1,229 @@
#!/usr/bin/env python3
"""
backfill_embeddings.py — generador de embeddings por lotes
============================================================
Genera embeddings MiniLM para CUALQUIER colección de nodos (torrents, wikipedia,
noticias...) de forma NO DESTRUCTIVA, idempotente y rápida.
Mejoras frente al embedder original (coconews/workers/embedder.py):
- Generalizado a cualquier colección (no solo source_type:'rss').
- Chunking por VENTANAS DE PALABRAS (~tokens), no texto[:512] caracteres.
- L2-normalización en escritura (normalize_embeddings=True) → coseno = dot.
- Gate de calidad `apto_embedding` (descarta blobs hex/keys/basura).
- bulk_write + SIN sleep(2) → backfill de horas, no de días.
- Chunks en colección SEPARADA `embeddings_chunks` (no infla los docs nodo).
- NO sobreescribe embeddings existentes salvo --reembed (preserva lo ya computado).
Uso (con mongod ARRIBA):
# subconjunto curado de leaks para calibrar
/usr/bin/python3 backfill_embeddings.py --coleccion torrents --limit 500
# contar pendientes sin escribir nada
/usr/bin/python3 backfill_embeddings.py --coleccion torrents --dry-run
# todo wikipedia
/usr/bin/python3 backfill_embeddings.py --coleccion wikipedia
Requisitos: numpy, pymongo, sentence_transformers (ya instalados en /usr/bin/python3).
"""
import argparse
import logging
import os
import re
import sys
import time
MONGO_URL = os.getenv("MONGO_URL", "mongodb://localhost:27017")
DB_NAME = os.getenv("DB_NAME", "FLUJOS_DATOS")
MODEL_NAME = os.getenv("EMBEDDER_MODEL", "paraphrase-multilingual-MiniLM-L12-v2")
CHUNKS_COL = os.getenv("CHUNKS_COL", "embeddings_chunks")
# --- parámetros de chunking / calidad (ajustables por env) ---
CHUNK_WORDS = int(os.getenv("CHUNK_WORDS", "110")) # ~128 tokens MiniLM
CHUNK_OVERLAP = int(os.getenv("CHUNK_OVERLAP", "20"))
MAX_CHUNKS = int(os.getenv("MAX_CHUNKS", "40")) # cap por doc: si excede, muestreo uniforme (NO truncado)
MIN_TOKENS = int(os.getenv("MIN_TOKENS", "15")) # < esto = no apto
MIN_ALPHA = float(os.getenv("MIN_ALPHA", "0.50")) # ratio de caracteres alfabéticos mínimo
logging.basicConfig(level=logging.INFO, format="%(asctime)s [backfill] %(message)s")
log = logging.getLogger(__name__)
_HEX_RE = re.compile(r"^[0-9a-fA-F]{16,}$")
# ----------------------------------------------------------------------------- #
# Gate de calidad: ¿este texto merece un embedding?
# ----------------------------------------------------------------------------- #
def es_apto(texto):
"""Devuelve (apto: bool, motivo: str). Descarta blobs hex/keys/basura."""
if not texto or not texto.strip():
return False, "vacio"
toks = texto.split()
if len(toks) < MIN_TOKENS:
return False, f"pocas_palabras({len(toks)})"
alpha = sum(c.isalpha() for c in texto)
if alpha / max(len(texto), 1) < MIN_ALPHA:
return False, "ratio_alfa_bajo"
# demasiados tokens-hash / claves largas → firmware/keys, no prosa
hex_like = sum(1 for t in toks if _HEX_RE.match(t))
long_tok = sum(1 for t in toks if len(t) > 40)
if (hex_like + long_tok) / len(toks) > 0.30:
return False, "hex_o_keys"
return True, "ok"
# ----------------------------------------------------------------------------- #
# Chunking por ventanas de palabras, con cap por MUESTREO uniforme (no truncado)
# ----------------------------------------------------------------------------- #
def trocear(texto):
palabras = texto.split()
if not palabras:
return []
paso = max(1, CHUNK_WORDS - CHUNK_OVERLAP)
ventanas = [palabras[i:i + CHUNK_WORDS] for i in range(0, len(palabras), paso)]
ventanas = [v for v in ventanas if len(v) >= 5]
if len(ventanas) > MAX_CHUNKS:
# muestreo uniforme a lo largo de TODO el documento (no los primeros N)
idx = [round(i * (len(ventanas) - 1) / (MAX_CHUNKS - 1)) for i in range(MAX_CHUNKS)]
ventanas = [ventanas[i] for i in sorted(set(idx))]
return [" ".join(v) for v in ventanas]
def conectar():
from pymongo import MongoClient
from pymongo.errors import ServerSelectionTimeoutError
cli = MongoClient(MONGO_URL, serverSelectionTimeoutMS=4000)
try:
cli.admin.command("ping")
except ServerSelectionTimeoutError:
log.error("MongoDB no responde en %s — ¿está mongod arriba? Aborto sin tocar nada.", MONGO_URL)
sys.exit(2)
return cli[DB_NAME]
def main():
ap = argparse.ArgumentParser(description="Backfill de embeddings no destructivo")
ap.add_argument("--coleccion", required=True, help="torrents | wikipedia | noticias")
ap.add_argument("--campo-texto", default="texto")
ap.add_argument("--limit", type=int, default=0, help="0 = todos los pendientes")
ap.add_argument("--batch", type=int, default=128)
ap.add_argument("--reembed", action="store_true",
help="re-embeber incluso los que YA tienen embedding (por defecto NO, preserva lo existente)")
ap.add_argument("--dry-run", action="store_true", help="solo cuenta pendientes, no escribe")
args = ap.parse_args()
db = conectar()
col = db[args.coleccion]
chunks_col = db[CHUNKS_COL]
filtro = {} if args.reembed else {"procesado_embedding": {"$ne": True}}
pendientes = col.count_documents(filtro)
total = col.estimated_document_count()
log.info("Colección '%s': %d docs totales, %d pendientes de embedding%s.",
args.coleccion, total, pendientes, " (--reembed)" if args.reembed else "")
if args.dry_run:
log.info("[dry-run] No se escribe nada. Saldría a embeber %d docs.", pendientes)
return
if pendientes == 0:
log.info("Nada que hacer. ✅")
return
log.info("Cargando modelo %s ...", MODEL_NAME)
from sentence_transformers import SentenceTransformer
from pymongo import UpdateOne
model = SentenceTransformer(MODEL_NAME)
log.info("Modelo listo. Empezando backfill (batch=%d)…", args.batch)
# Materializar los _id pendientes de una sola vez (consulta corta) para NO mantener
# un cursor abierto durante el embedding lento → evita CursorNotFound (timeout de cursor).
proj_ids = col.find(filtro, {"_id": 1})
if args.limit:
proj_ids = proj_ids.limit(args.limit)
ids_pendientes = [d["_id"] for d in proj_ids]
def iter_docs():
LOTE = 500
for i in range(0, len(ids_pendientes), LOTE):
sub = ids_pendientes[i:i + LOTE]
for doc in col.find({"_id": {"$in": sub}},
{"_id": 1, "archivo": 1, "source_type": 1, args.campo_texto: 1}):
yield doc
import numpy as np
t0 = time.time()
procesados = no_aptos = chunks_escritos = 0
buf_docs = [] # (doc, [textos_chunk])
chunks_planos = [] # textos para encode en bloque
mapa = [] # (idx_doc_en_buf, idx_chunk)
def flush():
nonlocal procesados, no_aptos, chunks_escritos, buf_docs, chunks_planos, mapa
if not buf_docs:
return
vecs = (model.encode(chunks_planos, batch_size=args.batch, show_progress_bar=False,
normalize_embeddings=True) if chunks_planos else np.zeros((0, 384)))
# agrupar vectores por doc
por_doc = {i: [] for i in range(len(buf_docs))}
for (di, ci), v in zip(mapa, vecs):
por_doc[di].append(v)
ops_nodo, ops_chunk = [], []
for di, (doc, ctexts) in enumerate(buf_docs):
vs = por_doc[di]
archivo = doc.get("archivo", str(doc["_id"]))
st = doc.get("source_type", args.coleccion)
if not vs: # no apto → marcar y seguir
ops_nodo.append(UpdateOne({"_id": doc["_id"]},
{"$set": {"procesado_embedding": True, "apto_embedding": False}}))
continue
arr = np.asarray(vs, dtype=np.float32)
centroide = arr.mean(axis=0)
n = np.linalg.norm(centroide)
centroide = (centroide / n) if n else centroide
ops_nodo.append(UpdateOne({"_id": doc["_id"]}, {"$set": {
"embedding": centroide.tolist(),
"embedding_model": MODEL_NAME,
"embedding_dim": 384,
"n_chunks": len(vs),
"procesado_embedding": True,
"apto_embedding": True,
}}))
for ci, v in enumerate(vs):
ops_chunk.append(UpdateOne(
{"archivo": archivo, "idx": ci},
{"$set": {"archivo": archivo, "idx": ci, "source_type": st,
"coleccion": args.coleccion, "embedding": v.tolist()}},
upsert=True))
if ops_nodo:
col.bulk_write(ops_nodo, ordered=False)
if ops_chunk:
chunks_col.bulk_write(ops_chunk, ordered=False)
chunks_escritos += len(ops_chunk)
procesados += len(buf_docs)
no_aptos += sum(1 for _, c in buf_docs if not c)
buf_docs, chunks_planos, mapa = [], [], []
DOCS_POR_FLUSH = max(args.batch, 64)
for doc in iter_docs():
texto = doc.get(args.campo_texto, "") or ""
apto, _motivo = es_apto(texto)
ctexts = trocear(texto) if apto else []
di = len(buf_docs)
buf_docs.append((doc, ctexts))
for ci, ct in enumerate(ctexts):
chunks_planos.append(ct)
mapa.append((di, ci))
if len(buf_docs) >= DOCS_POR_FLUSH:
flush()
if procesados % (DOCS_POR_FLUSH * 5) == 0:
vel = procesados / max(time.time() - t0, 1e-9)
log.info("… %d docs (%d no aptos, %d chunks) — %.1f docs/s",
procesados, no_aptos, chunks_escritos, vel)
flush()
dt = time.time() - t0
log.info("✅ Backfill '%s' completado: %d docs en %.0fs (%d no aptos, %d chunks). %.1f docs/s",
args.coleccion, procesados, dt, no_aptos, chunks_escritos, procesados / max(dt, 1e-9))
if __name__ == "__main__":
main()

45
coconews/workers/backfill_imagenes.py Normal file
View file

@ -0,0 +1,45 @@
#!/usr/bin/env python3
"""Rellena imagen_url (via og:image) en noticias RSS ya guardadas SIN imagen.
No destructivo: solo escribe donde falta. Por defecto, ultimos 14 dias.
python3 backfill_imagenes.py [dias] [limite]
"""
import os
import sys
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timedelta, timezone
from pymongo import MongoClient
# Reutiliza og_image() del ingestor sin ejecutar su __main__
_src = open(os.path.join(os.path.dirname(__file__), "ingestor.py")).read().split("if __name__")[0]
_ns = {}
exec(_src, _ns)
og_image = _ns["og_image"]
DIAS = int(sys.argv[1]) if len(sys.argv) > 1 else 14
LIMIT = int(sys.argv[2]) if len(sys.argv) > 2 else 1000000
db = MongoClient("mongodb://localhost:27017").FLUJOS_DATOS
desde = datetime.now(timezone.utc) - timedelta(days=DIAS)
q = {"source_type": "rss", "fecha": {"$gte": desde},
"$or": [{"imagen_url": None}, {"imagen_url": ""}], "url": {"$exists": True}}
docs = list(db.noticias.find(q, {"_id": 1, "url": 1}).limit(LIMIT))
total = len(docs)
print(f"[backfill-img] {total} noticias sin imagen en los ultimos {DIAS} dias", flush=True)
st = {"ok": 0, "no": 0}
def trabaja(d):
img = og_image(d["url"], timeout=5)
if img:
db.noticias.update_one({"_id": d["_id"]}, {"$set": {"imagen_url": img}})
st["ok"] += 1
else:
st["no"] += 1
n = st["ok"] + st["no"]
if n % 250 == 0:
print(f" {n}/{total} rescatadas={st['ok']}", flush=True)
if total:
with ThreadPoolExecutor(max_workers=12) as ex:
list(ex.map(trabaja, docs))
print(f"[backfill-img] HECHO: {st['ok']} rescatadas, {st['no']} sin og:image, de {total}", flush=True)

51
coconews/workers/embedder.py Normal file
View file

@ -0,0 +1,51 @@
#!/usr/bin/env python3
import logging
import os
import time
from pymongo import MongoClient
from sentence_transformers import SentenceTransformer
MONGO_URL = os.getenv("MONGO_URL", "mongodb://localhost:27017")
DB_NAME = os.getenv("DB_NAME", "FLUJOS_DATOS")
BATCH_SIZE = int(os.getenv("EMBEDDER_BATCH", "20"))
SLEEP_INTERVAL = int(os.getenv("EMBEDDER_SLEEP", "60"))
MODEL_NAME = os.getenv("EMBEDDER_MODEL", "paraphrase-multilingual-MiniLM-L12-v2")
logging.basicConfig(level=logging.INFO, format="%(asctime)s [embedder] %(message)s")
log = logging.getLogger(__name__)
log.info(f"Cargando modelo {MODEL_NAME}...")
model = SentenceTransformer(MODEL_NAME)
client = MongoClient(MONGO_URL)
db = client[DB_NAME]
log.info("Modelo listo")
def process_batch():
docs = list(db.noticias.find(
{"traducido": True, "procesado_embedding": False, "source_type": "rss"},
{"_id": 1, "texto": 1}
).limit(BATCH_SIZE))
if not docs:
return 0
textos = [d.get("texto", "")[:512] for d in docs]
embeddings = model.encode(textos, show_progress_bar=False)
for doc, emb in zip(docs, embeddings):
db.noticias.update_one(
{"_id": doc["_id"]},
{"$set": {"embedding": emb.tolist(), "procesado_embedding": True}}
)
log.info(f"Embeddings: {len(docs)} docs procesados")
return len(docs)
if __name__ == "__main__":
from _resiliencia import bucle_resiliente
# Si mongod cae, reintenta con backoff en vez de morir (evita el crash-loop
# que causó el incidente 2026-05-18). El modelo ya está cargado a nivel módulo.
bucle_resiliente(process_batch, sleep_trabajo=2, sleep_vacio=SLEEP_INTERVAL)

101
coconews/workers/import_feeds.py Normal file
View file

@ -0,0 +1,101 @@
#!/usr/bin/env python3
"""
Importa feeds desde data/feeds.csv de coconews a MongoDB coconews_feeds.
Mapea categorías de coconews → temas de flujos.
Ejecutar una sola vez (o para re-sincronizar).
"""
import csv
import os
import sys
from pathlib import Path
from pymongo import MongoClient
from pymongo.errors import DuplicateKeyError
MONGO_URL = os.getenv("MONGO_URL", "mongodb://localhost:27017")
DB_NAME = os.getenv("DB_NAME", "FLUJOS_DATOS")
CSV_PATH = Path(__file__).parent.parent / "feeds.csv"
# Mapeo categoría coconews → tema flujos
CATEGORIA_TEMA = {
"Internacional": "guerra global",
"Conflictos": "guerra global",
"Geopolítica": "guerra global",
"Seguridad": "inteligencia y seguridad",
"Inteligencia": "inteligencia y seguridad",
"Espionaje": "inteligencia y seguridad",
"Medio Ambiente": "cambio climático",
"Clima": "cambio climático",
"Energía": "cambio climático",
"Economía": "economía y corporaciones",
"Finanzas": "economía y corporaciones",
"Tecnología": "economía y corporaciones",
"Sociedad": "demografía y sociedad",
"Derechos Humanos": "demografía y sociedad",
"Política": "demografía y sociedad",
}
def map_tema(categoria: str) -> str:
if not categoria:
return "otros"
for key, tema in CATEGORIA_TEMA.items():
if key.lower() in categoria.lower():
return tema
return "otros"
def main():
if not CSV_PATH.exists():
print(f"ERROR: No se encuentra {CSV_PATH}")
print("Copia feeds.csv al directorio coconews/ antes de ejecutar este script")
sys.exit(1)
client = MongoClient(MONGO_URL, serverSelectionTimeoutMS=5000)
db = client[DB_NAME]
db.coconews_feeds.create_index([("url", 1)], unique=True)
imported = skipped = updated = 0
with open(CSV_PATH, newline="", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row in reader:
url = row.get("url", "").strip()
nombre = row.get("nombre", "").strip()
categoria= row.get("categoria", "").strip()
pais = row.get("pais", "").strip()
idioma = row.get("idioma", "").strip()[:2] if row.get("idioma") else "en"
activo_s = row.get("activo", "True")
activo = str(activo_s).strip().lower() in ("true", "1", "yes")
fallos = int(row.get("fallos", 0) or 0)
if not url or not nombre:
skipped += 1
continue
doc = {
"nombre": nombre,
"url": url,
"tema": map_tema(categoria),
"subtema": pais or "Internacional",
"idioma": idioma,
"activo": activo and fallos < 10,
"fallos": min(fallos, 10),
"last_fetch": None,
"last_error": None,
}
result = db.coconews_feeds.update_one(
{"url": url},
{"$setOnInsert": doc},
upsert=True
)
if result.upserted_id:
imported += 1
else:
skipped += 1
print(f"Importados: {imported} | Ya existían: {skipped}")
total = db.coconews_feeds.count_documents({})
activos = db.coconews_feeds.count_documents({"activo": True})
print(f"Total en BD: {total} feeds ({activos} activos)")
if __name__ == "__main__":
main()

225
coconews/workers/ingestor.py Normal file
View file

@ -0,0 +1,225 @@
#!/usr/bin/env python3
import hashlib
import logging
import os
import socket
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import datetime, timezone
import re
import feedparser
import requests
from bs4 import BeautifulSoup
from pymongo import MongoClient, ASCENDING
from pymongo.errors import DuplicateKeyError
MONGO_URL = os.getenv("MONGO_URL", "mongodb://localhost:27017")
DB_NAME = os.getenv("DB_NAME", "FLUJOS_DATOS")
SLEEP_INTERVAL = int(os.getenv("INGESTOR_SLEEP", "900"))
MAX_WORKERS = int(os.getenv("INGESTOR_WORKERS", "20"))
# feedparser.parse() NO acepta timeout= y usa el del socket, que por defecto es
# infinito: el 2026-08-18 un feed (Rappler) acepto la conexion y no respondio
# nunca, el hilo quedo bloqueado, el ThreadPoolExecutor no cerro y el ciclo no
# volvio a terminar -> 5 dias sin ingerir nada. Un timeout global lo evita.
FEED_TIMEOUT = int(os.getenv("INGESTOR_FEED_TIMEOUT", "20"))
socket.setdefaulttimeout(FEED_TIMEOUT)
# Backfill de imagenes integrado: cada ciclo rellena un lote de noticias
# recientes sin imagen via og:image (auto-reparacion del historico).
BACKFILL_IMG = os.getenv("INGESTOR_BACKFILL_IMG", "1") == "1"
BACKFILL_DIAS = int(os.getenv("INGESTOR_BACKFILL_DIAS", "90"))
BACKFILL_LOTE = int(os.getenv("INGESTOR_BACKFILL_LOTE", "300"))
logging.basicConfig(level=logging.INFO, format="%(asctime)s [ingestor] %(message)s")
log = logging.getLogger(__name__)
client = MongoClient(MONGO_URL)
db = client[DB_NAME]
def ensure_indexes():
db.noticias.create_index("archivo", unique=True)
db.noticias.create_index([("fecha", ASCENDING)])
db.noticias.create_index("source_type")
db.noticias.create_index("traducido")
db.noticias.create_index("procesado_ner")
db.noticias.create_index("procesado_embedding")
db.noticias.create_index("tema")
def parse_date(entry):
for attr in ("published_parsed", "updated_parsed"):
t = getattr(entry, attr, None)
if t:
return datetime(*t[:6], tzinfo=timezone.utc)
return datetime.now(timezone.utc)
_IMG_RE = re.compile(r'''<img[^>]+src=["\']([^"\']+)["\']''', re.I)
_HEADERS = {"User-Agent": "Mozilla/5.0 (compatible; coconews/1.0)"}
def _img_de_html(html):
"""Primera <img> con http(s) dentro de un fragmento HTML (summary/content)."""
if not html:
return None
m = _IMG_RE.search(html)
if m and m.group(1).strip().startswith("http"):
return m.group(1).strip()
return None
def image_from_entry(entry):
"""Imagen SIN tocar la red: media_*, enclosures, links o <img> del resumen."""
if getattr(entry, "media_thumbnail", None):
u = entry.media_thumbnail[0].get("url")
if u:
return u
if getattr(entry, "media_content", None):
u = entry.media_content[0].get("url")
if u:
return u
for lnk in getattr(entry, "links", None) or []:
if lnk.get("type", "").startswith("image/") and lnk.get("href"):
return lnk["href"]
u = _img_de_html(getattr(entry, "summary", "") or "")
if u:
return u
for c in getattr(entry, "content", None) or []:
u = _img_de_html(c.get("value", ""))
if u:
return u
return None
def og_image(url, timeout=4):
"""Fallback CON red: og:image / twitter:image de la pagina del articulo.
Timeout corto y todo en try/except: nunca debe frenar ni romper el ingestor."""
try:
r = requests.get(url, headers=_HEADERS, timeout=timeout)
if r.status_code != 200 or "html" not in r.headers.get("Content-Type", ""):
return None
soup = BeautifulSoup(r.text, "html.parser")
for prop in ("og:image", "twitter:image", "og:image:url"):
tag = (soup.find("meta", property=prop) or soup.find("meta", attrs={"name": prop}))
if tag and tag.get("content", "").strip().startswith("http"):
return tag["content"].strip()
except Exception:
pass
return None
def get_image(entry, link=None, con_red=True):
"""Imagen del entry; si el feed no trae, cae a og:image de la pagina."""
u = image_from_entry(entry)
if u:
return u
if con_red and link:
return og_image(link)
return None
def process_feed(feed_doc):
url = feed_doc["url"]
tema = feed_doc.get("tema", "general")
sub = feed_doc.get("subtema", "")
try:
parsed = feedparser.parse(url, request_headers={"User-Agent": "Mozilla/5.0"})
if getattr(parsed, "bozo", 0) and not parsed.entries:
raise RuntimeError(getattr(parsed, "bozo_exception", "feed ilegible"))
inserted = 0
for entry in parsed.entries:
link = getattr(entry, "link", None)
if not link:
continue
archivo = hashlib.md5(link.encode()).hexdigest() + ".txt"
# Saltar duplicados ANTES del og:image (evita un GET de red inutil)
if db.noticias.find_one({"archivo": archivo}, {"_id": 1}):
continue
resumen = getattr(entry, "summary", "") or ""
doc = {
"archivo": archivo,
"url": link,
"titulo_original": getattr(entry, "title", ""),
"resumen_original": resumen[:2000],
"texto": "",
"fecha": parse_date(entry),
"imagen_url": get_image(entry, link),
"tema": tema,
"subtema": sub,
"fuente": feed_doc.get("nombre", ""),
"lang": feed_doc.get("idioma", "en"),
"source_type": "rss",
"traducido": False,
"procesado_ner": False,
"procesado_embedding": False,
}
try:
db.noticias.insert_one(doc)
inserted += 1
except DuplicateKeyError:
pass
if inserted:
log.info(f"{feed_doc['nombre']}: {inserted} nuevas")
db.coconews_feeds.update_one(
{"_id": feed_doc["_id"]},
{"$set": {"ultimo_ok": datetime.now(timezone.utc), "fallos": 0}}
)
except Exception as e:
log.warning(f"{feed_doc['nombre']} ERROR: {e}")
db.coconews_feeds.update_one(
{"_id": feed_doc["_id"]},
{"$inc": {"fallos": 1}}
)
def backfill_imagenes(dias, lote):
"""Rellena imagen_url (og:image) en un lote de noticias recientes sin imagen.
Se ejecuta al final de cada ciclo del ingestor: el historico se auto-repara
progresivamente sin depender de ejecuciones manuales."""
from datetime import timedelta
desde = datetime.now(timezone.utc) - timedelta(days=dias)
q = {"source_type": "rss", "fecha": {"$gte": desde},
"$or": [{"imagen_url": None}, {"imagen_url": ""}], "url": {"$exists": True}}
docs = list(db.noticias.find(q, {"_id": 1, "url": 1}).limit(lote))
if not docs:
return
rescatadas = 0
def _una(d):
nonlocal rescatadas
img = og_image(d["url"], timeout=5)
if img:
db.noticias.update_one({"_id": d["_id"]}, {"$set": {"imagen_url": img}})
rescatadas += 1
with ThreadPoolExecutor(max_workers=min(MAX_WORKERS, 12)) as ex:
list(ex.map(_una, docs))
if rescatadas:
log.info(f"backfill imagenes: {rescatadas}/{len(docs)} rescatadas (ventana {dias}d)")
def run_cycle():
feeds = list(db.coconews_feeds.find({"activo": True}))
log.info(f"Procesando {len(feeds)} feeds...")
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as ex:
ex.map(process_feed, feeds)
log.info("Ciclo completado")
if BACKFILL_IMG:
try:
backfill_imagenes(BACKFILL_DIAS, BACKFILL_LOTE)
except Exception as e:
log.warning(f"backfill imagenes ERROR: {e}")
if __name__ == "__main__":
from _resiliencia import bucle_resiliente
_idx = {"done": False}
def paso():
if not _idx["done"]: # crea índices la 1ª vez; si Mongo está caído,
ensure_indexes() # el helper reintenta sin matar el proceso
_idx["done"] = True
run_cycle()
return 1
bucle_resiliente(paso, sleep_trabajo=SLEEP_INTERVAL, sleep_vacio=SLEEP_INTERVAL)

77
coconews/workers/ner.py Normal file
View file

@ -0,0 +1,77 @@
#!/usr/bin/env python3
import logging
import os
import time
from collections import Counter
from datetime import datetime, timezone
import spacy
from pymongo import MongoClient
MONGO_URL = os.getenv("MONGO_URL", "mongodb://localhost:27017")
DB_NAME = os.getenv("DB_NAME", "FLUJOS_DATOS")
BATCH_SIZE = int(os.getenv("NER_BATCH", "30"))
SLEEP_INTERVAL = int(os.getenv("NER_SLEEP", "30"))
logging.basicConfig(level=logging.INFO, format="%(asctime)s [ner] %(message)s")
log = logging.getLogger(__name__)
nlp = spacy.load("es_core_news_sm")
client = MongoClient(MONGO_URL)
db = client[DB_NAME]
TIPO_MAP = {"PER": "persona", "ORG": "organizacion", "LOC": "lugar", "MISC": "misc"}
def extract_entities(texto):
doc = nlp(texto[:10000])
entidades = []
seen = set()
for ent in doc.ents:
tipo = TIPO_MAP.get(ent.label_, None)
if not tipo or tipo == "misc":
continue
key = (ent.text.strip(), tipo)
if key in seen or len(ent.text.strip()) < 2:
continue
seen.add(key)
entidades.append({"texto": ent.text.strip(), "tipo": tipo})
return entidades
def update_entidades_collection(entidades, tema):
for ent in entidades:
db.coconews_entidades.update_one(
{"texto": ent["texto"], "tipo": ent["tipo"]},
{
"$inc": {"count": 1},
"$addToSet": {"temas": tema},
"$set": {"ultima_vez": datetime.now(timezone.utc)},
},
upsert=True,
)
def process_batch():
docs = list(db.noticias.find(
{"traducido": True, "procesado_ner": False, "source_type": "rss"},
{"_id": 1, "texto": 1, "tema": 1}
).limit(BATCH_SIZE))
for doc in docs:
texto = doc.get("texto", "")
entidades = extract_entities(texto)
tema = doc.get("tema", "")
update_entidades_collection(entidades, tema)
db.noticias.update_one(
{"_id": doc["_id"]},
{"$set": {"entidades": entidades, "procesado_ner": True}}
)
log.info(f"NER: {len(entidades)} entidades en doc {doc['_id']}")
return len(docs)
if __name__ == "__main__":
from _resiliencia import bucle_resiliente
bucle_resiliente(process_batch, sleep_trabajo=1, sleep_vacio=SLEEP_INTERVAL)

56
coconews/workers/translator.py Normal file
View file

@ -0,0 +1,56 @@
#!/usr/bin/env python3
import logging
import os
import time
from deep_translator import GoogleTranslator, MyMemoryTranslator
from pymongo import MongoClient
MONGO_URL = os.getenv("MONGO_URL", "mongodb://localhost:27017")
DB_NAME = os.getenv("DB_NAME", "FLUJOS_DATOS")
BATCH_SIZE = int(os.getenv("TRANSLATOR_BATCH", "20"))
SLEEP_INTERVAL = int(os.getenv("TRANSLATOR_SLEEP", "60"))
logging.basicConfig(level=logging.INFO, format="%(asctime)s [translator] %(message)s")
log = logging.getLogger(__name__)
client = MongoClient(MONGO_URL)
db = client[DB_NAME]
def translate(text, src="auto"):
if not text or not text.strip():
return ""
text = text[:4500]
try:
return GoogleTranslator(source=src, target="es").translate(text) or text
except Exception:
try:
return MyMemoryTranslator(source="en-US", target="es-ES").translate(text) or text
except Exception:
return text
def process_batch():
docs = list(db.noticias.find(
{"traducido": False, "source_type": "rss"},
{"_id": 1, "titulo_original": 1, "resumen_original": 1, "lang": 1}
).limit(BATCH_SIZE))
for doc in docs:
src = doc.get("lang", "auto") or "auto"
tit = translate(doc.get("titulo_original", ""), src)
res = translate(doc.get("resumen_original", ""), src)
texto = f"{tit}\n\n{res}".strip()
db.noticias.update_one(
{"_id": doc["_id"]},
{"$set": {"texto": texto, "traducido": True, "procesado_ner": False}}
)
log.info(f"Traducida: {tit[:60]}")
return len(docs)
if __name__ == "__main__":
from _resiliencia import bucle_resiliente
bucle_resiliente(process_batch, sleep_trabajo=2, sleep_vacio=SLEEP_INTERVAL)