Integração com Python

Retorno ilustrativoCopiar

{"id":"UUID_DA_EXECUCAO","status":"COMPLETED","arquivos":[{"id":"UUID_DO_ARQUIVO","path":"resultado-UUID.webp","bytes":12345}],"falhas":[]}

duckfile_integration.pyCopiar

import os, time, uuid, tempfilefrom pathlib import Pathfrom contextlib import ExitStackfrom urllib.parse import urlsplitimport requests # Exemplo para incorporar ao seu servidor; não é um SDK oficial.def executar_automacao(entrada, idempotency_key=None, output_dir='.'):    base = os.environ['DUCKFILE_API_URL'].rstrip('/')    url = urlsplit(base)    if url.scheme != 'https' or not url.hostname or url.username or url.password or url.query or url.fragment or not url.path.endswith('/api/v1/external'):        raise ValueError('Configure a base HTTPS da API.')    key = os.environ['DUCKFILE_API_KEY']    if not key or '\r' in key or '\n' in key:        raise ValueError('Configure a chave no servidor.')    automation = str(uuid.UUID(os.environ['DUCKFILE_AUTOMATION_ID']))    version = int(os.environ['DUCKFILE_AUTOMATION_VERSION'])    if version < 1:        raise ValueError('Informe a versão publicada no painel.')    paths = entrada if isinstance(entrada, (list, tuple)) else [entrada]    if not paths or len(paths) > 100:        raise ValueError('Informe um arquivo ou lote.')    types = {'.png': 'image/png', '.jpg': 'image/jpeg', '.jpeg': 'image/jpeg', '.webp': 'image/webp', '.avif': 'image/avif'}    request_id = idempotency_key or str(uuid.uuid4())    headers = {'Authorization': 'Bearer ' + key}    def checked(response):        if not 200 <= response.status_code < 300:            error = RuntimeError('API: HTTP ' + str(response.status_code) + '. Preserve a chave de idempotência em caso de reenvio.')            error.idempotency_key = request_id            raise error        return response    with ExitStack() as stack:        files, total = [], 0        for value in paths:            path = Path(value)            size, mime = path.stat().st_size, types.get(path.suffix.lower())            total += size            if not mime or not 0 < size <= 25000000 or total > 79000000:                raise ValueError('Use imagens de até 25 MB e respeite 80 MB com o formulário.')            files.append(('files', (path.name, stack.enter_context(path.open('rb')), mime)))        try:            response = requests.post(base + '/automations/' + automation + '/runs',                headers={**headers, 'Idempotency-Key': request_id},                data={'version': str(version)}, files=files, timeout=(10, 150), allow_redirects=False)        except requests.RequestException as error:            error.idempotency_key = request_id            raise        run = checked(response).json()['data']    run_id = str(uuid.UUID(run['id']))    deadline = time.monotonic() + 35 * 60    while run['status'] not in ['COMPLETED', 'PARTIAL', 'FAILED', 'CANCELLED', 'EXPIRED']:        if time.monotonic() >= deadline:            raise TimeoutError('Acompanhe a execução ' + run_id + ' antes de reenviar.')        time.sleep(2)        run = checked(requests.get(base + '/runs/' + run_id, headers=headers,            timeout=(10, 30), allow_redirects=False)).json()['data']    output = Path(output_dir)    output.mkdir(parents=True, exist_ok=True)    arquivos = []    for file in run.get('files', []):        result = file.get('result')        if not result:            continue        file_id = str(uuid.UUID(file['id']))        path = '/runs/' + run_id + '/files/' + file_id        if result['path'] != path:            raise ValueError('Caminho de resultado inválido.')        suffix = {'image/png': 'png', 'image/jpeg': 'jpg', 'image/webp': 'webp', 'image/avif': 'avif'}.get(result['mimeType'])        if not suffix:            raise ValueError('Formato de resultado inválido.')        if not 0 < int(result['bytes']) <= 100000000:            raise ValueError('Tamanho de resultado inválido.')        target = output / ('resultado-' + file_id + '.' + suffix)        descriptor, partial_name = tempfile.mkstemp(dir=output, prefix='duckfile-', suffix='.part')        partial = Path(partial_name)        size = 0        try:            with os.fdopen(descriptor, 'wb') as destination, requests.get(base + path, headers=headers, timeout=(10, 120), stream=True, allow_redirects=False) as response:                checked(response)                for chunk in response.iter_content(65536):                    size += len(chunk)                    if size > int(result['bytes']):                        raise ValueError('Tamanho de resultado inválido.')                    destination.write(chunk)            if size != int(result['bytes']):                raise ValueError('Arquivo recebido incompleto.')            os.link(partial, target)  # Falha se o destino já existir; não sobrescreve.        finally:            partial.unlink(missing_ok=True)        arquivos.append({'id': file_id, 'path': str(target), 'bytes': size})    if not arquivos:        raise RuntimeError('Nenhum arquivo disponível: ' + run['status'])    return {'id': run_id, 'status': run['status'], 'idempotencyKey': request_id, 'arquivos': arquivos,        'falhas': [{'id': f['id'], 'errorCode': f.get('errorCode')} for f in run.get('files', []) if f['status'] != 'COMPLETED' or f.get('partial')]} if __name__ == '__main__':    import sys, json    print(json.dumps(executar_automacao(sys.argv[1:]), ensure_ascii=False))