"""RuralZap: Python 3.10+, somente biblioteca padrão, execução no servidor. python read-transactions.py 2026-01-01 2026-01-31 Configure RURALZAP_BASE_URL e RURALZAP_API_KEY no ambiente privado. O exemplo valida dados e imprime somente a contagem; não persiste lançamentos. """ import json import os import random import re import sys import time from datetime import date from decimal import Decimal from urllib.error import HTTPError, URLError from urllib.parse import urlencode, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener MAX_ATTEMPTS = 4 MAX_PAGES = 10_000 TIMEOUT_SECONDS = 30 MONEY = re.compile(r"^-?\d+\.\d{4}$") class IntegrationError(Exception): """Mensagens controladas, sem credenciais ou respostas financeiras.""" class NoRedirects(HTTPRedirectHandler): def redirect_request(self, request, response, code, message, headers, new_url): # Não encaminhar a chave para uma origem indicada por redirecionamento. return None def configuration(): origin = os.environ.get("RURALZAP_BASE_URL", "") key = os.environ.get("RURALZAP_API_KEY", "") if not origin or not key: raise IntegrationError("Configure RURALZAP_BASE_URL e RURALZAP_API_KEY.") try: parsed = urlsplit(origin) local = parsed.hostname in {"localhost", "127.0.0.1", "::1"} valid_transport = parsed.scheme == "https" or (local and parsed.scheme == "http") valid_origin = parsed.hostname and not ( parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in {"", "/"} ) except ValueError: raise IntegrationError("RURALZAP_BASE_URL deve ser uma origem válida.") from None if not valid_transport or not valid_origin: raise IntegrationError("Use uma origem HTTPS sem caminho, credenciais ou query. HTTP só é aceito em localhost.") if len(sys.argv) != 3: raise IntegrationError("Uso: python read-transactions.py YYYY-MM-DD YYYY-MM-DD") start_date, end_date = sys.argv[1:] try: if not all(re.fullmatch(r"\d{4}-\d{2}-\d{2}", value) for value in (start_date, end_date)): raise ValueError() days = (date.fromisoformat(end_date) - date.fromisoformat(start_date)).days + 1 except ValueError: raise IntegrationError("Use datas válidas no formato YYYY-MM-DD.") from None if not 1 <= days <= 366: raise IntegrationError("O período deve ser ordenado e ter até 366 dias inclusivos.") return origin.rstrip("/"), key, start_date, end_date def parse_page(value): if not isinstance(value, dict) or not isinstance(value.get("data"), list) or not isinstance(value.get("meta"), dict): raise IntegrationError("Resposta incompatível com o contrato de paginação.") has_more = value["meta"].get("hasMore") next_cursor = value["meta"].get("nextCursor") if type(has_more) is not bool or (next_cursor is not None and not isinstance(next_cursor, str)) or (has_more and not next_cursor): raise IntegrationError("Metadados de paginação inválidos.") rows = [] for row in value["data"]: if not isinstance(row, dict): raise IntegrationError("Lançamento inválido.") identifier, amount, currency, financial_date = (row.get(field) for field in ("id", "amount", "currency", "date")) if not ( isinstance(identifier, str) and identifier and isinstance(amount, str) and MONEY.fullmatch(amount) and isinstance(currency, str) and re.fullmatch(r"[A-Z]{3}", currency) and isinstance(financial_date, str) and re.fullmatch(r"\d{4}-\d{2}-\d{2}", financial_date) ): raise IntegrationError("Lançamento incompatível com o contrato esperado.") try: date.fromisoformat(financial_date) except ValueError: raise IntegrationError("Data financeira inválida na resposta.") from None # Decimal recebe a string original: nunca passar por float. rows.append({"id": identifier, "amount": Decimal(amount), "currency": currency, "date": financial_date}) return rows, has_more, next_cursor def retry_delay(error, attempt): raw = error.headers.get("Retry-After") if raw is not None and re.fullmatch(r"\d+", raw): delay = int(raw) else: delay = 60 if error.code == 429 else 2 ** attempt if delay > 120: raise IntegrationError("Servidor pediu espera longa. Reexecute depois do prazo de Retry-After.") return delay + random.uniform(0, 0.25) def fetch_page(opener, url, key): for attempt in range(MAX_ATTEMPTS): request = Request(url, headers={"Authorization": f"Bearer {key}", "Accept": "application/json"}, method="GET") try: with opener.open(request, timeout=TIMEOUT_SECONDS) as response: return parse_page(json.load(response)) except HTTPError as error: # Nunca ler ou registrar body, URL ou headers que contenham credenciais. request_id = error.headers.get("X-Request-Id", "") safe_id = f"; requestId={request_id}" if re.fullmatch(r"[a-fA-F0-9-]{36}", request_id) else "" error.close() if error.code in {429, 503} and attempt < MAX_ATTEMPTS - 1: time.sleep(retry_delay(error, attempt)) continue raise IntegrationError(f"HTTP {error.code}{safe_id}. Confira o guia de erros; a execução foi encerrada.") from None except (URLError, TimeoutError): if attempt == MAX_ATTEMPTS - 1: raise IntegrationError("Falha de conexão ou timeout após tentativas limitadas.") from None time.sleep(2 ** attempt + random.uniform(0, 0.25)) raise IntegrationError("Limite de tentativas atingido.") def main(): origin, key, start_date, end_date = configuration() parameters = {"startDate": start_date, "endDate": end_date, "limit": "100"} opener = build_opener(NoRedirects()) cursors = set() count = 0 for page_number in range(1, MAX_PAGES + 1): url = f"{origin}/api/v1/integration/transactions?{urlencode(parameters)}" rows, has_more, next_cursor = fetch_page(opener, url, key) for transaction in rows: # Integração real: upsert pelo transaction["id"] no seu armazenamento. # Para salvar amount: coluna DECIMAL(…,4) ou format(transaction["amount"], ".4f"). # Este exemplo não grava nem imprime dados financeiros. count += 1 if not has_more: print(f"Consulta concluída: {count} lançamentos recebidos em {page_number} página(s).") return if not next_cursor or next_cursor in cursors: raise IntegrationError("Cursor ausente ou repetido; consulta interrompida.") cursors.add(next_cursor) parameters["cursor"] = next_cursor raise IntegrationError("Limite local de páginas atingido. Divida a consulta em períodos menores.") if __name__ == "__main__": try: # Releia o período para conciliar alterações/exclusões por id; não é feed # incremental nem snapshot. Para receitas/saldos, use os relatórios canônicos. main() except IntegrationError as error: print(str(error), file=sys.stderr) sys.exit(1) except Exception: print("Falha local ao processar a resposta. Nenhum payload foi registrado.", file=sys.stderr) sys.exit(1)