#!/usr/bin/env python3 """Senden API, Python >=3.10 standard library. Default command: discovery only. No automatic retries. Keys stay in environment; do not put them in prompts/logs. """ import json import os from pathlib import Path import urllib.error import urllib.parse import urllib.request OPERATIONS = json.loads("{\"account.settings.get\":false,\"chats.archive\":true,\"chats.expiration\":true,\"chats.reactions\":false,\"chats.typing\":true,\"chats.unarchive\":true,\"contacts.create\":true,\"contacts.delete\":true,\"contacts.get\":false,\"contacts.list\":false,\"contacts.lookup\":false,\"contacts.update\":true,\"groups.admins.add\":true,\"groups.admins.remove\":true,\"groups.create\":true,\"groups.get\":false,\"groups.leave\":true,\"groups.members.add\":true,\"groups.members.remove\":true,\"groups.rename\":true,\"groups.settings\":true,\"messages.contact\":true,\"messages.delete\":true,\"messages.edit\":true,\"messages.forward\":true,\"messages.get\":false,\"messages.history\":false,\"messages.location\":true,\"messages.poll\":true,\"messages.react\":true,\"messages.read\":true,\"profile.avatar.get\":false}") MAX_FILE = 20971520 MAX_RESPONSE = 32 * 1024 * 1024 def _segment(value): if not isinstance(value, str) or not value.strip() or len(value) > 200: raise ValueError('A nonempty resource ID is required') return urllib.parse.quote(value, safe='') def _key(value): if not isinstance(value, str) or not 8 <= len(value.strip()) <= 128 or any(c in value for c in '\r\n\0'): raise ValueError('Supply an explicit stable Idempotency-Key (8..128 characters)') return value class SendenApiError(Exception): def __init__(self, status, code, message): super().__init__(message) self.status, self.code = status, code class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None class SendenClient: def __init__(self, base=None, api_key=None, timeout_seconds=30): value = base or os.environ.get('SENDEN_BASE_URL', "https://api-senden.ru/api/v1") parsed = urllib.parse.urlsplit(value) if parsed.scheme not in ('https', 'http') or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ('', '/', '/api/v1', '/api/v1/'): raise ValueError('Use a canonical HTTPS origin or /api/v1 base') if parsed.scheme == 'http' and parsed.hostname not in ('127.0.0.1', 'localhost', '::1'): raise ValueError('HTTP is allowed only on loopback') self.base = urllib.parse.urlunsplit((parsed.scheme, parsed.netloc, '/api/v1', '', '')) self._key = api_key or os.environ.get('SENDEN_API_KEY', '') if not self._key.strip() or any(c in self._key for c in '\r\n\0'): raise ValueError('Set SENDEN_API_KEY to your named Senden key') if not isinstance(timeout_seconds, (int, float)) or not 1 <= timeout_seconds <= 120: raise ValueError('timeout_seconds must be 1..120') self.timeout = timeout_seconds self.opener = urllib.request.build_opener(_NoRedirect()) def request(self, method, path, body=None, data=None, headers=None, binary=False): if not path.startswith('/') or path.startswith('//') or '..' in path: raise ValueError('Use a relative Senden API path') supplied = {key: value for key, value in (headers or {}).items() if key.lower() != 'authorization'} supplied['Authorization'] = 'Bearer ' + self._key if body is not None: data = json.dumps(body, ensure_ascii=False).encode('utf-8') supplied['Content-Type'] = 'application/json' request = urllib.request.Request(self.base + path, data=data, headers=supplied, method=method) try: response = self.opener.open(request, timeout=self.timeout) except urllib.error.HTTPError as error: response = error with response: raw = response.read(MAX_RESPONSE + 1) if len(raw) > MAX_RESPONSE: raise ValueError('Senden response exceeds the client limit') status = response.code if binary and 200 <= status < 300: return raw try: result = json.loads(raw) if raw else None except (ValueError, UnicodeDecodeError): raise SendenApiError(status, 'invalid_response', 'The server did not return JSON') from None if not 200 <= status < 300: error = result.get('error', {}) if isinstance(result, dict) else {} raise SendenApiError(status, error.get('code', 'http_error'), error.get('message', 'Senden request failed')) return result def list_instances(self): return self.request('GET', '/instances') def get_manifest(self): return self.request('GET', '/integrations/manifest') def get_commands(self): return self.get_manifest()['commands'] def get_connected_messengers(self): instances = [{name: item.get(name) for name in ('id', 'name', 'messenger', 'state')} for item in self.list_instances()['instances'] if item.get('state') == 'connected'] return {'instances': instances, 'messengers': sorted({item['messenger'] for item in instances})} def get_methods(self): return self.request('GET', '/methods') def get_chats(self, instance_id): return self.request('GET', '/instances/' + _segment(instance_id) + '/chats') def get_history(self, instance_id, limit=50, cursor=None, chat_id=None): query = {'limit': limit} if cursor: query['cursor'] = cursor if chat_id: query['chatId'] = chat_id return self.request('GET', '/instances/' + _segment(instance_id) + '/messages?' + urllib.parse.urlencode(query)) def get_queue(self, instance_id): return self.request('GET', '/instances/' + _segment(instance_id) + '/queue') def send_text(self, instance_id, to, text, idempotency_key): _key(idempotency_key); _segment(to) if not isinstance(text, str) or not text.strip() or len(text) > 16000: raise ValueError('text must be nonempty and at most 16000 characters') return self.request('POST', '/instances/' + _segment(instance_id) + '/messages', body={'to': to, 'text': text}, headers={'Idempotency-Key': idempotency_key}) def send_file(self, instance_id, to, file_id, caption, idempotency_key): _key(idempotency_key); _segment(to); _segment(file_id) if not isinstance(caption, str) or len(caption) > 16000: raise ValueError('caption must be a string of at most 16000 characters') return self.request('POST', '/instances/' + _segment(instance_id) + '/messages', body={'to': to, 'text': caption, 'fileId': file_id}, headers={'Idempotency-Key': idempotency_key}) def upload_file(self, path, name=None, mime='application/octet-stream'): path = Path(path); info = path.stat(); name = name or path.name if not path.is_file() or not 1 <= info.st_size <= MAX_FILE: raise ValueError('Choose a nonempty file <=20 MiB') if not isinstance(name, str) or not name or len(name) > 180 or any(c in name for c in '\r\n\0/\\'): raise ValueError('Use a plain filename <=180 characters') data = path.read_bytes() if len(data) != info.st_size: raise ValueError('File changed while reading') return self.request('POST', '/files', data=data, headers={'Content-Type': 'application/octet-stream', 'X-Filename': urllib.parse.quote(name, safe=''), 'X-File-Mime': mime}) def download_file(self, file_id): return self.request('GET', '/files/' + _segment(file_id), binary=True) def operation(self, instance_id, operation_id, args=None, idempotency_key=None): if operation_id not in OPERATIONS: raise ValueError('Unknown operation; consult GET /methods') headers = {'Idempotency-Key': _key(idempotency_key)} if OPERATIONS[operation_id] else {} return self.request('POST', '/instances/' + _segment(instance_id) + '/operations/' + operation_id, body={} if args is None else args, headers=headers) def get_native_history(self, instance_id, chat_id, count=50): return self.operation(instance_id, 'messages.history', {'chatId': chat_id, 'count': count}) if __name__ == '__main__': client = SendenClient() manifest = client.get_manifest() instances = [{key: item.get(key) for key in ('id', 'name', 'messenger', 'state', 'connected', 'supportedCommands')} for item in manifest['instances']] print(json.dumps({'baseUrl': manifest.get('baseUrl'), 'permissions': manifest.get('permissions'), 'connectedMessengers': manifest.get('connectedMessengers'), 'instances': instances, 'commands': manifest['commands']}, ensure_ascii=False, indent=2))