#!/usr/bin/env python3 import argparse import base64 import csv import io import json import os import re import secrets import shutil import sqlite3 import time from html import escape as html_escape_lib from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import parse_qs, unquote, urlencode, urljoin, urlparse from urllib.request import Request, urlopen SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") LOCAL_URL_RE = re.compile(r"^https?://(?:127\.0\.0\.1|localhost)(?::\d+)?(?:/.*)?$", re.IGNORECASE) ASSET_SLOTS = {"hero", "event", "quote"} ASSET_EXTENSIONS = {".jpg", ".jpeg", ".png", ".webp", ".svg"} MAX_ASSET_BYTES = 12 * 1024 * 1024 SQLITE_FILENAME = "invitaciones.sqlite3" SPOTIFY_TOKEN_CACHE = {"access_token": "", "expires_at": 0} CLIENT_PANEL_FIELDS = { "tipo", "nombres", "fechaCorta", "fechaISO", "frase", "firma", "footer", "salon", "direccion", "hora", "dresscode", "dresscodeNote", "dresscodeColor1Enabled", "dresscodeColor1Name", "dresscodeColor1", "dresscodeColor2Enabled", "dresscodeColor2Name", "dresscodeColor2", "dresscodeColor3Enabled", "dresscodeColor3Name", "dresscodeColor3", "dresscodeAvoid1Enabled", "dresscodeAvoid1Name", "dresscodeAvoid1", "dresscodeAvoid2Enabled", "dresscodeAvoid2Name", "dresscodeAvoid2", "mapsUrl", "wazeUrl", "whatsapp", "shareUrl", "whatsappRsvpText", "whatsappShareText", "whatsappLocationText", "whatsappOrganizerText", "alias", "cbu", "titular", "heroImage", "eventImage", "heroImagePositionX", "heroImagePositionY", "heroImageZoom", "eventImagePositionX", "eventImagePositionY", "eventImageZoom", "publicUrl", "ogTitle", "ogDescription", "ogImage", "favicon", "paletteColor1", "paletteColor2", "paletteColor3", "paletteColor4", "paletteColor5", } class AdminError(Exception): def __init__(self, status, message): self.status = status self.message = message super().__init__(message) class AdminHandler(SimpleHTTPRequestHandler): repo_root: Path def end_headers(self): self.send_header("Cache-Control", "no-store") self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type, X-Filename") super().end_headers() def do_OPTIONS(self): self.send_response(204) self.end_headers() def do_GET(self): try: parsed = urlparse(self.path) route = unquote(parsed.path) if admin_auth_required(route) and not valid_admin_auth(self.headers): self.send_admin_auth_required() return if ( route == "/clientes/_access" or route.startswith("/clientes/_access/") or route == "/clientes/_data" or route.startswith("/clientes/_data/") ): raise AdminError(404, "Archivo no encontrado") if route == "/api/templates": self.send_json({"templates": list_templates(self.repo_root)}) return if route == "/api/clientes": self.send_json({"clientes": list_clients(self.repo_root)}) return if route == "/api/archivados": self.send_json({"archivados": list_archived_clients(self.repo_root)}) return if route == "/api/spotify/search": self.send_json(spotify_search_response(parsed)) return dashboard_match = re.fullmatch(r"/api/cliente/([^/]+)/dashboard", route) if dashboard_match: slug = validate_slug(dashboard_match.group(1)) require_client_token(self.repo_root, slug, token_from_query(parsed)) self.send_json(client_dashboard(self.repo_root, slug)) return csv_match = re.fullmatch(r"/api/cliente/([^/]+)/dashboard.csv", route) if csv_match: slug = validate_slug(csv_match.group(1)) require_client_token(self.repo_root, slug, token_from_query(parsed)) self.send_text(client_dashboard_csv(self.repo_root, slug), "text/csv; charset=utf-8") return xls_match = re.fullmatch(r"/api/cliente/([^/]+)/dashboard.xls", route) if xls_match: slug = validate_slug(xls_match.group(1)) require_client_token(self.repo_root, slug, token_from_query(parsed)) self.send_text(client_dashboard_xls(self.repo_root, slug), "application/vnd.ms-excel; charset=utf-8") return client_panel_match = re.fullmatch(r"/api/cliente/([^/]+)", route) if client_panel_match: slug = validate_slug(client_panel_match.group(1)) require_client_token(self.repo_root, slug, token_from_query(parsed)) config = read_client_config(self.repo_root, slug) self.send_json({ "slug": slug, "config": filter_client_config(config), "previewUrl": client_preview_url(slug), }) return match = re.fullmatch(r"/api/clientes/([^/]+)/config", route) if match: slug = validate_slug(match.group(1)) config_path = client_dir(self.repo_root, slug) / "config.json" if not config_path.exists(): raise AdminError(404, "Config no encontrado") self.send_json(read_json(config_path)) return super().do_GET() except AdminError as error: self.send_json({"ok": False, "error": error.message}, status=error.status) def do_POST(self): try: parsed = urlparse(self.path) route = unquote(parsed.path) if admin_auth_required(route) and not valid_admin_auth(self.headers): self.send_admin_auth_required() return if route == "/api/clientes": self.create_client() return public_rsvp_match = re.fullmatch(r"/api/public/([^/]+)/(rsvp|song)", route) if public_rsvp_match: self.save_public_entry(public_rsvp_match.group(1), public_rsvp_match.group(2)) return client_asset_match = re.fullmatch(r"/api/cliente/([^/]+)/assets/([^/]+)", route) if client_asset_match: slug = validate_slug(client_asset_match.group(1)) require_client_token(self.repo_root, slug, token_from_query(parsed)) self.save_asset(slug, client_asset_match.group(2)) return client_config_match = re.fullmatch(r"/api/cliente/([^/]+)/config", route) if client_config_match: self.save_client_panel_config(client_config_match.group(1), token_from_query(parsed)) return asset_match = re.fullmatch(r"/api/clientes/([^/]+)/assets/([^/]+)", route) if asset_match: self.save_asset(asset_match.group(1), asset_match.group(2)) return config_match = re.fullmatch(r"/api/clientes/([^/]+)/config", route) if config_match: self.save_config(config_match.group(1)) return deploy_match = re.fullmatch(r"/api/clientes/([^/]+)/(deploy|restart)", route) if deploy_match: self.run_compose(deploy_match.group(1), deploy_match.group(2)) return restore_match = re.fullmatch(r"/api/archivados/([^/]+)/restore", route) if restore_match: self.restore_client(restore_match.group(1)) return self.send_json({"ok": False, "error": "Endpoint no encontrado"}, status=404) except AdminError as error: self.send_json({"ok": False, "error": error.message}, status=error.status) def do_DELETE(self): try: parsed = urlparse(self.path) route = unquote(parsed.path) if admin_auth_required(route) and not valid_admin_auth(self.headers): self.send_admin_auth_required() return match = re.fullmatch(r"/api/clientes/([^/]+)", route) if not match: self.send_json({"ok": False, "error": "Endpoint no encontrado"}, status=404) return slug = validate_slug(match.group(1)) source = client_dir(self.repo_root, slug) if not source.exists(): raise AdminError(404, "Cliente no encontrado") archive_root = self.repo_root / "clientes" / "_archivados" archive_root.mkdir(parents=True, exist_ok=True) target = archive_root / f"{slug}-{time.strftime('%Y%m%d-%H%M%S')}" shutil.move(str(source), str(target)) self.send_json({ "ok": True, "archivedPath": str(target.relative_to(self.repo_root)), }) except AdminError as error: self.send_json({"ok": False, "error": error.message}, status=error.status) def read_json_body(self): length = int(self.headers.get("Content-Length", "0")) raw_body = self.rfile.read(length) try: return json.loads(raw_body.decode("utf-8") or "{}") except json.JSONDecodeError: raise AdminError(400, "JSON invalido") def send_json(self, payload, status=200): 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.end_headers() self.wfile.write(body) def send_text(self, text, content_type="text/plain; charset=utf-8", status=200): body = text.encode("utf-8") self.send_response(status) self.send_header("Content-Type", content_type) self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def send_admin_auth_required(self): body = b"Autenticacion requerida\n" self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="Invitaciones Admin"') self.send_header("Content-Type", "text/plain; charset=utf-8") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) def create_client(self): payload = self.read_json_body() slug = validate_slug(payload.get("slug", "")) template = validate_slug(payload.get("template", "")) display_name = str(payload.get("name", "")).strip() template_path = self.repo_root / "templates" / template if not template_path.exists(): raise AdminError(404, "Template no encontrado") target = client_dir(self.repo_root, slug) if target.exists(): raise AdminError(409, "El cliente ya existe") shutil.copytree(template_path, target) config_path = target / "config.json" config = read_json(config_path) config["clientSlug"] = slug if display_name: config["nombres"] = display_name config["seoTitle"] = f"{display_name} - Invitacion" config["footer"] = display_name normalize_client_urls(config, slug, self) prepare_social_preview(target, config) write_json(config_path, config) self.send_json({ "ok": True, "cliente": client_summary(self.repo_root, slug), "url": client_public_url(slug), }, status=201) def save_config(self, slug): slug = validate_slug(slug) payload = self.read_json_body() target_dir = client_dir(self.repo_root, slug) if not target_dir.exists(): raise AdminError(404, "Cliente no encontrado") target = target_dir / "config.json" normalize_client_urls(payload, slug, self) prepare_social_preview(target_dir, payload) write_json(target, payload) self.send_json({"ok": True, "path": str(target.relative_to(self.repo_root))}) def save_client_panel_config(self, slug, token): slug = validate_slug(slug) require_client_token(self.repo_root, slug, token) payload = self.read_json_body() current_config = read_client_config(self.repo_root, slug) for key, value in payload.items(): if key in CLIENT_PANEL_FIELDS: current_config[key] = value target = client_dir(self.repo_root, slug) / "config.json" normalize_client_urls(current_config, slug, self) prepare_social_preview(client_dir(self.repo_root, slug), current_config) write_json(target, current_config) self.send_json({"ok": True, "config": filter_client_config(current_config)}) def save_public_entry(self, slug, kind): slug = validate_slug(slug) if not client_dir(self.repo_root, slug).exists(): raise AdminError(404, "Cliente no encontrado") payload = self.read_json_body() payload["fecha"] = payload.get("fecha") or time.strftime("%Y-%m-%dT%H:%M:%S") store_public_entry(self.repo_root, slug, kind, payload) self.send_json({"ok": True}) def save_asset(self, slug, slot): slug = validate_slug(slug) if slot not in ASSET_SLOTS: raise AdminError(400, "Slot de imagen invalido") target_dir = client_dir(self.repo_root, slug) if not target_dir.exists(): raise AdminError(404, "Cliente no encontrado") original_name = self.headers.get("X-Filename", "") extension = Path(original_name).suffix.lower() if extension not in ASSET_EXTENSIONS: raise AdminError(400, "Formato no permitido. Usar jpg, png, webp o svg.") length = int(self.headers.get("Content-Length", "0")) if length <= 0 or length > MAX_ASSET_BYTES: raise AdminError(400, "Archivo vacio o demasiado grande") assets_dir = target_dir / "assets" assets_dir.mkdir(exist_ok=True) target = assets_dir / f"{slot}{extension}" tmp_target = assets_dir / f".{slot}{extension}.tmp" tmp_target.write_bytes(self.rfile.read(length)) os.replace(tmp_target, target) self.send_json({"ok": True, "path": f"assets/{target.name}"}) def run_compose(self, slug, action): slug = validate_slug(slug) if not client_dir(self.repo_root, slug).exists(): raise AdminError(404, "Cliente no encontrado") self.send_json({ "ok": True, "output": "No requiere publicacion manual por cliente: el contenedor web sirve la carpeta automaticamente.", }) def restore_client(self, archive_name): archive_name = validate_archive_name(archive_name) archive_dir = archive_dir_for_name(self.repo_root, archive_name) if not archive_dir.exists(): raise AdminError(404, "Archivado no encontrado") slug = slug_from_archive_name(archive_name) target = client_dir(self.repo_root, slug) if target.exists(): raise AdminError(409, "Ya existe un cliente activo con ese slug") shutil.move(str(archive_dir), str(target)) config_path = target / "config.json" if config_path.exists(): config = read_json(config_path) normalize_client_urls(config, slug, self) prepare_social_preview(target, config) write_json(config_path, config) self.send_json({ "ok": True, "cliente": client_summary(self.repo_root, slug), "url": client_public_url(slug), }) def validate_slug(value): slug = str(value or "").strip() if not SLUG_RE.fullmatch(slug): raise AdminError(400, "Slug invalido") return slug def validate_archive_name(value): archive_name = str(value or "").strip() if not re.fullmatch(r"[a-z0-9][a-z0-9-]*-\d{8}-\d{6}", archive_name): raise AdminError(400, "Nombre de archivado invalido") return archive_name def slug_from_archive_name(archive_name): match = re.fullmatch(r"(.+)-\d{8}-\d{6}", archive_name) if not match: raise AdminError(400, "Nombre de archivado invalido") return validate_slug(match.group(1)) def client_dir(repo_root, slug): path = (repo_root / "clientes" / slug).resolve() clients_root = (repo_root / "clientes").resolve() if clients_root not in path.parents: raise AdminError(400, "Ruta de cliente invalida") return path def archive_dir_for_name(repo_root, archive_name): path = (repo_root / "clientes" / "_archivados" / archive_name).resolve() archive_root = (repo_root / "clientes" / "_archivados").resolve() if archive_root not in path.parents: raise AdminError(400, "Ruta de archivado invalida") return path def read_json(path): return json.loads(path.read_text(encoding="utf-8")) def read_client_config(repo_root, slug): config_path = client_dir(repo_root, slug) / "config.json" if not config_path.exists(): raise AdminError(404, "Config no encontrado") return read_json(config_path) def write_json(path, payload): tmp_target = path.with_name(f".{path.name}.tmp") tmp_target.write_text( json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8", ) os.replace(tmp_target, path) def filter_client_config(config): return {key: config.get(key, "") for key in sorted(CLIENT_PANEL_FIELDS)} def html_attr(value): return html_escape_lib(str(value or ""), quote=True) def social_title(config): return ( config.get("ogTitle") or " - ".join(part for part in [config.get("nombres"), config.get("fechaCorta")] if part) or config.get("seoTitle") or "Invitacion" ) def social_description(config): return ( config.get("ogDescription") or config.get("seoDescription") or "Te invitamos a celebrar. Confirma asistencia y mira todos los detalles." ) def initials_from_name(value): words = re.findall(r"[A-Za-zÁÉÍÓÚÜÑáéíóúüñ0-9]+", str(value or "")) if not words: return "IV" if len(words) == 1: return words[0][:2].upper() return (words[0][0] + words[-1][0]).upper() def public_asset_url(config, path): path = str(path or "") public_url = str(config.get("publicUrl") or "").strip() if not public_url or re.match(r"^https?://", path): return path return urljoin(public_url.rstrip("/") + "/", path) def ensure_social_config_defaults(config): config.setdefault("publicUrl", "") config.setdefault("ogTitle", "") config.setdefault("ogDescription", "") config.setdefault("ogImage", "assets/og-image.svg") config.setdefault("favicon", "assets/favicon.svg") def prepare_social_preview(client_path, config): ensure_social_config_defaults(config) write_social_assets(client_path, config) update_social_head(client_path / "index.html", config) def write_social_assets(client_path, config): assets_dir = client_path / "assets" assets_dir.mkdir(exist_ok=True) favicon = str(config.get("favicon") or "") if favicon.startswith("assets/") and Path(favicon).suffix.lower() == ".svg": (client_path / favicon).write_text(render_favicon_svg(config), encoding="utf-8") og_image = str(config.get("ogImage") or "") if og_image.startswith("assets/") and Path(og_image).suffix.lower() == ".svg": (client_path / og_image).write_text(render_og_image_svg(config), encoding="utf-8") def render_favicon_svg(config): initials = html_attr(initials_from_name(config.get("nombres"))) return f''' {initials} ''' def render_og_image_svg(config): title = html_attr(social_title(config)) event_type = html_attr(config.get("tipo") or "Invitacion") date = html_attr(config.get("fechaCorta") or "") place = html_attr(config.get("salon") or "") return f''' {title} Imagen para compartir invitacion. {event_type} {title} {date} {place} ''' def update_social_head(index_path, config): if not index_path.exists(): return html = index_path.read_text(encoding="utf-8") title = social_title(config) description = social_description(config) og_image = public_asset_url(config, config.get("ogImage")) favicon = config.get("favicon") or "assets/favicon.svg" public_url = str(config.get("publicUrl") or "").strip() og_url = public_url.rstrip("/") + "/" if public_url else "" head_block = social_head_block(title, description, og_image, og_url, favicon) start = html.find(" ") end_markers = [ html.find(" <link rel=\"preconnect\"", start), html.find(" <link rel=\"stylesheet\"", start), html.find(" <style", start), html.find("</head>", start), ] end_candidates = [index for index in end_markers if index != -1] if start != -1 and end_candidates: end = min(end_candidates) updated = html[:start] + head_block + html[end:] else: updated = html.replace( " <meta name=\"viewport\" content=\"width=device-width, initial-scale=1, viewport-fit=cover\">\n", " <meta name=\"viewport\" content=\"width=device-width, initial-scale=1, viewport-fit=cover\">\n" + head_block, 1, ) index_path.write_text(updated, encoding="utf-8") def social_head_block(title, description, og_image, og_url, favicon): lines = [ f" <title>{html_attr(title)}", f" ", " ", f" ", f" ", ] if og_url: lines.append(f" ") if og_image: lines.append(f" ") lines.extend([ " ", f" ", f" ", ]) if og_image: lines.append(f" ") lines.extend([ f" ", " ", ]) return "\n".join(lines) + "\n" def token_from_query(parsed_url): return parse_qs(parsed_url.query).get("token", [""])[0] def admin_password(): return os.environ.get("ADMIN_PASSWORD", "") def admin_user(): return os.environ.get("ADMIN_USER", "admin") def admin_auth_required(route): return route == "/admin" or route.startswith("/admin/") or is_admin_api_route(route) def is_admin_api_route(route): return ( route == "/api/templates" or route == "/api/clientes" or route.startswith("/api/clientes/") or route == "/api/archivados" or route.startswith("/api/archivados/") ) def valid_admin_auth(headers): password = admin_password() if not password: return True auth_header = headers.get("Authorization", "") scheme, _, token = auth_header.partition(" ") if scheme.lower() != "basic" or not token: return False try: decoded = base64.b64decode(token, validate=True).decode("utf-8") except (ValueError, UnicodeDecodeError): return False user, separator, supplied_password = decoded.partition(":") if not separator: return False return secrets.compare_digest(user, admin_user()) and secrets.compare_digest(supplied_password, password) def is_local_or_empty_url(value): url = str(value or "").strip() return not url or LOCAL_URL_RE.fullmatch(url) is not None def normalize_client_urls(config, slug, handler=None): admin_url = os.environ.get("ADMIN_PUBLIC_URL", "").strip().rstrip("/") if not admin_url and handler: admin_url = admin_base_url(handler) if admin_url and is_local_or_empty_url(config.get("backendApiUrl")): config["backendApiUrl"] = admin_url public_url = configured_client_public_url(slug) if not public_url: return if is_local_or_empty_url(config.get("publicUrl")): config["publicUrl"] = public_url if is_local_or_empty_url(config.get("shareUrl")): config["shareUrl"] = public_url def admin_base_url(handler): configured = os.environ.get("ADMIN_PUBLIC_URL", "").strip() if configured: return configured.rstrip("/") proto = handler.headers.get("X-Forwarded-Proto", "http").split(",")[0].strip() or "http" host = handler.headers.get("X-Forwarded-Host") or handler.headers.get("Host", "127.0.0.1:8099") return f"{proto}://{host}" def configured_client_public_url(slug): base = os.environ.get("CLIENTS_PUBLIC_URL", "").strip() if not base: return "" return f"{base.rstrip('/')}/clientes/{slug}/" def client_public_url(slug): return configured_client_public_url(slug) or f"http://127.0.0.1:8092/clientes/{slug}/" def client_preview_url(slug): return f"/clientes/{slug}/index.html" def sqlite_path(repo_root): data_root = repo_root / "clientes" / "_data" data_root.mkdir(parents=True, exist_ok=True) return data_root / SQLITE_FILENAME def open_db(repo_root): conn = sqlite3.connect(sqlite_path(repo_root), timeout=10) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") ensure_db_schema(conn) return conn def ensure_db_schema(conn): conn.executescript( """ CREATE TABLE IF NOT EXISTS access_tokens ( slug TEXT PRIMARY KEY, token TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE TABLE IF NOT EXISTS public_entries ( id INTEGER PRIMARY KEY AUTOINCREMENT, slug TEXT NOT NULL, kind TEXT NOT NULL CHECK (kind IN ('rsvp', 'song')), fecha TEXT NOT NULL, payload_json TEXT NOT NULL, created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP ); CREATE INDEX IF NOT EXISTS idx_public_entries_slug_kind_id ON public_entries (slug, kind, id); """ ) def initialize_storage(repo_root): with open_db(repo_root) as conn: migrate_legacy_access_store(repo_root, conn) migrate_legacy_public_entries(repo_root, conn) def migrate_legacy_access_store(repo_root, conn): path = repo_root / "clientes" / "_access" / "client-panel.json" if not path.exists(): return try: access = read_json(path) except json.JSONDecodeError: return if not isinstance(access, dict): return for slug, token in access.items(): if SLUG_RE.fullmatch(str(slug)) and token: conn.execute( "INSERT OR IGNORE INTO access_tokens (slug, token) VALUES (?, ?)", (str(slug), str(token)), ) def migrate_legacy_public_entries(repo_root, conn): data_root = repo_root / "clientes" / "_data" if not data_root.exists(): return for slug_dir in sorted(data_root.iterdir()): if not slug_dir.is_dir() or not SLUG_RE.fullmatch(slug_dir.name): continue for kind, filename in (("rsvp", "rsvps.json"), ("song", "songs.json")): if conn.execute( "SELECT 1 FROM public_entries WHERE slug = ? AND kind = ? LIMIT 1", (slug_dir.name, kind), ).fetchone(): continue path = slug_dir / filename if not path.exists(): continue try: entries = read_json(path) except json.JSONDecodeError: continue if not isinstance(entries, list): continue for payload in entries: if isinstance(payload, dict): store_public_entry_in_conn(conn, slug_dir.name, kind, payload) def store_public_entry_in_conn(conn, slug, kind, payload): payload = dict(payload) fecha = str(payload.get("fecha") or time.strftime("%Y-%m-%dT%H:%M:%S")) payload["fecha"] = fecha conn.execute( """ INSERT INTO public_entries (slug, kind, fecha, payload_json) VALUES (?, ?, ?, ?) """, (slug, kind, fecha, json.dumps(payload, ensure_ascii=False)), ) def ensure_client_token(repo_root, slug): with open_db(repo_root) as conn: row = conn.execute("SELECT token FROM access_tokens WHERE slug = ?", (slug,)).fetchone() if row: return row["token"] token = secrets.token_urlsafe(24) conn.execute( "INSERT INTO access_tokens (slug, token) VALUES (?, ?)", (slug, token), ) return token def require_client_token(repo_root, slug, token): with open_db(repo_root) as conn: row = conn.execute("SELECT token FROM access_tokens WHERE slug = ?", (slug,)).fetchone() expected = row["token"] if row else "" if not expected or not secrets.compare_digest(str(token), expected): raise AdminError(403, "Acceso no autorizado") def read_public_entries(repo_root, slug, kind): with open_db(repo_root) as conn: rows = conn.execute( """ SELECT payload_json FROM public_entries WHERE slug = ? AND kind = ? ORDER BY id """, (slug, kind), ).fetchall() entries = [] for row in rows: try: payload = json.loads(row["payload_json"]) except json.JSONDecodeError: continue if isinstance(payload, dict): entries.append(payload) return entries def store_public_entry(repo_root, slug, kind, payload): with open_db(repo_root) as conn: store_public_entry_in_conn(conn, slug, kind, payload) def spotify_search_response(parsed): params = parse_qs(parsed.query) query = (params.get("q", [""])[0] or "").strip() if len(query) < 2: raise AdminError(400, "Escribi al menos 2 caracteres") if len(query) > 120: raise AdminError(400, "La busqueda es demasiado larga") try: limit = int(params.get("limit", ["6"])[0] or 6) except ValueError: limit = 6 limit = min(max(limit, 1), 8) return {"ok": True, "tracks": spotify_search_tracks(query, limit)} def spotify_search_tracks(query, limit): token = spotify_access_token() market = os.environ.get("SPOTIFY_MARKET", "AR").strip().upper() or "AR" url = "https://api.spotify.com/v1/search?" + urlencode({ "q": query, "type": "track", "market": market, "limit": str(limit), }) request = Request(url, headers={"Authorization": f"Bearer {token}"}) try: with urlopen(request, timeout=8) as response: data = json.loads(response.read().decode("utf-8")) except HTTPError as error: message = spotify_error_message(error) raise AdminError(502, f"Spotify no respondio correctamente: {message}") except (URLError, TimeoutError, json.JSONDecodeError) as error: raise AdminError(502, f"No se pudo consultar Spotify: {error}") tracks = data.get("tracks", {}).get("items", []) return [spotify_track_payload(track) for track in tracks if track.get("uri")] def spotify_access_token(): client_id = os.environ.get("SPOTIFY_CLIENT_ID", "").strip() client_secret = os.environ.get("SPOTIFY_CLIENT_SECRET", "").strip() if not client_id or not client_secret: raise AdminError(503, "Spotify no esta configurado") now = int(time.time()) if SPOTIFY_TOKEN_CACHE["access_token"] and SPOTIFY_TOKEN_CACHE["expires_at"] > now + 60: return SPOTIFY_TOKEN_CACHE["access_token"] credentials = base64.b64encode(f"{client_id}:{client_secret}".encode("utf-8")).decode("ascii") request = Request( "https://accounts.spotify.com/api/token", data=urlencode({"grant_type": "client_credentials"}).encode("utf-8"), headers={ "Authorization": f"Basic {credentials}", "Content-Type": "application/x-www-form-urlencoded", }, method="POST", ) try: with urlopen(request, timeout=8) as response: data = json.loads(response.read().decode("utf-8")) except HTTPError as error: message = spotify_error_message(error) raise AdminError(502, f"No se pudo autenticar Spotify: {message}") except (URLError, TimeoutError, json.JSONDecodeError) as error: raise AdminError(502, f"No se pudo autenticar Spotify: {error}") access_token = data.get("access_token") if not access_token: raise AdminError(502, "Spotify no devolvio access_token") SPOTIFY_TOKEN_CACHE["access_token"] = access_token SPOTIFY_TOKEN_CACHE["expires_at"] = now + int(data.get("expires_in", 3600)) return access_token def spotify_error_message(error): try: payload = json.loads(error.read().decode("utf-8")) except Exception: return str(error) spotify_error = payload.get("error", {}) if isinstance(spotify_error, dict): return spotify_error.get("message") or str(spotify_error.get("status") or error) return payload.get("error_description") or str(spotify_error) or str(error) def spotify_track_payload(track): album = track.get("album") or {} artists = [artist.get("name", "") for artist in track.get("artists", []) if artist.get("name")] images = album.get("images") or [] return { "id": track.get("id", ""), "name": track.get("name", ""), "artist": ", ".join(artists), "album": album.get("name", ""), "image": images[-1]["url"] if images else "", "spotify_uri": track.get("uri", ""), "spotify_url": (track.get("external_urls") or {}).get("spotify", ""), "duration_ms": track.get("duration_ms", 0), } def client_dashboard(repo_root, slug): local_rsvps = read_public_entries(repo_root, slug, "rsvp") local_songs = read_public_entries(repo_root, slug, "song") return build_dashboard_payload(local_rsvps, local_songs, "local") def build_dashboard_payload(rsvps, songs, source): confirmed = [] declined = [] menus = {} total_people = 0 for row in rsvps: asistencia = str(row.get("asistencia", "")).strip().lower() is_declined = asistencia in {"no", "no puedo", "no asisto"} if is_declined: declined.append(row) continue confirmed.append(row) try: total_people += int(row.get("cantidad") or 1) except (TypeError, ValueError): total_people += 1 menu = str(row.get("menu", "") or "Sin menu").strip() or "Sin menu" menus[menu] = menus.get(menu, 0) + 1 return { "ok": True, "source": source, "summary": { "confirmedResponses": len(confirmed), "totalPeople": total_people, "declinedResponses": len(declined), "songCount": len(songs), }, "menus": [{"name": name, "count": count} for name, count in sorted(menus.items())], "rsvps": rsvps, "songs": songs, } def client_dashboard_csv(repo_root, slug): dashboard = client_dashboard(repo_root, slug) output = io.StringIO() writer = csv.writer(output) writer.writerow(["Fecha", "Nombre", "Asistencia", "Cantidad", "Menu", "Mensaje"]) for row in dashboard["rsvps"]: writer.writerow([ row.get("fecha", ""), row.get("nombre", ""), row.get("asistencia", ""), row.get("cantidad", ""), row.get("menu", ""), row.get("mensaje", ""), ]) writer.writerow([]) writer.writerow(["Canciones sugeridas"]) writer.writerow(["Fecha", "Cancion", "Artista", "Album", "Spotify URI", "Spotify URL"]) for row in dashboard["songs"]: writer.writerow([ row.get("fecha", ""), row.get("song", ""), row.get("artist", ""), row.get("album", ""), row.get("spotify_uri", ""), row.get("spotify_url", ""), ]) return output.getvalue() def client_dashboard_xls(repo_root, slug): dashboard = client_dashboard(repo_root, slug) rows = [ "", "

