"""HTTP server, REST API, tracking endpoints, and scheduler for MailPulse.""" import base64 import csv import io import json import mimetypes import os import threading import time from email.header import Header from email.mime.multipart import MIMEMultipart from email.mime.text import MIMEText from http.cookies import SimpleCookie from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from urllib.parse import parse_qs, urlparse from db import Database from mailer import MailerService, build_contact_context, html_to_plain_text, render_template_string from pro_templates import STOCK_PHOTOS, STUDIO_BLOCKS, wrap_for_outlook STATIC_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static") # 1x1 transparent GIF bytes TRANSPARENT_GIF = base64.b64decode( "R0lGODlhAQABAIAAAAAAAP///yH5BAEAAAAALAAAAAABAAEAAAIBRAA7" ) def parse_csv_contacts(csv_text: str) -> list: """Parse CSV text with flexible Spanish/English headers into contact dicts.""" cleaned = (csv_text or "").strip() if not cleaned: return [] # Detect delimiter (; or , or tab) first_line = cleaned.splitlines()[0] delimiter = ";" if first_line.count(";") > first_line.count(",") else "," if "\t" in first_line and first_line.count("\t") >= first_line.count(delimiter): delimiter = "\t" reader = csv.DictReader(io.StringIO(cleaned), delimiter=delimiter) if not reader.fieldnames: return [] alias_map = { "email": "email", "correo": "email", "mail": "email", "e-mail": "email", "nombre": "first_name", "first_name": "first_name", "firstname": "first_name", "apellido": "last_name", "apellidos": "last_name", "last_name": "last_name", "lastname": "last_name", "empresa": "company", "compania": "company", "compañía": "company", "company": "company", "telefono": "phone", "teléfono": "phone", "phone": "phone", "celular": "phone", "etiquetas": "tags", "tags": "tags", } parsed = [] for row in reader: contact = {"custom_fields": {}} for raw_key, raw_val in row.items(): if raw_key is None: continue key_norm = raw_key.strip().lower() val = (raw_val or "").strip() mapped = alias_map.get(key_norm) if mapped: contact[mapped] = val elif val: contact["custom_fields"][key_norm] = val if contact.get("email") and "@" in contact["email"]: parsed.append(contact) return parsed class AppContext: """Shared application state injected into HTTP handlers.""" def __init__(self, db_path: str = None, base_url: str = "", require_auth: bool = None): self.db = Database(db_path) if db_path else Database() self.mailer = MailerService(self.db) self.base_url = base_url or os.environ.get("BASE_URL", "") if require_auth is None: self.require_auth = os.environ.get("MAILPULSE_AUTH", "1") != "0" else: self.require_auth = require_auth class MailPulseHandler(BaseHTTPRequestHandler): """HTTP request handler for API routes, tracking pixels, and static UI.""" app: AppContext = None # Set by factory def log_message(self, format, *args): # Suppress noisy access logs during automated tests if env is set if os.environ.get("MAILPULSE_QUIET") == "1": return super().log_message(format, *args) def _base_url(self) -> str: if self.app.base_url: return self.app.base_url proto = self.headers.get("X-Forwarded-Proto") or "http" host = self.headers.get("Host") or "localhost:8765" return f"{proto}://{host}" def _get_session_token(self) -> str: raw_cookie = self.headers.get("Cookie") or "" if not raw_cookie: return "" cookie = SimpleCookie() try: cookie.load(raw_cookie) except Exception: return "" morsel = cookie.get("mp_session") return morsel.value if morsel else "" def _current_user(self) -> dict: if not self.app.require_auth: return {"id": "u-admin", "username": "admin", "display_name": "Administrador", "role": "admin"} token = self._get_session_token() return self.app.db.get_session_user(token) def _require_auth_api(self) -> dict: user = self._current_user() if not user: self._send_json({"error": "Sesión expirada o no autenticado"}, 401) return None return user def _redirect(self, location: str): self.send_response(302) self.send_header("Location", location) self.send_header("Content-Length", "0") self.end_headers() def _read_json(self) -> dict: length = int(self.headers.get("Content-Length") or 0) if length <= 0: return {} raw = self.rfile.read(length).decode("utf-8") return json.loads(raw) if raw.strip() else {} def _send_json(self, payload, status: int = 200, extra_headers: dict = None): body = json.dumps(payload, ensure_ascii=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.send_header("Cache-Control", "no-store") if extra_headers: for k, v in extra_headers.items(): self.send_header(k, v) self.end_headers() self.wfile.write(body) def _send_html(self, html_str: str, status: int = 200): body = html_str.encode("utf-8") self.send_response(status) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def _serve_static(self, rel_path: str): safe = os.path.normpath(rel_path).lstrip("/") if safe == "" or safe == ".": safe = "index.html" full_path = os.path.abspath(os.path.join(STATIC_DIR, safe)) if not full_path.startswith(os.path.abspath(STATIC_DIR)) or not os.path.isfile(full_path): self.send_error(404, "Archivo no encontrado") return ctype, _ = mimetypes.guess_type(full_path) with open(full_path, "rb") as f: data = f.read() self.send_response(200) self.send_header("Content-Type", f"{ctype or 'application/octet-stream'}; charset=utf-8" if (ctype or "").startswith("text/") or ctype == "application/javascript" else (ctype or "application/octet-stream")) self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) # ---------- HTTP Verbs ---------- def do_GET(self): parsed = urlparse(self.path) path = parsed.path qs = parse_qs(parsed.query) try: # Tracking pixel if path.startswith("/t/o/") and path.endswith(".gif"): tracking_id = path[len("/t/o/") : -4] self.app.db.record_open(tracking_id) self.send_response(200) self.send_header("Content-Type", "image/gif") self.send_header("Content-Length", str(len(TRANSPARENT_GIF))) self.send_header("Cache-Control", "no-store, no-cache, must-revalidate") self.end_headers() self.wfile.write(TRANSPARENT_GIF) return # Unsubscribe page if path.startswith("/unsubscribe/"): contact_id = path[len("/unsubscribe/") :].strip("/") ok = self.app.db.unsubscribe_contact(contact_id) msg = ( "Tu suscripción ha sido cancelada correctamente. No recibirás más campañas masivas." if ok else "El enlace de baja no es válido o el contacto ya no existe." ) self._send_html( f"""Baja de suscripción

