143 lines
4.2 KiB
Python
143 lines
4.2 KiB
Python
import sqlite3
|
|
|
|
DB_NAME = "data/emails.db"
|
|
|
|
# ========================================================================
|
|
def init_db():
|
|
"""
|
|
Inicializa la base de datos SQLite y crea la tabla de correos si no existe.
|
|
"""
|
|
conn = sqlite3.connect(DB_NAME)
|
|
cursor = conn.cursor()
|
|
|
|
# Crear tabla con id como llave primaria para evitar duplicados
|
|
cursor.execute('''
|
|
CREATE TABLE IF NOT EXISTS emails (
|
|
id TEXT PRIMARY KEY,
|
|
fecha TEXT,
|
|
sender TEXT,
|
|
subject TEXT,
|
|
has_attach INTEGER,
|
|
fecha_descarga TEXT
|
|
)
|
|
''')
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
# ========================================================================
|
|
def save_emails(emails_list):
|
|
"""
|
|
Guarda una lista de correos en la base de datos.
|
|
Retorna la cantidad de registros nuevos insertados.
|
|
"""
|
|
if not emails_list:
|
|
return 0
|
|
|
|
conn = sqlite3.connect(DB_NAME)
|
|
cursor = conn.cursor()
|
|
|
|
from datetime import datetime
|
|
fecha_ahora = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
|
|
|
# Mapear lista de diccionarios a tuplas para executemany
|
|
# has_attach se guarda como 1 (True) o 0 (False)
|
|
data = [
|
|
(
|
|
e['id'],
|
|
e['fecha'],
|
|
e['sender'],
|
|
e['subject'],
|
|
1 if e['has_attach'] else 0,
|
|
fecha_ahora
|
|
)
|
|
for e in emails_list
|
|
]
|
|
|
|
# INSERT OR IGNORE evita errores si el ID ya existe
|
|
cursor.executemany('''
|
|
INSERT OR IGNORE INTO emails (id, fecha, sender, subject, has_attach, fecha_descarga)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
''', data)
|
|
|
|
conn.commit()
|
|
rows_affected = cursor.rowcount
|
|
conn.close()
|
|
return rows_affected
|
|
|
|
# ========================================================================
|
|
def get_existing_ids(ids_list):
|
|
"""
|
|
Recibe una lista de IDs y devuelve una lista con aquellos que ya existen en la BD.
|
|
"""
|
|
if not ids_list:
|
|
return []
|
|
|
|
conn = sqlite3.connect(DB_NAME)
|
|
cursor = conn.cursor()
|
|
|
|
existing_ids = []
|
|
# Procesar por lotes para evitar límites de parámetros en SQLite
|
|
for i in range(0, len(ids_list), 900):
|
|
chunk = ids_list[i:i + 900]
|
|
placeholders = ','.join(['?'] * len(chunk))
|
|
query = f"SELECT id FROM emails WHERE id IN ({placeholders})"
|
|
cursor.execute(query, chunk)
|
|
existing_ids.extend([row[0] for row in cursor.fetchall()])
|
|
|
|
conn.close()
|
|
return existing_ids
|
|
|
|
# ========================================================================
|
|
def delete_email_by_id(email_id):
|
|
"""
|
|
Elimina un correo de la base de datos por su ID.
|
|
Retorna True si se eliminó el registro, False si no se encontró.
|
|
"""
|
|
if not email_id:
|
|
return False
|
|
|
|
conn = sqlite3.connect(DB_NAME)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("DELETE FROM emails WHERE id = ?", (email_id,))
|
|
|
|
conn.commit()
|
|
rows_affected = cursor.rowcount
|
|
conn.close()
|
|
|
|
return rows_affected > 0
|
|
|
|
# ========================================================================
|
|
def get_ids_by_fecha(fecha):
|
|
"""
|
|
Retorna una lista de IDs cuya fecha sea >= al parámetro recibido.
|
|
Formato esperado de fecha: 'YYYY-MM-DD' o 'YYYY-MM-DD HH:MM:SS'
|
|
"""
|
|
if not fecha:
|
|
return []
|
|
|
|
conn = sqlite3.connect(DB_NAME)
|
|
cursor = conn.cursor()
|
|
|
|
cursor.execute("SELECT id FROM emails WHERE fecha >= ?", (fecha,))
|
|
ids = [row[0] for row in cursor.fetchall()]
|
|
|
|
conn.close()
|
|
return ids
|
|
# ========================================================================
|
|
if __name__ == "__main__":
|
|
fecha_inicio = '2026-07-31'
|
|
print(f"\nObteniendo IDs mayores o iguales a {fecha_inicio}:")
|
|
|
|
ids_encontrados = get_ids_by_fecha(fecha_inicio)
|
|
|
|
if ids_encontrados:
|
|
print(f"✓ Se encontraron {len(ids_encontrados)} correos desde esa fecha:")
|
|
for i, email_id in enumerate(ids_encontrados, 1):
|
|
print(f" {i}. {email_id}")
|
|
# delete_email_by_id(email_id)
|
|
if len(ids_encontrados) > 10:
|
|
print(f" ... y {len(ids_encontrados) - 10} más")
|
|
else:
|
|
print("✗ No se encontraron correos desde esa fecha.") |