import os import queue import uuid import json import random import requests from flask import request, jsonify, Response, current_app as app from . import bp_api from ..utils.response import ok, fail def _sanitize_ascii(text: str) -> str: return "".join(c for c in (text or "") if ord(c) < 128) def _get_api_key() -> str: raw_key = os.getenv("AIML_API_KEY", "").strip() clean_key = _sanitize_ascii(raw_key) if not clean_key: raise RuntimeError("AIML_API_KEY não encontrado ou contém apenas caracteres inválidos.") return clean_key def _get_base_url() -> str: default = "https://api.aimlapi.com/v1" raw_url = os.getenv("AIML_BASE_URL", default).strip().rstrip("/") clean_url = _sanitize_ascii(raw_url) if not clean_url: raise RuntimeError("AIML_BASE_URL não encontrado ou contém apenas caracteres inválidos.") return clean_url AIML_BASE = _get_base_url() DEFAULT_MODEL = "meta-llama/Meta-Llama-3.1-8B-Instruct-Turbo" FREE_FALLBACK = "gpt-4o" IMAGE_MODEL = "imagen-3.0-generate-002" HEADERS = { "Authorization": f"Bearer {_get_api_key()}", "Content-Type": "application/json", } def _call_aiml(path: str, payload: dict, timeout: int = 60): url = f"{AIML_BASE}{path}" try: r = requests.post(url, headers=HEADERS, json=payload, timeout=timeout) r.raise_for_status() except requests.HTTPError as e: if r.status_code == 403 and "verification" in r.text.lower(): if payload.get("model") != FREE_FALLBACK: payload["model"] = FREE_FALLBACK return _call_aiml(path, payload, timeout) raise return r.json() class _Session: __slots__ = ("prompt", "model", "events", "last_url", "fields") def __init__(self, prompt: str, model: str, fields: list[str]): self.prompt = prompt self.model = model self.events = queue.Queue() self.last_url = None self.fields = fields def push_event(self, evt: dict): self.events.put(evt) _sessions: dict[str, _Session] = {} def _new_session(prompt: str, model: str, fields: list[str]) -> str: sid = uuid.uuid4().hex _sessions[sid] = _Session(prompt, model, fields) return sid def _parse_json_text(text: str): txt = (text or "").strip() if txt.startswith("```"): txt = txt[7:] if txt.startswith("```json") else txt[3:] txt = txt[:-3] if txt.endswith("```") else txt txt = txt.strip() return json.loads(txt) def _slugify(name: str) -> str: s = _sanitize_ascii(name.lower()) return "-".join(s.split()) def _truncate(text: str, limit: int = 1000) -> str: """Corta o texto em até `limit` caracteres sem quebrar palavras no meio.""" txt = (text or "").strip() if len(txt) <= limit: return txt cut = txt[:limit].rsplit(" ", 1)[0] return cut.strip() + "…" @bp_api.get("/post/categories") def list_post_categories(): """Retorna até 20 categorias aleatórias de turismo para posts de blog. Query params: n (int) – quantidade de categorias (1-20). Padrão 10. Response: [ {"id": 1, "name": "Ecoturismo", "slug": "ecoturismo", "icone": "icon-ecoturismo"}, ... ] """ try: n = int(request.args.get("n", 10)) except ValueError: return fail("'n' deve ser número inteiro", status=400) n = max(1, min(n, 20)) prompt = f""" Crie aleatoriamente {n} categorias curtas, criativas e distintas EXCLUSIVAMENTE ligadas ao setor de turismo (e.g., tipos de viagem, experiências turísticas, segmentos de público ou destinos). FORMATO DE SAÍDA (obrigatório): Retorne APENAS um JSON em UMA LINHA contendo um array de objetos. Cada objeto deve ter "name" e "slug" (slug kebab-case, sem acentos). Exemplo: [{{"name":"Ecoturismo","slug":"ecoturismo"}},{{"name":"Turismo Gastronomico","slug":"turismo-gastronomico"}}] Não inclua nada além do JSON. """.strip() payload = { "model": DEFAULT_MODEL, "messages": [ {"role": "system", "content": "Você gera dados estruturados. Sempre retorne JSON válido em uma linha."}, {"role": "user", "content": prompt}, ], "max_tokens": 300, "temperature": 1.0, "stream": False, } try: data = _call_aiml("/chat/completions", payload) content = data["choices"][0]["message"]["content"] cats = _parse_json_text(content) out = [] for idx, c in enumerate(cats[:n], 1): name = str(c.get("name", "")).strip() or f"Categoria {idx}" slug = str(c.get("slug", "")).strip() or _slugify(name) icone = f"icon-{slug}" out.append({"id": idx, "name": name, "slug": slug, "icone": icone}) if out: return ok(out) except Exception: pass fallback_pool = [ "Ecoturismo", "Turismo de Aventura", "Turismo Cultural", "Turismo Gastronômico", "Praias Paradisíacas", "Turismo Histórico", "Turismo Religioso", "Turismo Rural", "Turismo de Luxo", "Turismo Sustentável", "Viagens em Família", "Destinos Românticos", "Observação de Fauna", "Trilhas e Caminhadas", "Cruzeiros", "Road Trips", "Turismo Urbano", "Parques Nacionais", "Turismo de Compras", "Bem-Estar & Spa", ] chosen = random.sample(fallback_pool, k=min(n, len(fallback_pool))) out = [ { "id": i + 1, "name": name, "slug": _slugify(name), "icone": f"icon-{_slugify(name)}", } for i, name in enumerate(chosen) ] return ok(out) @bp_api.post("/post/examples") def generate_post_examples(): """Gera até 5 exemplos (title, category, local, description) para posts de turismo.""" body = request.get_json(silent=True) or {} title = (body.get("title") or "").strip() category = (body.get("category") or "").strip() local_ = (body.get("local") or "").strip() n = int(body.get("n", 5)) n = max(1, min(n, 5)) if not (title or category or local_): return fail("informe ao menos 'title', 'category' ou 'local'", status=422) prompt = f""" Você é um redator especialista em turismo. Crie {n} ideias diferentes de post de blog de turismo. Para cada ideia forneça: • "title" (máx. 120 caracteres), • "category" (pode repetir ou variar), • "local" (destino ou região), • "description" (até 1000 caracteres, parágrafo único, português, sem emojis/hashtags/HTML). Dados de referência: - Título base: {title or "(não fornecido)"} - Categoria base: {category or "(não fornecido)"} - Local base: {local_ or "(não fornecido)"} FORMATO DE SAÍDA (obrigatório): Retorne APENAS um JSON em UMA linha contendo um array de objetos: [ {{"title":"...","category":"...","local":"...","description":"..."}}, ... ] """.strip() payload = { "model": DEFAULT_MODEL, "messages": [ {"role": "system", "content": "Você gera dados estruturados. Sempre retorne JSON válido em uma linha."}, {"role": "user", "content": prompt}, ], "max_tokens": 1200, "temperature": 0.9, "top_p": 0.9, "presence_penalty": 0.2, "frequency_penalty": 0.2, "stream": False, } try: data = _call_aiml("/chat/completions", payload) txt = data["choices"][0]["message"]["content"] if txt.startswith("```"): txt = txt[7:] if txt.startswith("```json") else txt[3:] if txt.endswith("```"): txt = txt[:-3] ideas = json.loads(txt.strip()) out = [] for idx, idea in enumerate(ideas[:n], 1): t = str(idea.get("title", "")).strip() or f"Idea {idx}" c = str(idea.get("category", "")).strip() or (category or "Turismo") l = str(idea.get("local", "")).strip() or (local_ or "Destino") d = _truncate(str(idea.get("description", "")).strip(), 1000) out.append({"title": t, "category": c, "local": l, "description": d}) if out: return ok(out) except Exception: pass base_cat = category or "Turismo" base_loc = local_ or "um destino imperdível" fallback_titles = [ f"Descobrindo {base_loc} a Dois", f"Aventura Romântica em {base_loc}", f"{base_loc}: Guia Completo para Casais", f"Roteiro de {base_cat} em {base_loc}", f"Top Experiências de {base_cat} em {base_loc}", ] out = [] for i in range(n): t = fallback_titles[i % len(fallback_titles)] d_parts = [ t + ".", f"Explore o melhor que {base_loc} oferece.", "Vivencie momentos inesquecíveis lado a lado!", ] out.append({ "title": t, "category": base_cat, "local": base_loc, "description": _truncate(" ".join(d_parts), 1000), }) return ok(out) @bp_api.post("/service/description") def service_description(): body = request.get_json(silent=True) or {} nome = (body.get("nome") or "").strip() local_ = (body.get("local") or "").strip() tipo = (body.get("tipo") or "").strip() missing = [k for k, v in {"nome": nome, "local": local_, "tipo": tipo}.items() if not v] if missing: return fail("parâmetros obrigatórios ausentes no corpo da requisição", status=422, details={"missing_fields": missing}) tom = (body.get("tom") or "").strip() publico = (body.get("publico") or "").strip() keywords = body.get("keywords") if isinstance(keywords, list): keywords = ", ".join([str(k) for k in keywords if k]) else: keywords = (keywords or "").strip() extras = [] if tom: extras.append(f'Tom desejado: "{tom}"') if publico: extras.append(f'Público-alvo: "{publico}"') if keywords: extras.append(f"Palavras-chave: {keywords}") extras_txt = "\n".join(extras) prompt = f""" Você é um redator turístico sênior. Crie DESCRIÇÕES CURTAS e CRIATIVAS para o serviço abaixo, em três idiomas. FORMATO DE SAÍDA (obrigatório): Retorne APENAS um JSON plano, em UMA LINHA, com as chaves "pt", "en" e "es". Cada valor DEVE ser UMA ÚNICA STRING (sem objetos, sem arrays, sem quebras de linha, sem chaves "{{}}"). Exemplo: {{"pt":"Frase completa PT.","en":"Full sentence EN.","es":"Frase completa ES."}} CONTEÚDO: - Máx. 500 caracteres por idioma. - Combine: 1) gancho sensorial curto; 2) diferencial específico do {nome} em {local_}; 3) chamada para ação direta. - Verbo ativo, substantivos concretos. Evite clichês. - Sem emojis/hashtags. DADOS: Nome: {nome} Local: {local_} Tipo: {tipo} {extras_txt} """.strip() payload = { "model": DEFAULT_MODEL, "messages": [ { "role": "system", "content": "Você é um redator turístico sênior. Sempre retorne JSON válido, plano, em uma linha, com valores de string. Nunca retorne objetos/arrays por idioma.", }, {"role": "user", "content": prompt}, ], "max_tokens": 450, "temperature": float(body.get("temperature", 0.95)), "top_p": float(body.get("top_p", 0.9)), "presence_penalty": float(body.get("presence_penalty", 0.2)), "frequency_penalty": float(body.get("frequency_penalty", 0.2)), "stream": False, } def _default(): return { "pt": f"{nome} em {local_}. {tipo.capitalize()} com proposta única no destino. Garanta sua vaga agora.", "en": f"{nome} in {local_}. A {tipo} with a clear local edge. Book now.", "es": f"{nome} en {local_}. {tipo.capitalize()} con un diferencial local. Reserva hoy.", } def _parse_json_text(content: str): txt = (content or "").strip() if txt.startswith("```"): if txt.startswith("```json"): txt = txt[7:] else: txt = txt[3:] if txt.endswith("```"): txt = txt[:-3] txt = txt.strip() return json.loads(txt) def _flatten_value(v) -> str: if isinstance(v, dict): keys_pref = ("gancho", "hook", "diferencial", "unique", "cta", "chamada", "call_to_action") parts = [str(v[k]) for k in keys_pref if k in v and v[k]] if not parts: parts = [str(x) for x in v.values() if x] s = ". ".join(parts) elif isinstance(v, list): s = ". ".join(str(x) for x in v if x) else: s = str(v or "") s = s.replace("{", "").replace("}", "") s = " ".join(s.split())[:500] return s try: data = _call_aiml("/chat/completions", payload) content = data["choices"][0]["message"]["content"].strip() try: parsed = _parse_json_text(content) if isinstance(parsed, dict): pt = _flatten_value(parsed.get("pt", "")) en = _flatten_value(parsed.get("en", "")) es = _flatten_value(parsed.get("es", "")) if any([pt, en, es]): return ok({"pt": pt, "en": en, "es": es}) return ok(_default()) except (json.JSONDecodeError, TypeError, ValueError): return ok(_default()) except requests.RequestException: return ok(_default()) @bp_api.post("/chat") def chat_completion(): body = request.get_json(silent=True) or {} prompt = body.get("prompt") if not prompt: return {"error": "campo 'prompt' é obrigatório"}, 400 payload = { "model": body.get("model", DEFAULT_MODEL), "messages": [{"role": "user", "content": prompt}], "max_tokens": body.get("max_tokens", 512), "stream": False, } try: return jsonify(_call_aiml("/chat/completions", payload)) except requests.HTTPError as err: return {"error": "falha ao contatar modelo de chat", "detail": err.response.text}, err.response.status_code except requests.RequestException as err: return {"error": "falha de rede", "detail": str(err)}, 502 @bp_api.post("/image") def image_generation(): body = request.get_json(silent=True) or {} prompt = body.get("prompt") if not prompt: return {"error": "campo 'prompt' é obrigatório"}, 400 fields = body.get("fields", []) if not isinstance(fields, list): return {"error": "campo 'fields' deve ser lista"}, 400 fields_str = ", ".join(fields) full_prompt = f"{prompt}. Campos do formulário: {fields_str}" if fields else prompt model = body.get("model", IMAGE_MODEL) payload = {"model": model, "prompt": full_prompt, "convert_base64_to_url": True} try: data = _call_aiml("/images/generations", payload, timeout=120) first_url = data["data"][0]["url"] sid = _new_session(full_prompt, model, fields) sess = _sessions[sid] sess.last_url = first_url sess.push_event({"type": "image", "data": data}) base_path = "/api/sessions" return { "session_id": sid, "initial_image": data, "events_url": f"{base_path}/{sid}/events", "command_url": f"{base_path}/{sid}/command", "html_url": f"{base_path}/{sid}/html", } except requests.HTTPError as err: return {"error": "falha ao gerar imagem", "detail": err.response.text}, err.response.status_code except requests.RequestException as err: return {"error": "falha de rede", "detail": str(err)}, 502 def _event_stream(sess: _Session): while True: yield "data: " + json.dumps(sess.events.get()) + "\n\n" @bp_api.get("/sessions//events") def listen_events(sid: str): sess = _sessions.get(sid) if not sess: return {"error": "sessão não encontrada"}, 404 return Response(_event_stream(sess), mimetype="text/event-stream") @bp_api.post("/sessions//command") def handle_command(sid: str): sess = _sessions.get(sid) if not sess: return {"error": "sessão não encontrada"}, 404 body = request.get_json(silent=True) or {} cmd = body.get("command") if not cmd: return {"error": "campo 'command' é obrigatório"}, 400 try: if cmd == "regenerate": payload = { "model": sess.model, "prompt": body.get("prompt", sess.prompt), "convert_base64_to_url": True, } data = _call_aiml("/images/generations", payload, timeout=120) sess.last_url = data["data"][0]["url"] sess.push_event({"type": "image", "data": data}) return {"status": "generated"}, 200 return {"error": "comando inválido"}, 400 except requests.HTTPError as err: return {"error": "falha ao gerar imagem", "detail": err.response.text}, err.response.status_code except requests.RequestException as err: return {"error": "falha de rede", "detail": str(err)}, 502 @bp_api.get("/sessions//html") def html_snippet(sid: str): sess = _sessions.get(sid) if not sess: return {"error": "sessão não encontrada"}, 404 count = len(sess.fields) inputs_html = "".join( f'
' f'
' for f in sess.fields ) html = f""" Formulário – {count} campos

Reserve sua viagem perfeita

{inputs_html}
""" return Response(html, mimetype="text/html")