268 lines
12 KiB
Python
268 lines
12 KiB
Python
import logging
|
|
import json
|
|
|
|
# Importar MongoDBArticleCache desde mongodb_manager
|
|
from managers.mongodb_manager import MongoDBArticleCache
|
|
|
|
NOTIFIED_ARTICLE_TTL = 7 * 24 * 60 * 60 # TTL de 7 días en segundos para artículos notificados (mantener para compatibilidad)
|
|
|
|
class RedisArticleCache:
|
|
"""Maneja el cache de artículos notificados usando Redis"""
|
|
|
|
def __init__(self, redis_host='localhost', redis_port=6379, redis_db=0, redis_password=None):
|
|
self.logger = logging.getLogger(__name__)
|
|
|
|
# Inicializar conexión Redis
|
|
try:
|
|
self._redis_client = redis.Redis(
|
|
host=redis_host,
|
|
port=redis_port,
|
|
db=redis_db,
|
|
password=redis_password,
|
|
decode_responses=True,
|
|
socket_connect_timeout=5,
|
|
socket_timeout=5
|
|
)
|
|
# Verificar conexión
|
|
self._redis_client.ping()
|
|
self.logger.info(f"Conectado a Redis en {redis_host}:{redis_port} (db={redis_db})")
|
|
except (redis.ConnectionError, redis.TimeoutError) as e:
|
|
self.logger.error(f"Error conectando a Redis: {e}")
|
|
self.logger.error("Redis no está disponible. El sistema no podrá evitar duplicados sin Redis.")
|
|
raise
|
|
except Exception as e:
|
|
self.logger.error(f"Error inesperado inicializando Redis: {e}")
|
|
raise
|
|
|
|
def _get_article_key(self, article):
|
|
"""Genera una clave única para un artículo en Redis"""
|
|
return f"notified:{article.get_platform()}:{article.get_id()}"
|
|
|
|
def is_article_notified(self, article):
|
|
"""Verifica si un artículo ya ha sido notificado"""
|
|
try:
|
|
key = self._get_article_key(article)
|
|
return self._redis_client.exists(key) > 0
|
|
except Exception as e:
|
|
self.logger.error(f"Error verificando artículo en Redis: {e}")
|
|
return False
|
|
|
|
def mark_article_as_notified(self, article, username=None, worker_name=None):
|
|
"""Marca un artículo como notificado en Redis con TTL, guardando toda la información del artículo"""
|
|
try:
|
|
key = self._get_article_key(article)
|
|
# Guardar toda la información del artículo como JSON
|
|
# Verificar si el artículo ya existe para mantener el estado de favorito, username y worker_name
|
|
existing_value = self._redis_client.get(key)
|
|
is_favorite = False
|
|
existing_username = None
|
|
existing_worker_name = None
|
|
if existing_value:
|
|
try:
|
|
existing_data = json.loads(existing_value)
|
|
is_favorite = existing_data.get('is_favorite', False)
|
|
existing_username = existing_data.get('username')
|
|
existing_worker_name = existing_data.get('worker_name')
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Mantener username y worker_name existentes si ya existen, o usar los nuevos si se proporcionan
|
|
final_username = existing_username if existing_username else username
|
|
final_worker_name = existing_worker_name if existing_worker_name else worker_name
|
|
|
|
article_data = {
|
|
'id': article.get_id(),
|
|
'title': article.get_title(),
|
|
'description': article._description, # Acceder al campo privado para obtener la descripción completa
|
|
'price': article.get_price(),
|
|
'currency': article.get_currency(),
|
|
'location': article.get_location(),
|
|
'allows_shipping': article._allows_shipping, # Acceder al campo privado para obtener el valor booleano
|
|
'url': article.get_url(),
|
|
'images': article.get_images(),
|
|
'modified_at': article.get_modified_at(),
|
|
'platform': article.get_platform(),
|
|
'is_favorite': is_favorite, # Mantener el estado de favorito
|
|
}
|
|
|
|
# Añadir username y worker_name si están disponibles
|
|
if final_username:
|
|
article_data['username'] = final_username
|
|
if final_worker_name:
|
|
article_data['worker_name'] = final_worker_name
|
|
|
|
self._redis_client.setex(key, NOTIFIED_ARTICLE_TTL, json.dumps(article_data))
|
|
except Exception as e:
|
|
self.logger.error(f"Error marcando artículo como notificado en Redis: {e}")
|
|
|
|
def mark_articles_as_notified(self, articles, username=None, worker_name=None):
|
|
"""Añade múltiples artículos a la lista de artículos ya notificados en Redis"""
|
|
article_list = articles if isinstance(articles, list) else [articles]
|
|
|
|
try:
|
|
# Verificar qué artículos ya existen antes de añadirlos
|
|
# Usar pipeline para mejor rendimiento al verificar múltiples artículos
|
|
pipe = self._redis_client.pipeline()
|
|
keys_to_check = []
|
|
for article in article_list:
|
|
key = self._get_article_key(article)
|
|
keys_to_check.append((article, key))
|
|
pipe.exists(key)
|
|
|
|
# Ejecutar las verificaciones
|
|
exists_results = pipe.execute()
|
|
|
|
# Ahora añadir solo los artículos que no existen
|
|
pipe = self._redis_client.pipeline()
|
|
added_count = 0
|
|
skipped_count = 0
|
|
|
|
for (article, key), exists in zip(keys_to_check, exists_results):
|
|
if exists > 0:
|
|
# El artículo ya existe, no hacer nada
|
|
skipped_count += 1
|
|
continue
|
|
|
|
# El artículo no existe, añadirlo
|
|
# Guardar toda la información del artículo como JSON
|
|
article_data = {
|
|
'id': article.get_id(),
|
|
'title': article.get_title(),
|
|
'description': article._description, # Acceder al campo privado para obtener la descripción completa
|
|
'price': article.get_price(),
|
|
'currency': article.get_currency(),
|
|
'location': article.get_location(),
|
|
'allows_shipping': article._allows_shipping, # Acceder al campo privado para obtener el valor booleano
|
|
'url': article.get_url(),
|
|
'images': article.get_images(),
|
|
'modified_at': article.get_modified_at(),
|
|
'platform': article.get_platform(),
|
|
'is_favorite': False, # Por defecto no es favorito
|
|
}
|
|
|
|
# Añadir username y worker_name si están disponibles
|
|
if username:
|
|
article_data['username'] = username
|
|
if worker_name:
|
|
article_data['worker_name'] = worker_name
|
|
|
|
pipe.setex(key, NOTIFIED_ARTICLE_TTL, json.dumps(article_data))
|
|
added_count += 1
|
|
|
|
# Ejecutar solo si hay artículos para añadir
|
|
if added_count > 0:
|
|
pipe.execute()
|
|
|
|
self.logger.debug(f"{added_count} artículos añadidos, {skipped_count} ya existían en Redis")
|
|
except Exception as e:
|
|
self.logger.error(f"Error añadiendo artículos a Redis: {e}")
|
|
|
|
def set_favorite(self, platform, article_id, is_favorite=True):
|
|
"""Marca o desmarca un artículo como favorito en Redis"""
|
|
try:
|
|
key = f"notified:{platform}:{article_id}"
|
|
value = self._redis_client.get(key)
|
|
if value:
|
|
article_data = json.loads(value)
|
|
article_data['is_favorite'] = is_favorite
|
|
# Mantener el TTL existente o usar el default
|
|
ttl = self._redis_client.ttl(key)
|
|
if ttl > 0:
|
|
self._redis_client.setex(key, ttl, json.dumps(article_data))
|
|
else:
|
|
self._redis_client.setex(key, NOTIFIED_ARTICLE_TTL, json.dumps(article_data))
|
|
return True
|
|
return False
|
|
except Exception as e:
|
|
self.logger.error(f"Error marcando favorito en Redis: {e}")
|
|
return False
|
|
|
|
def get_favorites(self):
|
|
"""Obtiene todos los artículos marcados como favoritos"""
|
|
try:
|
|
keys = self._redis_client.keys('notified:*')
|
|
favorites = []
|
|
for key in keys:
|
|
value = self._redis_client.get(key)
|
|
if value:
|
|
try:
|
|
article_data = json.loads(value)
|
|
if article_data.get('is_favorite', False):
|
|
favorites.append(article_data)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
return favorites
|
|
except Exception as e:
|
|
self.logger.error(f"Error obteniendo favoritos de Redis: {e}")
|
|
return []
|
|
|
|
def is_favorite(self, platform, article_id):
|
|
"""Verifica si un artículo es favorito"""
|
|
try:
|
|
key = f"notified:{platform}:{article_id}"
|
|
value = self._redis_client.get(key)
|
|
if value:
|
|
article_data = json.loads(value)
|
|
return article_data.get('is_favorite', False)
|
|
return False
|
|
except Exception as e:
|
|
self.logger.error(f"Error verificando favorito en Redis: {e}")
|
|
return False
|
|
|
|
def clear_cache(self):
|
|
"""Elimina toda la caché de artículos notificados en Redis"""
|
|
try:
|
|
# Obtener todas las claves que empiezan con 'notified:'
|
|
keys = self._redis_client.keys('notified:*')
|
|
|
|
if not keys:
|
|
self.logger.info("Cache de Redis ya está vacío")
|
|
return 0
|
|
|
|
# Eliminar todas las claves usando pipeline para mejor rendimiento
|
|
count = len(keys)
|
|
pipe = self._redis_client.pipeline()
|
|
for key in keys:
|
|
pipe.delete(key)
|
|
pipe.execute()
|
|
|
|
self.logger.info(f"Cache de Redis limpiado: {count} artículos eliminados")
|
|
return count
|
|
except Exception as e:
|
|
self.logger.error(f"Error limpiando cache de Redis: {e}")
|
|
return 0
|
|
|
|
|
|
def create_article_cache(cache_type='mongodb', **kwargs):
|
|
"""
|
|
Factory function para crear el cache de artículos usando MongoDB.
|
|
|
|
Args:
|
|
cache_type: 'mongodb' (solo MongoDB está soportado)
|
|
**kwargs: Argumentos para MongoDB:
|
|
- mongodb_host: host de MongoDB (default: 'localhost')
|
|
- mongodb_port: puerto de MongoDB (default: 27017)
|
|
- mongodb_database: base de datos (default: 'wallabicher')
|
|
- mongodb_username: usuario de MongoDB (opcional)
|
|
- mongodb_password: contraseña de MongoDB (opcional)
|
|
- mongodb_auth_source: base de datos para autenticación (default: 'admin')
|
|
|
|
Returns:
|
|
MongoDBArticleCache
|
|
"""
|
|
if cache_type == 'mongodb':
|
|
return MongoDBArticleCache(
|
|
mongodb_host=kwargs.get('mongodb_host', 'localhost'),
|
|
mongodb_port=kwargs.get('mongodb_port', 27017),
|
|
mongodb_database=kwargs.get('mongodb_database', 'wallabicher'),
|
|
mongodb_username=kwargs.get('mongodb_username'),
|
|
mongodb_password=kwargs.get('mongodb_password'),
|
|
mongodb_auth_source=kwargs.get('mongodb_auth_source', 'admin')
|
|
)
|
|
elif cache_type == 'redis':
|
|
# Mantener compatibilidad con Redis (deprecated)
|
|
raise ValueError("Redis ya no está soportado. Por favor, usa MongoDB.")
|
|
else:
|
|
raise ValueError(f"Tipo de cache desconocido: {cache_type}. Solo se soporta 'mongodb'")
|
|
|