Confirmaciones RSVP

", "", "", ] for row in dashboard["rsvps"]: rows.append( "" f"" f"" f"" f"" f"" f"" "" ) rows.extend([ "
FechaNombreAsistenciaCantidadMenuMensaje
{html_escape(row.get('fecha', ''))}{html_escape(row.get('nombre', ''))}{html_escape(row.get('asistencia', ''))}{html_escape(row.get('cantidad', ''))}{html_escape(row.get('menu', ''))}{html_escape(row.get('mensaje', ''))}
", "

Canciones sugeridas

", "", "", ]) for row in dashboard["songs"]: rows.append( "" f"" f"" f"" f"" f"" f"" "" ) rows.append("
FechaCancionArtistaAlbumSpotify URISpotify URL
{html_escape(row.get('fecha', ''))}{html_escape(row.get('song', ''))}{html_escape(row.get('artist', ''))}{html_escape(row.get('album', ''))}{html_escape(row.get('spotify_uri', ''))}{html_escape(row.get('spotify_url', ''))}
") return "\n".join(rows) def html_escape(value): return str(value or "").replace("&", "&").replace("<", "<").replace(">", ">").replace('"', """) def list_templates(repo_root): templates_root = repo_root / "templates" if not templates_root.exists(): return [] return [ {"name": path.name, "hasConfig": (path / "config.json").exists()} for path in sorted(templates_root.iterdir()) if path.is_dir() ] def list_clients(repo_root): clients_root = repo_root / "clientes" clients_root.mkdir(exist_ok=True) return [ client_summary(repo_root, path.name) for path in sorted(clients_root.iterdir()) if path.is_dir() and not path.name.startswith("_") ] def list_archived_clients(repo_root): archive_root = repo_root / "clientes" / "_archivados" if not archive_root.exists(): return [] archives = [] for path in sorted(archive_root.iterdir()): if not path.is_dir() or not re.fullmatch(r"[a-z0-9][a-z0-9-]*-\d{8}-\d{6}", path.name): continue slug = slug_from_archive_name(path.name) config_path = path / "config.json" config = {} config_valid = False if config_path.exists(): try: config = read_json(config_path) config_valid = True except json.JSONDecodeError: config_valid = False archives.append({ "archiveName": path.name, "slug": slug, "name": config.get("nombres", slug), "tipo": config.get("tipo", ""), "path": str(path.relative_to(repo_root)), "configValid": config_valid, "archivedAt": path.name[-15:], }) return archives def client_summary(repo_root, slug): path = client_dir(repo_root, slug) config_path = path / "config.json" config = {} config_valid = False if config_path.exists(): try: config = read_json(config_path) config_valid = True except json.JSONDecodeError: config_valid = False token = ensure_client_token(repo_root, slug) return { "slug": slug, "name": config.get("nombres", slug), "tipo": config.get("tipo", ""), "url": client_public_url(slug), "path": str(path.relative_to(repo_root)), "configValid": config_valid, "hasAssets": (path / "assets").exists(), "isPublished": True, "previewUrl": client_preview_url(slug), "clientPanelUrl": f"/cliente/?slug={slug}&token={token}", } def main(): parser = argparse.ArgumentParser(description="Servidor admin local para administrar invitaciones.") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8099) args = parser.parse_args() repo_root = Path(__file__).resolve().parents[1] os.chdir(repo_root) initialize_storage(repo_root) AdminHandler.repo_root = repo_root server = ThreadingHTTPServer((args.host, args.port), AdminHandler) print(f"Admin: http://{args.host}:{args.port}/admin/") print("Cortar con Ctrl+C") server.serve_forever() if __name__ == "__main__": main()