"""Database layer for MailPulse (Brevo-like email campaign & 1-to-1 sender).""" import hashlib import hmac import json import os import secrets import sqlite3 import uuid from contextlib import contextmanager from datetime import datetime, timedelta, timezone from pro_templates import PRO_TEMPLATES DEFAULT_DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "mailpulse.db") def utc_now() -> str: """Return current UTC ISO-8601 timestamp string.""" return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") def hash_password(password: str, salt: str = None) -> str: """Hash a password with PBKDF2-HMAC-SHA256 and a random salt.""" salt = salt or secrets.token_hex(16) dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), salt.encode("utf-8"), 120_000) return f"{salt}${dk.hex()}" def verify_password(password: str, stored_hash: str) -> bool: """Verify a password against a stored PBKDF2-HMAC-SHA256 hash.""" if not password or not stored_hash or "$" not in stored_hash: return False salt, _ = stored_hash.split("$", 1) expected = hash_password(password, salt=salt) return hmac.compare_digest(expected, stored_hash) class Database: """Thread-safe SQLite wrapper and repository for MailPulse.""" def __init__(self, db_path: str = DEFAULT_DB_PATH): self.db_path = db_path self.init_db() @contextmanager def connect(self): conn = sqlite3.connect(self.db_path, check_same_thread=False, timeout=30.0) conn.row_factory = sqlite3.Row conn.execute("PRAGMA foreign_keys = ON;") try: yield conn conn.commit() except Exception: conn.rollback() raise finally: conn.close() def init_db(self): with self.connect() as conn: conn.executescript( """ CREATE TABLE IF NOT EXISTS smtp_configs ( id TEXT PRIMARY KEY, name TEXT NOT NULL, mode TEXT NOT NULL DEFAULT 'sandbox', -- 'sandbox' | 'smtp' host TEXT DEFAULT '', port INTEGER DEFAULT 587, security TEXT DEFAULT 'tls', -- 'tls' | 'ssl' | 'none' username TEXT DEFAULT '', password TEXT DEFAULT '', from_name TEXT NOT NULL, from_email TEXT NOT NULL, reply_to TEXT DEFAULT '', is_default INTEGER DEFAULT 0, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS lists ( id TEXT PRIMARY KEY, name TEXT NOT NULL, description TEXT DEFAULT '', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS contacts ( id TEXT PRIMARY KEY, email TEXT NOT NULL UNIQUE, first_name TEXT DEFAULT '', last_name TEXT DEFAULT '', company TEXT DEFAULT '', phone TEXT DEFAULT '', tags TEXT DEFAULT '', custom_fields TEXT DEFAULT '{}', status TEXT NOT NULL DEFAULT 'subscribed', -- 'subscribed' | 'unsubscribed' | 'bounced' created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS list_contacts ( list_id TEXT NOT NULL, contact_id TEXT NOT NULL, added_at TEXT NOT NULL, PRIMARY KEY (list_id, contact_id), FOREIGN KEY (list_id) REFERENCES lists(id) ON DELETE CASCADE, FOREIGN KEY (contact_id) REFERENCES contacts(id) ON DELETE CASCADE ); CREATE TABLE IF NOT EXISTS templates ( id TEXT PRIMARY KEY, name TEXT NOT NULL, category TEXT DEFAULT 'General', subject TEXT NOT NULL, preheader TEXT DEFAULT '', html_content TEXT NOT NULL, text_content TEXT DEFAULT '', updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS campaigns ( id TEXT PRIMARY KEY, name TEXT NOT NULL, subject TEXT NOT NULL, preheader TEXT DEFAULT '', sender_id TEXT, list_id TEXT, template_id TEXT, html_content TEXT NOT NULL, text_content TEXT DEFAULT '', status TEXT NOT NULL DEFAULT 'draft', -- 'draft' | 'scheduled' | 'sending' | 'sent' | 'failed' scheduled_at TEXT DEFAULT '', throttle_ms INTEGER DEFAULT 150, track_opens INTEGER DEFAULT 1, total_recipients INTEGER DEFAULT 0, sent_count INTEGER DEFAULT 0, failed_count INTEGER DEFAULT 0, open_count INTEGER DEFAULT 0, created_at TEXT NOT NULL, sent_at TEXT DEFAULT '' ); CREATE TABLE IF NOT EXISTS email_logs ( id TEXT PRIMARY KEY, tracking_id TEXT UNIQUE NOT NULL, send_type TEXT NOT NULL, -- 'individual' | 'campaign' campaign_id TEXT DEFAULT '', contact_id TEXT DEFAULT '', sender_id TEXT DEFAULT '', from_email TEXT NOT NULL, to_email TEXT NOT NULL, to_name TEXT DEFAULT '', cc TEXT DEFAULT '', bcc TEXT DEFAULT '', subject TEXT NOT NULL, html_body TEXT NOT NULL, text_body TEXT DEFAULT '', attachments_json TEXT DEFAULT '[]', status TEXT NOT NULL, -- 'sent' | 'simulated' | 'failed' error_message TEXT DEFAULT '', opened INTEGER DEFAULT 0, opened_at TEXT DEFAULT '', open_count INTEGER DEFAULT 0, created_at TEXT NOT NULL ); CREATE INDEX IF NOT EXISTS idx_email_logs_tracking ON email_logs(tracking_id); CREATE INDEX IF NOT EXISTS idx_email_logs_campaign ON email_logs(campaign_id); CREATE INDEX IF NOT EXISTS idx_email_logs_to ON email_logs(to_email); CREATE INDEX IF NOT EXISTS idx_contacts_email ON email_logs(to_email); CREATE TABLE IF NOT EXISTS users ( id TEXT PRIMARY KEY, username TEXT NOT NULL UNIQUE, display_name TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'member', -- 'admin' | 'member' password_hash TEXT NOT NULL, is_active INTEGER DEFAULT 1, created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS sessions ( token TEXT PRIMARY KEY, user_id TEXT NOT NULL, expires_at TEXT NOT NULL, created_at TEXT NOT NULL, FOREIGN KEY (user_id) REFERENCES users(id) ON DELETE CASCADE ); """ ) self._seed_defaults(conn) def _seed_defaults(self, conn): now = utc_now() # Seed initial private team users if empty count_users = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0] if count_users == 0: admin_user = os.environ.get("MAILPULSE_ADMIN_USER", "admin").strip().lower() admin_pass = os.environ.get("MAILPULSE_ADMIN_PASS", "MailPulse2026!") colleague_user = os.environ.get("MAILPULSE_COLLEAGUE_USER", "maryori.vargas").strip().lower() colleague_pass = os.environ.get("MAILPULSE_COLLEAGUE_PASS", "Maryori2026!") conn.execute( """ INSERT INTO users (id, username, display_name, role, password_hash, is_active, created_at) VALUES (?, ?, ?, 'admin', ?, 1, ?) """, ("u-admin", admin_user, "Administrador", hash_password(admin_pass), now), ) conn.execute( """ INSERT INTO users (id, username, display_name, role, password_hash, is_active, created_at) VALUES (?, ?, ?, 'member', ?, 1, ?) """, ("u-maryori", colleague_user, "Maryori Vargas", hash_password(colleague_pass), now), ) # Seed default sandbox SMTP profile if none exists count_smtp = conn.execute("SELECT COUNT(*) FROM smtp_configs").fetchone()[0] if count_smtp == 0: conn.execute( """ INSERT INTO smtp_configs (id, name, mode, host, port, security, username, password, from_name, from_email, reply_to, is_default, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( "smtp-sandbox-default", "Modo Simulación Local (Pruebas)", "sandbox", "localhost", 1025, "none", "", "", "Equipo MailPulse", "hola@tudominio.com", "hola@tudominio.com", 1, now, ), ) # Seed default lists & sample contacts if empty count_lists = conn.execute("SELECT COUNT(*) FROM lists").fetchone()[0] if count_lists == 0: list_general = "list-general" list_vip = "list-vip" conn.execute( "INSERT INTO lists (id, name, description, created_at) VALUES (?, ?, ?, ?)", (list_general, "Suscriptores Newsletter", "Lista principal de novedades y promociones", now), ) conn.execute( "INSERT INTO lists (id, name, description, created_at) VALUES (?, ?, ?, ?)", (list_vip, "Clientes VIP", "Clientes activos para seguimiento personalizado", now), ) sample_contacts = [ ("c-1", "ana.martinez@ejemplo.com", "Ana", "Martínez", "Estudio Creativo Sur", "+56 9 8123 4567", "vip,diseño", [list_general, list_vip]), ("c-2", "carlos.rojas@ejemplo.com", "Carlos", "Rojas", "TechNorte SpA", "+56 9 7654 3210", "saas,b2b", [list_general, list_vip]), ("c-3", "sofia.lopez@ejemplo.com", "Sofía", "López", "Boutique Verde", "+56 9 9112 2334", "retail", [list_general]), ] for cid, email, fn, ln, comp, phone, tags, lists in sample_contacts: conn.execute( """ INSERT OR IGNORE INTO contacts (id, email, first_name, last_name, company, phone, tags, custom_fields, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, '{}', 'subscribed', ?) """, (cid, email, fn, ln, comp, phone, tags, now), ) for lid in lists: conn.execute( "INSERT OR IGNORE INTO list_contacts (list_id, contact_id, added_at) VALUES (?, ?, ?)", (lid, cid, now), ) # Seed or upgrade professional Outlook-compatible templates has_pro = conn.execute("SELECT 1 FROM templates WHERE id = 'tpl-pro-launch'").fetchone() if not has_pro: conn.execute("DELETE FROM templates WHERE id IN ('tpl-welcome', 'tpl-newsletter', 'tpl-followup-1a1')") for tpl in PRO_TEMPLATES: conn.execute( """ INSERT OR IGNORE INTO templates (id, name, category, subject, preheader, html_content, text_content, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( tpl["id"], tpl["name"], tpl["category"], tpl["subject"], tpl["preheader"], tpl["html_content"], tpl["text_content"], now, ), ) # ---------- Helpers ---------- @staticmethod def _row_to_dict(row) -> dict: if row is None: return None return dict(row) # ---------- SMTP Profiles ---------- def list_smtp_configs(self) -> list: with self.connect() as conn: rows = conn.execute("SELECT * FROM smtp_configs ORDER BY is_default DESC, created_at ASC").fetchall() return [dict(r) for r in rows] def get_smtp_config(self, config_id: str = None) -> dict: with self.connect() as conn: if config_id: row = conn.execute("SELECT * FROM smtp_configs WHERE id = ?", (config_id,)).fetchone() if row: return dict(row) row = conn.execute("SELECT * FROM smtp_configs WHERE is_default = 1 LIMIT 1").fetchone() if row: return dict(row) row = conn.execute("SELECT * FROM smtp_configs LIMIT 1").fetchone() return dict(row) if row else None def upsert_smtp_config(self, data: dict) -> dict: cid = data.get("id") or f"smtp-{uuid.uuid4().hex[:8]}" now = utc_now() is_default = 1 if data.get("is_default") else 0 with self.connect() as conn: if is_default: conn.execute("UPDATE smtp_configs SET is_default = 0") existing = conn.execute("SELECT id, password FROM smtp_configs WHERE id = ?", (cid,)).fetchone() password = data.get("password") if existing and (password is None or password == "••••••••"): password = existing["password"] if existing: conn.execute( """ UPDATE smtp_configs SET name=?, mode=?, host=?, port=?, security=?, username=?, password=?, from_name=?, from_email=?, reply_to=?, is_default=? WHERE id=? """, ( data.get("name", "SMTP"), data.get("mode", "smtp"), data.get("host", ""), int(data.get("port") or 587), data.get("security", "tls"), data.get("username", ""), password or "", data.get("from_name", ""), data.get("from_email", ""), data.get("reply_to", ""), is_default, cid, ), ) else: conn.execute( """ INSERT INTO smtp_configs (id, name, mode, host, port, security, username, password, from_name, from_email, reply_to, is_default, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( cid, data.get("name", "SMTP"), data.get("mode", "smtp"), data.get("host", ""), int(data.get("port") or 587), data.get("security", "tls"), data.get("username", ""), password or "", data.get("from_name", ""), data.get("from_email", ""), data.get("reply_to", ""), is_default, now, ), ) return self.get_smtp_config(cid) def delete_smtp_config(self, config_id: str) -> bool: with self.connect() as conn: cur = conn.execute("DELETE FROM smtp_configs WHERE id = ?", (config_id,)) # Ensure one default remains rem = conn.execute("SELECT id FROM smtp_configs WHERE is_default = 1").fetchone() if not rem: first = conn.execute("SELECT id FROM smtp_configs LIMIT 1").fetchone() if first: conn.execute("UPDATE smtp_configs SET is_default = 1 WHERE id = ?", (first["id"],)) return cur.rowcount > 0 # ---------- Lists ---------- def list_lists(self) -> list: with self.connect() as conn: rows = conn.execute( """ SELECT l.*, COUNT(DISTINCT CASE WHEN c.status = 'subscribed' THEN lc.contact_id END) AS subscribed_count, COUNT(DISTINCT lc.contact_id) AS total_count FROM lists l LEFT JOIN list_contacts lc ON lc.list_id = l.id LEFT JOIN contacts c ON c.id = lc.contact_id GROUP BY l.id ORDER BY l.created_at DESC """ ).fetchall() return [dict(r) for r in rows] def create_list(self, name: str, description: str = "") -> dict: lid = f"list-{uuid.uuid4().hex[:8]}" now = utc_now() with self.connect() as conn: conn.execute( "INSERT INTO lists (id, name, description, created_at) VALUES (?, ?, ?, ?)", (lid, name.strip(), description.strip(), now), ) row = conn.execute("SELECT * FROM lists WHERE id = ?", (lid,)).fetchone() res = dict(row) res["subscribed_count"] = 0 res["total_count"] = 0 return res def delete_list(self, list_id: str) -> bool: with self.connect() as conn: cur = conn.execute("DELETE FROM lists WHERE id = ?", (list_id,)) return cur.rowcount > 0 # ---------- Contacts ---------- def list_contacts(self, list_id: str = None, search: str = None, status: str = None) -> list: query = """ SELECT DISTINCT c.* FROM contacts c """ params = [] wheres = [] if list_id: query += " INNER JOIN list_contacts lc ON lc.contact_id = c.id " wheres.append("lc.list_id = ?") params.append(list_id) if status: wheres.append("c.status = ?") params.append(status) if search: like = f"%{search.strip()}%" wheres.append("(c.email LIKE ? OR c.first_name LIKE ? OR c.last_name LIKE ? OR c.company LIKE ? OR c.tags LIKE ?)") params.extend([like, like, like, like, like]) if wheres: query += " WHERE " + " AND ".join(wheres) query += " ORDER BY c.created_at DESC" with self.connect() as conn: rows = conn.execute(query, params).fetchall() contacts = [] for r in rows: d = dict(r) try: d["custom_fields"] = json.loads(d.get("custom_fields") or "{}") except Exception: d["custom_fields"] = {} list_rows = conn.execute( "SELECT l.id, l.name FROM lists l INNER JOIN list_contacts lc ON lc.list_id = l.id WHERE lc.contact_id = ?", (d["id"],), ).fetchall() d["lists"] = [dict(lr) for lr in list_rows] contacts.append(d) return contacts def get_contact(self, contact_id: str = None, email: str = None) -> dict: with self.connect() as conn: if contact_id: row = conn.execute("SELECT * FROM contacts WHERE id = ?", (contact_id,)).fetchone() elif email: row = conn.execute("SELECT * FROM contacts WHERE LOWER(email) = LOWER(?)", (email.strip(),)).fetchone() else: return None if not row: return None d = dict(row) try: d["custom_fields"] = json.loads(d.get("custom_fields") or "{}") except Exception: d["custom_fields"] = {} list_rows = conn.execute( "SELECT l.id, l.name FROM lists l INNER JOIN list_contacts lc ON lc.list_id = l.id WHERE lc.contact_id = ?", (d["id"],), ).fetchall() d["lists"] = [dict(lr) for lr in list_rows] return d def upsert_contact(self, data: dict) -> dict: email = (data.get("email") or "").strip().lower() if not email or "@" not in email: raise ValueError("Correo electrónico inválido") now = utc_now() cid = data.get("id") list_ids = data.get("list_ids") custom_fields = data.get("custom_fields") or {} if isinstance(custom_fields, str): try: custom_fields = json.loads(custom_fields) except Exception: custom_fields = {} with self.connect() as conn: existing = None if cid: existing = conn.execute("SELECT * FROM contacts WHERE id = ?", (cid,)).fetchone() if not existing: existing = conn.execute("SELECT * FROM contacts WHERE LOWER(email) = ?", (email,)).fetchone() if existing: cid = existing["id"] conn.execute( """ UPDATE contacts SET email=?, first_name=?, last_name=?, company=?, phone=?, tags=?, custom_fields=?, status=? WHERE id=? """, ( email, data.get("first_name", existing["first_name"]), data.get("last_name", existing["last_name"]), data.get("company", existing["company"]), data.get("phone", existing["phone"]), data.get("tags", existing["tags"]), json.dumps(custom_fields, ensure_ascii=False), data.get("status", existing["status"]), cid, ), ) else: cid = cid or f"c-{uuid.uuid4().hex[:8]}" conn.execute( """ INSERT INTO contacts (id, email, first_name, last_name, company, phone, tags, custom_fields, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( cid, email, data.get("first_name", ""), data.get("last_name", ""), data.get("company", ""), data.get("phone", ""), data.get("tags", ""), json.dumps(custom_fields, ensure_ascii=False), data.get("status", "subscribed"), now, ), ) if list_ids is not None: if data.get("replace_lists"): conn.execute("DELETE FROM list_contacts WHERE contact_id = ?", (cid,)) for lid in list_ids: if lid: conn.execute( "INSERT OR IGNORE INTO list_contacts (list_id, contact_id, added_at) VALUES (?, ?, ?)", (lid, cid, now), ) return self.get_contact(contact_id=cid) def delete_contact(self, contact_id: str) -> bool: with self.connect() as conn: cur = conn.execute("DELETE FROM contacts WHERE id = ?", (contact_id,)) return cur.rowcount > 0 def unsubscribe_contact(self, contact_id: str) -> bool: with self.connect() as conn: cur = conn.execute("UPDATE contacts SET status = 'unsubscribed' WHERE id = ?", (contact_id,)) return cur.rowcount > 0 # ---------- Templates ---------- def list_templates(self) -> list: with self.connect() as conn: rows = conn.execute("SELECT * FROM templates ORDER BY updated_at DESC").fetchall() return [dict(r) for r in rows] def get_template(self, template_id: str) -> dict: with self.connect() as conn: row = conn.execute("SELECT * FROM templates WHERE id = ?", (template_id,)).fetchone() return dict(row) if row else None def upsert_template(self, data: dict) -> dict: tid = data.get("id") or f"tpl-{uuid.uuid4().hex[:8]}" now = utc_now() with self.connect() as conn: existing = conn.execute("SELECT id FROM templates WHERE id = ?", (tid,)).fetchone() if existing: conn.execute( """ UPDATE templates SET name=?, category=?, subject=?, preheader=?, html_content=?, text_content=?, updated_at=? WHERE id=? """, ( data.get("name", "Plantilla"), data.get("category", "General"), data.get("subject", ""), data.get("preheader", ""), data.get("html_content", ""), data.get("text_content", ""), now, tid, ), ) else: conn.execute( """ INSERT INTO templates (id, name, category, subject, preheader, html_content, text_content, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?) """, ( tid, data.get("name", "Plantilla"), data.get("category", "General"), data.get("subject", ""), data.get("preheader", ""), data.get("html_content", ""), data.get("text_content", ""), now, ), ) return self.get_template(tid) def delete_template(self, template_id: str) -> bool: with self.connect() as conn: cur = conn.execute("DELETE FROM templates WHERE id = ?", (template_id,)) return cur.rowcount > 0 # ---------- Campaigns ---------- def list_campaigns(self) -> list: with self.connect() as conn: rows = conn.execute( """ SELECT ca.*, l.name AS list_name, s.from_email AS sender_email, s.from_name AS sender_name FROM campaigns ca LEFT JOIN lists l ON l.id = ca.list_id LEFT JOIN smtp_configs s ON s.id = ca.sender_id ORDER BY ca.created_at DESC """ ).fetchall() return [dict(r) for r in rows] def get_campaign(self, campaign_id: str) -> dict: with self.connect() as conn: row = conn.execute( """ SELECT ca.*, l.name AS list_name, s.from_email AS sender_email, s.from_name AS sender_name FROM campaigns ca LEFT JOIN lists l ON l.id = ca.list_id LEFT JOIN smtp_configs s ON s.id = ca.sender_id WHERE ca.id = ? """, (campaign_id,), ).fetchone() return dict(row) if row else None def upsert_campaign(self, data: dict) -> dict: cid = data.get("id") or f"camp-{uuid.uuid4().hex[:8]}" now = utc_now() with self.connect() as conn: existing = conn.execute("SELECT * FROM campaigns WHERE id = ?", (cid,)).fetchone() if existing: conn.execute( """ UPDATE campaigns SET name=?, subject=?, preheader=?, sender_id=?, list_id=?, template_id=?, html_content=?, text_content=?, status=?, scheduled_at=?, throttle_ms=?, track_opens=? WHERE id=? """, ( data.get("name", existing["name"]), data.get("subject", existing["subject"]), data.get("preheader", existing["preheader"]), data.get("sender_id", existing["sender_id"]), data.get("list_id", existing["list_id"]), data.get("template_id", existing["template_id"]), data.get("html_content", existing["html_content"]), data.get("text_content", existing["text_content"]), data.get("status", existing["status"]), data.get("scheduled_at", existing["scheduled_at"]), int(data.get("throttle_ms") if data.get("throttle_ms") is not None else existing["throttle_ms"]), int(data.get("track_opens") if data.get("track_opens") is not None else existing["track_opens"]), cid, ), ) else: conn.execute( """ INSERT INTO campaigns (id, name, subject, preheader, sender_id, list_id, template_id, html_content, text_content, status, scheduled_at, throttle_ms, track_opens, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( cid, data.get("name", "Nueva Campaña"), data.get("subject", ""), data.get("preheader", ""), data.get("sender_id", ""), data.get("list_id", ""), data.get("template_id", ""), data.get("html_content", ""), data.get("text_content", ""), data.get("status", "draft"), data.get("scheduled_at", ""), int(data.get("throttle_ms") or 150), 1 if data.get("track_opens", True) else 0, now, ), ) return self.get_campaign(cid) def update_campaign_progress(self, campaign_id: str, **fields): if not fields: return allowed = {"status", "total_recipients", "sent_count", "failed_count", "open_count", "sent_at"} cols = [] vals = [] for k, v in fields.items(): if k in allowed: cols.append(f"{k}=?") vals.append(v) if not cols: return vals.append(campaign_id) with self.connect() as conn: conn.execute(f"UPDATE campaigns SET {', '.join(cols)} WHERE id = ?", vals) def delete_campaign(self, campaign_id: str) -> bool: with self.connect() as conn: cur = conn.execute("DELETE FROM campaigns WHERE id = ?", (campaign_id,)) return cur.rowcount > 0 def get_due_scheduled_campaigns(self) -> list: now = utc_now() with self.connect() as conn: rows = conn.execute( "SELECT * FROM campaigns WHERE status = 'scheduled' AND scheduled_at != '' AND scheduled_at <= ?", (now,), ).fetchall() return [dict(r) for r in rows] # ---------- Email Logs & Tracking ---------- def create_email_log(self, data: dict) -> dict: lid = data.get("id") or f"log-{uuid.uuid4().hex[:10]}" tracking_id = data.get("tracking_id") or uuid.uuid4().hex now = utc_now() with self.connect() as conn: conn.execute( """ INSERT INTO email_logs (id, tracking_id, send_type, campaign_id, contact_id, sender_id, from_email, to_email, to_name, cc, bcc, subject, html_body, text_body, attachments_json, status, error_message, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( lid, tracking_id, data.get("send_type", "individual"), data.get("campaign_id", ""), data.get("contact_id", ""), data.get("sender_id", ""), data.get("from_email", ""), data.get("to_email", ""), data.get("to_name", ""), data.get("cc", ""), data.get("bcc", ""), data.get("subject", ""), data.get("html_body", ""), data.get("text_body", ""), json.dumps(data.get("attachments", []), ensure_ascii=False), data.get("status", "simulated"), data.get("error_message", ""), now, ), ) row = conn.execute("SELECT * FROM email_logs WHERE id = ?", (lid,)).fetchone() return dict(row) def list_email_logs(self, send_type: str = None, status: str = None, campaign_id: str = None, search: str = None, limit: int = 150) -> list: query = "SELECT * FROM email_logs" wheres = [] params = [] if send_type: wheres.append("send_type = ?") params.append(send_type) if status: wheres.append("status = ?") params.append(status) if campaign_id: wheres.append("campaign_id = ?") params.append(campaign_id) if search: like = f"%{search.strip()}%" wheres.append("(to_email LIKE ? OR to_name LIKE ? OR subject LIKE ?)") params.extend([like, like, like]) if wheres: query += " WHERE " + " AND ".join(wheres) query += " ORDER BY created_at DESC LIMIT ?" params.append(limit) with self.connect() as conn: rows = conn.execute(query, params).fetchall() return [dict(r) for r in rows] def get_email_log(self, log_id: str) -> dict: with self.connect() as conn: row = conn.execute("SELECT * FROM email_logs WHERE id = ?", (log_id,)).fetchone() return dict(row) if row else None def record_open(self, tracking_id: str) -> bool: now = utc_now() with self.connect() as conn: row = conn.execute("SELECT id, opened, campaign_id FROM email_logs WHERE tracking_id = ?", (tracking_id,)).fetchone() if not row: return False first_open = row["opened"] == 0 conn.execute( """ UPDATE email_logs SET opened = 1, opened_at = CASE WHEN opened_at = '' THEN ? ELSE opened_at END, open_count = open_count + 1 WHERE id = ? """, (now, row["id"]), ) if first_open and row["campaign_id"]: conn.execute( "UPDATE campaigns SET open_count = open_count + 1 WHERE id = ?", (row["campaign_id"],), ) return True def get_dashboard_stats(self) -> dict: with self.connect() as conn: total_contacts = conn.execute("SELECT COUNT(*) FROM contacts").fetchone()[0] subscribed_contacts = conn.execute("SELECT COUNT(*) FROM contacts WHERE status = 'subscribed'").fetchone()[0] total_lists = conn.execute("SELECT COUNT(*) FROM lists").fetchone()[0] total_campaigns = conn.execute("SELECT COUNT(*) FROM campaigns").fetchone()[0] total_templates = conn.execute("SELECT COUNT(*) FROM templates").fetchone()[0] total_sent = conn.execute("SELECT COUNT(*) FROM email_logs WHERE status IN ('sent', 'simulated')").fetchone()[0] total_individual = conn.execute("SELECT COUNT(*) FROM email_logs WHERE send_type = 'individual' AND status IN ('sent', 'simulated')").fetchone()[0] total_campaign_emails = conn.execute("SELECT COUNT(*) FROM email_logs WHERE send_type = 'campaign' AND status IN ('sent', 'simulated')").fetchone()[0] total_failed = conn.execute("SELECT COUNT(*) FROM email_logs WHERE status = 'failed'").fetchone()[0] total_opened = conn.execute("SELECT COUNT(*) FROM email_logs WHERE opened = 1").fetchone()[0] open_rate = round((total_opened / total_sent) * 100, 1) if total_sent > 0 else 0.0 recent_logs = conn.execute( "SELECT id, send_type, to_email, to_name, subject, status, opened, created_at FROM email_logs ORDER BY created_at DESC LIMIT 8" ).fetchall() return { "total_contacts": total_contacts, "subscribed_contacts": subscribed_contacts, "total_lists": total_lists, "total_campaigns": total_campaigns, "total_templates": total_templates, "total_sent": total_sent, "total_individual": total_individual, "total_campaign_emails": total_campaign_emails, "total_failed": total_failed, "total_opened": total_opened, "open_rate": open_rate, "recent_logs": [dict(r) for r in recent_logs], } # ---------- Auth & Users ---------- def authenticate_user(self, username: str, password: str) -> dict: uname = (username or "").strip().lower() if not uname or not password: return None with self.connect() as conn: row = conn.execute( "SELECT * FROM users WHERE LOWER(username) = ? AND is_active = 1", (uname,), ).fetchone() if not row or not verify_password(password, row["password_hash"]): return None return { "id": row["id"], "username": row["username"], "display_name": row["display_name"], "role": row["role"], "is_active": row["is_active"], } def create_session(self, user_id: str, days: int = 14) -> str: token = secrets.token_urlsafe(32) now = utc_now() exp = (datetime.now(timezone.utc) + timedelta(days=days)).strftime("%Y-%m-%d %H:%M:%S") with self.connect() as conn: conn.execute( "INSERT INTO sessions (token, user_id, expires_at, created_at) VALUES (?, ?, ?, ?)", (token, user_id, exp, now), ) return token def get_session_user(self, token: str) -> dict: if not token: return None now = utc_now() with self.connect() as conn: row = conn.execute( """ SELECT u.id, u.username, u.display_name, u.role, u.is_active FROM sessions s INNER JOIN users u ON u.id = s.user_id WHERE s.token = ? AND s.expires_at > ? AND u.is_active = 1 """, (token, now), ).fetchone() return dict(row) if row else None def delete_session(self, token: str): if not token: return with self.connect() as conn: conn.execute("DELETE FROM sessions WHERE token = ?", (token,)) def list_users(self) -> list: with self.connect() as conn: rows = conn.execute( "SELECT id, username, display_name, role, is_active, created_at FROM users ORDER BY created_at ASC" ).fetchall() return [dict(r) for r in rows] def upsert_user(self, data: dict) -> dict: uid = data.get("id") username = (data.get("username") or "").strip().lower() display_name = (data.get("display_name") or "").strip() or username role = data.get("role") if data.get("role") in ("admin", "member") else "member" password = data.get("password") or "" is_active = 1 if data.get("is_active", True) else 0 now = utc_now() if not username: raise ValueError("El nombre de usuario es obligatorio.") with self.connect() as conn: existing = None if uid: existing = conn.execute("SELECT * FROM users WHERE id = ?", (uid,)).fetchone() if not existing: existing = conn.execute("SELECT * FROM users WHERE LOWER(username) = ?", (username,)).fetchone() if existing: uid = existing["id"] pw_hash = hash_password(password) if password else existing["password_hash"] conn.execute( "UPDATE users SET username=?, display_name=?, role=?, password_hash=?, is_active=? WHERE id=?", (username, display_name, role, pw_hash, is_active, uid), ) else: if len(password) < 4: raise ValueError("La contraseña debe tener al menos 4 caracteres.") uid = uid or f"u-{uuid.uuid4().hex[:8]}" conn.execute( """ INSERT INTO users (id, username, display_name, role, password_hash, is_active, created_at) VALUES (?, ?, ?, ?, ?, ?, ?) """, (uid, username, display_name, role, hash_password(password), is_active, now), ) row = conn.execute( "SELECT id, username, display_name, role, is_active, created_at FROM users WHERE id = ?", (uid,), ).fetchone() return dict(row) def delete_user(self, user_id: str) -> bool: with self.connect() as conn: cur = conn.execute("DELETE FROM users WHERE id = ?", (user_id,)) return cur.rowcount > 0