#!/usr/bin/env python3 """ Clinic Content Studio — servidor local Endpoints: Packs (existente): POST /api/save-pack — persiste pack JSON em ./packs/ GET /api/list-packs — lista packs salvos GET /api/load-pack/{f} — carrega um pack específico Meta integration (Fase 3+4): GET /api/meta/config — retorna app_id (sem secret) + redirect_uri POST /api/meta/config — salva app_id + app_secret em meta_config.json GET /api/meta/auth/start — redireciona para OAuth Facebook GET /api/meta/auth/callback — recebe code, troca por token, salva POST /api/meta/disconnect — limpa tokens GET /api/meta/status — info da conexão (páginas, contas IG) POST /api/meta/post — publica post no Instagram (image/reel/story) GET /api/meta/insights/{id} — métricas de um media específico GET /api/meta/leads/{form_id} — leads de um form específico Arquivos sensíveis (não-commitados): meta_config.json — { app_id, app_secret, redirect_uri } meta_tokens.json — { user_token, expires_at, pages: [{...}] } Uso: python3 server.py Acesse: http://localhost:8080/clinic_content_studio%20(5).html """ import http.server import json import os import datetime import socketserver import sys import urllib.parse import urllib.request import urllib.error import re # Python 3.13+ removeu cgi. Parser multipart próprio. cgi = None # mantido para compatibilidade de referência def _parse_multipart(fp, headers): """Parser multipart/form-data sem dependência de cgi.""" import email.parser, email.policy ctype = headers.get("Content-Type", "") content_length = int(headers.get("Content-Length", 0)) body = fp.read(content_length) raw = ("Content-Type: " + ctype + "\r\n\r\n").encode() + body msg = email.parser.BytesParser(policy=email.policy.compat32).parsebytes(raw) fields = {} for part in msg.walk(): disp = part.get("Content-Disposition", "") if not disp: continue pdict = {} for seg in disp.split(";")[1:]: seg = seg.strip() if "=" in seg: k, v = seg.split("=", 1) pdict[k.strip()] = v.strip().strip('"') name = pdict.get("name", "") fname = pdict.get("filename", "") payload = part.get_payload(decode=True) if payload is not None: fields[name] = (fname, payload) return fields import mimetypes import uuid # Backend de autenticação (magic-link, sessions, whitelist beta) import auth_module # Leitura de insights (Meta + Vista Social) do Postgres VPS import insights_module import catalogo_module import sync_module import whatsapp_sender PORT = int(os.environ.get("STUDIO_PORT", "8080")) BIND = os.environ.get("STUDIO_BIND", "") # "" = todas interfaces; "127.0.0.1" = só loopback PACKS_DIR = "packs" UPLOADS_DIR = "uploads" MAX_UPLOAD_BYTES = 100 * 1024 * 1024 # 100 MB ALLOWED_UPLOAD_EXT = { ".jpg", ".jpeg", ".png", ".webp", ".gif", ".mp4", ".mov", ".m4v", ".webm", } META_CONFIG_FILE = "meta_config.json" META_TOKENS_FILE = "meta_tokens.json" META_GRAPH_VERSION = "v21.0" META_GRAPH_BASE = f"https://graph.facebook.com/{META_GRAPH_VERSION}" DEFAULT_REDIRECT_URI = f"http://localhost:{PORT}/api/meta/auth/callback" META_SCOPES = [ "pages_show_list", "pages_read_engagement", ] # ───────────────────────────────────────────────────────────────────────── # Helpers para Meta (separados da classe handler para testabilidade) # ───────────────────────────────────────────────────────────────────────── def meta_load_config(): if not os.path.exists(META_CONFIG_FILE): return None try: with open(META_CONFIG_FILE, encoding="utf-8") as f: return json.load(f) except Exception: return None def meta_save_config(app_id, app_secret, redirect_uri=None): cfg = { "app_id": app_id.strip(), "app_secret": app_secret.strip(), "redirect_uri": (redirect_uri or DEFAULT_REDIRECT_URI).strip(), } with open(META_CONFIG_FILE, "w", encoding="utf-8") as f: json.dump(cfg, f, indent=2) return cfg def meta_load_tokens(): if not os.path.exists(META_TOKENS_FILE): return None try: with open(META_TOKENS_FILE, encoding="utf-8") as f: return json.load(f) except Exception: return None def meta_save_tokens(tokens): with open(META_TOKENS_FILE, "w", encoding="utf-8") as f: json.dump(tokens, f, indent=2) def meta_clear_tokens(): if os.path.exists(META_TOKENS_FILE): os.remove(META_TOKENS_FILE) def meta_graph_get(path, params=None): url = f"{META_GRAPH_BASE}{path}" if params: url += "?" + urllib.parse.urlencode(params) req = urllib.request.Request(url, headers={"Accept": "application/json"}) with urllib.request.urlopen(req, timeout=20) as r: return json.loads(r.read().decode("utf-8")) def meta_graph_post(path, params=None, data=None): url = f"{META_GRAPH_BASE}{path}" if params: url += "?" + urllib.parse.urlencode(params) body = urllib.parse.urlencode(data or {}).encode() req = urllib.request.Request( url, data=body, headers={ "Accept": "application/json", "Content-Type": "application/x-www-form-urlencoded", }, ) with urllib.request.urlopen(req, timeout=30) as r: return json.loads(r.read().decode("utf-8")) def meta_exchange_short_token(app_id, app_secret, redirect_uri, code): """Troca o code do OAuth por um short-lived user access token.""" return meta_graph_get( "/oauth/access_token", { "client_id": app_id, "client_secret": app_secret, "redirect_uri": redirect_uri, "code": code, }, ) def meta_exchange_long_lived(app_id, app_secret, short_token): """Troca short-lived por long-lived (60 dias).""" return meta_graph_get( "/oauth/access_token", { "grant_type": "fb_exchange_token", "client_id": app_id, "client_secret": app_secret, "fb_exchange_token": short_token, }, ) def meta_fetch_pages_with_ig(user_token): """Lista páginas do usuário + tokens de página + contas IG vinculadas.""" pages_resp = meta_graph_get( "/me/accounts", { "fields": "id,name,access_token,instagram_business_account{id,username,profile_picture_url}", "access_token": user_token, }, ) out = [] for p in pages_resp.get("data", []): ig = p.get("instagram_business_account") or {} out.append( { "id": p.get("id"), "name": p.get("name"), "access_token": p.get("access_token"), "ig_user_id": ig.get("id"), "ig_username": ig.get("username"), "ig_profile_picture_url": ig.get("profile_picture_url"), } ) return out # ───────────────────────────────────────────────────────────────────────── # HTTP handler # ───────────────────────────────────────────────────────────────────────── class StudioHandler(http.server.SimpleHTTPRequestHandler): def log_message(self, fmt, *args): if self.path.startswith("/api/"): print(f" [API] {self.path} — {args[1] if len(args) > 1 else ''}") def do_OPTIONS(self): self.send_response(200) self._cors_headers() self.end_headers() def do_GET(self): # Packs (existing) if self.path == "/api/list-packs": self._list_packs() elif self.path.startswith("/api/load-pack/"): filename = self.path.split("/api/load-pack/", 1)[1] self._load_pack(filename) # Uploaded media — served from /uploads/{filename} with correct Content-Type elif self.path.startswith("/uploads/"): self._serve_upload(self.path[len("/uploads/"):]) elif self.path == "/api/list-uploads": self._list_uploads() # Meta elif self.path == "/api/meta/config": self._meta_get_config() elif self.path.startswith("/api/meta/auth/start"): self._meta_auth_start() elif self.path.startswith("/api/meta/auth/callback"): self._meta_auth_callback() elif self.path == "/api/meta/status": self._meta_status() elif self.path.startswith("/api/meta/insights/"): media_id = self.path.split("/api/meta/insights/", 1)[1].split("?")[0] self._meta_insights(media_id) elif self.path.startswith("/api/meta/leads/"): form_id = self.path.split("/api/meta/leads/", 1)[1].split("?")[0] self._meta_leads(form_id) # Auth elif self.path.startswith("/api/social/"): qs = ("?" + self.path.split("?", 1)[1]) if "?" in self.path else "" hub_path = self.path.replace("/api/social", "", 1).split("?")[0] self._social_hub_proxy("GET", hub_path + qs) elif self.path == "/api/auth/me": self._auth_me() # Insights (auth obrigatório) elif self.path.startswith("/api/insights/"): self._handle_insights() # Catálogo CE (auth obrigatório) elif self.path.startswith("/api/catalogo"): self._handle_catalogo() # Backup remoto Six Hype (Sprint M) elif self.path == "/api/sync/status": self._handle_sync_status() elif self.path == "/api/sync/pull": self._handle_sync_pull() else: super().do_GET() def do_POST(self): if self.path == "/api/save-pack": self._save_pack() elif self.path == "/api/meta/config": self._meta_set_config() elif self.path == "/api/meta/disconnect": self._meta_disconnect() elif self.path == "/api/meta/post": self._meta_post() elif self.path == "/api/upload": self._upload_media() elif self.path.startswith("/api/uploads/delete/"): filename = self.path.split("/api/uploads/delete/", 1)[1] self._delete_upload(filename) # Backup remoto Six Hype (Sprint M) elif self.path == "/api/sync/push": self._handle_sync_push() # Notif reviewer WhatsApp (Sprint J slice 2) elif self.path == "/api/notify/reviewer": self._handle_notify_reviewer() # Auth elif self.path == "/api/auth/request-link": self._auth_request_link() elif self.path == "/api/auth/verify": self._auth_verify() elif self.path == "/api/auth/onboarding": self._auth_onboarding() elif self.path == "/api/auth/logout": self._auth_logout() elif self.path.startswith("/api/social/"): body = self._read_raw() hub_path = self.path.replace("/api/social", "", 1) self._social_hub_proxy("POST", hub_path, body) else: self.send_error(404, "Endpoint não encontrado") # ── Packs ───────────────────────────────────────────────────────────── def _save_pack(self): try: data = self._read_json() os.makedirs(PACKS_DIR, exist_ok=True) ts = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S") week = data.get("week", "") filename = f"pack_{week}_{ts}.json" filepath = os.path.join(PACKS_DIR, filename) with open(filepath, "w", encoding="utf-8") as f: json.dump(data, f, ensure_ascii=False, indent=2) size_kb = os.path.getsize(filepath) // 1024 self._json_response(200, {"filename": filename, "size_kb": size_kb}) except Exception as e: self._json_response(500, {"error": str(e)}) def _list_packs(self): try: os.makedirs(PACKS_DIR, exist_ok=True) files = sorted( [f for f in os.listdir(PACKS_DIR) if f.endswith(".json")], reverse=True, ) packs = [] for f in files: fp = os.path.join(PACKS_DIR, f) size_kb = os.path.getsize(fp) // 1024 try: with open(fp, encoding="utf-8") as fh: meta = json.load(fh) pack_items = meta.get("pack", []) workflow_statuses = [ pi.get("workflow", {}).get("status", "rascunho") for pi in pack_items if isinstance(pi, dict) ] workflow_summary = workflow_statuses[0] if workflow_statuses else "" packs.append( { "filename": f, "size_kb": size_kb, "week": meta.get("week", ""), "year": meta.get("year", ""), "date": meta.get("date", ""), "themes": meta.get("themes", []), "briefingFocus": meta.get("briefingFocus", ""), "profileId": meta.get("profileId", ""), "profileName": meta.get("profileName", ""), "workflowStatus": workflow_summary, } ) except Exception: packs.append({"filename": f, "size_kb": size_kb}) self._json_response(200, {"packs": packs}) except Exception as e: self._json_response(500, {"error": str(e)}) def _load_pack(self, filename): try: filename = os.path.basename(filename) filepath = os.path.join(PACKS_DIR, filename) if not os.path.exists(filepath): self._json_response(404, {"error": "Pack não encontrado"}) return with open(filepath, encoding="utf-8") as f: data = json.load(f) self._json_response(200, data) except Exception as e: self._json_response(500, {"error": str(e)}) # ── Meta config ─────────────────────────────────────────────────────── def _meta_get_config(self): cfg = meta_load_config() if not cfg: self._json_response( 200, { "configured": False, "redirect_uri": DEFAULT_REDIRECT_URI, "scopes": META_SCOPES, }, ) return # Não retorna o secret — só sinaliza que existe self._json_response( 200, { "configured": True, "app_id": cfg.get("app_id"), "redirect_uri": cfg.get("redirect_uri", DEFAULT_REDIRECT_URI), "has_secret": bool(cfg.get("app_secret")), "scopes": META_SCOPES, }, ) def _meta_set_config(self): try: data = self._read_json() app_id = (data.get("app_id") or "").strip() app_secret = (data.get("app_secret") or "").strip() if not app_id or not app_secret: self._json_response(400, {"error": "app_id e app_secret são obrigatórios"}) return cfg = meta_save_config(app_id, app_secret, data.get("redirect_uri")) self._json_response( 200, { "configured": True, "app_id": cfg["app_id"], "redirect_uri": cfg["redirect_uri"], }, ) except Exception as e: self._json_response(500, {"error": str(e)}) # ── Meta OAuth ──────────────────────────────────────────────────────── def _meta_auth_start(self): cfg = meta_load_config() if not cfg: self._json_response(400, {"error": "Configure app_id/app_secret primeiro"}) return params = { "client_id": cfg["app_id"], "redirect_uri": cfg["redirect_uri"], "scope": ",".join(META_SCOPES), "response_type": "code", "state": "studio_meta_oauth", } url = "https://www.facebook.com/" + META_GRAPH_VERSION + "/dialog/oauth?" + urllib.parse.urlencode(params) # Redireciona o navegador self.send_response(302) self.send_header("Location", url) self._cors_headers() self.end_headers() def _meta_auth_callback(self): try: qs = urllib.parse.urlparse(self.path).query params = urllib.parse.parse_qs(qs) error = params.get("error", [None])[0] if error: self._send_html_close( f"

