andaquenosabesna/server.py

"""Endpoint del formulario de contacto: POST /api/contacto -> correo por SMTP.

Configuración solo por variables de entorno (nunca en el repo):
  SMTP_HOST, SMTP_PORT (587), SMTP_USER, SMTP_PASS, SMTP_TLS_NAME (nombre del certificado, por defecto SMTP_HOST),
  MAIL_FROM, MAIL_TO
No guarda los mensajes en disco: solo los reenvía por correo.
"""
import datetime
from zoneinfo import ZoneInfo
import json
import os
from urllib.parse import urlparse, parse_qs
import re
import smtplib
import ssl
import sys
import time
from collections import defaultdict, deque
from email.message import EmailMessage
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

MOTIVOS = {
    'venir': 'Quiero venir al programa',
    'tema': 'Propongo un tema',
    'otro': 'Otro asunto',
}
TIPOS = {
    'mencion': 'Anunciarme / que me mencionéis en el programa',
    'patrocinio': 'Patrocinar episodios o una temporada',
    'colab': 'Colaboración o canje',
    'otro': 'Otra propuesta',
}
EMAIL_RE = re.compile(r'^[^@\s]{1,64}@[^@\s]{1,190}\.[^@\s]{2,}$')
MAX_BODY = 16 * 1024
RATE_MAX, RATE_WINDOW = 5, 3600          # 5 mensajes por IP y hora
hits = defaultdict(deque)


def clean(s, n):
    return re.sub(r'[\r\n\t]+', ' ', str(s or '')).strip()[:n]


def rate_ok(ip):
    now = time.time()
    q = hits[ip]
    while q and now - q[0] > RATE_WINDOW:
        q.popleft()
    if len(q) >= RATE_MAX:
        return False
    q.append(now)
    return True


def send(subject, reply_to, body, to=None):
    host, user, pwd = os.environ.get('SMTP_HOST'), os.environ.get('SMTP_USER'), os.environ.get('SMTP_PASS')
    if not (host and user and pwd and os.environ.get('MAIL_TO')):
        raise RuntimeError('SMTP no configurado')
    msg = EmailMessage()
    msg['Subject'] = subject
    msg['From'] = os.environ.get('MAIL_FROM', user)
    msg['To'] = to or os.environ['MAIL_TO']
    if reply_to:
        msg['Reply-To'] = reply_to
    msg.set_content(body)
    s = smtplib.SMTP(host, int(os.environ.get('SMTP_PORT', '587')), timeout=20)
    try:
        s.ehlo()
        s._host = os.environ.get('SMTP_TLS_NAME', host)   # el certificado se valida contra este nombre
        s.starttls(context=ssl.create_default_context())
        s.ehlo()
        s.login(user, pwd)
        s.send_message(msg)
    finally:
        try:
            s.quit()
        except Exception:
            pass


def load_guest():
    """Preguntas de la ficha (copia del formulario de Google) -> [{id, sec, t, req}]."""
    p = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'guest_form.json')
    try:
        form = json.load(open(p, encoding='utf-8'))
    except Exception:
        return []
    out, sec, n = [], '', 0
    for it in form['items']:
        if it['type'] == 8:
            sec = it['title']
        elif it['type'] in (0, 1):
            n += 1
            out.append({'id': f'q{n:02d}', 'sec': sec, 't': it['title'], 'req': it['required']})
    return out


GUEST_Q = load_guest()
check_hits = defaultdict(deque)
guest_hits = defaultdict(deque)


def limited(bucket, ip, maxn, window=3600):
    now = time.time()
    q = bucket[ip]
    while q and now - q[0] > window:
        q.popleft()
    if len(q) >= maxn:
        return True
    q.append(now)
    return False


confirmados = set()
DIAS = ['lunes', 'martes', 'miércoles', 'jueves', 'viernes', 'sábado', 'domingo']
MESES = ['enero', 'febrero', 'marzo', 'abril', 'mayo', 'junio', 'julio', 'agosto', 'septiembre', 'octubre', 'noviembre', 'diciembre']


def cita_txt(ts):
    d = datetime.datetime.fromtimestamp(ts, ZoneInfo('Europe/Madrid'))
    return f'{DIAS[d.weekday()]} {d.day} de {MESES[d.month - 1]} de {d.year}, a las {d:%H:%M}'


