"""SDK oficial da API ImobyFlow — Python 3.9+, só biblioteca padrão (versão do contrato 1.0.0). Official ImobyFlow API SDK — Python 3.9+, standard library only. Gerado do mesmo contrato do portal: https://imobyflow.com.br/desenvolvedores import os from imobyflow import Client api = Client(os.environ["IMOBYFLOW_API_KEY"]) print(api.get_me()["data"]) for imovel in api.iter_list_properties(status="ACTIVE"): print(imovel["id"]) A chave é SÓ do servidor. / The key is SERVER-SIDE only. """ import base64 import hashlib import hmac import json import random import time import urllib.error import urllib.parse import urllib.request import uuid API_VERSION = "1.0.0" DEFAULT_BASE_URL = "https://api.imobyflow.com.br/v1" _RETRY_STATUS = {429, 502, 503, 504} _OPS = { "getMe": { "method": "GET", "path": "/me", "query": [], "dryRun": False, "idempotency": False }, "getPropertyTypes": { "method": "GET", "path": "/reference/property-types", "query": [], "dryRun": False, "idempotency": False }, "getPurposes": { "method": "GET", "path": "/reference/purposes", "query": [], "dryRun": False, "idempotency": False }, "getPropertyStatuses": { "method": "GET", "path": "/reference/property-statuses", "query": [], "dryRun": False, "idempotency": False }, "getLeadStatuses": { "method": "GET", "path": "/reference/lead-statuses", "query": [], "dryRun": False, "idempotency": False }, "getPartnershipStatuses": { "method": "GET", "path": "/reference/partnership-statuses", "query": [], "dryRun": False, "idempotency": False }, "getConstructionStages": { "method": "GET", "path": "/reference/construction-stages", "query": [], "dryRun": False, "idempotency": False }, "getScopes": { "method": "GET", "path": "/reference/scopes", "query": [], "dryRun": False, "idempotency": False }, "getEventTypes": { "method": "GET", "path": "/reference/event-types", "query": [], "dryRun": False, "idempotency": False }, "listCities": { "method": "GET", "path": "/geo/cities", "query": [ "uf", "q", "limit", "cursor" ], "dryRun": False, "idempotency": False }, "listNeighborhoods": { "method": "GET", "path": "/geo/cities/{cityId}/neighborhoods", "query": [ "limit", "cursor" ], "dryRun": False, "idempotency": False }, "listProperties": { "method": "GET", "path": "/properties", "query": [ "limit", "cursor", "status", "purpose", "type", "citySlug", "updatedSince", "radar" ], "dryRun": False, "idempotency": False }, "getPropertyByRef": { "method": "GET", "path": "/properties/by-ref/{ref}", "query": [], "dryRun": False, "idempotency": False }, "getProperty": { "method": "GET", "path": "/properties/{propertyId}", "query": [], "dryRun": False, "idempotency": False }, "listPropertyMatches": { "method": "GET", "path": "/properties/{propertyId}/matches", "query": [], "dryRun": False, "idempotency": False }, "listLeads": { "method": "GET", "path": "/leads", "query": [ "limit", "cursor", "state", "status", "ownerAccountId", "updatedSince", "radar" ], "dryRun": False, "idempotency": False }, "getLead": { "method": "GET", "path": "/leads/{leadId}", "query": [], "dryRun": False, "idempotency": False }, "getLeadByRef": { "method": "GET", "path": "/leads/by-ref/{ref}", "query": [], "dryRun": False, "idempotency": False }, "listLeadMatches": { "method": "GET", "path": "/leads/{leadId}/matches", "query": [], "dryRun": False, "idempotency": False }, "listMatches": { "method": "GET", "path": "/matches", "query": [ "limit", "cursor", "status", "scope", "ownerAccountId" ], "dryRun": False, "idempotency": False }, "listPartnerships": { "method": "GET", "path": "/partnerships", "query": [ "status", "updatedSince" ], "dryRun": False, "idempotency": False }, "getPartnership": { "method": "GET", "path": "/partnerships/{partnershipId}", "query": [], "dryRun": False, "idempotency": False }, "listTeam": { "method": "GET", "path": "/team", "query": [], "dryRun": False, "idempotency": False }, "getFeed": { "method": "GET", "path": "/feed", "query": [], "dryRun": False, "idempotency": False }, "syncFeed": { "method": "POST", "path": "/feed/sync", "query": [], "dryRun": True, "idempotency": True }, "listLaunches": { "method": "GET", "path": "/launches", "query": [ "limit", "cursor", "citySlug", "neighborhood", "type", "stage", "priceMin", "priceMax", "q" ], "dryRun": False, "idempotency": False }, "putPropertyByRef": { "method": "PUT", "path": "/properties/by-ref/{ref}", "query": [], "dryRun": True, "idempotency": True }, "createProperty": { "method": "POST", "path": "/properties", "query": [], "dryRun": True, "idempotency": True }, "patchProperty": { "method": "PATCH", "path": "/properties/{propertyId}", "query": [], "dryRun": True, "idempotency": True }, "setPropertyStatus": { "method": "POST", "path": "/properties/{propertyId}/status", "query": [], "dryRun": True, "idempotency": True }, "confirmProperties": { "method": "POST", "path": "/properties/confirm", "query": [], "dryRun": True, "idempotency": True }, "putLeadByRef": { "method": "PUT", "path": "/leads/by-ref/{ref}", "query": [], "dryRun": True, "idempotency": True }, "createLead": { "method": "POST", "path": "/leads", "query": [], "dryRun": True, "idempotency": True }, "patchLead": { "method": "PATCH", "path": "/leads/{leadId}", "query": [], "dryRun": True, "idempotency": True }, "setLeadState": { "method": "POST", "path": "/leads/{leadId}/state", "query": [], "dryRun": True, "idempotency": True }, "addLeadInteraction": { "method": "POST", "path": "/leads/{leadId}/interactions", "query": [], "dryRun": True, "idempotency": True }, "deleteProperty": { "method": "DELETE", "path": "/properties/{propertyId}", "query": [], "dryRun": True, "idempotency": True }, "deleteLead": { "method": "DELETE", "path": "/leads/{leadId}", "query": [], "dryRun": True, "idempotency": True }, "dismissMatch": { "method": "POST", "path": "/matches/{matchId}/dismiss", "query": [], "dryRun": True, "idempotency": True }, "restoreMatch": { "method": "POST", "path": "/matches/{matchId}/restore", "query": [], "dryRun": True, "idempotency": True }, "setPartnershipStatus": { "method": "POST", "path": "/partnerships/{partnershipId}/status", "query": [], "dryRun": True, "idempotency": True }, "listCaptacaoOffers": { "method": "GET", "path": "/captacao/offers", "query": [], "dryRun": False, "idempotency": False }, "getCaptacaoOffer": { "method": "GET", "path": "/captacao/offers/{offerId}", "query": [], "dryRun": False, "idempotency": False }, "listCaptacaoOwners": { "method": "GET", "path": "/captacao/owners", "query": [ "tab", "cursor" ], "dryRun": False, "idempotency": False }, "acceptCaptacaoOffer": { "method": "POST", "path": "/captacao/offers/{offerId}/accept", "query": [], "dryRun": True, "idempotency": True }, "listEvents": { "method": "GET", "path": "/events", "query": [ "limit", "cursor", "since", "types" ], "dryRun": False, "idempotency": False }, "listWebhooks": { "method": "GET", "path": "/webhooks", "query": [], "dryRun": False, "idempotency": False }, "createWebhook": { "method": "POST", "path": "/webhooks", "query": [], "dryRun": False, "idempotency": False }, "updateWebhook": { "method": "PATCH", "path": "/webhooks/{webhookId}", "query": [], "dryRun": False, "idempotency": True }, "deleteWebhook": { "method": "DELETE", "path": "/webhooks/{webhookId}", "query": [], "dryRun": False, "idempotency": True }, "rotateWebhookSecret": { "method": "POST", "path": "/webhooks/{webhookId}/rotate-secret", "query": [], "dryRun": False, "idempotency": False }, "testWebhook": { "method": "POST", "path": "/webhooks/{webhookId}/test", "query": [], "dryRun": False, "idempotency": True }, "listWebhookDeliveries": { "method": "GET", "path": "/webhooks/{webhookId}/deliveries", "query": [ "limit", "cursor" ], "dryRun": False, "idempotency": False }, "retryWebhookDelivery": { "method": "POST", "path": "/webhooks/{webhookId}/deliveries/{deliveryId}/retry", "query": [], "dryRun": False, "idempotency": True }, "createPropertyBatch": { "method": "POST", "path": "/properties/batch", "query": [], "dryRun": True, "idempotency": True }, "getPropertyBatch": { "method": "GET", "path": "/properties/batch/{batchId}", "query": [], "dryRun": False, "idempotency": False }, "createLeadBatch": { "method": "POST", "path": "/leads/batch", "query": [], "dryRun": True, "idempotency": True }, "getLeadBatch": { "method": "GET", "path": "/leads/batch/{batchId}", "query": [], "dryRun": False, "idempotency": False } } class ImobyFlowError(Exception): """O erro da API (application/problem+json). Decida pelo `code`, nunca pelo texto.""" def __init__(self, status, body, request_id=None): p = body if isinstance(body, dict) else {} self.status = status self.code = p.get("code") self.title = p.get("title") self.type = p.get("type") self.request_id = p.get("requestId") or request_id self.detail = p.get("detail") self.invalid_params = p.get("invalidParams") or [] self.body = body super().__init__(f"{status} {self.code or ''}: {self.title or ''}".strip()) class Client: """Cliente da API ImobyFlow. Novas tentativas seguras em 429, 502, 503, 504 e falha de rede.""" def __init__(self, api_key, *, base_url=DEFAULT_BASE_URL, language="pt-BR", timeout=30, max_retries=2, retry_base_delay=0.5): chave = (api_key or "").strip() if chave.lower().startswith("bearer "): chave = chave[7:].strip() if not chave or any(ch.isspace() for ch in chave): raise ValueError("ImobyFlow: chave da API ausente ou inválida.") self._key = chave self.base_url = base_url.rstrip("/") self.language = language self.timeout = timeout self.max_retries = max_retries self.retry_base_delay = retry_base_delay def _call(self, op, path, params, body=None, dry_run=False, idempotency_key=None): desconhecidos = sorted(set(params) - set(op["query"])) if desconhecidos: raise TypeError(f"ImobyFlow: filtro desconhecido nesta rota: {', '.join(desconhecidos)}") caminho = op["path"] for nome, valor in path.items(): if valor is None or str(valor) == "": raise ValueError(f"ImobyFlow: {nome} é obrigatório.") caminho = caminho.replace("{" + nome + "}", urllib.parse.quote(str(valor), safe="")) qs = [(k, str(v).lower() if isinstance(v, bool) else str(v)) for k, v in params.items() if v is not None and v != ""] if op["dryRun"] and dry_run: qs.append(("dryRun", "true")) url = self.base_url + caminho + ("?" + urllib.parse.urlencode(qs) if qs else "") headers = { "Authorization": f"Bearer {self._key}", "Accept": "application/json", "Accept-Language": self.language, "User-Agent": f"imobyflow-sdk-python/{API_VERSION}", } # A MESMA chave em todas as tentativas: a re-tentativa devolve a resposta da primeira. chave = (idempotency_key or str(uuid.uuid4())) if op["idempotency"] else None if chave: headers["Idempotency-Key"] = chave dados = None if body is not None: dados = json.dumps(body, ensure_ascii=False).encode("utf-8") headers["Content-Type"] = "application/json" pode_repetir = op["method"] == "GET" or bool(chave) tentativa = 0 while True: pedido = urllib.request.Request(url, data=dados, headers=headers, method=op["method"]) try: with urllib.request.urlopen(pedido, timeout=self.timeout) as resp: texto = resp.read().decode("utf-8") return json.loads(texto) if texto else None except urllib.error.HTTPError as erro: texto = erro.read().decode("utf-8", "replace") try: corpo = json.loads(texto) if texto else None except ValueError: corpo = texto if pode_repetir and erro.code in _RETRY_STATUS and tentativa < self.max_retries: depois = erro.headers.get("Retry-After") time.sleep(float(depois) if depois and depois.isdigit() else self.retry_base_delay * (2 ** tentativa) + random.random() * 0.01) tentativa += 1 continue raise ImobyFlowError(erro.code, corpo, erro.headers.get("X-Request-Id")) from None except (urllib.error.URLError, TimeoutError, ConnectionError): if pode_repetir and tentativa < self.max_retries: time.sleep(self.retry_base_delay * (2 ** tentativa)) tentativa += 1 continue raise def get_me(self): """Quem sou eu · Who am I GET /v1/me """ return self._call(_OPS["getMe"], {}, {}, None, False, None) def get_property_types(self): """Tipos de imóvel · Property types GET /v1/reference/property-types """ return self._call(_OPS["getPropertyTypes"], {}, {}, None, False, None) def get_purposes(self): """Finalidades · Purposes GET /v1/reference/purposes """ return self._call(_OPS["getPurposes"], {}, {}, None, False, None) def get_property_statuses(self): """Status do imóvel · Property statuses GET /v1/reference/property-statuses """ return self._call(_OPS["getPropertyStatuses"], {}, {}, None, False, None) def get_lead_statuses(self): """Etapas do cliente · Client stages GET /v1/reference/lead-statuses """ return self._call(_OPS["getLeadStatuses"], {}, {}, None, False, None) def get_partnership_statuses(self): """Etapas da parceria · Partnership stages GET /v1/reference/partnership-statuses """ return self._call(_OPS["getPartnershipStatuses"], {}, {}, None, False, None) def get_construction_stages(self): """Estágios da obra · Construction stages GET /v1/reference/construction-stages """ return self._call(_OPS["getConstructionStages"], {}, {}, None, False, None) def get_scopes(self): """Permissões · Permissions GET /v1/reference/scopes """ return self._call(_OPS["getScopes"], {}, {}, None, False, None) def get_event_types(self): """Tipos de evento · Event types GET /v1/reference/event-types """ return self._call(_OPS["getEventTypes"], {}, {}, None, False, None) def list_cities(self, **params): """Cidades · Cities GET /v1/geo/cities Filtros / filters: uf, q, limit, cursor """ return self._call(_OPS["listCities"], {}, params, None, False, None) def iter_list_cities(self, **params): """Todas as páginas de list_cities, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_cities(**params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def list_neighborhoods(self, city_id, **params): """Bairros de uma cidade · A city's neighborhoods GET /v1/geo/cities/{cityId}/neighborhoods Filtros / filters: limit, cursor """ return self._call(_OPS["listNeighborhoods"], {"cityId": city_id}, params, None, False, None) def iter_list_neighborhoods(self, city_id, **params): """Todas as páginas de list_neighborhoods, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_neighborhoods(city_id, **params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def list_properties(self, **params): """Listar a carteira · List the portfolio GET /v1/properties — permissão properties:read Filtros / filters: limit, cursor, status, purpose, type, citySlug, updatedSince, radar """ return self._call(_OPS["listProperties"], {}, params, None, False, None) def iter_list_properties(self, **params): """Todas as páginas de list_properties, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_properties(**params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def get_property_by_ref(self, ref): """Um imóvel pelo seu código · One property by your code GET /v1/properties/by-ref/{ref} — permissão properties:read """ return self._call(_OPS["getPropertyByRef"], {"ref": ref}, {}, None, False, None) def get_property(self, property_id): """Um imóvel · One property GET /v1/properties/{propertyId} — permissão properties:read """ return self._call(_OPS["getProperty"], {"propertyId": property_id}, {}, None, False, None) def list_property_matches(self, property_id): """Oportunidades de um imóvel · A property's opportunities GET /v1/properties/{propertyId}/matches — permissão radar:read """ return self._call(_OPS["listPropertyMatches"], {"propertyId": property_id}, {}, None, False, None) def list_leads(self, **params): """Listar clientes · List clients GET /v1/leads — permissão leads:read Filtros / filters: limit, cursor, state, status, ownerAccountId, updatedSince, radar """ return self._call(_OPS["listLeads"], {}, params, None, False, None) def iter_list_leads(self, **params): """Todas as páginas de list_leads, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_leads(**params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def get_lead(self, lead_id): """Um cliente · One client GET /v1/leads/{leadId} — permissão leads:read """ return self._call(_OPS["getLead"], {"leadId": lead_id}, {}, None, False, None) def get_lead_by_ref(self, ref): """Um cliente pelo seu código · One client by your code GET /v1/leads/by-ref/{ref} — permissão leads:read """ return self._call(_OPS["getLeadByRef"], {"ref": ref}, {}, None, False, None) def list_lead_matches(self, lead_id): """Oportunidades de um cliente · A client's opportunities GET /v1/leads/{leadId}/matches — permissão radar:read """ return self._call(_OPS["listLeadMatches"], {"leadId": lead_id}, {}, None, False, None) def list_matches(self, **params): """Oportunidades da conta · The account's opportunities GET /v1/matches — permissão radar:read Filtros / filters: limit, cursor, status, scope, ownerAccountId """ return self._call(_OPS["listMatches"], {}, params, None, False, None) def iter_list_matches(self, **params): """Todas as páginas de list_matches, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_matches(**params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def list_partnerships(self, **params): """Parcerias da equipe · The team's partnerships GET /v1/partnerships — permissão partnerships:read Filtros / filters: status, updatedSince """ return self._call(_OPS["listPartnerships"], {}, params, None, False, None) def get_partnership(self, partnership_id): """Uma parceria · One partnership GET /v1/partnerships/{partnershipId} — permissão partnerships:read """ return self._call(_OPS["getPartnership"], {"partnershipId": partnership_id}, {}, None, False, None) def list_team(self): """A equipe · The team GET /v1/team — permissão team:read """ return self._call(_OPS["listTeam"], {}, {}, None, False, None) def get_feed(self): """Estado da sincronização do feed · Feed sync status GET /v1/feed — permissão properties:read """ return self._call(_OPS["getFeed"], {}, {}, None, False, None) def sync_feed(self, *, dry_run=False, idempotency_key=None): """Sincronizar o feed agora · Sync the feed now POST /v1/feed/sync — permissão properties:write """ return self._call(_OPS["syncFeed"], {}, {}, None, dry_run, idempotency_key) def list_launches(self, **params): """Lançamentos da cidade · The city's new developments GET /v1/launches — permissão launches:read Filtros / filters: limit, cursor, citySlug, neighborhood, type, stage, priceMin, priceMax, q """ return self._call(_OPS["listLaunches"], {}, params, None, False, None) def iter_list_launches(self, **params): """Todas as páginas de list_launches, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_launches(**params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def put_property_by_ref(self, ref, body, *, dry_run=False, idempotency_key=None): """Criar ou atualizar pelo seu código · Create or update by your code PUT /v1/properties/by-ref/{ref} — permissão properties:write """ return self._call(_OPS["putPropertyByRef"], {"ref": ref}, {}, body, dry_run, idempotency_key) def create_property(self, body, *, dry_run=False, idempotency_key=None): """Cadastrar um imóvel · Create a property POST /v1/properties — permissão properties:write """ return self._call(_OPS["createProperty"], {}, {}, body, dry_run, idempotency_key) def patch_property(self, property_id, body, *, dry_run=False, idempotency_key=None): """Alterar parte de um imóvel · Partially update a property PATCH /v1/properties/{propertyId} — permissão properties:write """ return self._call(_OPS["patchProperty"], {"propertyId": property_id}, {}, body, dry_run, idempotency_key) def set_property_status(self, property_id, body, *, dry_run=False, idempotency_key=None): """Mudar o status · Change the status POST /v1/properties/{propertyId}/status — permissão properties:write """ return self._call(_OPS["setPropertyStatus"], {"propertyId": property_id}, {}, body, dry_run, idempotency_key) def confirm_properties(self, body, *, dry_run=False, idempotency_key=None): """Confirmar disponibilidade em lote · Confirm availability in bulk POST /v1/properties/confirm — permissão properties:write """ return self._call(_OPS["confirmProperties"], {}, {}, body, dry_run, idempotency_key) def put_lead_by_ref(self, ref, body, *, dry_run=False, idempotency_key=None): """Criar, atualizar ou vincular pelo seu código · Create, update or link by your code PUT /v1/leads/by-ref/{ref} — permissão leads:write """ return self._call(_OPS["putLeadByRef"], {"ref": ref}, {}, body, dry_run, idempotency_key) def create_lead(self, body, *, dry_run=False, idempotency_key=None): """Cadastrar um cliente · Create a client POST /v1/leads — permissão leads:write """ return self._call(_OPS["createLead"], {}, {}, body, dry_run, idempotency_key) def patch_lead(self, lead_id, body, *, dry_run=False, idempotency_key=None): """Alterar parte de um cliente · Partially update a client PATCH /v1/leads/{leadId} — permissão leads:write """ return self._call(_OPS["patchLead"], {"leadId": lead_id}, {}, body, dry_run, idempotency_key) def set_lead_state(self, lead_id, body, *, dry_run=False, idempotency_key=None): """Arquivar ou reativar · Archive or reactivate POST /v1/leads/{leadId}/state — permissão leads:write """ return self._call(_OPS["setLeadState"], {"leadId": lead_id}, {}, body, dry_run, idempotency_key) def add_lead_interaction(self, lead_id, body, *, dry_run=False, idempotency_key=None): """Registrar um atendimento · Log an interaction POST /v1/leads/{leadId}/interactions — permissão leads:write """ return self._call(_OPS["addLeadInteraction"], {"leadId": lead_id}, {}, body, dry_run, idempotency_key) def delete_property(self, property_id, *, dry_run=False, idempotency_key=None): """Mandar para a lixeira · Move to the trash DELETE /v1/properties/{propertyId} — permissão properties:write """ return self._call(_OPS["deleteProperty"], {"propertyId": property_id}, {}, None, dry_run, idempotency_key) def delete_lead(self, lead_id, *, dry_run=False, idempotency_key=None): """Mandar para a lixeira · Move to the trash DELETE /v1/leads/{leadId} — permissão leads:write """ return self._call(_OPS["deleteLead"], {"leadId": lead_id}, {}, None, dry_run, idempotency_key) def dismiss_match(self, match_id, *, dry_run=False, idempotency_key=None): """Descartar oportunidade · Dismiss an opportunity POST /v1/matches/{matchId}/dismiss — permissão radar:write """ return self._call(_OPS["dismissMatch"], {"matchId": match_id}, {}, None, dry_run, idempotency_key) def restore_match(self, match_id, *, dry_run=False, idempotency_key=None): """Restaurar oportunidade · Restore an opportunity POST /v1/matches/{matchId}/restore — permissão radar:write """ return self._call(_OPS["restoreMatch"], {"matchId": match_id}, {}, None, dry_run, idempotency_key) def set_partnership_status(self, partnership_id, body, *, dry_run=False, idempotency_key=None): """Mover a etapa da parceria · Move the partnership stage POST /v1/partnerships/{partnershipId}/status — permissão partnerships:write """ return self._call(_OPS["setPartnershipStatus"], {"partnershipId": partnership_id}, {}, body, dry_run, idempotency_key) def list_captacao_offers(self): """Proprietários na vez da conta · Owners offered to the account GET /v1/captacao/offers — permissão captacao:read """ return self._call(_OPS["listCaptacaoOffers"], {}, {}, None, False, None) def get_captacao_offer(self, offer_id): """Um proprietário · One owner GET /v1/captacao/offers/{offerId} — permissão captacao:read """ return self._call(_OPS["getCaptacaoOffer"], {"offerId": offer_id}, {}, None, False, None) def list_captacao_owners(self, **params): """Proprietários aceitos · Accepted owners GET /v1/captacao/owners — permissão captacao:read Filtros / filters: tab, cursor """ return self._call(_OPS["listCaptacaoOwners"], {}, params, None, False, None) def iter_list_captacao_owners(self, **params): """Todas as páginas de list_captacao_owners, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_captacao_owners(**params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def accept_captacao_offer(self, offer_id, *, dry_run=False, idempotency_key=None): """Aceitar um proprietário · Accept an owner POST /v1/captacao/offers/{offerId}/accept — permissão captacao:accept """ return self._call(_OPS["acceptCaptacaoOffer"], {"offerId": offer_id}, {}, None, dry_run, idempotency_key) def list_events(self, **params): """O que mudou · What changed GET /v1/events — permissão events:read Filtros / filters: limit, cursor, since, types """ return self._call(_OPS["listEvents"], {}, params, None, False, None) def iter_list_events(self, **params): """Todas as páginas de list_events, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_events(**params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def list_webhooks(self): """Endereços cadastrados · Registered endpoints GET /v1/webhooks — permissão webhooks:manage """ return self._call(_OPS["listWebhooks"], {}, {}, None, False, None) def create_webhook(self, body): """Cadastrar um endereço · Register an endpoint POST /v1/webhooks — permissão webhooks:manage """ return self._call(_OPS["createWebhook"], {}, {}, body, False, None) def update_webhook(self, webhook_id, body, *, idempotency_key=None): """Alterar um endereço · Update an endpoint PATCH /v1/webhooks/{webhookId} — permissão webhooks:manage """ return self._call(_OPS["updateWebhook"], {"webhookId": webhook_id}, {}, body, False, idempotency_key) def delete_webhook(self, webhook_id, *, idempotency_key=None): """Apagar um endereço · Delete an endpoint DELETE /v1/webhooks/{webhookId} — permissão webhooks:manage """ return self._call(_OPS["deleteWebhook"], {"webhookId": webhook_id}, {}, None, False, idempotency_key) def rotate_webhook_secret(self, webhook_id): """Trocar o segredo · Rotate the secret POST /v1/webhooks/{webhookId}/rotate-secret — permissão webhooks:manage """ return self._call(_OPS["rotateWebhookSecret"], {"webhookId": webhook_id}, {}, None, False, None) def test_webhook(self, webhook_id, *, idempotency_key=None): """Mandar um evento de teste · Send a test event POST /v1/webhooks/{webhookId}/test — permissão webhooks:manage """ return self._call(_OPS["testWebhook"], {"webhookId": webhook_id}, {}, None, False, idempotency_key) def list_webhook_deliveries(self, webhook_id, **params): """Entregas de um endereço · An endpoint's deliveries GET /v1/webhooks/{webhookId}/deliveries — permissão webhooks:manage Filtros / filters: limit, cursor """ return self._call(_OPS["listWebhookDeliveries"], {"webhookId": webhook_id}, params, None, False, None) def iter_list_webhook_deliveries(self, webhook_id, **params): """Todas as páginas de list_webhook_deliveries, item a item · every page, item by item.""" params.pop("cursor", None) vistos = set() cursor = None while True: if cursor: params["cursor"] = cursor pagina = self.list_webhook_deliveries(webhook_id, **params) for item in pagina.get("data") or []: yield item cursor = pagina.get("nextCursor") if not cursor or cursor in vistos: # cursor repetido: para em vez de rodar para sempre return vistos.add(cursor) def retry_webhook_delivery(self, webhook_id, delivery_id, *, idempotency_key=None): """Reenviar uma entrega · Redeliver POST /v1/webhooks/{webhookId}/deliveries/{deliveryId}/retry — permissão webhooks:manage """ return self._call(_OPS["retryWebhookDelivery"], {"webhookId": webhook_id, "deliveryId": delivery_id}, {}, None, False, idempotency_key) def create_property_batch(self, body, *, dry_run=False, idempotency_key=None): """Lote de imóveis · Batch of properties POST /v1/properties/batch — permissão properties:write """ return self._call(_OPS["createPropertyBatch"], {}, {}, body, dry_run, idempotency_key) def get_property_batch(self, batch_id): """Acompanhar o lote · Track the batch GET /v1/properties/batch/{batchId} — permissão properties:read """ return self._call(_OPS["getPropertyBatch"], {"batchId": batch_id}, {}, None, False, None) def create_lead_batch(self, body, *, dry_run=False, idempotency_key=None): """Lote de clientes · Batch of clients POST /v1/leads/batch — permissão leads:write """ return self._call(_OPS["createLeadBatch"], {}, {}, body, dry_run, idempotency_key) def get_lead_batch(self, batch_id): """Acompanhar o lote · Track the batch GET /v1/leads/batch/{batchId} — permissão leads:read """ return self._call(_OPS["getLeadBatch"], {"batchId": batch_id}, {}, None, False, None) def verify_webhook(secret, msg_id, timestamp, signature_header, raw_body): """Confere a assinatura de um webhook (Standard Webhooks). raw_body: os BYTES do corpo como chegaram (Flask: request.get_data(); Django: request.body). Aceita qualquer uma das assinaturas (na troca de segredo vêm duas) e recusa mensagem com mais de 5 minutos. """ if not timestamp.isdigit() or abs(time.time() - int(timestamp)) > 5 * 60: return False if isinstance(raw_body, str): raw_body = raw_body.encode("utf-8") key = base64.b64decode(secret[len("whsec_"):]) signed = f"{msg_id}.{timestamp}.".encode() + raw_body expected = base64.b64encode(hmac.new(key, signed, hashlib.sha256).digest()).decode() for part in signature_header.split(" "): version, _, sig = part.partition(",") if version == "v1" and hmac.compare_digest(sig, expected): return True return False