#!/usr/bin/env python3 """Doomio API v1 client. Python 3.10+, standard library only; see /developers.""" import argparse import json import mimetypes import os from pathlib import Path import secrets import sys import time from urllib.error import HTTPError from urllib.parse import urlencode, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): # Never send an API credential to a redirect target. return None class Doomio: def __init__(self, base, key): parts = urlsplit(base) if parts.scheme != 'https' or not parts.hostname or parts.username or parts.password or parts.query or parts.fragment: raise ValueError('API base URL must be HTTPS without credentials, query or fragment.') if not key.startswith('doomio_') or len(key) != 71: raise ValueError('A valid DOOMIO_API_KEY is required.') self.base = base.rstrip('/') + '/api/v1' if not base.rstrip('/').endswith('/api/v1') else base.rstrip('/') self.key = key self.http = build_opener(NoRedirect) def request(self, method, path, data=None, content_type='application/json', etag=None): headers = {'Authorization': 'Bearer ' + self.key, 'Accept': 'application/json', 'User-Agent': 'doomio-python-client/1.0'} if data is not None: headers['Content-Type'] = content_type if etag: headers['If-Match'] = etag body = data if isinstance(data, bytes) else (json.dumps(data, ensure_ascii=False).encode('utf-8') if data is not None else None) for attempt in range(3): try: with self.http.open(Request(self.base + path, data=body, method=method, headers=headers), timeout=90) as response: return json.load(response), response.headers.get('ETag') except HTTPError as error: if error.code == 429 and attempt < 2: time.sleep(min(120, max(1, int(error.headers.get('Retry-After', '60'))))) continue detail = error.read(65536).decode('utf-8', 'replace') raise RuntimeError(f'HTTP {error.code}; request ID {error.headers.get("X-Request-ID", "unknown")}: {detail}') from None def listing(self, external_id): response, _ = self.request('GET', '/properties?' + urlencode({'externalId': external_id})) return response['data'][0] if response['data'] else None def sync(self, records): if not isinstance(records, list) or not all(isinstance(item, dict) and isinstance(item.get('externalId'), str) for item in records): raise ValueError('Input must be an array of listing objects with externalId.') ids = [item['externalId'] for item in records] if len(ids) != len(set(ids)): raise ValueError('Input contains duplicate external IDs.') results = [] for item in records: existing = self.listing(item['externalId']) if existing is None: response, _ = self.request('POST', '/properties', item) else: current, etag = self.request('GET', '/properties/' + str(existing['id'])) changes = {k: v for k, v in item.items() if k != 'externalId' and current['data'].get(k) != v} response, _ = self.request('PATCH', '/properties/' + str(existing['id']), changes, etag=etag) if changes else (current, etag) results.append({'externalId': item['externalId'], 'id': response['data']['id'], 'status': response['data']['status']}) return results def upload(self, property_id, file): path = Path(file) if path.stat().st_size > 8 * 1024 * 1024: raise ValueError('Maximum image size is 8 MiB.') mime = mimetypes.guess_type(path.name)[0] if mime not in ('image/jpeg', 'image/png', 'image/webp'): raise ValueError('Use a JPEG, PNG or WebP file.') boundary = 'doomio-' + secrets.token_hex(24) body = (f'--{boundary}\r\nContent-Disposition: form-data; name="file"; filename="image{path.suffix}"\r\nContent-Type: {mime}\r\n\r\n').encode() + path.read_bytes() + f'\r\n--{boundary}--\r\n'.encode() return self.request('POST', f'/properties/{property_id}/images', body, 'multipart/form-data; boundary=' + boundary)[0] def settings(file): values = {} if file: for line in Path(file).read_text(encoding='utf-8').splitlines(): if not line.strip() or line.lstrip().startswith('#'): continue name, sep, value = line.partition('=') if sep and name.strip() in ('DOOMIO_API_BASE_URL', 'DOOMIO_API_KEY'): values[name.strip()] = value.strip() return os.environ.get('DOOMIO_API_BASE_URL', values.get('DOOMIO_API_BASE_URL', 'https://doomio.cz')), os.environ.get('DOOMIO_API_KEY', values.get('DOOMIO_API_KEY', '')) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--env', help='Path to the downloaded key .env file (not executed as shell code).') sub = parser.add_subparsers(dest='command', required=True) sub.add_parser('list') sync = sub.add_parser('sync'); sync.add_argument('file', help='JSON array of listing objects.') for command in ('upload', 'publish', 'withdraw'): arg = sub.add_parser(command); arg.add_argument('id', type=int) if command == 'upload': arg.add_argument('file') args = parser.parse_args() try: api = Doomio(*settings(args.env)) if args.command == 'sync': result = api.sync(json.loads(Path(args.file).read_text(encoding='utf-8'))) elif args.command == 'upload': result = api.upload(args.id, args.file) elif args.command in ('publish', 'withdraw'): result = api.request('POST' if args.command == 'publish' else 'DELETE', f'/properties/{args.id}' + ('/publish' if args.command == 'publish' else ''))[0] else: result = []; after = 0 while True: page, _ = api.request('GET', '/properties?' + urlencode({'after': after, 'limit': 100})) result.extend(page['data']); after = page['pagination']['nextAfter'] if after is None: break print(json.dumps(result, ensure_ascii=False, indent=2)) except (ValueError, OSError, RuntimeError) as error: print(str(error), file=sys.stderr) return 1 return 0 if __name__ == '__main__': sys.exit(main())