Falha no OAuth

{error}: {params.get('error_description', [''])[0]}

" ) return code = params.get("code", [None])[0] if not code: self._send_html_close("

OAuth sem code

") return cfg = meta_load_config() if not cfg: self._send_html_close("

Config Meta ausente

") return # 1) short-lived short = meta_exchange_short_token( cfg["app_id"], cfg["app_secret"], cfg["redirect_uri"], code ) short_token = short.get("access_token") if not short_token: self._send_html_close(f"

Sem short token

{json.dumps(short)}
") return # 2) long-lived long_resp = meta_exchange_long_lived(cfg["app_id"], cfg["app_secret"], short_token) long_token = long_resp.get("access_token") or short_token expires_in = long_resp.get("expires_in", 60 * 24 * 3600) expires_at = ( datetime.datetime.utcnow() + datetime.timedelta(seconds=int(expires_in)) ).isoformat() + "Z" # 3) páginas + IG pages = meta_fetch_pages_with_ig(long_token) tokens = { "user_token": long_token, "expires_at": expires_at, "pages": pages, "saved_at": datetime.datetime.utcnow().isoformat() + "Z", } meta_save_tokens(tokens) # 4) Fecha a janela e avisa o app self._send_html_close( "

✓ Conectado ao Meta

Você já pode fechar esta janela.

