""" Base de datos MySQL para el catálogo de productos. Tú cargas productos a mano (con add_producto.py) y el bot los muestra a los usuarios desde acá. """ from contextlib import contextmanager from datetime import datetime import pymysql import pymysql.cursors from config import MYSQL_HOST, MYSQL_USER, MYSQL_PASSWORD, MYSQL_DATABASE @contextmanager def get_conn(): conn = pymysql.connect( host=MYSQL_HOST, user=MYSQL_USER, password=MYSQL_PASSWORD, database=MYSQL_DATABASE, charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, ) try: yield conn conn.commit() finally: conn.close() def init_db(): """Crea las tablas si no existen. Llamar una vez al arrancar.""" with get_conn() as conn: with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS productos ( id INT AUTO_INCREMENT PRIMARY KEY, titulo VARCHAR(255) NOT NULL, precio DECIMAL(12,2) NOT NULL, categoria_id INT NULL, link_afiliado TEXT NOT NULL, imagen_url TEXT, activo TINYINT(1) DEFAULT 1, creado_en DATETIME NOT NULL, INDEX idx_categoria_id (categoria_id), INDEX idx_activo_creado (activo, creado_en) ) """) cur.execute(""" CREATE TABLE IF NOT EXISTS usuarios ( chat_id BIGINT PRIMARY KEY, categoria_preferida VARCHAR(100), creado_en DATETIME NOT NULL ) """) _asegurar_columna(cur, "productos", "ml_item_id", "VARCHAR(20) NULL") _asegurar_columna(cur, "productos", "ultima_verificacion", "DATETIME NULL") _asegurar_columna(cur, "productos", "url_producto_ml", "TEXT NULL") _asegurar_columna(cur, "usuarios", "nombre", "VARCHAR(255) NULL") _asegurar_columna(cur, "usuarios", "username", "VARCHAR(100) NULL") _asegurar_columna(cur, "usuarios", "idioma", "VARCHAR(10) NULL") cur.execute(""" CREATE TABLE IF NOT EXISTS categorias ( id INT AUTO_INCREMENT PRIMARY KEY, nombre VARCHAR(100) NOT NULL UNIQUE, creado_en DATETIME NOT NULL ) """) _asegurar_columna(cur, "productos", "categoria_id", "INT NULL") _asegurar_columna(cur, "productos", "precio_anterior", "DECIMAL(12,2) NULL") _asegurar_columna(cur, "categorias", "icono", "VARCHAR(10) NULL") _asegurar_columna(cur, "productos", "canal_message_id", "INT NULL") _asegurar_columna(cur, "productos", "descripcion", "TEXT NULL") cur.execute(""" CREATE TABLE IF NOT EXISTS ml_tokens ( id INT PRIMARY KEY DEFAULT 1, access_token TEXT NOT NULL, refresh_token TEXT NOT NULL, expira_en DATETIME NOT NULL, actualizado_en DATETIME NOT NULL ) """) cur.execute(""" CREATE TABLE IF NOT EXISTS banner_hero ( id INT PRIMARY KEY DEFAULT 1, eyebrow VARCHAR(100) NULL, titulo VARCHAR(255) NULL, subtitulo VARCHAR(255) NULL, boton_texto VARCHAR(100) NULL, boton_link VARCHAR(255) NULL, imagen_url TEXT NULL, actualizado_en DATETIME NOT NULL ) """) cur.execute(""" CREATE TABLE IF NOT EXISTS notificaciones_pendientes ( id INT AUTO_INCREMENT PRIMARY KEY, producto_id INT NOT NULL, precio_viejo DECIMAL(12,2) NOT NULL, precio_nuevo DECIMAL(12,2) NOT NULL, detectado_en DATETIME NOT NULL, enviado TINYINT(1) DEFAULT 0, enviado_en DATETIME NULL, INDEX idx_producto_pendiente (producto_id, enviado) ) """) _asegurar_columna(cur, "notificaciones_pendientes", "canal_message_id", "INT NULL") def _asegurar_columna(cur, tabla, columna, definicion): """Agrega una columna si todavía no existe (migración liviana, sin depender de sintaxis 'IF NOT EXISTS' que no todas las versiones de MySQL/MariaDB soportan igual).""" cur.execute( """SELECT COUNT(*) AS c FROM information_schema.COLUMNS WHERE TABLE_SCHEMA = DATABASE() AND TABLE_NAME = %s AND COLUMN_NAME = %s""", (tabla, columna) ) if cur.fetchone()["c"] == 0: cur.execute(f"ALTER TABLE {tabla} ADD COLUMN {columna} {definicion}") def agregar_producto(titulo, precio, categoria_id, link_afiliado, imagen_url=None, ml_item_id=None, url_producto_ml=None): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """INSERT INTO productos (titulo, precio, categoria_id, link_afiliado, imagen_url, ml_item_id, url_producto_ml, creado_en) VALUES (%s, %s, %s, %s, %s, %s, %s, %s)""", (titulo, precio, categoria_id, link_afiliado, imagen_url, ml_item_id, url_producto_ml, datetime.utcnow()) ) return cur.lastrowid def listar_todas_categorias(): with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT id, nombre FROM categorias ORDER BY nombre") return cur.fetchall() def listar_productos(categoria=None, limite=10, offset=0): with get_conn() as conn: with conn.cursor() as cur: if categoria: cur.execute( """SELECT p.* FROM productos p JOIN categorias c ON c.id = p.categoria_id WHERE p.activo=1 AND c.nombre=%s ORDER BY p.creado_en DESC LIMIT %s OFFSET %s""", (categoria.lower(), limite, offset) ) else: cur.execute( """SELECT * FROM productos WHERE activo=1 ORDER BY creado_en DESC LIMIT %s OFFSET %s""", (limite, offset) ) return cur.fetchall() def buscar_productos(palabra_clave, limite=10): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """SELECT * FROM productos WHERE activo=1 AND titulo LIKE %s ORDER BY creado_en DESC LIMIT %s""", (f"%{palabra_clave}%", limite) ) return cur.fetchall() def listar_categorias(): """Categorías que tienen al menos un producto activo (para el bot).""" with get_conn() as conn: with conn.cursor() as cur: cur.execute(""" SELECT DISTINCT c.nombre FROM productos p JOIN categorias c ON c.id = p.categoria_id WHERE p.activo=1 """) return [r["nombre"] for r in cur.fetchall()] def registrar_usuario(chat_id, nombre=None, username=None, idioma=None): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """INSERT INTO usuarios (chat_id, nombre, username, idioma, creado_en) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE nombre=VALUES(nombre), username=VALUES(username), idioma=VALUES(idioma)""", (chat_id, nombre, username, idioma, datetime.utcnow()) ) def listar_todos_activos(): """Para el script de verificación de precios: todos los productos activos, con el id de Mercado Libre si ya se resolvió antes.""" with get_conn() as conn: with conn.cursor() as cur: cur.execute( "SELECT id, titulo, precio, link_afiliado, ml_item_id FROM productos WHERE activo=1" ) return cur.fetchall() def actualizar_desde_ml(producto_id, precio=None, activo=None, ml_item_id=None): campos = ["ultima_verificacion=%s"] valores = [datetime.utcnow()] if precio is not None: campos.append("precio=%s") valores.append(precio) if activo is not None: campos.append("activo=%s") valores.append(1 if activo else 0) if ml_item_id is not None: campos.append("ml_item_id=%s") valores.append(ml_item_id) valores.append(producto_id) with get_conn() as conn: with conn.cursor() as cur: cur.execute(f"UPDATE productos SET {', '.join(campos)} WHERE id=%s", valores) def obtener_tokens_ml(): with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT * FROM ml_tokens WHERE id=1") return cur.fetchone() def guardar_tokens_ml(access_token, refresh_token, expira_en): with get_conn() as conn: with conn.cursor() as cur: cur.execute( """INSERT INTO ml_tokens (id, access_token, refresh_token, expira_en, actualizado_en) VALUES (1, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE access_token=VALUES(access_token), refresh_token=VALUES(refresh_token), expira_en=VALUES(expira_en), actualizado_en=VALUES(actualizado_en)""", (access_token, refresh_token, expira_en, datetime.utcnow()) ) def contar_usuarios(): with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT COUNT(*) as c FROM usuarios") return cur.fetchone()["c"] def encolar_notificacion(producto_id, precio_viejo, precio_nuevo): """Registra una bajada de precio pendiente de notificar. Si ya hay una pendiente sin enviar para el mismo producto, solo actualiza el precio nuevo (conserva el precio_viejo original de esa tanda).""" with get_conn() as conn: with conn.cursor() as cur: cur.execute( "SELECT id FROM notificaciones_pendientes WHERE producto_id=%s AND enviado=0", (producto_id,) ) pendiente = cur.fetchone() if pendiente: cur.execute( "UPDATE notificaciones_pendientes SET precio_nuevo=%s WHERE id=%s", (precio_nuevo, pendiente["id"]) ) else: cur.execute( """INSERT INTO notificaciones_pendientes (producto_id, precio_viejo, precio_nuevo, detectado_en) VALUES (%s, %s, %s, %s)""", (producto_id, precio_viejo, precio_nuevo, datetime.utcnow()) ) def listar_notificaciones_pendientes(): with get_conn() as conn: with conn.cursor() as cur: cur.execute(""" SELECT np.id, np.producto_id, np.precio_viejo, np.precio_nuevo, p.titulo, p.imagen_url, p.link_afiliado FROM notificaciones_pendientes np JOIN productos p ON p.id = np.producto_id WHERE np.enviado = 0 """) return cur.fetchall() def marcar_notificacion_enviada(notificacion_id): with get_conn() as conn: with conn.cursor() as cur: cur.execute( "UPDATE notificaciones_pendientes SET enviado=1, enviado_en=%s WHERE id=%s", (datetime.utcnow(), notificacion_id) ) def guardar_canal_message_id_notificacion(notificacion_id, message_id): with get_conn() as conn: with conn.cursor() as cur: cur.execute( "UPDATE notificaciones_pendientes SET canal_message_id=%s WHERE id=%s", (message_id, notificacion_id) ) def guardar_canal_message_id_producto(producto_id, message_id): with get_conn() as conn: with conn.cursor() as cur: cur.execute( "UPDATE productos SET canal_message_id=%s WHERE id=%s", (message_id, producto_id) ) def listar_chat_ids(): with get_conn() as conn: with conn.cursor() as cur: cur.execute("SELECT chat_id FROM usuarios") return [fila["chat_id"] for fila in cur.fetchall()]