Preferencias de Correo

{msg}

""" ) return # Public login page if path == "/login": return self._serve_static("login.html") # Enforce login for private app & API routes if self.app.require_auth: if path.startswith("/api/"): user = self._require_auth_api() if not user: return elif path in ("/", "/index.html"): if not self._current_user(): return self._redirect("/login") # API routes if path == "/api/auth/me": return self._send_json({"user": self._current_user()}) if path == "/api/users": return self._send_json({"users": self.app.db.list_users()}) if path == "/api/dashboard": return self._send_json(self.app.db.get_dashboard_stats()) if path == "/api/smtp": configs = self.app.db.list_smtp_configs() for c in configs: if c.get("password"): c["password"] = "••••••••" return self._send_json({"configs": configs}) if path == "/api/lists": return self._send_json({"lists": self.app.db.list_lists()}) if path == "/api/contacts": list_id = qs.get("list_id", [None])[0] search = qs.get("search", [None])[0] status = qs.get("status", [None])[0] return self._send_json({"contacts": self.app.db.list_contacts(list_id=list_id, search=search, status=status)}) if path == "/api/templates": return self._send_json({"templates": self.app.db.list_templates()}) if path == "/api/studio-assets": return self._send_json({"blocks": STUDIO_BLOCKS, "photos": STOCK_PHOTOS}) if path == "/api/campaigns": return self._send_json({"campaigns": self.app.db.list_campaigns()}) if path.startswith("/api/campaigns/"): cid = path[len("/api/campaigns/") :].strip("/") camp = self.app.db.get_campaign(cid) if not camp: return self._send_json({"error": "Campaña no encontrada"}, 404) logs = self.app.db.list_email_logs(campaign_id=cid, limit=300) return self._send_json({"campaign": camp, "logs": logs}) if path == "/api/logs": send_type = qs.get("send_type", [None])[0] status = qs.get("status", [None])[0] campaign_id = qs.get("campaign_id", [None])[0] search = qs.get("search", [None])[0] limit = int(qs.get("limit", [150])[0]) return self._send_json( {"logs": self.app.db.list_email_logs(send_type=send_type, status=status, campaign_id=campaign_id, search=search, limit=limit)} ) if path.startswith("/api/logs/"): lid = path[len("/api/logs/") :].strip("/") log = self.app.db.get_email_log(lid) if not log: return self._send_json({"error": "Registro no encontrado"}, 404) return self._send_json({"log": log}) if path.startswith("/api/"): return self._send_json({"error": "Endpoint no encontrado"}, 404) # Static files self._serve_static(path) except Exception as exc: self._send_json({"error": str(exc)}, 500) def do_POST(self): parsed = urlparse(self.path) path = parsed.path try: data = self._read_json() # Public auth endpoints if path == "/api/auth/login": user = self.app.db.authenticate_user(data.get("username", ""), data.get("password", "")) if not user: return self._send_json({"error": "Usuario o contraseña incorrectos."}, 401) token = self.app.db.create_session(user["id"]) return self._send_json( {"user": user}, 200, extra_headers={"Set-Cookie": f"mp_session={token}; Path=/; HttpOnly; SameSite=Lax; Max-Age=1209600"}, ) if path == "/api/auth/logout": token = self._get_session_token() self.app.db.delete_session(token) return self._send_json( {"ok": True}, 200, extra_headers={"Set-Cookie": "mp_session=; Path=/; HttpOnly; SameSite=Lax; Max-Age=0"}, ) current_user = self._require_auth_api() if self.app.require_auth else self._current_user() if not current_user: return if path == "/api/users": if current_user.get("role") != "admin" and data.get("id") != current_user.get("id"): return self._send_json({"error": "Solo un administrador puede crear o editar otros usuarios."}, 403) if current_user.get("role") != "admin": data["role"] = current_user.get("role") saved = self.app.db.upsert_user(data) return self._send_json({"user": saved}, 201) if path == "/api/smtp": cfg = self.app.db.upsert_smtp_config(data) if cfg.get("password"): cfg["password"] = "••••••••" return self._send_json({"config": cfg}) if path == "/api/smtp/test": cfg_data = dict(data) if cfg_data.get("id") and (not cfg_data.get("password") or cfg_data.get("password") == "••••••••"): existing = self.app.db.get_smtp_config(cfg_data["id"]) if existing: cfg_data["password"] = existing.get("password", "") res = self.app.mailer.test_smtp_connection(cfg_data) return self._send_json(res, 200 if res.get("ok") else 400) if path == "/api/lists": if not (data.get("name") or "").strip(): return self._send_json({"error": "El nombre de la lista es obligatorio"}, 400) lst = self.app.db.create_list(data["name"], data.get("description", "")) return self._send_json({"list": lst}, 201) if path == "/api/contacts": contact = self.app.db.upsert_contact(data) return self._send_json({"contact": contact}, 201) if path == "/api/contacts/import": csv_text = data.get("csv_text", "") list_id = data.get("list_id") rows = parse_csv_contacts(csv_text) if not rows: return self._send_json({"error": "No se encontraron contactos válidos en el CSV (asegúrate de incluir la columna email)."}, 400) imported = 0 for row in rows: if list_id: row["list_ids"] = [list_id] self.app.db.upsert_contact(row) imported += 1 return self._send_json({"imported": imported}) if path == "/api/templates": tpl = self.app.db.upsert_template(data) return self._send_json({"template": tpl}, 201) if path == "/api/templates/export-outlook": tpl = self.app.db.get_template(data["template_id"]) if data.get("template_id") else None raw_subject = data.get("subject") if data.get("subject") is not None else (tpl["subject"] if tpl else "") raw_html = data.get("html_content") if data.get("html_content") is not None else (tpl["html_content"] if tpl else "") bg_color = data.get("bg_color") or "#f1f5f9" contact = self.app.db.get_contact(contact_id=data["contact_id"]) if data.get("contact_id") else None ctx = build_contact_context(contact, extra_vars=data.get("extra_vars")) rendered_subject = render_template_string(raw_subject, ctx) rendered_body = render_template_string(raw_html, ctx) outlook_doc = wrap_for_outlook(rendered_body, subject=rendered_subject, bg_color=bg_color) plain = html_to_plain_text(rendered_body) eml = MIMEMultipart("alternative") eml["Subject"] = str(Header(rendered_subject, "utf-8")) eml["X-Unsent"] = "1" if data.get("to_email"): eml["To"] = data["to_email"] eml.attach(MIMEText(plain, "plain", "utf-8")) eml.attach(MIMEText(outlook_doc, "html", "utf-8")) return self._send_json( { "subject": rendered_subject, "body_html": rendered_body, "outlook_html": outlook_doc, "plain_text": plain, "eml_content": eml.as_string(), } ) if path == "/api/campaigns": camp = self.app.db.upsert_campaign(data) return self._send_json({"campaign": camp}, 201) if path.startswith("/api/campaigns/") and path.endswith("/send"): cid = path[len("/api/campaigns/") : -len("/send")].strip("/") res = self.app.mailer.start_campaign_async(cid, base_url=self._base_url()) return self._send_json(res) if path == "/api/send-individual": log = self.app.mailer.send_single_email( to_email=data.get("to_email", ""), subject=data.get("subject", ""), html_content=data.get("html_content", ""), text_content=data.get("text_content", ""), to_name=data.get("to_name", ""), cc=data.get("cc", ""), bcc=data.get("bcc", ""), sender_id=data.get("sender_id"), contact_id=data.get("contact_id"), extra_vars=data.get("extra_vars"), attachments=data.get("attachments"), send_type="individual", track_opens=bool(data.get("track_opens", True)), base_url=self._base_url(), include_unsubscribe=False, ) status_code = 200 if log["status"] in ("sent", "simulated") else 502 return self._send_json({"log": log}, status_code) if path == "/api/preview": contact = None if data.get("contact_id"): contact = self.app.db.get_contact(contact_id=data["contact_id"]) elif data.get("to_email"): contact = self.app.db.get_contact(email=data["to_email"]) ctx = build_contact_context(contact, extra_vars=data.get("extra_vars")) if not ctx["email"] and data.get("to_email"): ctx["email"] = data["to_email"] if not ctx["nombre"] and data.get("to_name"): ctx["nombre"] = data["to_name"].split()[0] return self._send_json( { "subject": render_template_string(data.get("subject", ""), ctx), "html": render_template_string(data.get("html_content", ""), ctx), } ) if path.startswith("/api/logs/") and path.endswith("/simulate-open"): tracking_id = path[len("/api/logs/") : -len("/simulate-open")].strip("/") ok = self.app.db.record_open(tracking_id) return self._send_json({"ok": ok}, 200 if ok else 404) return self._send_json({"error": "Endpoint no encontrado"}, 404) except ValueError as exc: return self._send_json({"error": str(exc)}, 400) except Exception as exc: return self._send_json({"error": str(exc)}, 500) def do_DELETE(self): path = urlparse(self.path).path try: current_user = self._require_auth_api() if self.app.require_auth else self._current_user() if not current_user: return if path.startswith("/api/users/"): uid = path[len("/api/users/") :].strip("/") if current_user.get("role") != "admin": return self._send_json({"error": "Solo un administrador puede eliminar usuarios."}, 403) if uid == current_user.get("id"): return self._send_json({"error": "No puedes eliminar tu propio usuario activo."}, 400) return self._send_json({"deleted": self.app.db.delete_user(uid)}) if path.startswith("/api/smtp/"): cid = path[len("/api/smtp/") :].strip("/") return self._send_json({"deleted": self.app.db.delete_smtp_config(cid)}) if path.startswith("/api/lists/"): lid = path[len("/api/lists/") :].strip("/") return self._send_json({"deleted": self.app.db.delete_list(lid)}) if path.startswith("/api/contacts/"): cid = path[len("/api/contacts/") :].strip("/") return self._send_json({"deleted": self.app.db.delete_contact(cid)}) if path.startswith("/api/templates/"): tid = path[len("/api/templates/") :].strip("/") return self._send_json({"deleted": self.app.db.delete_template(tid)}) if path.startswith("/api/campaigns/"): cid = path[len("/api/campaigns/") :].strip("/") return self._send_json({"deleted": self.app.db.delete_campaign(cid)}) return self._send_json({"error": "Endpoint no encontrado"}, 404) except Exception as exc: return self._send_json({"error": str(exc)}, 500) def make_handler_class(app_ctx: AppContext): """Create a bound RequestHandler subclass carrying the AppContext.""" class BoundHandler(MailPulseHandler): app = app_ctx return BoundHandler def start_scheduler_loop(app_ctx: AppContext, stop_event: threading.Event): """Background loop checking scheduled campaigns every 15 seconds.""" while not stop_event.is_set(): try: due = app_ctx.db.get_due_scheduled_campaigns() for camp in due: app_ctx.mailer.start_campaign_async(camp["id"], base_url=app_ctx.base_url) except Exception: pass stop_event.wait(15) def run_server(host: str = "127.0.0.1", port: int = 8765, db_path: str = None): base_url = f"http://{host}:{port}" app_ctx = AppContext(db_path=db_path, base_url=base_url) handler_cls = make_handler_class(app_ctx) server = ThreadingHTTPServer((host, port), handler_cls) stop_event = threading.Event() sched = threading.Thread(target=start_scheduler_loop, args=(app_ctx, stop_event), daemon=True) sched.start() print(f"MailPulse escuchando en {base_url}") try: server.serve_forever() except KeyboardInterrupt: print("\nDeteniendo MailPulse...") finally: stop_event.set() server.server_close() if __name__ == "__main__": port = int(os.environ.get("PORT", "8765")) host = os.environ.get("HOST", "127.0.0.1") run_server(host=host, port=port)