pietre/rss2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

mejora: rendimiento captura RSS y velocidad de traducción

Browse source 
- Ingestor: reactivar ETag/Last-Modified (304), http.Client compartido, rate-limit por host y Retry-After
- Traducción: batch 128, cache en Redis con TTL (efímera), commit por lote
- Scraper: worker pool paralelo (SCRAPER_WORKERS)
- Config: compose lee RSS_/TRANSLATOR_ variables del .env (fuente única)
This commit is contained in:
jlimolina 2026-08-10 18:27:34 +02:00
parent 71d345ec35
commit e3936fd4cf
6 changed files with 239 additions and 48 deletions
Download patch file Download diff file Expand all files Collapse all files

1
.gitignore vendored
View file

@ -67,6 +67,7 @@ models/nllb-ct2-1.3b/
# Wikipedia images - downloaded by wiki_worker
data/wiki_images/
data/backups/
backups/
# Database storage (PostgreSQL)
data/pgdata/

3
Dockerfile.translator
View file

@ -23,7 +23,8 @@ RUN pip install --no-cache-dir \
protobuf==3.20.3 \
"numpy<2" \
psycopg2-binary \
langdetect
langdetect \
redis
# === ARREGLAR EL EXECUTABLE STACK ===
RUN find /usr/local/lib/python3.11/site-packages/ctranslate2* \

54
backend/cmd/scraper/main.go
View file

