# -*- coding: utf-8 -*- import logging import sys from functools import wraps import pandas as pd import pymysql import pymysql.cursors from flask import Flask, Response, jsonify, redirect, request from flasgger import Swagger, swag_from from config import API_TOKEN, DB_CONFIG, DEBUG, DOCS_PASSWORD, DOCS_USERNAME LOG_LEVEL = logging.DEBUG if DEBUG else logging.INFO logging.basicConfig( level=LOG_LEVEL, format="%(asctime)s - %(levelname)s - %(message)s", stream=sys.stderr, ) logger = logging.getLogger("API_ZEMPO_IA") app = Flask(__name__) template = { "swagger": "2.0", "info": { "title": "Zempo IA Query API", "description": ( "API Flask otimizada para consultas do Zempo por IA/MCP, com " "endpoints explícitos agrupados por domínio e rotas concretas " "para recursos, catálogos, detalhes e estatísticas." ), "version": "3.1.0", }, "securityDefinitions": { "Bearer": { "type": "apiKey", "name": "Authorization", "in": "header", "description": "Insira o token no formato: Bearer ", } }, "security": [{"Bearer": []}], } swagger = Swagger(app, template=template) MAX_LIMIT = 500 DEFAULT_LIMIT = 50 DYNAMIC_ENDPOINTS_REGISTERED = False GROUP_DESCRIPTIONS = { "cadastros": "Cadastros principais de pessoas, clubes, federações e instituições.", "competicoes": "Competições, inscrições, lutas, fases, áreas e estruturas competitivas.", "eventos": "Eventos, atividades, inscrições, adesões e arquivos correlatos.", "ranking": "Rankings esportivos, pontos, regras e recortes anuais.", "solicitacoes": "Fluxos administrativos, pedidos, status e notificações.", "avaliacoes": "Avaliações, perguntas, respostas e indicadores relacionados.", "conteudo": "Agenda, avisos, FAQ, banners, downloads e mídia.", "operacional": "Recursos auxiliares relevantes que não se enquadram nos demais grupos.", } GROUP_TITLES = { "cadastros": "Cadastros", "competicoes": "Competições", "eventos": "Eventos", "ranking": "Ranking", "solicitacoes": "Solicitações", "avaliacoes": "Avaliações", "conteudo": "Conteúdo", "operacional": "Operacional", } CATALOG_TABLES = { "ambitos", "anuidades", "cargos", "categorias", "categorias_arbitros", "categorias_arquivos", "categorias_tecnicos", "cidades", "civis", "classes", "delegacoes_funcoes", "escolaridades", "estados", "funcoes_arbitros", "funcoes_arbitros_funcional", "golpes", "graduacoes", "niveis", "niveis_conhecimento", "outras_funcoes", "paises", "pix_tipos", "regioes", "sexo", "situacoes", "solicitacoes_status", "solicitacoes_tipos", "staff", "status", "status_verificacoes", "subcategorias_arbitros", "tipos_arquivos", "tipos_comentarios", "tipos_sorteios", "tokuiwaza", } SKIP_PREFIXES = ( "zgestao_", "mblog_", "chat_", "vw_", "view_", "migracao_", ) SKIP_CONTAINS = ( "_bkp", "_temp", "_teste", "_old", "_view", ) SKIP_EXACT = { "acessos", "atualizacoes_bd_express", "conciliacao_bancaria_historico", "confirmacoes", "leads_confirmacao", "manutencao", "mapa", "temp", "usuario_online", "view_mblog_msg", "view_ranking", "view_usuario", "z_emails_zempo", } RESOURCE_DESCRIPTIONS = { "agenda": "Compromissos e tarefas registradas.", "avaliacoes": "Avaliações e processos avaliativos.", "avisos": "Comunicados oficiais publicados.", "clubes": "Clubes e entidades filiadas.", "clubes_contatos": "Contatos vinculados aos clubes.", "competicoes": "Competições cadastradas no Zempo.", "competicoes_categorias": "Categorias habilitadas em cada competição.", "competicoes_eventos": "Eventos relacionados a uma competição.", "competicoes_inscricoes": "Inscrições de atletas em competições.", "competicoes_inscricoes_arbitros": "Inscrições de árbitros em competições.", "competicoes_inscricoes_delegacoes": "Inscrições de delegações em competições.", "competicoes_inscricoes_staff": "Inscrições de staff em competições.", "competicoes_lutas": "Lutas registradas em competições.", "competicoes_lutas_ocorrencias": "Ocorrências detalhadas das lutas.", "delegacoes": "Delegações cadastradas.", "downloads": "Arquivos disponibilizados para download.", "eventos": "Eventos cadastrados no sistema.", "eventos_atividades": "Atividades vinculadas aos eventos.", "eventos_inscricoes": "Inscrições realizadas em eventos.", "faq": "Perguntas frequentes e respostas.", "federacoes": "Federações vinculadas à CBJ.", "federacoes_contatos": "Contatos de federações.", "instituicoes": "Instituições de ensino ou governo.", "pessoas": "Cadastros de atletas, técnicos, árbitros e demais pessoas.", "pessoas_arquivos": "Arquivos anexados ao cadastro de pessoas.", "ranking": "Ranking consolidado por atleta.", "ranking_competicoes": "Competições consideradas nos rankings.", "ranking_dados": "Parâmetros e regras do ranking.", "ranking_pontos": "Detalhamento de pontos de ranking.", "solicitacoes": "Solicitações administrativas e esportivas.", "solicitacoes_notificacoes": "Notificações ligadas às solicitações.", "videos": "Vídeos cadastrados no sistema.", } RESERVED_QUERY_PARAMS = { "limit", "offset", "sort_by", "sort_order", "fields", "q", "group", "granularity", "date_field", "top", } @app.before_request def check_docs_auth(): if request.path.startswith("/apidocs"): auth = request.authorization if not auth or not ( auth.username == DOCS_USERNAME and auth.password == DOCS_PASSWORD ): return Response( "Acesso restrito à documentação.", 401, {"WWW-Authenticate": 'Basic realm="Login Required"'}, ) def token_required(func): @wraps(func) def decorated(*args, **kwargs): auth_header = request.headers.get("Authorization", "") if not auth_header: return jsonify({"error": "Acesso negado. Token de autenticação ausente."}), 401 parts = auth_header.split(" ", 1) if len(parts) != 2 or parts[0] != "Bearer" or not parts[1].strip(): return jsonify({"error": "Token Bearer mal formatado."}), 401 if not API_TOKEN or parts[1].strip() != API_TOKEN: return jsonify({"error": "Acesso negado. Token inválido ou expirado."}), 401 return func(*args, **kwargs) return decorated def execute_query(query, params=None, fetch_all=True): conn = None try: conn = pymysql.connect(**DB_CONFIG) with conn.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute(query, params or ()) return cursor.fetchall() if fetch_all else cursor.fetchone() except pymysql.MySQLError as exc: logger.error("Erro SQL: %s", exc) return None finally: if conn: conn.close() def infer_column_kind(raw_type): if raw_type.startswith(("int", "decimal", "float", "double", "year")): return "number" if raw_type.startswith(("date", "datetime", "timestamp", "time")): return "date" return "text" def should_skip_table(table_name): if table_name in SKIP_EXACT: return True if table_name in CATALOG_TABLES: return False if table_name.startswith(SKIP_PREFIXES): return True return any(part in table_name for part in SKIP_CONTAINS) def classify_group(table_name): if table_name == "pessoas" or table_name.startswith( ("clubes", "federacoes", "pessoas_", "delegacoes", "instituicoes") ): return "cadastros" if table_name == "competicoes" or table_name.startswith("competicoes_"): return "competicoes" if table_name == "eventos" or table_name.startswith("eventos_"): return "eventos" if table_name == "ranking" or table_name.startswith("ranking_"): return "ranking" if table_name == "solicitacoes" or table_name.startswith("solicitacoes_"): return "solicitacoes" if table_name == "avaliacoes" or table_name.startswith("avaliacoes_"): return "avaliacoes" if table_name in {"agenda", "avisos", "downloads", "faq", "videos", "banners"}: return "conteudo" return "operacional" def get_searchable_fields(columns): preferred = [] for name, meta in columns.items(): if meta["kind"] != "text": continue if any( token in name for token in [ "nome", "codigo", "sigla", "titulo", "pergunta", "texto", "descricao", "obs", "email", "telefone", ] ): preferred.append(name) return preferred[:6] def get_default_order(columns): for field in ["data_atualizacao", "data_cadastro", "data", "inicio", "id"]: if field in columns: return field, "desc" first_col = next(iter(columns), "id") return first_col, "asc" def build_resource_registry(schema): resources = {} catalogs = {} for table_name, columns in schema.items(): if should_skip_table(table_name): continue metadata = { "table": table_name, "group": classify_group(table_name), "columns": columns, "column_names": list(columns.keys()), "searchable_fields": get_searchable_fields(columns), "has_id": "id" in columns, "description": RESOURCE_DESCRIPTIONS.get( table_name, f"Recurso baseado na tabela `{table_name}`.", ), } order_field, order_direction = get_default_order(columns) metadata["default_order_field"] = order_field metadata["default_order_direction"] = order_direction if table_name in CATALOG_TABLES: catalogs[table_name] = metadata else: resources[table_name] = metadata return resources, catalogs def load_schema_from_database(): rows = execute_query( ( "SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE " "FROM INFORMATION_SCHEMA.COLUMNS " "WHERE TABLE_SCHEMA = DATABASE();" ) ) if rows is None: raise RuntimeError("Não foi possível carregar o schema a partir do banco de dados.") schema = {} for row in rows: table_name = row["TABLE_NAME"] column_name = row["COLUMN_NAME"] raw_type = row["DATA_TYPE"].lower() schema.setdefault(table_name, {}) schema[table_name][column_name] = { "name": column_name, "raw_type": raw_type, "kind": infer_column_kind(raw_type), } return schema SCHEMA = {} RESOURCE_REGISTRY = {} CATALOG_REGISTRY = {} def refresh_resource_registry(): global SCHEMA, RESOURCE_REGISTRY, CATALOG_REGISTRY SCHEMA = load_schema_from_database() RESOURCE_REGISTRY, CATALOG_REGISTRY = build_resource_registry(SCHEMA) def ensure_registry_loaded(): if RESOURCE_REGISTRY or CATALOG_REGISTRY: return refresh_resource_registry() def get_resource_or_404(resource_name): try: ensure_registry_loaded() except Exception as exc: logger.error("Falha ao carregar registry de recursos: %s", exc) return None, (jsonify({"error": "Não foi possível carregar o schema da API."}), 500) resource = RESOURCE_REGISTRY.get(resource_name) if not resource: return None, (jsonify({"error": f"Recurso '{resource_name}' não encontrado."}), 404) return resource, None def get_catalog_or_404(catalog_name): try: ensure_registry_loaded() except Exception as exc: logger.error("Falha ao carregar registry de catálogos: %s", exc) return None, (jsonify({"error": "Não foi possível carregar o schema da API."}), 500) catalog = CATALOG_REGISTRY.get(catalog_name) if not catalog: return None, (jsonify({"error": f"Catálogo '{catalog_name}' não encontrado."}), 404) return catalog, None def parse_requested_fields(requested_fields, resource): if not requested_fields: return resource["column_names"] fields = [field.strip() for field in requested_fields.split(",") if field.strip()] invalid = [field for field in fields if field not in resource["columns"]] if invalid: raise ValueError(f"Campos inválidos: {', '.join(invalid)}") return fields def build_where_clauses(args, resource): clauses = [] params = [] invalid_filters = [] query_text = args.get("q", "").strip() if query_text and resource["searchable_fields"]: search_clauses = [] for field in resource["searchable_fields"]: search_clauses.append(f"`{field}` LIKE %s") params.append(f"%{query_text}%") clauses.append("(" + " OR ".join(search_clauses) + ")") for key in args.keys(): if key in RESERVED_QUERY_PARAMS: continue if "__" in key: field_name, operator = key.split("__", 1) else: field_name, operator = key, "eq" if field_name not in resource["columns"]: invalid_filters.append(key) continue values = args.getlist(key) if operator == "in": raw_items = [] for value in values: raw_items.extend([item.strip() for item in value.split(",") if item.strip()]) if not raw_items: continue placeholders = ", ".join(["%s"] * len(raw_items)) clauses.append(f"`{field_name}` IN ({placeholders})") params.extend(raw_items) continue if operator == "between": if not values: continue parts = [item.strip() for item in values[0].split(",") if item.strip()] if len(parts) != 2: invalid_filters.append(key) continue clauses.append(f"`{field_name}` BETWEEN %s AND %s") params.extend(parts) continue if operator == "isnull": value = values[0].strip().lower() if values else "true" is_null = value in {"1", "true", "yes"} clauses.append(f"`{field_name}` IS {'NULL' if is_null else 'NOT NULL'}") continue operator_map = { "eq": "=", "ne": "<>", "gt": ">", "gte": ">=", "lt": "<", "lte": "<=", "like": "LIKE", } sql_operator = operator_map.get(operator) if not sql_operator: invalid_filters.append(key) continue for value in values: param_value = f"%{value}%" if operator == "like" else value clauses.append(f"`{field_name}` {sql_operator} %s") params.append(param_value) return clauses, params, invalid_filters def build_list_query(resource, args): requested_fields = parse_requested_fields(args.get("fields"), resource) select_clause = ", ".join(f"`{field}`" for field in requested_fields) clauses, params, invalid_filters = build_where_clauses(args, resource) if invalid_filters: raise ValueError("Filtros inválidos: " + ", ".join(sorted(set(invalid_filters)))) where_sql = f" WHERE {' AND '.join(clauses)}" if clauses else "" sort_by = args.get("sort_by", resource["default_order_field"]) if sort_by not in resource["columns"]: raise ValueError(f"Campo de ordenação inválido: {sort_by}") sort_order = args.get("sort_order", resource["default_order_direction"]).lower() if sort_order not in {"asc", "desc"}: raise ValueError("sort_order deve ser 'asc' ou 'desc'") try: limit = min(max(int(args.get("limit", DEFAULT_LIMIT)), 1), MAX_LIMIT) offset = max(int(args.get("offset", 0)), 0) except ValueError as exc: raise ValueError("limit e offset devem ser inteiros válidos.") from exc query = ( f"SELECT {select_clause} FROM `{resource['table']}`" f"{where_sql} ORDER BY `{sort_by}` {sort_order.upper()} LIMIT %s OFFSET %s" ) params.extend([limit, offset]) return query, params, requested_fields, limit, offset def detect_groupable_fields(resource): candidates = [] for field_name, meta in resource["columns"].items(): if meta["kind"] == "date": continue if any( token in field_name for token in [ "status", "federacao", "clube", "categoria", "classe", "sexo", "tipo", "competicao", "evento", "ano", "pago", "credenciado", "funcao", "atividade", ] ): candidates.append(field_name) return candidates[:6] def detect_date_fields(resource): return [ field_name for field_name, meta in resource["columns"].items() if meta["kind"] == "date" ] def resource_metadata_payload(resource): return { "resource": resource["table"], "group": resource["group"], "description": resource["description"], "has_id": resource["has_id"], "columns": [ {"name": meta["name"], "type": meta["raw_type"], "kind": meta["kind"]} for meta in resource["columns"].values() ], "searchable_fields": resource["searchable_fields"], "groupable_fields": detect_groupable_fields(resource), "date_fields": detect_date_fields(resource), "default_order": { "field": resource["default_order_field"], "direction": resource["default_order_direction"], }, } def run_resource_list(resource_name, expected_group=None): resource, error = get_resource_or_404(resource_name) if error: return error if expected_group and resource["group"] != expected_group: return ( jsonify( { "error": ( f"Recurso '{resource_name}' não pertence ao grupo " f"'{expected_group}'." ) } ), 404, ) try: query, params, requested_fields, limit, offset = build_list_query(resource, request.args) records = execute_query(query, params=params, fetch_all=True) except ValueError as exc: return jsonify({"error": str(exc)}), 400 except pymysql.MySQLError: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 if records is None: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 return ( jsonify( { "resource": resource_name, "group": resource["group"], "filters": request.args.to_dict(flat=False), "pagination": {"limit": limit, "offset": offset, "returned": len(records)}, "fields": requested_fields, "data": records, } ), 200, ) def run_resource_detail(resource_name, item_id, expected_group=None): resource, error = get_resource_or_404(resource_name) if error: return error if expected_group and resource["group"] != expected_group: return ( jsonify( { "error": ( f"Recurso '{resource_name}' não pertence ao grupo " f"'{expected_group}'." ) } ), 404, ) if not resource["has_id"]: return jsonify({"error": f"Recurso '{resource_name}' não possui chave 'id' consultável."}), 400 try: record = execute_query( f"SELECT * FROM `{resource['table']}` WHERE `id` = %s LIMIT 1", params=[item_id], fetch_all=False, ) except pymysql.MySQLError: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 if not record: return jsonify({"error": "Registro não encontrado."}), 404 return jsonify({"resource": resource_name, "data": record}), 200 def run_resource_summary(resource_name, expected_group=None): resource, error = get_resource_or_404(resource_name) if error: return error if expected_group and resource["group"] != expected_group: return ( jsonify( { "error": ( f"Recurso '{resource_name}' não pertence ao grupo " f"'{expected_group}'." ) } ), 404, ) try: clauses, params, invalid_filters = build_where_clauses(request.args, resource) if invalid_filters: return jsonify({"error": "Filtros inválidos: " + ", ".join(sorted(set(invalid_filters)))}), 400 where_sql = f" WHERE {' AND '.join(clauses)}" if clauses else "" total_row = execute_query( f"SELECT COUNT(*) AS total FROM `{resource['table']}`{where_sql}", params=params, fetch_all=False, ) if total_row is None: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 summary = { "resource": resource_name, "group": resource["group"], "total_registros": total_row["total"] if total_row else 0, "groupings": {}, } for field_name in detect_groupable_fields(resource): rows = execute_query( ( f"SELECT `{field_name}` AS valor, COUNT(*) AS total " f"FROM `{resource['table']}`{where_sql} " f"GROUP BY `{field_name}` ORDER BY total DESC LIMIT 10" ), params=params, fetch_all=True, ) summary["groupings"][field_name] = rows or [] return jsonify(summary), 200 except pymysql.MySQLError: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 def run_resource_group_by(resource_name, field_name, expected_group=None): resource, error = get_resource_or_404(resource_name) if error: return error if expected_group and resource["group"] != expected_group: return ( jsonify( { "error": ( f"Recurso '{resource_name}' não pertence ao grupo " f"'{expected_group}'." ) } ), 404, ) if field_name not in resource["columns"]: return jsonify({"error": f"Campo '{field_name}' não existe no recurso '{resource_name}'."}), 400 try: top = min(max(int(request.args.get("top", 20)), 1), 100) except ValueError: return jsonify({"error": "Parâmetro 'top' inválido."}), 400 try: clauses, params, invalid_filters = build_where_clauses(request.args, resource) if invalid_filters: return jsonify({"error": "Filtros inválidos: " + ", ".join(sorted(set(invalid_filters)))}), 400 where_sql = f" WHERE {' AND '.join(clauses)}" if clauses else "" rows = execute_query( ( f"SELECT `{field_name}` AS valor, COUNT(*) AS total " f"FROM `{resource['table']}`{where_sql} " f"GROUP BY `{field_name}` ORDER BY total DESC LIMIT %s" ), params=params + [top], fetch_all=True, ) if rows is None: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 return jsonify({"resource": resource_name, "field": field_name, "data": rows}), 200 except pymysql.MySQLError: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 def run_resource_time_series(resource_name, expected_group=None): resource, error = get_resource_or_404(resource_name) if error: return error if expected_group and resource["group"] != expected_group: return ( jsonify( { "error": ( f"Recurso '{resource_name}' não pertence ao grupo " f"'{expected_group}'." ) } ), 404, ) date_fields = detect_date_fields(resource) date_field = request.args.get("date_field") or (date_fields[0] if date_fields else None) if not date_field: return jsonify({"error": f"Recurso '{resource_name}' não possui campo temporal para série."}), 400 if date_field not in resource["columns"]: return jsonify({"error": f"Campo temporal '{date_field}' inválido."}), 400 granularity = request.args.get("granularity", "month").lower() format_map = {"day": "%Y-%m-%d", "month": "%Y-%m", "year": "%Y"} if granularity not in format_map: return jsonify({"error": "granularity deve ser 'day', 'month' ou 'year'."}), 400 try: clauses, params, invalid_filters = build_where_clauses(request.args, resource) if invalid_filters: return jsonify({"error": "Filtros inválidos: " + ", ".join(sorted(set(invalid_filters)))}), 400 where_sql = f" WHERE {' AND '.join(clauses)}" if clauses else "" rows = execute_query( ( f"SELECT DATE_FORMAT(`{date_field}`, %s) AS periodo, COUNT(*) AS total " f"FROM `{resource['table']}`{where_sql} " f"GROUP BY periodo ORDER BY periodo ASC" ), params=[format_map[granularity]] + params, fetch_all=True, ) if rows is None: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 return ( jsonify( { "resource": resource_name, "date_field": date_field, "granularity": granularity, "data": rows, } ), 200, ) except pymysql.MySQLError: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 def run_catalog_list(catalog_name): catalog, error = get_catalog_or_404(catalog_name) if error: return error display_fields = [field for field in ["id", "nome", "codigo", "sigla", "ordem"] if field in catalog["columns"]] if not display_fields: display_fields = catalog["column_names"][:5] try: query, params, requested_fields, limit, offset = build_list_query(catalog, request.args) rows = execute_query(query, params=params, fetch_all=True) except ValueError as exc: return jsonify({"error": str(exc)}), 400 except pymysql.MySQLError: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 if rows is None: return jsonify({"error": "Erro ao consultar o banco de dados."}), 500 return ( jsonify( { "catalog": catalog_name, "display_fields": display_fields, "pagination": {"limit": limit, "offset": offset, "returned": len(rows)}, "fields": requested_fields, "data": rows, } ), 200, ) def sanitize_endpoint_name(value): return value.replace("/", "_").replace("-", "_").replace(".", "_") def build_auth_security(): return [{"Bearer": []}] def build_standard_error_responses(include_not_found=True): responses = { 400: {"description": "Parâmetros ou filtros inválidos"}, 401: {"description": "Token ausente ou inválido"}, 500: {"description": "Erro interno ou falha ao consultar o banco"}, } if include_not_found: responses[404] = {"description": "Recurso não encontrado"} return responses def build_list_parameters(resource): return [ { "name": "q", "in": "query", "type": "string", "required": False, "description": ( "Busca textual nos campos indexados para pesquisa: " + ", ".join(resource["searchable_fields"]) if resource["searchable_fields"] else "Busca textual quando o recurso possuir campos textuais relevantes." ), }, { "name": "fields", "in": "query", "type": "string", "required": False, "description": ( "Lista de colunas separadas por vírgula. Colunas disponíveis: " + ", ".join(resource["column_names"]) ), }, { "name": "sort_by", "in": "query", "type": "string", "required": False, "description": ( "Campo de ordenação. Padrão: " + resource["default_order_field"] ), }, { "name": "sort_order", "in": "query", "type": "string", "required": False, "enum": ["asc", "desc"], "default": resource["default_order_direction"], "description": "Direção da ordenação.", }, { "name": "limit", "in": "query", "type": "integer", "required": False, "default": DEFAULT_LIMIT, "description": f"Quantidade máxima de registros por página. Máximo: {MAX_LIMIT}.", }, { "name": "offset", "in": "query", "type": "integer", "required": False, "default": 0, "description": "Deslocamento da paginação.", }, ] def build_filter_parameters(resource): return [ { "name": "q", "in": "query", "type": "string", "required": False, "description": ( "Busca textual nos campos indexados para pesquisa: " + ", ".join(resource["searchable_fields"]) if resource["searchable_fields"] else "Busca textual quando o recurso possuir campos textuais relevantes." ), } ] def build_detail_parameters(): return [ { "name": "item_id", "in": "path", "type": "integer", "required": True, "description": "ID do registro.", } ] def build_time_series_parameters(resource): date_fields = detect_date_fields(resource) params = [ { "name": "granularity", "in": "query", "type": "string", "required": False, "enum": ["day", "month", "year"], "default": "month", "description": "Granularidade da série temporal.", } ] if date_fields: params.insert( 0, { "name": "date_field", "in": "query", "type": "string", "required": False, "enum": date_fields, "description": ( "Campo temporal utilizado na série. " f"Padrão: {date_fields[0]}." ), }, ) return params def build_group_by_parameters(): return [ { "name": "top", "in": "query", "type": "integer", "required": False, "default": 20, "description": "Número máximo de grupos retornados.", } ] def build_resource_list_spec(resource): return { "tags": [f"{GROUP_TITLES[resource['group']]} - Recursos"], "summary": f"Lista registros de {resource['table']}", "description": ( f"{resource['description']} " "Aceita filtros dinâmicos por query string usando nomes reais de colunas e " "operadores como `__ne`, `__gt`, `__gte`, `__lt`, `__lte`, `__like`, " "`__in`, `__between` e `__isnull`." ), "security": build_auth_security(), "parameters": build_list_parameters(resource), "responses": { 200: {"description": "Registros retornados com sucesso."}, **build_standard_error_responses(), }, } def build_resource_detail_spec(resource): return { "tags": [f"{GROUP_TITLES[resource['group']]} - Recursos"], "summary": f"Consulta um registro de {resource['table']} por ID", "description": resource["description"], "security": build_auth_security(), "parameters": build_detail_parameters(), "responses": { 200: {"description": "Registro encontrado."}, **build_standard_error_responses(), }, } def build_resource_summary_spec(resource): return { "tags": [f"{GROUP_TITLES[resource['group']]} - Stats"], "summary": f"Resumo quantitativo de {resource['table']}", "description": ( "Retorna total de registros e agrupamentos automáticos para os principais " "campos categóricos do recurso." ), "security": build_auth_security(), "parameters": build_filter_parameters(resource), "responses": { 200: {"description": "Resumo estatístico retornado com sucesso."}, **build_standard_error_responses(), }, } def build_resource_group_by_spec(resource, field_name): return { "tags": [f"{GROUP_TITLES[resource['group']]} - Stats"], "summary": f"Agrupa {resource['table']} por {field_name}", "description": ( f"Agrupamento explícito do recurso `{resource['table']}` pelo campo `{field_name}`." ), "security": build_auth_security(), "parameters": build_group_by_parameters() + build_filter_parameters(resource), "responses": { 200: {"description": "Agrupamento retornado com sucesso."}, **build_standard_error_responses(), }, } def build_resource_time_series_spec(resource): date_fields = detect_date_fields(resource) description = "Série temporal quantitativa do recurso." if date_fields: description += " Campos temporais disponíveis: " + ", ".join(date_fields) + "." return { "tags": [f"{GROUP_TITLES[resource['group']]} - Stats"], "summary": f"Série temporal de {resource['table']}", "description": description, "security": build_auth_security(), "parameters": build_time_series_parameters(resource) + build_filter_parameters(resource), "responses": { 200: {"description": "Série temporal retornada com sucesso."}, **build_standard_error_responses(), }, } def build_group_resource_list_spec(resource): group = resource["group"] return { "tags": [f"{GROUP_TITLES[group]} - Grupo"], "summary": f"Lista {resource['table']} dentro do grupo {group}", "description": resource["description"], "security": build_auth_security(), "parameters": build_list_parameters(resource), "responses": { 200: {"description": "Registros do grupo retornados com sucesso."}, **build_standard_error_responses(), }, } def build_group_resource_detail_spec(resource): group = resource["group"] return { "tags": [f"{GROUP_TITLES[group]} - Grupo"], "summary": f"Consulta um registro de {resource['table']} no grupo {group}", "description": resource["description"], "security": build_auth_security(), "parameters": build_detail_parameters(), "responses": { 200: {"description": "Registro do grupo encontrado."}, **build_standard_error_responses(), }, } def build_group_resource_summary_spec(resource): group = resource["group"] return { "tags": [f"{GROUP_TITLES[group]} - Grupo Stats"], "summary": f"Resumo de {resource['table']} no grupo {group}", "description": "Resumo quantitativo com contexto explícito do grupo.", "security": build_auth_security(), "parameters": build_filter_parameters(resource), "responses": { 200: {"description": "Resumo do grupo retornado com sucesso."}, **build_standard_error_responses(), }, } def build_group_resource_group_by_spec(resource, field_name): group = resource["group"] return { "tags": [f"{GROUP_TITLES[group]} - Grupo Stats"], "summary": f"Agrupa {resource['table']} por {field_name} no grupo {group}", "description": ( f"Agrupamento explícito do recurso `{resource['table']}` pelo campo " f"`{field_name}` dentro do grupo `{group}`." ), "security": build_auth_security(), "parameters": build_group_by_parameters() + build_filter_parameters(resource), "responses": { 200: {"description": "Agrupamento do grupo retornado com sucesso."}, **build_standard_error_responses(), }, } def build_group_resource_time_series_spec(resource): group = resource["group"] return { "tags": [f"{GROUP_TITLES[group]} - Grupo Stats"], "summary": f"Série temporal de {resource['table']} no grupo {group}", "description": "Série temporal do recurso no contexto explícito do grupo.", "security": build_auth_security(), "parameters": build_time_series_parameters(resource) + build_filter_parameters(resource), "responses": { 200: {"description": "Série temporal do grupo retornada com sucesso."}, **build_standard_error_responses(), }, } def build_catalog_spec(catalog): return { "tags": ["Catálogos"], "summary": f"Consulta registros do catálogo {catalog['table']}", "description": ( f"Consulta explícita do catálogo `{catalog['table']}`. " "Aceita os mesmos filtros dinâmicos disponíveis para listagens." ), "security": build_auth_security(), "parameters": build_list_parameters(catalog), "responses": { 200: {"description": "Registros do catálogo retornados com sucesso."}, **build_standard_error_responses(), }, } def build_meta_resource_spec(resource): return { "tags": ["Meta"], "summary": f"Detalha a estrutura do recurso {resource['table']}", "description": resource["description"], "security": build_auth_security(), "responses": { 200: {"description": "Metadados do recurso."}, 401: {"description": "Token ausente ou inválido"}, 404: {"description": "Recurso não encontrado"}, 500: {"description": "Falha ao carregar o schema"}, }, } def build_group_resources_spec(group_name): return { "tags": [f"{GROUP_TITLES[group_name]} - Grupo"], "summary": f"Lista os recursos do grupo {group_name}", "description": GROUP_DESCRIPTIONS.get(group_name, ""), "security": build_auth_security(), "responses": { 200: {"description": "Lista de recursos do grupo."}, 401: {"description": "Token ausente ou inválido"}, 404: {"description": "Grupo não encontrado"}, 500: {"description": "Falha ao carregar o schema"}, }, } def register_documented_rule(rule, endpoint, view_func, methods, spec): documented_view = token_required(view_func) documented_view.__name__ = endpoint documented_view = swag_from(spec)(documented_view) app.add_url_rule(rule, endpoint=endpoint, view_func=documented_view, methods=methods) @app.route("/") def index(): # """ # Redireciona para a documentação # --- # tags: # - Sistema # responses: # 302: # description: Redirecionamento para /apidocs # """ return redirect("/apidocs") @app.route("/health", methods=["GET"]) def health_check(): """ Health Check Verifica o status da API e a conexão com o banco de dados. --- tags: - Health responses: 200: description: API está online e banco conectado. 503: description: Erro de conexão com o banco. """ try: conn = pymysql.connect(**DB_CONFIG) conn.close() return ( jsonify( { "status": "online", "database": "connected", "version": "3.1.0", "timestamp": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S"), } ), 200, ) except Exception as exc: logger.error("Erro ao verificar conexão com o banco de dados: %s", exc) return ( jsonify( { "status": "degraded", "database": "disconnected", "error": str(exc), } ), 503, ) @app.route("/api/meta/overview", methods=["GET"]) @token_required def meta_overview(): """ Visão geral da API --- tags: - Meta security: - Bearer: [] responses: 200: description: Resumo com grupos, contagem de recursos e operadores suportados. 401: description: Token ausente ou inválido. 500: description: Falha ao carregar o schema. """ ensure_registry_loaded() grouped = {} for resource in RESOURCE_REGISTRY.values(): grouped.setdefault(resource["group"], []).append(resource["table"]) return ( jsonify( { "service": "Zempo IA Query API", "resource_count": len(RESOURCE_REGISTRY), "catalog_count": len(CATALOG_REGISTRY), "groups": { group: { "description": GROUP_DESCRIPTIONS.get(group, ""), "resources": sorted(resources), } for group, resources in sorted(grouped.items()) }, "query_operators": [ "eq", "ne", "gt", "gte", "lt", "lte", "like", "in", "between", "isnull", ], } ), 200, ) @app.route("/api/meta/groups", methods=["GET"]) @token_required def meta_groups(): """ Lista os grupos de recursos --- tags: - Meta security: - Bearer: [] responses: 200: description: Lista de grupos disponíveis. 401: description: Token ausente ou inválido. 500: description: Falha ao carregar o schema. """ ensure_registry_loaded() grouped = {} for resource in RESOURCE_REGISTRY.values(): grouped.setdefault(resource["group"], []).append(resource["table"]) data = [ { "group": group, "description": GROUP_DESCRIPTIONS.get(group, ""), "resources": sorted(resources), } for group, resources in sorted(grouped.items()) ] return jsonify(data), 200 @app.route("/api/meta/resources", methods=["GET"]) @token_required def meta_resources(): """ Lista os recursos disponíveis --- tags: - Meta security: - Bearer: [] parameters: - name: group in: query type: string required: false description: Filtra os recursos por grupo. responses: 200: description: Lista de recursos com metadados. 401: description: Token ausente ou inválido. 500: description: Falha ao carregar o schema. """ ensure_registry_loaded() group_filter = request.args.get("group") resources = [ resource_metadata_payload(resource) for resource in RESOURCE_REGISTRY.values() if not group_filter or resource["group"] == group_filter ] return jsonify(sorted(resources, key=lambda item: (item["group"], item["resource"]))), 200 @app.route("/api/catalogs", methods=["GET"]) @token_required def list_catalogs(): """ Lista os catálogos categóricos --- tags: - Catálogos security: - Bearer: [] responses: 200: description: Lista de catálogos disponíveis. 401: description: Token ausente ou inválido. 500: description: Falha ao carregar o schema. """ ensure_registry_loaded() payload = [ { "catalog": name, "description": metadata["description"], "columns": metadata["column_names"], } for name, metadata in sorted(CATALOG_REGISTRY.items()) ] return jsonify(payload), 200 @app.route("/api/meta/resources/", methods=["GET"]) @token_required def meta_resource_detail_fallback(resource_name): return meta_resource_detail_handler(resource_name) @app.route("/api/catalogs/", methods=["GET"]) @token_required def catalog_detail_fallback(catalog_name): return run_catalog_list(catalog_name) @app.route("/api/resources/", methods=["GET"]) @token_required def resource_list_fallback(resource_name): return run_resource_list(resource_name) @app.route("/api/resources//", methods=["GET"]) @token_required def resource_detail_fallback(resource_name, item_id): return run_resource_detail(resource_name, item_id) @app.route("/api/resources//stats/summary", methods=["GET"]) @token_required def resource_summary_fallback(resource_name): return run_resource_summary(resource_name) @app.route("/api/resources//stats/group-by/", methods=["GET"]) @token_required def resource_group_by_fallback(resource_name, field_name): return run_resource_group_by(resource_name, field_name) @app.route("/api/resources//stats/time-series", methods=["GET"]) @token_required def resource_time_series_fallback(resource_name): return run_resource_time_series(resource_name) @app.route("/api/groups//resources", methods=["GET"]) @token_required def group_resources_fallback(group_name): return group_resources_handler(group_name) @app.route("/api/groups//", methods=["GET"]) @token_required def group_resource_list_fallback(group_name, resource_name): return run_resource_list(resource_name, expected_group=group_name) @app.route("/api/groups///", methods=["GET"]) @token_required def group_resource_detail_fallback(group_name, resource_name, item_id): return run_resource_detail(resource_name, item_id, expected_group=group_name) @app.route("/api/groups///stats/summary", methods=["GET"]) @token_required def group_resource_summary_fallback(group_name, resource_name): return run_resource_summary(resource_name, expected_group=group_name) @app.route("/api/groups///stats/group-by/", methods=["GET"]) @token_required def group_resource_group_by_fallback(group_name, resource_name, field_name): return run_resource_group_by(resource_name, field_name, expected_group=group_name) @app.route("/api/groups///stats/time-series", methods=["GET"]) @token_required def group_resource_time_series_fallback(group_name, resource_name): return run_resource_time_series(resource_name, expected_group=group_name) def meta_resource_detail_handler(resource_name): resource, error = get_resource_or_404(resource_name) if error: return error return jsonify(resource_metadata_payload(resource)), 200 def group_resources_handler(group_name): try: ensure_registry_loaded() except Exception as exc: logger.error("Falha ao carregar registry para grupos: %s", exc) return jsonify({"error": "Não foi possível carregar o schema da API."}), 500 if group_name not in GROUP_DESCRIPTIONS: return jsonify({"error": f"Grupo '{group_name}' não encontrado."}), 404 resources = [ resource_metadata_payload(resource) for resource in RESOURCE_REGISTRY.values() if resource["group"] == group_name ] return jsonify(sorted(resources, key=lambda item: item["resource"])), 200 def register_meta_resource_endpoint(resource): resource_name = resource["table"] def view(): return meta_resource_detail_handler(resource_name) endpoint = f"meta_resource_{sanitize_endpoint_name(resource_name)}" register_documented_rule( f"/api/meta/resources/{resource_name}", endpoint, view, ["GET"], build_meta_resource_spec(resource), ) def register_catalog_endpoint(catalog): catalog_name = catalog["table"] def view(): return run_catalog_list(catalog_name) endpoint = f"catalog_{sanitize_endpoint_name(catalog_name)}" register_documented_rule( f"/api/catalogs/{catalog_name}", endpoint, view, ["GET"], build_catalog_spec(catalog), ) def register_group_resources_endpoint(group_name): def view(): return group_resources_handler(group_name) endpoint = f"group_resources_{sanitize_endpoint_name(group_name)}" register_documented_rule( f"/api/groups/{group_name}/resources", endpoint, view, ["GET"], build_group_resources_spec(group_name), ) def register_resource_endpoints(resource): resource_name = resource["table"] group_name = resource["group"] def list_view(): return run_resource_list(resource_name) def detail_view(item_id): return run_resource_detail(resource_name, item_id) def summary_view(): return run_resource_summary(resource_name) def time_series_view(): return run_resource_time_series(resource_name) def group_list_view(): return run_resource_list(resource_name, expected_group=group_name) def group_detail_view(item_id): return run_resource_detail(resource_name, item_id, expected_group=group_name) def group_summary_view(): return run_resource_summary(resource_name, expected_group=group_name) def group_time_series_view(): return run_resource_time_series(resource_name, expected_group=group_name) base_name = sanitize_endpoint_name(resource_name) register_documented_rule( f"/api/resources/{resource_name}", f"resource_list_{base_name}", list_view, ["GET"], build_resource_list_spec(resource), ) register_documented_rule( f"/api/resources/{resource_name}/", f"resource_detail_{base_name}", detail_view, ["GET"], build_resource_detail_spec(resource), ) register_documented_rule( f"/api/resources/{resource_name}/stats/summary", f"resource_summary_{base_name}", summary_view, ["GET"], build_resource_summary_spec(resource), ) register_documented_rule( f"/api/resources/{resource_name}/stats/time-series", f"resource_time_series_{base_name}", time_series_view, ["GET"], build_resource_time_series_spec(resource), ) register_documented_rule( f"/api/groups/{group_name}/{resource_name}", f"group_resource_list_{group_name}_{base_name}", group_list_view, ["GET"], build_group_resource_list_spec(resource), ) register_documented_rule( f"/api/groups/{group_name}/{resource_name}/", f"group_resource_detail_{group_name}_{base_name}", group_detail_view, ["GET"], build_group_resource_detail_spec(resource), ) register_documented_rule( f"/api/groups/{group_name}/{resource_name}/stats/summary", f"group_resource_summary_{group_name}_{base_name}", group_summary_view, ["GET"], build_group_resource_summary_spec(resource), ) register_documented_rule( f"/api/groups/{group_name}/{resource_name}/stats/time-series", f"group_resource_time_series_{group_name}_{base_name}", group_time_series_view, ["GET"], build_group_resource_time_series_spec(resource), ) for field_name in detect_groupable_fields(resource): def group_by_view(field=field_name): return run_resource_group_by(resource_name, field) def group_group_by_view(field=field_name): return run_resource_group_by(resource_name, field, expected_group=group_name) field_slug = sanitize_endpoint_name(field_name) register_documented_rule( f"/api/resources/{resource_name}/stats/group-by/{field_name}", f"resource_group_by_{base_name}_{field_slug}", group_by_view, ["GET"], build_resource_group_by_spec(resource, field_name), ) register_documented_rule( f"/api/groups/{group_name}/{resource_name}/stats/group-by/{field_name}", f"group_resource_group_by_{group_name}_{base_name}_{field_slug}", group_group_by_view, ["GET"], build_group_resource_group_by_spec(resource, field_name), ) def register_dynamic_endpoints(): global DYNAMIC_ENDPOINTS_REGISTERED if DYNAMIC_ENDPOINTS_REGISTERED: return ensure_registry_loaded() for group_name in sorted(GROUP_DESCRIPTIONS): register_group_resources_endpoint(group_name) for resource_name in sorted(RESOURCE_REGISTRY): resource = RESOURCE_REGISTRY[resource_name] register_meta_resource_endpoint(resource) register_resource_endpoints(resource) for catalog_name in sorted(CATALOG_REGISTRY): register_catalog_endpoint(CATALOG_REGISTRY[catalog_name]) DYNAMIC_ENDPOINTS_REGISTERED = True logger.info( "Rotas explícitas registradas com sucesso: %s recursos e %s catálogos.", len(RESOURCE_REGISTRY), len(CATALOG_REGISTRY), ) try: register_dynamic_endpoints() except Exception as exc: logger.warning( "Não foi possível registrar as rotas explícitas no startup: %s. " "As rotas fallback continuarão disponíveis.", exc, ) if __name__ == "__main__": if DEBUG: logger.debug("Iniciando API em modo DEBUG") app.run(port=5000, debug=True) else: app.run(debug=False)