NAME_RE = re.compile(r"^[^\W\d_][\w .'\-]{1,59}$", re.UNICODE)


def invitado(n, f):
    """Enlace = nombre + fecha de grabación como marca de tiempo Unix (segundos). Válido de 14 días antes a 10 días después."""
    n = ' '.join(str(n or '').split())
    if not NAME_RE.match(n):
        return None
    try:
        ts = int(str(f).strip())
        dia = datetime.datetime.fromtimestamp(ts, datetime.timezone.utc).date()
    except (ValueError, OverflowError, OSError):
        return None
    delta = (dia - datetime.date.today()).days
    return (n, ts) if -10 <= delta <= 14 else None


class H(BaseHTTPRequestHandler):
    server_version = 'aqnsn'

    def log_message(self, fmt, *a):          # sin IPs ni contenido en los logs
        pass

    def reply(self, code, obj):
        body = json.dumps(obj).encode()
        self.send_response(code)
        self.send_header('Content-Type', 'application/json')
        self.send_header('Content-Length', str(len(body)))
        self.send_header('Cache-Control', 'no-store')
        self.end_headers()
        self.wfile.write(body)

    def client_ip(self):
        return (self.headers.get('X-Forwarded-For') or self.client_address[0]).split(',')[0].strip()

    def do_GET(self):
        if urlparse(self.path).path != '/api/invitados/check':
            return self.reply(404, {'ok': False})
        if limited(check_hits, self.client_ip(), 40):
            return self.reply(429, {'ok': False})
        q = parse_qs(urlparse(self.path).query)
        inv = invitado(q.get('n', [''])[0], q.get('f', [''])[0])
        return self.reply(200, {'ok': True, 'nombre': inv[0], 'cita': inv[1]} if inv else {'ok': False})

    def invitados(self):
        try:
            n = int(self.headers.get('Content-Length', '0'))
            if n <= 0 or n > 128 * 1024:
                return self.reply(413, {'ok': False, 'error': 'La ficha es demasiado larga.'})
            d = json.loads(self.rfile.read(n))
        except Exception:
            return self.reply(400, {'ok': False, 'error': 'Petición no válida.'})
        if d.get('web') or (isinstance(d.get('t'), (int, float)) and d['t'] < 20000):
            return self.reply(200, {'ok': True})          # trampa o demasiado rápido: no se envía
        if limited(guest_hits, self.client_ip(), 8):
            return self.reply(429, {'ok': False, 'error': 'Demasiados envíos. Inténtalo más tarde.'})
        inv = invitado(d.get('n'), d.get('f'))
        nombre, ts = inv if inv else (None, None)
        if not nombre:
            return self.reply(403, {'ok': False, 'error': 'Enlace no válido o caducado.'})
        if d.get('consent') is not True:
            return self.reply(400, {'ok': False, 'error': 'Debes aceptar las condiciones de participación.'})
        resp = d.get('resp') if isinstance(d.get('resp'), dict) else {}
        email = clean(d.get('email'), 200)
        if not email or not EMAIL_RE.match(email):
            return self.reply(400, {'ok': False, 'error': 'El correo no parece válido.'})
        lineas, seccion = [], None
        for q in GUEST_Q:
            ans = str(resp.get(q['id']) or '').replace('\r', '').strip()[:3000]
            if q['req'] and not ans:
                return self.reply(400, {'ok': False, 'error': 'Faltan respuestas obligatorias.'})
            if q['sec'] != seccion:
                seccion = q['sec']
                lineas += ['', f'== {seccion} ==']
            lineas += ['', f"• {q['t']}", '  ' + (ans.replace('\n', '\n  ') if ans else '(sin respuesta)')]
        cabecera = [f'Ficha del invitado: {nombre}', 'Recibida: ' + time.strftime('%d/%m/%Y %H:%M', time.localtime()),
                    f'Cita: {cita_txt(ts)}', f'Correo del invitado: {email}',
                    'Aceptó las condiciones de participación y la política de privacidad: sí']
        try:
            send(f'[Invitado] {clean(nombre, 80)} · ficha del podcast', email, '\n'.join(cabecera + lineas) + '\n')
        except RuntimeError:
            return self.reply(503, {'ok': False, 'error': 'El formulario no está disponible ahora mismo.'})
        except Exception as exc:
            print('error enviando ficha:', type(exc).__name__, file=sys.stderr, flush=True)
            return self.reply(502, {'ok': False, 'error': 'No se pudo enviar. Inténtalo más tarde.'})
        key = (nombre, ts)
        if key not in confirmados:                         # un solo recordatorio por invitado y cita
            confirmados.add(key)
            try:
                send('Tu cita en Anda Que No Sabes Ná', None,
                     f'Hola, {nombre}:\n\nGracias por rellenar tu ficha. Recuerda que grabamos el '
                     f'{cita_txt(ts)} (hora de Madrid), y te enviaremos por correo la dirección del estudio.\n\n'
                     'Si no puedes venir o necesitas cambiar algo, responde a este correo.\n\n'
                     'Un abrazo,\nÁlex y Juan\nAnda Que No Sabes Ná\nhttps://andaquenosabesna.com\n', to=email)
            except Exception as exc:
                confirmados.discard(key)
                print('error enviando recordatorio:', type(exc).__name__, file=sys.stderr, flush=True)
        self.reply(200, {'ok': True})

    def do_POST(self):
        if self.path.split('?')[0] == '/api/invitados':
            return self.invitados()
        if self.path not in ('/api/contacto', '/api/patrocinio'):
            return self.reply(404, {'ok': False})
        patro = self.path == '/api/patrocinio'
        try:
            n = int(self.headers.get('Content-Length', '0'))
            if n <= 0 or n > MAX_BODY:
                return self.reply(413, {'ok': False, 'error': 'Mensaje demasiado largo.'})
            d = json.loads(self.rfile.read(n))
        except Exception:
            return self.reply(400, {'ok': False, 'error': 'Petición no válida.'})
        ip = (self.headers.get('X-Forwarded-For') or self.client_address[0]).split(',')[0].strip()
        # antispam: campo trampa relleno o envío en menos de 3 s => se responde "ok" sin enviar
        if d.get('web') or (isinstance(d.get('t'), (int, float)) and d['t'] < 3000):
            return self.reply(200, {'ok': True})
        if not rate_ok(ip):
            return self.reply(429, {'ok': False, 'error': 'Demasiados mensajes. Inténtalo más tarde.'})
        nombre, email = clean(d.get('nombre'), 100), clean(d.get('email'), 200)
        mensaje = str(d.get('mensaje') or '').replace('\r', '').strip()[:3000]
        bad = {'ok': False, 'error': 'Revisa los campos del formulario.'}
        if not nombre or not EMAIL_RE.match(email) or len(mensaje) < 10 or d.get('consent') is not True:
            return self.reply(400, bad)
        if patro:
            empresa, sitio = clean(d.get('empresa'), 120), clean(d.get('sitio'), 200)
            tipo = d.get('tipo') if d.get('tipo') in TIPOS else 'otro'
            if not empresa or (sitio and not re.match(r'^https?://', sitio)):
                return self.reply(400, bad)
            subject = f'[Web AQNSN · Patrocinio] {empresa} · {nombre}'
            body = (f'Empresa: {empresa}\nContacto: {nombre}\nEmail: {email}\nWeb: {sitio or "—"}\n'
                    f'Interés: {TIPOS[tipo]}\n\n{mensaje}\n')
        else:
            motivo = d.get('motivo') if d.get('motivo') in MOTIVOS else 'otro'
            subject = f'[Web AQNSN] {MOTIVOS[motivo]} · {nombre}'
            body = f'Motivo: {MOTIVOS[motivo]}\nNombre: {nombre}\nEmail: {email}\n\n{mensaje}\n'
        try:
            send(subject, email, body)
        except RuntimeError:
            return self.reply(503, {'ok': False, 'error': 'El formulario no está disponible ahora mismo.'})
        except Exception as exc:
            print('error enviando correo:', type(exc).__name__, file=sys.stderr, flush=True)
            return self.reply(502, {'ok': False, 'error': 'No se pudo enviar. Inténtalo más tarde.'})
        self.reply(200, {'ok': True})


if __name__ == '__main__':
    ThreadingHTTPServer(('127.0.0.1', 8081), H).serve_forever()