flowise1 / app.py
Persano's picture
Update app.py
c9cc1c7 verified
raw
history blame
12.2 kB
import os
import io
import base64
import pandas as pd
import matplotlib.pyplot as plt
from flask import Flask, render_template, request, send_file
from matplotlib.ticker import FuncFormatter
from xhtml2pdf import pisa
from datetime import datetime
import openai
import logging
import locale
from datetime import datetime
app = Flask(__name__)
# Setar locale para português do Brasil
try:
locale.setlocale(locale.LC_TIME, "pt_BR.UTF-8")
except locale.Error:
# fallback caso o locale não esteja instalado no servidor
locale.setlocale(locale.LC_TIME, "pt_BR")
# Configuração de logger
logging.basicConfig(level=logging.DEBUG)
openai.api_key = os.getenv("OPENAI_API_KEY")
def formatar_brl(valor):
"""Formata número float em string moeda BRL."""
return f"R$ {valor:,.2f}".replace(",", "X").replace(".", ",").replace("X", ".")
def gerar_analise(investimentos_finais, capital, patrimonio_studio_final):
melhor = max(investimentos_finais, key=investimentos_finais.get)
valor_melhor = investimentos_finais[melhor]
retorno_pct = ((valor_melhor - capital) / capital) * 100
texto = f"""
Após análise dos cenários projetados para 5 anos, o investimento <strong>{melhor}</strong> apresenta o melhor desempenho,
com um valor final estimado de <strong>{formatar_brl(valor_melhor)}</strong>, equivalente a um retorno de <strong>{retorno_pct:.1f}%</strong>.
<br><br>
O valor patrimonial do Studio, conforme normas contábeis e profissionais de investimento, representa o valor acumulado do imóvel sem considerar a renda gerada,
ou seja, é o aumento do patrimônio bruto estimado. Neste cenário, o valor patrimonial final do Studio é <strong>{formatar_brl(patrimonio_studio_final)}</strong>.
"""
return texto
def gerar_analise_ia(investimentos_finais, capital, patrimonio_studio_final):
prompt = f"""
Você é um analista financeiro experiente, e está redigindo um relatório para um cliente com capital inicial de R$ {capital:,.2f}. Após 5 anos, os resultados dos investimentos foram:
- Studio (imóvel com renda e valorização): R$ {investimentos_finais['Studio']:,.2f}
- Franquia: R$ {investimentos_finais['Franquia']:,.2f}
- Ações: R$ {investimentos_finais['Ações']:,.2f}
- Renda Fixa: R$ {investimentos_finais['Renda Fixa']:,.2f}
O valor patrimonial isolado do Studio (sem a renda gerada) ficou em R$ {patrimonio_studio_final:,.2f}.
Com base nesses dados, escreva um relatório claro, didático e consultivo, com no máximo 10 linhas, contendo:
1. Identificação do investimento com melhor desempenho e retorno percentual.
2. Análise do perfil de risco e tipo de investidor ideal para cada investimento.
3. Estratégias por trás de cada modelo.
4. Vantagens e desvantagens.
5. Sugestão de diversificação balanceada para perfil moderado.
Use quebras de linha para separar os tópicos.
"""
resposta = openai.ChatCompletion.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "Você é um analista financeiro experiente."},
{"role": "user", "content": prompt}
],
max_tokens=500,
temperature=0.7
)
texto = resposta.choices[0].message.content
# Troca quebras de linha por <br> para exibir no HTML
texto_formatado = texto.replace("\n", "<br>")
return texto_formatado
def render_pdf(template_src, context_dict):
html = render_template(template_src, **context_dict)
result = io.BytesIO()
pisa_status = pisa.CreatePDF(io.StringIO(html), dest=result)
if not pisa_status.err:
result.seek(0)
return result
logging.error(f"Erro pisa.CreatePDF: {pisa_status.err}")
return None
def safe_float(valor, nome, default=0.0):
try:
f = float(valor)
logging.debug(f"{nome} recebido e convertido: {f}")
return f
except Exception as e:
logging.error(f"Erro convertendo {nome} com valor '{valor}': {e}")
return default
def calcular_projecoes(capital, studio_ret, valorizacao, franquia_ret, acoes_ret, renda_fixa):
anos = list(range(1, 6))
patrimonio_studio = [capital * ((1 + valorizacao / 100) ** ano) for ano in anos]
renda_acumulada_studio = [capital * (((1 + studio_ret / 100) ** (12 * ano)) - 1) for ano in anos]
studio_total = [p + r for p, r in zip(patrimonio_studio, renda_acumulada_studio)]
franquia = [capital + (franquia_ret * ano) for ano in anos]
acoes = [capital * ((1 + acoes_ret / 100) ** ano) for ano in anos]
renda_fixa_valores = [capital * ((1 + renda_fixa / 100) ** ano) for ano in anos]
dados = {
"Ano": anos,
"Studio (Patrimônio + Renda)": studio_total,
"Franquia": franquia,
"Ações": acoes,
"Renda Fixa": renda_fixa_valores,
}
investimentos_finais = {
"Studio": studio_total[-1],
"Franquia": franquia[-1],
"Ações": acoes[-1],
"Renda Fixa": renda_fixa_valores[-1],
}
return dados, investimentos_finais
def gerar_grafico(anos, studio_total, franquia, acoes, renda_fixa_valores):
plt.figure(figsize=(8, 5))
plt.plot(anos, studio_total, label="Studio (Patrimônio + Renda)", marker="o")
plt.plot(anos, franquia, label="Franquia", marker="o")
plt.plot(anos, acoes, label="Ações", marker="o")
plt.plot(anos, renda_fixa_valores, label="Renda Fixa", marker="o")
plt.title("Projeção de Investimentos (5 anos)")
plt.xlabel("Ano")
plt.ylabel("Valor (R$)")
plt.legend()
plt.grid(True)
plt.gca().yaxis.set_major_formatter(FuncFormatter(lambda x, _: f"R${x:,.0f}".replace(",", ".")))
plt.tight_layout()
buf = io.BytesIO()
plt.savefig(buf, format="png")
buf.seek(0)
grafico_base64 = base64.b64encode(buf.getvalue()).decode("utf-8")
buf.close()
plt.close()
return grafico_base64
@app.route("/", methods=["GET", "POST"])
def index():
if request.method == "POST":
try:
logging.debug(f"Dados recebidos no POST: {request.form}")
capital = safe_float(request.form.get("capital"), "capital")
studio_ret = safe_float(request.form.get("studio_ret"), "studio_ret")
valorizacao = safe_float(request.form.get("valorizacao"), "valorizacao")
franquia_ret = safe_float(request.form.get("franquia_ret"), "franquia_ret")
acoes_ret = safe_float(request.form.get("acoes_ret"), "acoes_ret")
renda_fixa = safe_float(request.form.get("renda_fixa"), "renda_fixa")
inflacao = safe_float(request.form.get("inflacao"), "inflacao")
dados, investimentos_finais = calcular_projecoes(capital, studio_ret, valorizacao, franquia_ret, acoes_ret, renda_fixa)
df = pd.DataFrame(dados)
df_formatado = df.copy()
for col in df.columns:
if col != "Ano":
df_formatado[col] = df_formatado[col].apply(formatar_brl)
tabela = df_formatado.to_html(index=False, classes="table table-striped table-sm text-end", border=0)
anos = dados["Ano"]
studio_total = dados["Studio (Patrimônio + Renda)"]
franquia = dados["Franquia"]
acoes = dados["Ações"]
renda_fixa_valores = dados["Renda Fixa"]
grafico_base64 = gerar_grafico(anos, studio_total, franquia, acoes, renda_fixa_valores)
analise_final = gerar_analise(investimentos_finais, capital, studio_total[-1])
return render_template(
"index.html",
capital=capital,
tabela=tabela,
grafico=grafico_base64,
analise_final=analise_final
)
except Exception:
logging.error("Erro na rota /", exc_info=True)
return "Erro interno no servidor. Verifique os logs.", 500
return render_template("index.html")
import locale
from datetime import datetime
@app.route("/gerar-pdf", methods=["POST"])
def gerar_pdf():
try:
# --- Configurar locale para português do Brasil (formatação de mês em pt)
try:
locale.setlocale(locale.LC_TIME, "pt_BR.UTF-8")
except locale.Error:
try:
locale.setlocale(locale.LC_TIME, "pt_BR")
except locale.Error:
# Se falhar, continua com o locale padrão (possivelmente inglês)
pass
# --- Coleta e conversão dos dados do formulário
capital = float(request.form["capital"])
studio_ret = float(request.form["studio_ret"])
valorizacao = float(request.form["valorizacao"])
franquia_ret = float(request.form["franquia_ret"])
acoes_ret = float(request.form["acoes_ret"])
renda_fixa = float(request.form["renda_fixa"])
inflacao = float(request.form["inflacao"]) # Ainda não usado
anos = list(range(1, 6))
# --- Cálculo das projeções de investimento
patrimonio_studio = [capital * ((1 + valorizacao / 100) ** ano) for ano in anos]
renda_acumulada_studio = [capital * (((1 + studio_ret / 100) ** (12 * ano)) - 1) for ano in anos]
studio_total = [p + r for p, r in zip(patrimonio_studio, renda_acumulada_studio)]
patrimonio_final = studio_total[-1]
franquia = [capital + (franquia_ret * ano) for ano in anos]
acoes = [capital * ((1 + acoes_ret / 100) ** ano) for ano in anos]
renda_fixa_valores = [capital * ((1 + renda_fixa / 100) ** ano) for ano in anos]
# --- Monta o DataFrame
df = pd.DataFrame({
"Ano": anos,
"Studio (Patrimônio + Renda)": studio_total,
"Franquia": franquia,
"Ações": acoes,
"Renda Fixa": renda_fixa_valores
})
# --- Dicionário final de investimentos
investimentos_finais = {
"Studio": studio_total[-1],
"Franquia": franquia[-1],
"Ações": acoes[-1],
"Renda Fixa": renda_fixa_valores[-1],
}
# --- Geração do gráfico
buf = io.BytesIO()
plt.figure(figsize=(8, 5))
plt.plot(anos, studio_total, label="Studio (Patrimônio + Renda)", marker="o")
plt.plot(anos, franquia, label="Franquia", marker="o")
plt.plot(anos, acoes, label="Ações", marker="o")
plt.plot(anos, renda_fixa_valores, label="Renda Fixa", marker="o")
plt.title("Projeção de Investimentos (5 anos)")
plt.xlabel("Ano")
plt.ylabel("Valor (R$)")
plt.legend()
plt.grid(True)
plt.tight_layout()
plt.savefig(buf, format="png")
buf.seek(0)
grafico_base64 = base64.b64encode(buf.read()).decode("utf-8")
buf.close()
plt.close()
# --- Formata a tabela (função formatar_brl deve estar definida)
df_formatado = df.copy()
for col in df.columns:
if col != "Ano":
df_formatado[col] = df_formatado[col].apply(formatar_brl)
tabela_html = df_formatado.to_html(index=False, classes="tabela", border=0)
# --- Geração da análise com IA (função gerar_analise_ia deve estar definida)
analise_final_ia = gerar_analise_ia(investimentos_finais, capital, patrimonio_final)
# --- Formatar data para português
data_hoje = datetime.now().strftime("São Paulo, %d de %B de %Y")
# --- Contexto para o PDF
context = {
"capital": capital,
"grafico": grafico_base64,
"tabela": tabela_html,
"analise_final": analise_final_ia,
"data_hoje": data_hoje
}
# --- Renderiza e retorna o PDF
pdf = render_pdf("relatorio.html", context)
if pdf:
return send_file(pdf, mimetype="application/pdf", as_attachment=True, download_name="relatorio_investimento.pdf")
else:
return "Erro ao gerar PDF", 500
except Exception as e:
app.logger.error("Erro na rota /gerar-pdf", exc_info=True)
return "Erro interno ao gerar o PDF", 500
if __name__ == "__main__":
app.run(host="0.0.0.0", port=int(os.environ.get("PORT", 7860)), debug=True)