" "" ) except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") self._send_html_close(f"

Erro Meta {e.code}

{body}
") except Exception as e: self._send_html_close(f"

Erro

{e}
") def _meta_disconnect(self): meta_clear_tokens() self._json_response(200, {"disconnected": True}) def _meta_status(self): tokens = meta_load_tokens() if not tokens: self._json_response(200, {"connected": False}) return # Sanitiza tokens antes de enviar (não expõe access_token) safe_pages = [] for p in tokens.get("pages", []): safe_pages.append( { "id": p.get("id"), "name": p.get("name"), "ig_user_id": p.get("ig_user_id"), "ig_username": p.get("ig_username"), "ig_profile_picture_url": p.get("ig_profile_picture_url"), "has_token": bool(p.get("access_token")), } ) self._json_response( 200, { "connected": True, "expires_at": tokens.get("expires_at"), "saved_at": tokens.get("saved_at"), "pages": safe_pages, }, ) # ── Meta publish ────────────────────────────────────────────────────── def _meta_post(self): try: data = self._read_json() ig_user_id = (data.get("ig_user_id") or "").strip() page_id = (data.get("page_id") or "").strip() caption = data.get("caption") or "" media_url = (data.get("media_url") or "").strip() media_type = (data.get("media_type") or "IMAGE").upper() # IMAGE | REELS | STORIES tokens = meta_load_tokens() if not tokens: self._json_response(400, {"error": "Não conectado ao Meta"}) return # Encontra a página correspondente para pegar o page access_token pages = tokens.get("pages", []) page = next((p for p in pages if p.get("id") == page_id or p.get("ig_user_id") == ig_user_id), None) if not page: self._json_response(400, {"error": "Página/IG não encontrada nas conexões"}) return page_token = page.get("access_token") ig_user_id = page.get("ig_user_id") if not page_token or not ig_user_id: self._json_response(400, {"error": "Página sem token ou sem Instagram Business vinculado"}) return if not media_url: self._json_response(400, {"error": "media_url é obrigatório (URL pública da imagem/vídeo)"}) return # 1) Cria container container_params = {"caption": caption, "access_token": page_token} if media_type == "REELS": container_params["media_type"] = "REELS" container_params["video_url"] = media_url elif media_type == "STORIES": container_params["media_type"] = "STORIES" if media_url.lower().endswith((".mp4", ".mov")): container_params["video_url"] = media_url else: container_params["image_url"] = media_url else: # IMAGE container_params["image_url"] = media_url container = meta_graph_post(f"/{ig_user_id}/media", data=container_params) container_id = container.get("id") if not container_id: self._json_response(500, {"error": "Container não criado", "detail": container}) return # 2) Publica publish = meta_graph_post( f"/{ig_user_id}/media_publish", data={"creation_id": container_id, "access_token": page_token}, ) self._json_response( 200, { "container_id": container_id, "media_id": publish.get("id"), "published_at": datetime.datetime.utcnow().isoformat() + "Z", "ig_username": page.get("ig_username"), }, ) except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") try: err = json.loads(body) except Exception: err = {"raw": body} self._json_response(e.code, {"error": "Meta API error", "detail": err}) except Exception as e: self._json_response(500, {"error": str(e)}) # ── Meta insights ───────────────────────────────────────────────────── def _meta_insights(self, media_id): try: tokens = meta_load_tokens() if not tokens: self._json_response(400, {"error": "Não conectado ao Meta"}) return # Tenta cada page token até funcionar (o post pertence a uma das páginas conectadas) metrics = "impressions,reach,saved,likes,comments,shares,total_interactions" last_err = None for p in tokens.get("pages", []): t = p.get("access_token") if not t: continue try: resp = meta_graph_get( f"/{media_id}/insights", {"metric": metrics, "access_token": t}, ) self._json_response(200, resp) return except urllib.error.HTTPError as e: last_err = e.read().decode("utf-8", errors="replace") continue self._json_response(404, {"error": "Insights não disponíveis", "detail": last_err}) except Exception as e: self._json_response(500, {"error": str(e)}) # ── Meta leads ──────────────────────────────────────────────────────── def _meta_leads(self, form_id): try: tokens = meta_load_tokens() if not tokens: self._json_response(400, {"error": "Não conectado ao Meta"}) return last_err = None for p in tokens.get("pages", []): t = p.get("access_token") if not t: continue try: resp = meta_graph_get( f"/{form_id}/leads", {"access_token": t, "limit": 50}, ) self._json_response(200, resp) return except urllib.error.HTTPError as e: last_err = e.read().decode("utf-8", errors="replace") continue self._json_response(404, {"error": "Leads não disponíveis", "detail": last_err}) except Exception as e: self._json_response(500, {"error": str(e)}) # ── Upload de mídia ─────────────────────────────────────────────────── def _upload_media(self): """Recebe multipart/form-data com campo 'file'. Salva em ./uploads/ com nome UUID + extensão original. Retorna {filename, url, size_kb}. """ try: content_length = int(self.headers.get("Content-Length", 0)) if content_length > MAX_UPLOAD_BYTES: self._json_response(413, {"error": f"Arquivo > {MAX_UPLOAD_BYTES // (1024*1024)} MB"}) return ctype = self.headers.get("Content-Type", "") if not ctype.startswith("multipart/form-data"): self._json_response(400, {"error": "Esperado multipart/form-data"}) return # Parser próprio (sem cgi) fields = _parse_multipart(self.rfile, self.headers) if "file" not in fields: self._json_response(400, {"error": "Campo 'file' obrigatorio"}) return original_name, file_bytes = fields["file"] if not original_name: self._json_response(400, {"error": "Arquivo sem nome"}) return original_name = os.path.basename(original_name) ext = os.path.splitext(original_name)[1].lower() if ext not in ALLOWED_UPLOAD_EXT: self._json_response(400, {"error": f"Extensão não permitida: {ext}. Aceitas: {sorted(ALLOWED_UPLOAD_EXT)}"}) return os.makedirs(UPLOADS_DIR, exist_ok=True) # Nome único: timestamp + uuid curto + ext (preserva tipo) ts = datetime.datetime.now().strftime("%Y%m%d-%H%M%S") short_id = uuid.uuid4().hex[:8] base = re.sub(r"[^a-zA-Z0-9_-]", "_", os.path.splitext(original_name)[0])[:40] or "media" stored_name = f"{ts}_{short_id}_{base}{ext}" stored_path = os.path.join(UPLOADS_DIR, stored_name) with open(stored_path, "wb") as out: out.write(file_bytes) size_kb = os.path.getsize(stored_path) // 1024 self._json_response( 200, { "filename": stored_name, "original": original_name, "size_kb": size_kb, "url": f"/uploads/{stored_name}", "absolute_url_hint": f"http://localhost:{PORT}/uploads/{stored_name}", }, ) except Exception as e: self._json_response(500, {"error": str(e)}) def _serve_upload(self, filename): try: filename = os.path.basename(urllib.parse.unquote(filename)) filepath = os.path.join(UPLOADS_DIR, filename) if not os.path.exists(filepath) or not os.path.isfile(filepath): self.send_error(404, "Arquivo não encontrado") return ctype, _ = mimetypes.guess_type(filename) if not ctype: ctype = "application/octet-stream" size = os.path.getsize(filepath) self.send_response(200) self.send_header("Content-Type", ctype) self.send_header("Content-Length", str(size)) self.send_header("Cache-Control", "public, max-age=86400") self._cors_headers() self.end_headers() with open(filepath, "rb") as f: while True: chunk = f.read(64 * 1024) if not chunk: break self.wfile.write(chunk) except (BrokenPipeError, ConnectionResetError): pass except Exception as e: try: self.send_error(500, str(e)) except Exception: pass def _list_uploads(self): try: os.makedirs(UPLOADS_DIR, exist_ok=True) files = [] for f in sorted(os.listdir(UPLOADS_DIR), reverse=True): if f.startswith(".") or not os.path.isfile(os.path.join(UPLOADS_DIR, f)): continue ext = os.path.splitext(f)[1].lower() if ext not in ALLOWED_UPLOAD_EXT: continue fp = os.path.join(UPLOADS_DIR, f) files.append( { "filename": f, "size_kb": os.path.getsize(fp) // 1024, "url": f"/uploads/{f}", "kind": "video" if ext in {".mp4", ".mov", ".m4v", ".webm"} else "image", "modified_at": datetime.datetime.fromtimestamp(os.path.getmtime(fp)).isoformat(), } ) self._json_response(200, {"uploads": files}) except Exception as e: self._json_response(500, {"error": str(e)}) def _delete_upload(self, filename): try: filename = os.path.basename(urllib.parse.unquote(filename)) filepath = os.path.join(UPLOADS_DIR, filename) if not os.path.exists(filepath): self._json_response(404, {"error": "Arquivo não encontrado"}) return os.remove(filepath) self._json_response(200, {"deleted": filename}) except Exception as e: self._json_response(500, {"error": str(e)}) # ── Insights ────────────────────────────────────────────────────────── def _handle_catalogo(self): """Roteia /api/catalogo?tipo=X&busca=Y&apenas_ativos=Z&limit=N e /api/catalogo/sync-status (status do último sync CE). Tipos: procedimento, consulta, experiencia, protocolo, profissional, pacote, produto, todos. Requer sessão autenticada.""" if not self._current_user(): self._json_response(401, {"error": "Não autenticado."}) return if not catalogo_module.is_enabled(): self._json_response(503, { "error": "Catálogo desabilitado (SUPABASE_URL/SERVICE_ROLE não configurado).", "hint": "Defina UNI_DATA_SUPABASE_URL e UNI_DATA_SUPABASE_SERVICE_ROLE no .env", }) return path = self.path.split("?")[0] params = urllib.parse.parse_qs(self.path.split("?")[1] if "?" in self.path else "") # Sub-rota /sync-status if path == "/api/catalogo/sync-status": try: self._json_response(200, catalogo_module.sync_status()) except Exception as e: self._json_response(500, {"error": f"Erro lendo sync_status: {e}"}) return # Rota principal: lista catálogo tipo = params.get("tipo", ["todos"])[0] busca = params.get("busca", [""])[0] apenas_ativos = params.get("apenas_ativos", ["true"])[0].lower() in ("1", "true", "yes") try: limit = int(params.get("limit", ["50"])[0]) except ValueError: limit = 50 try: data = catalogo_module.listar( tipo=tipo, busca=busca, apenas_ativos=apenas_ativos, limit=limit, ) self._json_response(200, data) except catalogo_module.CatalogoError as e: self._json_response(400, {"error": str(e)}) except Exception as e: self._json_response(500, {"error": f"Erro lendo catálogo: {e}"}) def _handle_insights(self): """Roteia /api/insights/{persona_id}/{tipo}?limit=N. Tipos: top-posts, by-format, topics-demanded, heatmap. Requer sessão autenticada.""" if not self._current_user(): self._json_response(401, {"error": "Não autenticado."}) return if not insights_module.is_enabled(): self._json_response(503, {"error": "Insights desabilitado (INSIGHTS_DB_DSN não configurado)."}) return # Path: /api/insights/persona/{id}/{kind} path = self.path.split("?")[0] params = urllib.parse.parse_qs(self.path.split("?")[1] if "?" in self.path else "") parts = path.strip("/").split("/") # ["api", "insights", "persona", "{id}", "{kind}"] if len(parts) < 5 or parts[2] != "persona": self._json_response(404, {"error": "Use /api/insights/persona/{id}/{kind}"}) return persona_id = parts[3] kind = parts[4] limit = int(params.get("limit", ["10"])[0]) try: if kind == "top-posts": data = insights_module.top_posts(persona_id, limit=limit) elif kind == "by-format": data = insights_module.by_format(persona_id) elif kind == "topics-demanded": data = insights_module.topics_demanded(persona_id, limit=limit) elif kind == "heatmap": data = insights_module.heatmap(persona_id) else: self._json_response(404, {"error": f"Tipo '{kind}' não suportado."}) return self._json_response(200, {"data": data}) except Exception as e: self._json_response(500, {"error": f"Erro lendo insights: {e}"}) # ── Auth ────────────────────────────────────────────────────────────── AUTH_COOKIE_NAME = "studio_session" def _auth_request_link(self): try: data = self._read_json() email = (data.get("email") or "").strip().lower() if not email or "@" not in email: self._json_response(400, {"error": "Email inválido."}) return if not auth_module.is_email_whitelisted(email): self._json_response(403, {"error": "Acesso ao beta ainda não liberado para este email."}) return try: link = auth_module.create_magic_link(email) except auth_module.RateLimitExceeded as rl: # 429 Too Many Requests + Retry-After payload = json.dumps({ "error": f"Muitas tentativas. Tente novamente em {rl.retry_after_seconds}s.", "retry_after": rl.retry_after_seconds, }, ensure_ascii=False).encode("utf-8") self.send_response(429) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Retry-After", str(rl.retry_after_seconds)) self.send_header("Content-Length", len(payload)) self._cors_headers(allow_credentials=True) self.end_headers() self.wfile.write(payload) return base_url = self._public_base_url() auth_module.send_magic_link_email(email, link["token"], base_url) response = {"ok": True, "message": "Link enviado. Verifique seu email (válido por 15 min)."} if auth_module.DEV_MODE: response["dev_token"] = link["token"] response["dev_link"] = f"{base_url}/?token={link['token']}" self._json_response(200, response) except Exception as e: self._json_response(500, {"error": str(e)}) def _auth_verify(self): try: data = self._read_json() token = (data.get("token") or "").strip() if not token: self._json_response(400, {"error": "Token ausente."}) return email = auth_module.verify_magic_link(token) if not email: self._json_response(401, {"error": "Token inválido ou expirado."}) return user = auth_module.get_or_create_user(email) jwt_token = auth_module.create_session( user_id=user["id"], user_agent=self.headers.get("User-Agent", ""), ip=self.client_address[0], ) self.send_response(200) self._set_session_cookie(jwt_token) self._send_json_body({"ok": True, "user": user}) except Exception as e: self._json_response(500, {"error": str(e)}) def _auth_me(self): user = self._current_user() if not user: self._json_response(401, {"error": "Não autenticado."}) return self._json_response(200, {"user": user}) def _auth_onboarding(self): try: user = self._current_user() if not user: self._json_response(401, {"error": "Não autenticado."}) return data = self._read_json() name = (data.get("name") or "").strip() role = (data.get("role") or "").strip() if not name or not role: self._json_response(400, {"error": "Nome e papel são obrigatórios."}) return updated = auth_module.update_user(user["id"], name=name, role=role) self._json_response(200, {"ok": True, "user": updated}) except Exception as e: self._json_response(500, {"error": str(e)}) def _auth_logout(self): jwt_token = self._get_session_cookie() auth_module.revoke_session(jwt_token) self.send_response(204) self._clear_session_cookie() self._cors_headers(allow_credentials=True) self.end_headers() # ── Sync / Backup remoto (Sprint M) ─────────────────────────────────── def _handle_sync_status(self): """GET /api/sync/status — devolve {enabled, schema, config_present}. Não exige auth (cliente decide se mostra UI de sync).""" try: self._json_response(200, sync_module.status()) except Exception as e: self._json_response(500, {"error": str(e)}) def _handle_sync_push(self): """POST /api/sync/push — body { operators?, personagens?, campanhas?, generatedPacks? }. Faz upsert em six_hype.* com owner_user_id = sessão atual.""" user = self._current_user() if not user: self._json_response(401, {"error": "Não autenticado."}) return if not sync_module.is_enabled(): self._json_response(503, { "error": "Sync remoto desabilitado.", "hint": "Defina SIX_HYPE_SYNC_ENABLED=true + UNI_DATA_SUPABASE_URL/SERVICE_ROLE no .env", }) return try: length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}" payload = json.loads(raw) except Exception as e: self._json_response(400, {"error": f"Body inválido: {e}"}) return try: result = sync_module.push(user, payload) self._json_response(200, {"ok": True, **result}) except sync_module.SyncError as e: self._json_response(502, {"error": str(e)}) except Exception as e: self._json_response(500, {"error": f"Falha no push: {e}"}) def _handle_sync_pull(self): """GET /api/sync/pull — devolve estado remoto do owner pra restaurar.""" user = self._current_user() if not user: self._json_response(401, {"error": "Não autenticado."}) return if not sync_module.is_enabled(): self._json_response(503, {"error": "Sync remoto desabilitado."}) return try: data = sync_module.pull(user) self._json_response(200, data) except sync_module.SyncError as e: self._json_response(502, {"error": str(e)}) except Exception as e: self._json_response(500, {"error": f"Falha no pull: {e}"}) # ── Notif reviewer WhatsApp (Sprint J slice 2) ──────────────────────── def _handle_notify_reviewer(self): """POST /api/notify/reviewer body { phone, message, reviewerName? } Envia mensagem WhatsApp via Evolution (BETA_MODE força redirect pra Caue). Não exige sync remoto — só Evolution configurado. """ user = self._current_user() if not user: self._json_response(401, {"error": "Não autenticado."}) return if not whatsapp_sender.is_configured(): self._json_response(503, { "error": "Evolution não configurado.", "hint": "Setar EVOLUTION_BASE_URL + EVOLUTION_INSTANCE + EVOLUTION_API_KEY no .env", }) return try: length = int(self.headers.get("Content-Length", "0")) raw = self.rfile.read(length).decode("utf-8") if length > 0 else "{}" payload = json.loads(raw) except Exception as e: self._json_response(400, {"error": f"Body inválido: {e}"}) return phone = (payload.get("phone") or "").strip() message = (payload.get("message") or "").strip() reviewer_name = (payload.get("reviewerName") or "").strip() if not phone or not message: self._json_response(400, {"error": "phone e message obrigatórios"}) return ok, info = whatsapp_sender.send_whatsapp(phone, message, real_recipient=reviewer_name) if ok: self._json_response(200, {"ok": True, "info": info}) else: self._json_response(502, {"error": info}) # ── Cookie / session helpers ── def _get_session_cookie(self): cookie_header = self.headers.get("Cookie", "") for part in cookie_header.split(";"): k, _, v = part.strip().partition("=") if k == self.AUTH_COOKIE_NAME: return urllib.parse.unquote(v) return None def _current_user(self): return auth_module.verify_session(self._get_session_cookie()) def _cookie_attrs(self): """Atributos de cookie sensíveis ao ambiente. Em prod: SameSite=None + Secure + Domain=.uniatacado.com (cross-subdomain). Em dev: SameSite=Lax + sem Secure (localhost http://).""" if auth_module.DEV_MODE: return " SameSite=Lax;" # Domain configurável via env (default: .uniatacado.com) cookie_domain = os.environ.get("COOKIE_DOMAIN", ".uniatacado.com") return f" SameSite=None; Secure; Domain={cookie_domain};" def _set_session_cookie(self, jwt_token): max_age = auth_module.SESSION_TTL_DAYS * 24 * 60 * 60 cookie = ( f"{self.AUTH_COOKIE_NAME}={urllib.parse.quote(jwt_token)};" f" Path=/; HttpOnly; Max-Age={max_age};{self._cookie_attrs()}" ) self.send_header("Set-Cookie", cookie) def _clear_session_cookie(self): cookie = ( f"{self.AUTH_COOKIE_NAME}=; Path=/; HttpOnly;" f" Max-Age=0;{self._cookie_attrs()}" ) self.send_header("Set-Cookie", cookie) def _public_base_url(self): host = self.headers.get("Host", f"localhost:{PORT}") scheme = "https" if self.headers.get("X-Forwarded-Proto") == "https" else "http" return f"{scheme}://{host}" def _send_json_body(self, body): """Após self.send_response() + cookies, envia headers e corpo JSON.""" payload = json.dumps(body, ensure_ascii=False).encode("utf-8") self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", len(payload)) self._cors_headers(allow_credentials=True) self.end_headers() self.wfile.write(payload) # ── Helpers ─────────────────────────────────────────────────────────── def _read_json(self): length = int(self.headers.get("Content-Length", 0)) return json.loads(self.rfile.read(length)) def _cors_headers(self, allow_credentials=False): origin = self.headers.get("Origin") # Em DEV, ecoa o Origin (necessário pra cookies httpOnly cross-port); # em prod, restringe via CORS_ORIGINS env var. cors_origins = {o.strip() for o in os.environ.get("CORS_ORIGINS", "").split(",") if o.strip()} if origin and (auth_module.DEV_MODE or origin in cors_origins): self.send_header("Access-Control-Allow-Origin", origin) self.send_header("Vary", "Origin") elif allow_credentials: # Com credentials, NÃO pode usar "*". Se origin não bate, omite header # (browser bloqueia, que é o comportamento correto). pass else: self.send_header("Access-Control-Allow-Origin", "*") self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS") self.send_header("Access-Control-Allow-Headers", "Content-Type") if allow_credentials: self.send_header("Access-Control-Allow-Credentials", "true") def _json_response(self, code, body): payload = json.dumps(body, ensure_ascii=False).encode() self.send_response(code) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", len(payload)) self._cors_headers() self.end_headers() self.wfile.write(payload) def _send_html_close(self, html): body = ( "" "" + html ).encode("utf-8") self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.send_header("Content-Length", len(body)) self._cors_headers() self.end_headers() self.wfile.write(body) # ── Social Hub proxy ────────────────────────────────────────────────── def _social_hub_proxy(self, method, hub_path, body=None): """Proxy seguro: encaminha /api/social/* para social-hub (localhost:8092). SOCIAL_HUB_SECRET fica só no .env do servidor, nunca exposto no frontend. """ import os as _os import urllib.request as _ur import urllib.error as _ue hub_url = _os.environ.get("SOCIAL_HUB_URL", "http://127.0.0.1:8092") hub_secret = _os.environ.get("SOCIAL_HUB_SECRET", "") if not hub_secret: self.send_error(503, "SOCIAL_HUB_SECRET nao configurado") return url = hub_url.rstrip("/") + hub_path req = _ur.Request(url, method=method) req.add_header("Authorization", f"Bearer {hub_secret}") req.add_header("Content-Type", "application/json") if body: req.data = body if isinstance(body, bytes) else body.encode() try: with _ur.urlopen(req, timeout=30) as resp: data = resp.read() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(data) except _ue.HTTPError as e: err_body = e.read() self.send_response(e.code) self.send_header("Content-Type", "application/json") self.send_header("Access-Control-Allow-Origin", "*") self.end_headers() self.wfile.write(err_body) except Exception as ex: self.send_error(502, str(ex)) def _read_raw(self): length = int(self.headers.get("Content-Length", 0)) return self.rfile.read(length) if length else b"" class ReusableTCPServer(socketserver.TCPServer): allow_reuse_address = True def main(): os.chdir(os.path.dirname(os.path.abspath(__file__))) # Inicializa o DB de auth (cria tabelas se não existirem) auth_module.init_db() with ReusableTCPServer((BIND, PORT), StudioHandler) as httpd: url = f"http://localhost:{PORT}/clinic_content_studio%20(5).html" whitelist_count = len(auth_module.BETA_EMAILS) dev_label = "DEV" if auth_module.DEV_MODE else "PROD" print(f"\n Clinic Content Studio [{dev_label}]") print(f" ─────────────────────────────────────") print(f" Servidor: http://localhost:{PORT}") print(f" Acesse: {url}") print(f" Auth DB: {auth_module.AUTH_DB_PATH}") print(f" Beta whitelist: {whitelist_count} email(s) liberado(s)") if whitelist_count == 0: print(f" ⚠ Defina BETA_EMAILS=email1,email2 para liberar acesso.") print(f" ─────────────────────────────────────") print(f" Endpoints Auth:") print(f" POST /api/auth/request-link {'(retorna dev_token)' if auth_module.DEV_MODE else ''}") print(f" POST /api/auth/verify") print(f" GET /api/auth/me") print(f" POST /api/auth/onboarding") print(f" POST /api/auth/logout") print(f" Endpoints Meta:") print(f" Config: /api/meta/config") print(f" OAuth: /api/meta/auth/start") print(f" Status: /api/meta/status") print(f" Publicar: POST /api/meta/post") print(f" Endpoints Upload:") print(f" Upload: POST /api/upload (multipart/form-data, campo 'file')") print(f" Listar: GET /api/list-uploads") print(f" Servir: GET /uploads/{{filename}}") print(f" ─────────────────────────────────────") print(f" ⚠ Para Meta API: rode 'ngrok http {PORT}' em outro terminal e use a URL HTTPS pública") print(f" ─────────────────────────────────────") print(f" Ctrl+C para parar\n") try: httpd.serve_forever() except KeyboardInterrupt: print("\n Servidor encerrado.") sys.exit(0) if __name__ == "__main__": main()