@ -10,6 +10,7 @@ import (
"os/signal"
"strconv"
"strings"
"sync"
"syscall"
"time"
@ -25,6 +26,7 @@ var (
sleepInterval = 60
batchSize = 10
enrichLimit = 20
scraperWorkers = 5
)
type URLSource struct {
@ -63,6 +65,7 @@ func loadConfig() {
sleepInterval = getEnvInt("SCRAPER_SLEEP", 60)
batchSize = getEnvInt("SCRAPER_BATCH", 10)
enrichLimit = getEnvInt("SCRAPER_ENRICH_LIMIT", 20)
scraperWorkers = getEnvInt("SCRAPER_WORKERS", 5)
}
func getEnvInt(key string, defaultValue int) int {
@ -406,6 +409,41 @@ func processSource(ctx context.Context, source URLSource) {
}
}
// enrichConcurrent enriches noticias in parallel with a bounded worker pool.
func enrichConcurrent(ctx context.Context, noticias []Noticia) {
limiter := make(chan struct{}, scraperWorkers)
var wg sync.WaitGroup
for _, noticia := range noticias {
wg.Add(1)
limiter <- struct{}{}
go func(n Noticia) {
defer wg.Done()
defer func() { <-limiter }()
if processEnrichment(ctx, n) {
time.Sleep(1 * time.Second)
}
}(noticia)
}
wg.Wait()
}
// scrapeSources processes fuentes_url in parallel with a bounded worker pool.
func scrapeSources(ctx context.Context, sources []URLSource) {
limiter := make(chan struct{}, scraperWorkers)
var wg sync.WaitGroup
for _, source := range sources {
wg.Add(1)
limiter <- struct{}{}
go func(src URLSource) {
defer wg.Done()
defer func() { <-limiter }()
time.Sleep(1 * time.Second) // polite pacing across workers
processSource(ctx, src)
}(source)
}
wg.Wait()
}
func main() {
loadConfig()
logger.Println("Starting Scraper Worker")
@ -443,12 +481,8 @@ func main() {
if err != nil {
logger.Printf("Error fetching noticias to enrich: %v", err)
} else if len(noticias) > 0 {
logger.Printf("Enriching %d noticias", len(noticias))
for _, noticia := range noticias {
if processEnrichment(ctx, noticia) {
time.Sleep(3 * time.Second)
}
}
logger.Printf("Enriching %d noticias (%d workers)", len(noticias), scraperWorkers)
enrichConcurrent(ctx, noticias)
}
sources, err := getActiveURLs(ctx)
@ -462,12 +496,8 @@ func main() {
continue
}
logger.Printf("Processing %d sources", len(sources))
for _, source := range sources {
processSource(ctx, source)
time.Sleep(2 * time.Second) // Rate limiting
}
logger.Printf("Processing %d sources (%d workers)", len(sources), scraperWorkers)
scrapeSources(ctx, sources)
}
}
}

12
docker-compose.yml
View file

@ -92,8 +92,9 @@ services:
DB_NAME: ${DB_NAME:-rss}
DB_USER: ${DB_USER:-rss}
DB_PASS: ${DB_PASS}
RSS_MAX_WORKERS: 100
RSS_POKE_INTERVAL_MIN: 60
RSS_MAX_WORKERS: ${RSS_MAX_WORKERS:-100}
RSS_POKE_INTERVAL_MIN: ${RSS_POKE_INTERVAL_MIN:-60}
RSS_HOST_DELAY_MS: ${RSS_HOST_DELAY_MS:-500}
TZ: Europe/Madrid
networks:
- backend
@ -155,6 +156,7 @@ services:
SCRAPER_SLEEP: 60
SCRAPER_BATCH: 10
SCRAPER_ENRICH_LIMIT: 20
SCRAPER_WORKERS: ${SCRAPER_WORKERS:-5}
TZ: Europe/Madrid
networks:
- backend
@ -304,7 +306,7 @@ services:
DB_USER: ${DB_USER:-rss}
DB_PASS: ${DB_PASS}
TARGET_LANGS: es
TRANSLATOR_BATCH: 32
TRANSLATOR_BATCH: ${TRANSLATOR_BATCH:-128}
CT2_MODEL_PATH: /app/models/nllb-ct2
CT2_DEVICE: cpu
CT2_COMPUTE_TYPE: int8
@ -313,6 +315,7 @@ services:
HF_HOME: /app/hf_cache
TZ: Europe/Madrid
TRANSLATOR_ID: ${TRANSLATOR_ID:-}
REDIS_URL: redis://:${REDIS_PASSWORD:-rss_redis_pass_2024}@redis:6379
PYTORCH_ENABLE_MPS_FALLBACK: 1
volumes:
- ./workers:/app/workers
@ -341,7 +344,7 @@ services:
DB_USER: ${DB_USER:-rss}
DB_PASS: ${DB_PASS}
TARGET_LANGS: es
TRANSLATOR_BATCH: 32
TRANSLATOR_BATCH: ${TRANSLATOR_BATCH:-128}
CT2_MODEL_PATH: /app/models/nllb-ct2
CT2_DEVICE: cpu
CT2_COMPUTE_TYPE: int8
@ -350,6 +353,7 @@ services:
HF_HOME: /app/hf_cache
TZ: Europe/Madrid
TRANSLATOR_ID: ${TRANSLATOR_ID:-}
REDIS_URL: redis://:${REDIS_PASSWORD:-rss_redis_pass_2024}@redis:6379
PYTORCH_ENABLE_MPS_FALLBACK: 1
volumes:
- ./workers:/app/workers

91
rss-ingestor-go/main.go
View file

@ -7,6 +7,7 @@ import (
"fmt"
"log"
"net/http"
"net/url"
"os"
"strconv"
"strings"
@ -29,6 +30,7 @@ type Config struct {
MaxFailures int
PokeInterval time.Duration
FeedTimeout int
HostDelay time.Duration
}
// Feed represents a row in the feeds table
@ -57,10 +59,31 @@ type Noticia struct {
}
var (
db *sql.DB
config Config
db *sql.DB
config Config
httpClient *http.Client
)
// Per-host politeness limiter so parallel workers don't hammer one domain.
var (
rateMu sync.Mutex
lastRequest = make(map[string]time.Time)
)
func politeDelay(host string) {
if config.HostDelay <= 0 || host == "" {
return
}
rateMu.Lock()
now := time.Now()
next := lastRequest[host].Add(config.HostDelay)
lastRequest[host] = next
rateMu.Unlock()
if wait := next.Sub(now); wait > 0 {
time.Sleep(wait)
}
}
func loadConfig() {
config = Config{
DBHost: getEnv("DB_HOST", "localhost"),
@ -72,6 +95,19 @@ func loadConfig() {
MaxFailures: getEnvInt("RSS_MAX_FAILURES", 10),
PokeInterval: time.Duration(getEnvInt("RSS_POKE_INTERVAL_MIN", 8)) * time.Minute,
FeedTimeout: getEnvInt("RSS_FEED_TIMEOUT", 60),
HostDelay: time.Duration(getEnvInt("RSS_HOST_DELAY_MS", 500)) * time.Millisecond,
}
}
func initHTTPClient() {
httpClient = &http.Client{
Timeout: time.Duration(config.FeedTimeout) * time.Second,
Transport: &http.Transport{
MaxIdleConns: config.MaxWorkers * 2,
MaxIdleConnsPerHost: 2,
IdleConnTimeout: 90 * time.Second,
TLSHandshakeTimeout: 10 * time.Second,
},
}
}
@ -200,11 +236,9 @@ func extractImage(item *gofeed.Item) string {
}
func processFeed(fp *gofeed.Parser, feed Feed, results chan<- int) {
// Configure custom HTTP client with timeout and User-Agent
client := &http.Client{
Timeout: time.Duration(config.FeedTimeout) * time.Second,
}
// Shared HTTP client with connection pooling across workers
client := httpClient
// Create request to set User-Agent
req, err := http.NewRequest("GET", feed.URL, nil)
if err != nil {
@ -215,10 +249,21 @@ func processFeed(fp *gofeed.Parser, feed Feed, results chan<- int) {
}
req.Header.Set("User-Agent", "RSS2-Ingestor-Go/1.0")
// NOTE: We INTENTIONALLY SKIP ETag/Last-Modified headers based on user issues
// If needed in future, uncomment:
// if feed.LastEtag.Valid { req.Header.Set("If-None-Match", feed.LastEtag.String) }
// if feed.LastModified.Valid { req.Header.Set("If-Modified-Since", feed.LastModified.String) }
// Reuse ETag/Last-Modified so servers can reply 304 Not Modified,
// avoiding re-downloading unchanged feeds in every cycle.
if feed.LastEtag.Valid && feed.LastEtag.String != "" {
req.Header.Set("If-None-Match", feed.LastEtag.String)
}
if feed.LastModified.Valid && feed.LastModified.String != "" {
req.Header.Set("If-Modified-Since", feed.LastModified.String)
}
// Per-host politeness: pace parallel fetches to the same domain
host := feed.URL
if u, perr := url.Parse(feed.URL); perr == nil && u.Host != "" {
host = u.Host
}
politeDelay(host)
resp, err := client.Do(req)
if err != nil {
@ -229,10 +274,29 @@ func processFeed(fp *gofeed.Parser, feed Feed, results chan<- int) {
}
defer resp.Body.Close()
// Honor Retry-After on rate limits without punishing the feed with failures
if resp.StatusCode == 429 || resp.StatusCode == 503 {
ra, raErr := strconv.Atoi(resp.Header.Get("Retry-After"))
if raErr == nil && ra > 0 && ra <= 3600 {
log.Printf("[Feed %d] Rate limited (%d), sleeping %ds (Retry-After)", feed.ID, resp.StatusCode, ra)
time.Sleep(time.Duration(ra) * time.Second)
}
results <- 0
return
}
// On 304, keep the stored validators; update only if the server sends new ones.
if resp.StatusCode == 304 {
log.Printf("[Feed %d] Not Modified (304)", feed.ID)
// Update timestamp only? Or keep as is.
updateFeedStatus(feed.ID, feed.LastEtag.String, feed.LastModified.String, true, "")
newEtag := feed.LastEtag.String
newModified := feed.LastModified.String
if h := resp.Header.Get("ETag"); h != "" {
newEtag = h
}
if h := resp.Header.Get("Last-Modified"); h != "" {
newModified = h
}
updateFeedStatus(feed.ID, newEtag, newModified, true, "")
results <- 0
return
}
@ -433,6 +497,7 @@ func ingestCycle() {
func main() {
loadConfig()
initHTTPClient()
initDB()
// Run immediately on start

126
workers/ctranslator_worker.py
View file

@ -1,8 +1,9 @@
import os
import re
import time
import logging
import re
import fcntl
import hashlib
from typing import List, Optional
import psycopg2
@ -20,6 +21,62 @@ LOG = logging.getLogger("translator_ct2")
TRANSLATOR_ID = os.environ.get("TRANSLATOR_ID", "")
TRANSLATOR_TOTAL = int(os.environ.get("TRANSLATOR_TOTAL", "1"))
CACHE_TTL = int(os.environ.get("TRANSLATION_CACHE_TTL", str(30 * 24 * 3600)))
_redis_client = None
def get_redis():
"""Return a Redis client or None. Cache is best-effort: never blocks translation."""
global _redis_client
if _redis_client is not None:
return _redis_client if _redis_client is not False else None
url = os.environ.get("REDIS_URL", "redis://localhost:6379")
try:
import redis
client = redis.Redis.from_url(
url, decode_responses=True,
socket_connect_timeout=2, socket_timeout=2,
)
client.ping()
_redis_client = client
LOG.info("Redis translation cache enabled")
return client
except Exception as e:
LOG.warning(f"Redis cache unavailable, translating without cache: {e}")
_redis_client = False
return None
def cache_key(lang_from: str, lang_to: str, text: str) -> str:
digest = hashlib.md5((text or "").encode("utf-8", "ignore")).hexdigest()
return f"tr:{lang_from}:{lang_to}:{digest}"
def lookup_cache(keys: List[str]) -> dict:
r = get_redis()
if not r or not keys:
return {}
try:
vals = r.mget(keys)
return {k: v for k, v in zip(keys, vals) if v}
except Exception as e:
LOG.warning(f"Redis mget error: {e}")
return {}
def store_cache(pairs, ttl: int = CACHE_TTL):
r = get_redis()
if not r or not pairs:
return
try:
pipe = r.pipeline(transaction=False)
for k, v in pairs:
if v:
pipe.setex(k, ttl, v)
pipe.execute()
except Exception as e:
LOG.warning(f"Redis cache store error: {e}")
def clean_text(text: str) -> str:
if not text:
@ -403,33 +460,64 @@ def process_batch(conn, rows):
LOG.info(f"Translating {lang_from} -> {lang_to} ({len(items)} items)")
try:
# --- TITLES (served from Redis cache when possible) ---
titles = [i["titulo"] for i in items]
translated_titles = translate_texts(lang_from, lang_to, titles)
title_keys = [cache_key(lang_from, lang_to, t) for t in titles]
title_cache = lookup_cache(title_keys)
if title_cache:
LOG.info(f"Title cache hits: {len(title_cache)}/{len(titles)}")
# Collect all body chunks across all items for a single batched call
translated_titles = [title_cache.get(k) for k in title_keys]
title_miss = [i for i, k in enumerate(title_keys) if k not in title_cache]
if title_miss:
miss_texts = [titles[i] for i in title_miss]
miss_translated = translate_texts(lang_from, lang_to, miss_texts)
store_cache([(title_keys[i], tr) for i, tr in zip(title_miss, miss_translated)])
for i, tr in zip(title_miss, miss_translated):
translated_titles[i] = tr
# --- BODY chunks (single batched call for cache misses) ---
flat_chunks = []
flat_keys = []
flat_ck = []
for item in items:
body = (item["resumen"] or "").strip()
if body:
chunks = split_body_into_chunks(body)
flat_chunks.extend(chunks)
flat_keys.extend([(item["tr_id"], i) for i in range(len(chunks))])
flat_ck.extend([cache_key(lang_from, lang_to, c) for c in chunks])
translated_bodies = []
if flat_chunks:
try:
translated_bodies = translate_texts(lang_from, lang_to, flat_chunks)
except Exception as e:
LOG.error(f"Batch body translation error: {e}")
translated_bodies = flat_chunks
chunk_cache = lookup_cache(flat_ck)
if chunk_cache:
LOG.info(f"Body cache hits: {len(chunk_cache)}/{len(flat_ck)}")
body_parts = defaultdict(list)
for (tr_id, _), tr in zip(flat_keys, translated_bodies):
if tr is None:
continue
body_parts[tr_id].append(tr)
miss_idx = []
miss_texts = []
miss_keys = []
for idx, ck in enumerate(flat_ck):
tr_id, _ = flat_keys[idx]
if ck in chunk_cache:
body_parts[tr_id].append(chunk_cache[ck])
else:
miss_idx.append(idx)
miss_texts.append(flat_chunks[idx])
miss_keys.append(ck)
if miss_texts:
try:
miss_translated = translate_texts(lang_from, lang_to, miss_texts)
store_cache(list(zip(miss_keys, miss_translated)))
except Exception as e:
LOG.error(f"Batch body translation error: {e}")
miss_translated = miss_texts
for j, tr in enumerate(miss_translated):
if tr:
body_parts[flat_keys[miss_idx[j]][0]].append(tr)
# --- BATCH COMMIT: single transaction per group ---
updates = []
for idx, item in enumerate(items):
tt = clean_text((translated_titles[idx] or "").strip())
parts = body_parts.get(item["tr_id"])
@ -440,21 +528,23 @@ def process_batch(conn, rows):
if not tb:
tb = item["resumen"]
# 2. INDIVIDUAL COMMIT: Save each item as it's done
updates.append((tt, tb, item["tr_id"]))
if updates:
try:
cursor = conn.cursor()
cursor.execute(
cursor.executemany(
"""
UPDATE traducciones
SET titulo_trad = %s, resumen_trad = %s, status = 'done', locked_at = NULL
WHERE id = %s
""",
(tt, tb, item["tr_id"]),
updates,
)
conn.commit()
cursor.close()
except Exception as e:
LOG.error(f"Update error for ID {item['tr_id']}: {e}")
LOG.error(f"Batch update error (items stay pending for retry): {e}")
conn.rollback()
LOG.info(f"Finished group {lang_from} -> {lang_to}")