import os import json import requests from time import time from functools import wraps import base64 from base64 import b64decode from datetime import datetime, timedelta from flask import Flask, Blueprint, jsonify, request, Response from flask_mysqldb import MySQL from flask_cors import CORS from .auth import token_required from dotenv import load_dotenv from flask import send_file from io import BytesIO load_dotenv() # Configurações da API do Banco Inter CLIENT_ID = os.environ.get("CLIENT_ID") CLIENT_SECRET = os.environ.get("CLIENT_SECRET") CERT_PATH = os.environ.get("CERT_PATH") CERT_KEY_PATH = os.environ.get("CERT_KEY_PATH") ACCOUNT = os.environ.get("ACCOUNT", None) API_BASE_URL = "https://cdpj.partners.bancointer.com.br" boletos = Blueprint('boletos', __name__) app = Flask(__name__) cors = CORS(app, resources={r"boletos/*": {"origins": "*", "methods": ["GET", "POST", "PUT", "OPTIONS"]}}) app.config["CORS_HEADERS"] = ["Content-Type", "Authorization", "Content-Disposition", "Content-Length"] MYSQL_HOST = os.environ.get('MYSQL_HOST') MYSQL_PORT = os.environ.get('MYSQL_PORT') MYSQL_USER = os.environ.get('MYSQL_USER') MYSQL_PASSWORD = os.environ.get('MYSQL_PASSWORD') MYSQL_DB = os.environ.get('MYSQL_DB') MYSQL_CURSORCLASS = os.environ.get("MYSQL_CURSORCLASS") mysql = MySQL() token_cache = {} def intercept_options(f): @wraps(f) def decorated(*args, **kwargs): if request.method == "OPTIONS": response = jsonify({"status": "ok"}) response.headers.add("Access-Control-Allow-Origin", "*") response.headers.add("Access-Control-Allow-Methods", "GET,POST,PUT,OPTIONS") response.headers.add("Access-Control-Allow-Headers", "Content-Type,Authorization,Content-Disposition,Content-Length") return response, 200 return f(*args, **kwargs) return decorated def obter_token(scopes=["boleto-cobranca.write", "boleto-cobranca.read"]): """Obtem um token de acesso para a API do Banco Inter.""" scopes_param = " ".join(scopes) if token_cache.get(scopes_param): cached_token = token_cache.get(scopes_param) # token is still valid if cached_token.get("expires_at") > int(time()): return cached_token["token"] try: request_string = f"client_id={CLIENT_ID}&client_secret={CLIENT_SECRET}&scope={scopes_param}&grant_type=client_credentials" print("request_string", request_string) response = requests.post(url=f"{API_BASE_URL}/oauth/v2/token", headers={"Content-Type": "application/x-www-form-urlencoded"}, cert=(CERT_PATH, CERT_KEY_PATH), data=request_string) data_json = response.json() token = data_json.get("access_token") token_cache[scopes_param] = { "token": token, "expires_at": data_json.get("expires_in") + int(time()), } return token except Exception as e: print(f"Erro ao obter token de acesso: {e}") return None def emitir_cobranca(token, dados_cobranca): """Emite uma cobrança para o Banco Inter de forma assíncrona.""" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json", "Accept": "application/json" } try: response = requests.post( f"{API_BASE_URL}/cobranca/v3/cobrancas", headers=headers, cert=(CERT_PATH, CERT_KEY_PATH), json=dados_cobranca ) # Captura a resposta completa resposta_json = response.json() # Exibe detalhes da resposta para debug print("Código de status:", response.status_code) print("Resposta da API:", json.dumps(resposta_json, indent=4, ensure_ascii=False)) return resposta_json except Exception as e: print(f"Erro ao emitir cobrança: {e}") return None def obter_cobranca(token, codigo_solicitacao): """Obtem uma cobrança.""" headers = { "Authorization": f"Bearer {token}", "Content-Type": "Application/json", } try: response = requests.get(f"{API_BASE_URL}/cobranca/v3/cobrancas/{codigo_solicitacao}", headers=headers, cert=(CERT_PATH, CERT_KEY_PATH)) return response.json() except Exception as e: print(f"Erro ao emitir cobrança:{e}") return None def obter_cobranca_pdf(token, codigo_solicitacao): """Obtem o PDF de uma cobrança.""" headers = { "Authorization": f"Bearer {token}", "Content-Type": "Application/json", } try: # Certifique-se de que o codigo_solicitacao seja inserido corretamente na URL url = f"{API_BASE_URL}/cobranca/v3/cobrancas/{codigo_solicitacao}/pdf" response = requests.get(url, headers=headers, cert=(CERT_PATH, CERT_KEY_PATH)) return response.json() except Exception as e: print(f"Erro ao emitir cobrança:{e}") return None def processar_cobranca_cliente(cliente, token): """ Processa a emissão da cobrança para um cliente específico, retornando os dados relevantes. """ id_empresa = cliente["idempresa"] hoje = datetime.now() vencimento_dia = int(cliente["data_vencimento"]) vencimento = datetime(hoje.year, hoje.month, vencimento_dia).strftime("%Y-%m-%d") dados_cobranca = { "seuNumero": f"boleto-{id_empresa}", "valorNominal": str(cliente["valor_servico"]), "dataVencimento": vencimento, "numDiasAgenda": 30, "pagador": { "cpfCnpj": cliente["cpf_cnpj"], "tipoPessoa": cliente["tipo_pessoa"], "nome": cliente["nome"], "telefone": cliente["telefone"], "endereco": cliente["endereco"], "bairro": cliente["bairro"], "cidade": cliente["cidade"], "uf": cliente["uf"], "cep": cliente["cep"], }, "multa": { "taxa": 5.0, "codigo": "PERCENTUAL" }, "mora": { "taxa": 1.0, "codigo": "TAXAMENSAL" }, "formasRecebimento": ["BOLETO", "PIX"], } emissao_cobranca = emitir_cobranca(token, dados_cobranca) codigo_solicitacao = emissao_cobranca.get("codigoSolicitacao") if emissao_cobranca else None if not codigo_solicitacao: return { "cliente": cliente, "status": "erro", "mensagem": "Falha ao emitir cobrança." } cobranca = obter_cobranca(token, codigo_solicitacao) if not cobranca: return { "cliente": cliente, "status": "erro", "mensagem": "Falha ao obter cobrança." } boleto_data = cobranca.get("cobranca", {}).get("boleto", {}) codigo_barras = boleto_data.get("codigoBarras") linha_digitavel = cobranca.get("boleto", {}).get("linhaDigitavel") pix_data = cobranca.get("pix", {}) pix_copia_e_cola = pix_data.get("pixCopiaECola") cobranca_pdf = obter_cobranca_pdf(token, codigo_solicitacao) if not cobranca_pdf or "pdf" not in cobranca_pdf: return { "cliente": cliente, "status": "erro", "mensagem": "Falha ao obter PDF da cobrança." } return { "cliente": cliente, "status": "sucesso", "codigo_solicitacao": codigo_solicitacao, "codigo_barras": codigo_barras, "linha_digitavel": linha_digitavel, "pix_copia_e_cola": pix_copia_e_cola, "pdf": cobranca_pdf["pdf"], } @boletos.route('/boletos', methods=['POST', 'OPTIONS']) @intercept_options @token_required def gerar_boletos(data): domain = data.get('domain') escopos = data.get('scopes', []) escopos.append("boleto-cobranca.write") escopos.append("boleto-cobranca.read") if 'boleto-cobranca.write' not in escopos: return jsonify({"erro": "Permissão necessária: boleto-cobranca.write"}), 403 if not domain: return jsonify({"error": "Domínio não encontrado no token"}), 400 hoje = datetime.now() if hoje.day > 15: return jsonify({"erro": "Boletos só podem ser gerados até o dia 15 de cada mês."}), 403 try: cur = mysql.connection.cursor() cur.execute( "SELECT idempresa, nome, cpf_cnpj, tipo_pessoa, telefone, endereco, bairro, cidade, uf, cep, valor_servico, data_vencimento FROM empresa WHERE dominio = %s", (domain,) ) clientes = cur.fetchall() cur.close() if not clientes: return jsonify({"erro": "Nenhum cliente encontrado."}), 404 except Exception as e: return jsonify({"erro": "Erro ao acessar o banco de dados."}), 500 token = obter_token(escopos) if not token: return jsonify({"erro": "Falha ao obter o token de acesso."}), 401 resultados = [] for cliente in clientes: resultado = processar_cobranca_cliente(cliente, token) resultados.append(resultado) for resultado in resultados: if resultado.get("status") == "sucesso" and "pdf" in resultado: pdf_base64 = resultado["pdf"] pdf_data = base64.b64decode(pdf_base64) pdf_io = BytesIO(pdf_data) pdf_io.seek(0) return send_file( pdf_io, mimetype='application/pdf', as_attachment=True, download_name=f"boleto-{cliente['idempresa']}.pdf" ) if "erro" in resultado: return jsonify({"error": "Falha ao recuperar pdf"}), 500 return jsonify({"message": "Boletos gerados com sucesso!"}), 200 if __name__ == "__main__": app.register_blueprint(boletos) app.run(port=5000, debug=True)