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.
638 lines
24 KiB
Python
Executable File
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) |