import sys import ollama import os import warnings from urllib.parse import quote_plus import pandas as pd import openpyxl from openpyxl.styles import Font, PatternFill, Alignment, Border, Side from openpyxl.utils import get_column_letter from sqlalchemy import create_engine from langchain_community.utilities.sql_database import SQLDatabase from langchain_core.prompts import PromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_groq import ChatGroq warnings.filterwarnings("ignore", category=DeprecationWarning) if sys.platform == "win32": sys.stdout.reconfigure(encoding='utf-8', errors='replace') sys.stderr.reconfigure(encoding='utf-8', errors='replace') # Configuração do LLM #llm = ChatGroq( # model="llama-3.3-70b-versatile", # api_key="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" #) llm = ollama.chat( model="qwen2.5-coder:7b" ) # Configurações do Banco host = '127.0.0.1' port = 5432 user = 'postgres' password = 'ps@web' dbname = 'bdps01' try: password_encoded = quote_plus(password) url = f"postgresql+psycopg2://{user}:{password_encoded}@{host}:{port}/{dbname}" engine = create_engine( url, connect_args={ "client_encoding": "win1252" } ) tabelas_permitidas = [ 'vw_clientes', 'vw_vendas_langchain', 'vw_vendas_produtos_langchain', 'vw_produtos', 'vw_vendedores', ] db = SQLDatabase( engine, include_tables=tabelas_permitidas, sample_rows_in_table_info=2, view_support=True ) template = """Dada uma pergunta do usuário, crie uma query {dialect} sintaticamente correta para executar. Quando pedir o faturamento informe a quantidade e o valor. Senão informar a ordem order by nome_cliente na tabela vw_clientes o campo data_nota exibir no formato DD/MM/YYYY Quando pedir a nota mostre o numero da nota a data_nota e o valor Apenas retorne o código SQL limpo e sem erros para a versao PostgreSQL 9.5.3, compiled by Visual C++ build 1800, 64-bit, sem explicações e sem marcações markdown. Use a tabela vendas_produtos_langchain somente quando o propmt se referir a produto quando pedir o valor do produto informe também a quantidade do produto Quando pedir os que mais venderam ou faturaram informe na ordem do maior valor pro menor Use a estrutura de tabelas abaixo: {table_info} Pergunta: {question} Query SQL:""" prompt = PromptTemplate.from_template(template) chain = ( { "question": lambda x: x["question"], "table_info": lambda x: db.get_table_info(), "dialect": lambda x: db.dialect } | prompt | llm | StrOutputParser() ) pergunta_usuario = sys.argv[1] if len(sys.argv) > 1 else "Mostrar vendas" query_gerada = chain.invoke({"question": pergunta_usuario}) query_limpa = query_gerada.replace("```sql", "").replace("```", "").strip() #print("\n--- QUERY SENDO EXECUTADA ---") #print(query_limpa) #print("-----------------------------\n") df = pd.read_sql_query(query_limpa, con=engine) total_linhas = len(df) if total_linhas > 0: nome_arquivo = "resultado.xlsx" with pd.ExcelWriter(nome_arquivo, engine='openpyxl') as writer: df.to_excel(writer, index=False, sheet_name='Vendas_Produtos') workbook = writer.book worksheet = writer.sheets['Vendas_Produtos'] worksheet.views.sheetView[0].showGridLines = True header_font = Font(name="Arial", size=11, bold=True, color="FFFFFF") header_fill = PatternFill(start_color="1F497D", end_color="1F497D", fill_type="solid") thin_border = Border( left=Side(style='thin', color='D9D9D9'), right=Side(style='thin', color='D9D9D9'), top=Side(style='thin', color='D9D9D9'), bottom=Side(style='thin', color='D9D9D9') ) zebra_fill = PatternFill(start_color="F2F5F8", end_color="F2F5F8", fill_type="solid") # --- IDENTIFICAR A COLUNA BASE PARA O CÁLCULO DE % --- # Procura por colunas de valor/faturamento/total. Se não achar, usa a última coluna do DataFrame. col_valor_idx = None for idx, col_name in enumerate(df.columns, start=1): col_lc = str(col_name).strip().lower() if col_lc.startswith(("valor", "total", "faturamento")): col_valor_idx = idx # 1-based index no Excel if col_valor_idx is None: col_valor_idx = len(df.columns) col_valor_letter = get_column_letter(col_valor_idx) # A nova coluna de % ficará após a última coluna dos dados originais pct_col_idx = len(df.columns) + 1 pct_col_letter = get_column_letter(pct_col_idx) row_total = total_linhas + 2 # --- FORMATAÇÃO DO CABEÇALHO (Incluindo a nova coluna) --- for col_num in range(1, pct_col_idx + 1): cell = worksheet.cell(row=1, column=col_num) if col_num == pct_col_idx: cell.value = "% Relativo" cell.font = header_font cell.fill = header_fill cell.alignment = Alignment(horizontal="center", vertical="center") cell.border = thin_border # --- FORMATAÇÃO DO CORPO DOS DADOS --- for r_idx in range(2, total_linhas + 2): for c_idx in range(1, len(df.columns) + 1): cell = worksheet.cell(row=r_idx, column=c_idx) cell.border = thin_border if r_idx % 2 == 0: cell.fill = zebra_fill nome_coluna = str(df.columns[c_idx - 1]).strip().lower() if nome_coluna.startswith(("valor", "quantidade", "total", "faturamento")): cell.alignment = Alignment(horizontal="right", vertical="center") if nome_coluna.startswith(("valor", "total", "faturamento")): cell.number_format = '#,##0.00' else: cell.number_format = '#,##0' else: cell.alignment = Alignment(horizontal="left", vertical="center") # Preenche a célula de % Relativo para esta linha pct_cell = worksheet.cell(row=r_idx, column=pct_col_idx) # Fórmula Excel dividindo a célula atual da linha pelo total fixo ($R$N) pct_cell.value = f"={col_valor_letter}{r_idx}/{col_valor_letter}${row_total}" pct_cell.border = thin_border pct_cell.number_format = '0.00%' pct_cell.alignment = Alignment(horizontal="right", vertical="center") if r_idx % 2 == 0: pct_cell.fill = zebra_fill # --- LINHA DE TOTALIZAÇÃO (SOMA NO FINAL) --- total_font = Font(name="Arial", size=11, bold=True, color="000000") total_fill = PatternFill(start_color="D9E1F2", end_color="D9E1F2", fill_type="solid") total_border = Border( left=Side(style='thin', color='D9D9D9'), right=Side(style='thin', color='D9D9D9'), top=Side(style='thin', color='000000'), bottom=Side(style='double', color='000000') # Borda dupla inferior (estilo contábil) ) # Rótulo "Total" na primeira coluna worksheet.cell(row=row_total, column=1, value=f"Total: {total_linhas}").font = total_font for c_idx in range(1, pct_col_idx + 1): cell = worksheet.cell(row=row_total, column=c_idx) cell.font = total_font cell.fill = total_fill cell.border = total_border if c_idx <= len(df.columns): nome_coluna = str(df.columns[c_idx - 1]).strip().lower() col_letter = get_column_letter(c_idx) if nome_coluna.startswith(("valor", "quantidade", "total", "faturamento")): cell.value = f"=SUM({col_letter}2:{col_letter}{row_total - 1})" cell.alignment = Alignment(horizontal="right", vertical="center") if nome_coluna.startswith(("valor", "total", "faturamento")): cell.number_format = '#,##0.00' else: cell.number_format = '#,##0' else: if c_idx > 1: cell.value = "" cell.alignment = Alignment(horizontal="left", vertical="center") else: # Soma da coluna % na linha de total (deve resultar em 100,00%) cell.value = f"=SUM({pct_col_letter}2:{pct_col_letter}{row_total - 1})" cell.number_format = '0.00%' cell.alignment = Alignment(horizontal="right", vertical="center") # Ajuste dinâmico de largura das colunas for col in worksheet.columns: max_len = max(len(str(cell.value or '')) for cell in col) col_letter = get_column_letter(col[0].column) worksheet.column_dimensions[col_letter].width = max(max_len + 4, 12) primeira_linha = df.head(1).to_string(index=False) print(f"Primeira linha do resultado:\n{primeira_linha}\n") print(f"[OK] {total_linhas} registro(s) encontrado(s) e salvos em 'resultado.xlsx'.") else: print(f"Query SQL Executada:\n{query_limpa}\n\n A consulta não retornou nenhum resultado no banco de dados.") except Exception as e: sys.stderr.write(f"Erro ao processar consulta: {str(e)}\n") sys.exit(1)