Files
marcocosta10000 a74553dc1b fix(api): remove desligamento automático do servidor do endpoint /end
Remove a lógica de shutdown do servidor no endpoint de encerramento de chat
para evitar conflitos com o gerenciamento do ciclo de vida da aplicação.
Simplifica também o script de inicialização removendo a detecção de venv
e mantendo apenas o fluxo essencial de verificação e instalação de
dependências.
2026-10-11 19:28:16 -03:00

638 lines
24 KiB
Python
Executable File

import os
import json
import base64
import mimetypes
import requests
from datetime import date, datetime
from flask import Flask, render_template, request, jsonify, session, send_from_directory
from werkzeug.utils import secure_filename
from dotenv import load_dotenv
load_dotenv()
app = Flask(__name__)
app.secret_key = os.getenv("SECRET_KEY", "chave_secreta_flask_chat_2026")
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
CONTEXTO_PATH = os.path.join(BASE_DIR, "contexto.json")
PERFIL_IA_PATH = os.path.join(BASE_DIR, "perfil_ia.json")
PERFIL_USER_PATH = os.path.join(BASE_DIR, "perfil_user.json")
CONFIG_PATH = os.path.join(BASE_DIR, "config.json")
PROIBIDOS_PATH = os.path.join(BASE_DIR, "proibidos.json")
HISTORICO_DIR = os.path.join(BASE_DIR, "historico")
BACKUPS_DIR = os.path.join(BASE_DIR, "backups")
UPLOADS_DIR = os.path.join(BASE_DIR, "uploads")
os.makedirs(HISTORICO_DIR, exist_ok=True)
os.makedirs(BACKUPS_DIR, exist_ok=True)
os.makedirs(UPLOADS_DIR, exist_ok=True)
ALLOWED_EXTENSIONS = {
"png", "jpg", "jpeg", "gif", "webp", "svg", "bmp",
"pdf", "doc", "docx", "xls", "xlsx", "ppt", "pptx",
"txt", "csv", "json", "xml", "zip", "tar", "gz",
"mp3", "mp4", "mov", "avi",
}
IMAGE_EXTENSIONS = {"png", "jpg", "jpeg", "gif", "webp", "svg", "bmp"}
@app.route("/uploads/<path:filename>")
def uploaded_file(filename):
return send_from_directory(UPLOADS_DIR, filename)
# ── Dados padrão para factory reset ──
DEFAULT_CONFIG = {
"modelo": "gryphe/mythomax-l2-13b",
"api_key": "",
}
DEFAULT_CONTEXTO = {
"role": "system",
"content": (
"Você é um assistente de chat amigável e prestativo, especializado em ajudar "
"usuários em português do Brasil. Suas respostas devem ser:\n\n"
"- Sempre escritas em português do Brasil, de forma natural e coloquial.\n"
"- Claras, objetivas e bem estruturadas.\n"
"- Educadas, respeitosas e acolhedoras.\n"
"- Relevantes para a pergunta feita, sem desvios desnecessários.\n"
"- Em tom conversacional, evitando linguagem excessivamente formal ou robótica.\n\n"
"Quando não souber uma resposta ou a pergunta for ambígua, peça mais detalhes ao "
"usuário de forma educada. Nunca invente informações factuais e não faça suposições "
"sem justificativa. Seja útil, preciso e empático em todas as interações."
),
}
DEFAULT_PERFIL_IA = {
"nome": "Maria",
"descricao": "Assistente de IA especializada em comunicação e atendimento em português do Brasil.",
"personalidade": {
"tom": "amigavel e acolhedor",
"nivel_de_formalidade": "informal",
"empatia": "alta",
"proatividade": "moderada",
},
"especialidades": [
"Atendimento ao cliente",
"Comunicação escrita",
"Orientação de usuários",
"Resolução de dúvidas",
"Conversação natural em português",
],
"regras_de_comportamento": {
"sempre_responder_em": "portugues_brasil",
"evitar_jargoes": True,
"ser_conciso_quando_possivel": True,
"perguntar_quando_faltar_informacao": True,
"nao_inventar_dados": True,
"nao_fazer_suposicoes_sem_justificativa": True,
},
"formato_de_resposta": {
"paragrafos_curtos": True,
"usa_listas_quando_apropriado": True,
"destaca_pontos_importantes": True,
},
}
DEFAULT_PERFIL_USER = {
"nome": "Usuário",
"idioma_preferido": "portugues_brasil",
"horario_de_uso": "qualquer",
"preferencias": {
"respostas_curtas": True,
"exemplos_praticos": True,
"explicacoes_detalhadas_sob_demanda": True,
},
"contexto_adicional": (
"Perfil genérico de usuário. Pode ser editado livremente para refletir nome, "
"preferências, área de atuação ou qualquer informação relevante sobre o usuário."
),
}
DEFAULT_PROIBIDOS = {
"topicos_proibidos": [
"conteúdo adulto ou pornográfico",
"violência explícita ou incitação ao ódio",
"discriminação racial, religiosa, de gênero ou orientação sexual",
"instruções para atividades ilegais ou perigosas",
"disseminação de desinformação ou fake news",
],
"comportamentos_proibidos": [
"fingir ser um humano real",
"compartilhar informações pessoais de terceiros",
"responder em idiomas que não sejam português do Brasil",
"executar comandos ou códigos no sistema do usuário",
],
"palavras_proibidas": [],
}
DEFAULT_FILES = {
CONFIG_PATH: DEFAULT_CONFIG,
CONTEXTO_PATH: DEFAULT_CONTEXTO,
PERFIL_IA_PATH: DEFAULT_PERFIL_IA,
PERFIL_USER_PATH: DEFAULT_PERFIL_USER,
PROIBIDOS_PATH: DEFAULT_PROIBIDOS,
}
def factory_reset():
"""Restaura todos os arquivos de configuração ao padrão e limpa histórico."""
for path, data in DEFAULT_FILES.items():
with open(path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# Limpa histórico do dia
p = os.path.join(HISTORICO_DIR, hoje() + ".json")
if os.path.exists(p):
os.remove(p)
# Limpa sessão
session.pop("chat_history", None)
session.pop("chat_loaded", None)
def load_json(path, default):
try:
with open(path, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return default
def get_config():
cfg = load_json(CONFIG_PATH, {})
model = cfg.get("modelo") or os.getenv("OPENROUTER_MODEL", "gryphe/mythomax-l2-13b")
api_key = cfg.get("api_key") or os.getenv("OPENROUTER_API_KEY", "")
return model, api_key
def hoje():
return date.today().isoformat()
def fazer_backup_contexto():
ts = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
bak = os.path.join(BACKUPS_DIR, f"contexto_{ts}.json")
with open(CONTEXTO_PATH, "r", encoding="utf-8") as orig:
with open(bak, "w", encoding="utf-8") as dest:
dest.write(orig.read())
return bak
def listar_backups():
backups = []
for fname in os.listdir(BACKUPS_DIR):
if fname.startswith("contexto_") and fname.endswith(".json"):
path = os.path.join(BACKUPS_DIR, fname)
mtime = os.path.getmtime(path)
backups.append({"file": fname, "path": path, "mtime": datetime.fromtimestamp(mtime).isoformat()})
backups.sort(key=lambda x: x["mtime"], reverse=True)
return backups
def historico_path(data=None):
return os.path.join(HISTORICO_DIR, (data or hoje()) + ".json")
def carregar_historico_dia(data=None):
p = historico_path(data)
if os.path.exists(p):
try:
with open(p, "r", encoding="utf-8") as f:
return json.load(f)
except Exception:
return []
return []
def salvar_historico_dia(messages, data=None):
p = historico_path(data)
with open(p, "w", encoding="utf-8") as f:
json.dump(messages, f, ensure_ascii=False, indent=2)
def get_chat_history():
if "chat_loaded" not in session:
session["chat_history"] = carregar_historico_dia()
session["chat_loaded"] = True
return session.get("chat_history", [])
def save_chat_history(history):
session["chat_history"] = history
salvar_historico_dia(history)
def build_system_prompt():
ctx = load_json(CONTEXTO_PATH, {})
ia = load_json(PERFIL_IA_PATH, {})
user = load_json(PERFIL_USER_PATH, {})
prohib = load_json(PROIBIDOS_PATH, {})
cfg = load_json(CONFIG_PATH, {})
model, _ = cfg.get("modelo", "?"), cfg.get("api_key", "")
ctx_content = ctx.get("content", ctx.get("descricao", "Você é um assistente útil e amigável."))
ia_nome = ia.get("nome", "IA")
user_nome = user.get("nome", "Usuário")
regras = ia.get("regras_de_comportamento", {})
regras_txt = "\n".join(f"- {k.replace('_', ' ')}: {v}" for k, v in regras.items())
especialidades = ia.get("especialidades", [])
esp_txt = "\n".join(f"- {e}" for e in especialidades)
formato = ia.get("formato_de_resposta", {})
formato_txt = "\n".join(f"- {k.replace('_', ' ')}: {v}" for k, v in formato.items())
prohib_topicos = prohib.get("topicos_proibidos", [])
prohib_comp = prohib.get("comportamentos_proibidos", [])
prohib_palavras = prohib.get("palavras_proibidas", [])
prohib_txt = ""
if prohib_topicos:
prohib_txt += "\nTópicos proibidos:\n" + "\n".join(f"- {t}" for t in prohib_topicos)
if prohib_comp:
prohib_txt += "\nComportamentos proibidos:\n" + "\n".join(f"- {c}" for c in prohib_comp)
if prohib_palavras:
prohib_txt += "\nPalavras proibidas:\n" + "\n".join(f"- {p}" for p in prohib_palavras)
lines = [
ctx_content, "",
"--- SEU PERFIL ---",
f"Você se chama {ia_nome}.",
f"Personalidade: {ia.get('personalidade', {}).get('tom', 'amigável')}, formalidade: {ia.get('personalidade', {}).get('nivel_de_formalidade', 'informal')}.",
f"Empatia: {ia.get('personalidade', {}).get('empatia', 'alta')}.", "",
"Especialidades:", esp_txt, "",
"Regras de comportamento:", regras_txt, "",
"Formato de resposta:", formato_txt, "",
"--- PERFIL DO USUÁRIO ---",
f"O usuário se chama {user_nome}.",
f"Idioma: {user.get('idioma_preferido', 'português do Brasil')}.",
f"Contexto: {user.get('contexto_adicional', 'Nenhum.')}",
"--- PROIBIÇÕES ---",
prohib_txt,
]
return {"role": "system", "content": "\n".join(lines)}
def call_openrouter(messages):
model, api_key = get_config()
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
"HTTP-Referer": "http://localhost:5001",
"X-Title": "Chat Flask OpenRouter",
}
payload = {
"model": model,
"messages": messages,
"max_tokens": None,
"stream": False,
}
response = requests.post("https://openrouter.ai/api/v1/chat/completions",
headers=headers, json=payload, timeout=120)
response.raise_for_status()
return response.json()["choices"][0]["message"]["content"]
# ── Rotas ──
@app.route("/")
def index():
return render_template("chat.html")
HELP_TEXT = (
"Comandos disponíveis:\n"
"/help — Mostra esta lista de comandos\n"
"/reset — Restaura todos os arquivos ao padrão de fábrica (perfis, contexto, config, histórico)\n"
"/reiniciar — Reinicia a conversa do zero (mantém perfis e config)\n"
"/encerrar — Encerra o chat e desabilita o input\n"
"/edit_contexto — Abre o editor do arquivo contexto.json\n"
"/edit_perfil_ia — Abre o editor do arquivo perfil_ia.json\n"
"/edit_perfil_user — Abre o editor do arquivo perfil_user.json\n"
"/edit_model_ia — Abre o editor da configuração do modelo IA\n"
"/edit_key_provider — Abre o editor da chave da API\n"
"/edit_proibidos — Edita a lista de tópicos e comportamentos proibidos para a IA\n"
"/atualizar — IA examina perfis, contexto e histórico e melhora o contexto.json automaticamente\n"
"/restaurar_contexto — Restaura o contexto.json do último backup salvo\n"
"/stats — Mostra estatísticas da conversa\n"
"/export — Exporta a conversa atual\n"
)
@app.route("/help")
def help_cmd():
return jsonify({"reply": HELP_TEXT, "cmd_reply": True, "history": get_chat_history()})
PROFILE_FILES = {
"perfil_ia": PERFIL_IA_PATH,
"perfil_user": PERFIL_USER_PATH,
"contexto": CONTEXTO_PATH,
"config": CONFIG_PATH,
"proibidos": PROIBIDOS_PATH,
}
@app.route("/get_file/<filename>")
def get_file(filename):
path = PROFILE_FILES.get(filename)
if not path:
return jsonify({"error": "Arquivo não encontrado."}), 404
try:
with open(path, "r", encoding="utf-8") as f:
return jsonify({"content": f.read()})
except Exception as e:
return jsonify({"error": str(e)}), 500
@app.route("/save_file/<filename>", methods=["POST"])
def save_file(filename):
path = PROFILE_FILES.get(filename)
if not path:
return jsonify({"error": "Arquivo não encontrado."}), 404
try:
data = request.get_json()
content = data.get("content", "")
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return jsonify({"status": "salvo", "filename": filename})
except Exception as e:
return jsonify({"error": str(e)}), 500
EDIT_CMDS = {
"/edit_contexto": ("contexto", "Editor — contexto.json"),
"/editar_contexto": ("contexto", "Editor — contexto.json"),
"/edit_perfil_ia": ("perfil_ia", "Editor — perfil_ia.json"),
"/editar_perfil_ia": ("perfil_ia", "Editor — perfil_ia.json"),
"/edit_perfil_user": ("perfil_user", "Editor — perfil_user.json"),
"/editar_perfil_user":("perfil_user", "Editor — perfil_user.json"),
"/edit_model_ia": ("config", "Editor — Modelo de IA"),
"/edit_key_provider": ("config", "Editor — Chave da API"),
"/edit_proibidos": ("proibidos", "Editor — proibidos.json"),
"/editar_proibidos": ("proibidos", "Editor — proibidos.json"),
}
@app.route("/send", methods=["POST"])
def send_message():
user_message = request.json.get("message", "").strip()
image_url = request.json.get("image_url", "")
if not user_message and not image_url:
return jsonify({"error": "Mensagem vazia"}), 400
cmd = user_message.lower()
if cmd == "/help":
return help_cmd()
if cmd in ("/reset", "/reiniciar"):
if cmd == "/reset":
factory_reset()
text = "Sistema restaurado ao padrão de fábrica! Perfis, contexto, config e histórico foram resetados."
else:
save_chat_history([])
text = "Histórico zerado! Vamos começar uma nova conversa."
return jsonify({"reply": text, "cmd_reply": True, "history": []})
if cmd in ("/encerrar", "/encerar"):
save_chat_history([])
return jsonify({"reply": "Chat encerrado. Recarregue a página para começar novamente.",
"cmd_reply": True, "history": [], "end": True})
if cmd == "/atualizar":
return jsonify({"reply": "Analisando perfis, contexto e histórico para otimizar o contexto.json... Isso pode levar alguns segundos.",
"cmd_reply": True, "history": get_chat_history(), "optimizing": True})
if cmd == "/restaurar_contexto":
backups = listar_backups()
if not backups:
return jsonify({"reply": "Nenhum backup encontrado em backups/.", "cmd_reply": True,
"history": get_chat_history()})
with open(backups[0]["path"], "r", encoding="utf-8") as f:
conteudo = f.read()
with open(CONTEXTO_PATH, "w", encoding="utf-8") as f:
f.write(conteudo)
return jsonify({"reply": f"contexto.json restaurado do backup {backups[0]['file']}.",
"cmd_reply": True, "history": get_chat_history()})
if cmd == "/stats":
hist = get_chat_history()
n_user = sum(1 for m in hist if m.get("role") == "user")
n_ai = sum(1 for m in hist if m.get("role") == "assistant")
chars = sum(len(m.get("content", "")) for m in hist)
text = (f"Estatísticas da conversa:\n"
f"Mensagens do usuário: {n_user}\n"
f"Respostas da IA: {n_ai}\n"
f"Total de caracteres: {chars}")
return jsonify({"reply": text, "cmd_reply": True, "history": hist})
if cmd == "/export":
hist = get_chat_history()
export = json.dumps(hist, ensure_ascii=False, indent=2)
return jsonify({"reply": "```json\n" + export + "\n```", "cmd_reply": True, "history": hist})
if cmd in EDIT_CMDS:
filename, modal_title = EDIT_CMDS[cmd]
return jsonify({"reply": f"Abrindo editor de {os.path.basename(PROFILE_FILES[filename])}...", "cmd_reply": True,
"open_editor": True, "filename": filename, "modalTitle": modal_title, "history": get_chat_history()})
history = get_chat_history()
# Se veio imagem, guarda no histórico sem expô-la como texto no chat
if image_url:
history.append({"role": "user", "content": user_message or f"[imagem: ...]",
"image_url": image_url})
else:
history.append({"role": "user", "content": user_message})
system_msg = build_system_prompt()
# Monta mensagens no formato adequado (multimodal se houver image_url)
messages_with_context = [system_msg]
for msg in history:
entry = {"role": msg["role"]}
if msg.get("image_url"):
entry["content"] = [
{"type": "text", "text": msg["content"]},
{"type": "image_url", "image_url": {"url": msg["image_url"]}},
]
else:
entry["content"] = msg["content"]
messages_with_context.append(entry)
try:
ai_reply = call_openrouter(messages_with_context)
except requests.exceptions.HTTPError as e:
# Se falhou por conteúdo multimodal, reenvia só texto
if image_url:
# remove a image_url do histórico para não poluir futuras requisições
history[-1].pop("image_url", None)
hist_clean = [system_msg]
for msg in history:
hist_clean.append({"role": msg["role"], "content": msg["content"]})
try:
ai_reply = call_openrouter(hist_clean)
except requests.exceptions.HTTPError as e2:
return jsonify({"error": f"Erro na API: {str(e2)}"}), 500
else:
return jsonify({"error": f"Erro na API: {str(e)}"}), 500
except Exception as e:
return jsonify({"error": f"Erro: {str(e)}"}), 500
history.append({"role": "assistant", "content": ai_reply})
save_chat_history(history)
return jsonify({"reply": ai_reply, "history": history})
@app.route("/history", methods=["GET"])
def get_history():
return jsonify({"history": get_chat_history()})
@app.route("/reset", methods=["POST"])
def reset_chat():
factory_reset()
return jsonify({"status": "ok", "history": [], "factory": True})
@app.route("/clear", methods=["POST"])
def clear_history():
save_chat_history([])
return jsonify({"status": "ok", "history": []})
@app.route("/end", methods=["POST"])
def end_chat():
save_chat_history([])
return jsonify({"status": "encerrado", "history": []})
@app.route("/optimize", methods=["POST"])
def optimize_context():
try:
ctx = load_json(CONTEXTO_PATH, {})
ia = load_json(PERFIL_IA_PATH, {})
user = load_json(PERFIL_USER_PATH, {})
hist = carregar_historico_dia()
ctx_json = json.dumps(ctx, ensure_ascii=False, indent=2)
ia_json = json.dumps(ia, ensure_ascii=False, indent=2)
user_json = json.dumps(user, ensure_ascii=False, indent=2)
hist_amostra = json.dumps(hist[-20:], ensure_ascii=False, indent=2)
meta_prompt = (
"Você é um especialista em otimização de prompts para assistentes de IA.\n\n"
"Analise o contexto atual, os perfis e o histórico de conversas abaixo e produza "
"uma versão melhorada do arquivo de contexto. O objetivo é tornar o assistente "
"mais eficaz, engajador e alinhado com os padrões observados nas conversas.\n\n"
"--- CONTEXTO ATUAL (contexto.json) ---\n"
f"{ctx_json}\n\n"
"--- PERFIL DA IA (perfil_ia.json) ---\n"
f"{ia_json}\n\n"
"--- PERFIL DO USUÁRIO (perfil_user.json) ---\n"
f"{user_json}\n\n"
"--- HISTÓRICO RECENTE (últimas 20 mensagens) ---\n"
f"{hist_amostra}\n\n"
"Com base nessa análise, produza APENAS um JSON válido (sem markdown, sem código adicional) "
"EXATAMENTE no formato:\n"
'{"role": "system", "content": "..."}\n\n'
"O campo 'content' deve conter um prompt de sistema completo, bem escrito em português "
"do Brasil, que incorpore o que há de melhor no contexto atual mais as melhorias "
"identificadas nos padrões de conversa."
)
bak = fazer_backup_contexto()
messages = [
{"role": "system", "content": "Você é um especialista em otimização de prompts de IA. Sempre responda APENAS com JSON puro, sem formatação markdown."},
{"role": "user", "content": meta_prompt},
]
ai_reply = call_openrouter(messages)
# Log the raw response
import logging
logging.warning("RAW optimize response: %s", ai_reply[:500])
# Strip markdown fences
reply_clean = ai_reply.strip()
if "```" in reply_clean:
lines = reply_clean.split("\n")
cleaned = []
for line in lines:
stripped = line.strip()
if stripped.startswith("```"):
continue
cleaned.append(line)
reply_clean = "\n".join(cleaned).strip()
# Try to extract JSON from the response (find first { and last })
first_brace = reply_clean.find("{")
last_brace = reply_clean.rfind("}")
if first_brace != -1 and last_brace != -1 and last_brace > first_brace:
reply_clean = reply_clean[first_brace:last_brace + 1]
novo = json.loads(reply_clean)
if not isinstance(novo, dict) or "role" not in novo or "content" not in novo:
raise ValueError("Resposta não é um JSON válido para contexto.json")
with open(CONTEXTO_PATH, "w", encoding="utf-8") as f:
json.dump(novo, f, ensure_ascii=False, indent=2)
nome_bak = os.path.basename(bak)
return jsonify({
"reply": f"contexto.json otimizado com sucesso!\nBackup salvo: {nome_bak}",
"cmd_reply": True,
})
except json.JSONDecodeError:
return jsonify({"error": "A IA não retornou um JSON válido. O contexto não foi alterado. Tente novamente."}), 500
except Exception as e:
return jsonify({"error": f"Erro ao otimizar: {str(e)}"}), 500
# ── Upload de arquivos ──
def allowed_file(name):
return "." in name and name.rsplit(".", 1)[1].lower() in ALLOWED_EXTENSIONS
@app.route("/upload", methods=["POST"])
def upload_file():
if "file" not in request.files:
return jsonify({"error": "Nenhum arquivo enviado."}), 400
f = request.files["file"]
if f.filename == "":
return jsonify({"error": "Nome de arquivo vazio."}), 400
if not allowed_file(f.filename):
return jsonify({"error": "Tipo de arquivo não permitido."}), 400
try:
nome = secure_filename(f.filename)
# evitar sobrescrita
base, ext = os.path.splitext(nome)
contador = 1
caminho = os.path.join(UPLOADS_DIR, nome)
while os.path.exists(caminho):
caminho = os.path.join(UPLOADS_DIR, f"{base}_{contador}{ext}")
contador += 1
f.save(caminho)
nome_final = os.path.basename(caminho)
is_image = ext.lower().lstrip(".") in IMAGE_EXTENSIONS
result = {
"status": "ok",
"filename": nome_final,
"url": f"/uploads/{nome_final}",
"is_image": is_image,
}
if is_image:
mime = mimetypes.guess_type(nome_final)[0] or "image/png"
with open(caminho, "rb") as img:
b64 = base64.b64encode(img.read()).decode("ascii")
result["data_url"] = f"data:{mime};base64,{b64}"
return jsonify(result)
except Exception as e:
return jsonify({"error": str(e)}), 500
if __name__ == "__main__":
print(" Acesse: http://localhost:5001")
app.run(host="0.0.0.0", port=5001, debug=True, use_reloader=False)