Enviar Imagens diretamente para o Storage

Criada por Alexandre Pateis, Modificado em Qui, 2 Out na (o) 3:41 PM por Alexandre Pateis

Para grandes volumes de imagens e arquivos, é possível realizar o envio diretamente ao storage.

Antes do envio, é necessário que exista um pedido registrado no sistema. Para isso, você pode utilizar um ID de pedido já existente ou criar um novo por meio da API.

Ao criar um pedido, é obrigatório informar:

ID da clínica;
ID do status do pedido.

O sistema não aceitará requisições sem esses parâmetros ou caso os IDs informados não correspondam, respectivamente, a uma clínica válida e a um status de pedido pertencente ao usuário.

Com o ID do pedido e as informações de metadata dos arquivos, é possível solicitar ao servidor os dados necessários para o envio ao storage.

A requisição deve ser feita para a URL:

https://max.cfaz.net/api/v1/photos/signed_url

O corpo da requisição deve conter os seguintes parâmetros:

request_id → ID do pedido ao qual as imagens serão vinculadas.
file_name → O nome do arquivo.
file_size → Tamanho do arquivo em bytes.
content_type → Tipo do arquivo (ex.: image/jpeg).
checksum → Soma de verificação do arquivo (MD5 codificado em Base64).
width_px → Largura da imagem, em pixels.
height_px → Altura da imagem, em pixels.

É possível armazenar uma imagem associada a um arquivo DCM 2D. Para isso basta informar na mesma requisição as seguintes informações do arquivo DCM:

file_name_dicom → O nome do arquivo.
file_size_dicom → Tamanho do arquivo em bytes.
content_type_dicom → Tipo do arquivo (ex.: application/dicom).
checksum_dicom → Soma de verificação do arquivo (MD5 codificado em Base64).

Cálculo do checksum:

O cálculo do checksum deve ser realizado da seguinte maneira:

Ler o conteúdo binário do arquivo (não em texto, mas os bytes)
Calcular o hash MD5 desse conteúdo.
Codificar o resultado em Base64 (não em hexadecimal).
Enviar esse valor como checksum.

Exemplo com CURL:

curl -X POST "https://max.cfaz.net/api/v1/photos/signed_url" \
  -H "Authorization: Bearer 62ed05b2bd52b20e5a1eff01a0b862e6" \
  -d "request_id=300196" \
  -d "file_name=arquivo_1759424637008.jpg" \
  -d "file_size=99800" \
  -d "checksum=9giJyLnPMghVVGbUCSgTBw==" \
  -d "width_px=512" \
  -d "height_px=512" \
  -d "content_type=image/jpeg" \
  -d "file_name_dicom=arquivo" \
  -d "file_size_dicom=529058" \
  -d "checksum_dicom=KPnqX2hT1xqxyvR6BvaPjA==" \
  -d "content_type_dicom=application/dicom"

O retorno esperado é:

{
  "id": 875413,
  "headers": {
    "Content-MD5": "9giJyLnPMghVVGbUCSgTBw==",
    "Content-Disposition": "inline; filename=\"arquivo_1759424637008.jpg\"; filename*=UTF-8''arquivo_1759424637008.jpg",
    "Cache-Control": "public, max-age=31536000, immutable"
  },
  "signed_id": "eyJfcmFpbHMiOnsiZGF0YSI6MTI4MTUyNSwicHVyIjoiYmxvYl9pZCJ9fQ==--0f6f0c5a19f18be089f3192b40eb368beb296c12",
  "signed_url": "https://storage.googleapis.com",
  "signed_url_medium": "https://storage.googleapis.com",
  "download_url": "https://storage.googleapis.com",
  "signed_id_dicom": "eyJfcmFpbHMiOnsiZGF0YSI6MTI4MTUyNiwicHVyIjoiYmxvYl9pZCJ9fQ==--94959db76fee9abb2f2694340ac76fa4fb2e3b02",
  "headers_dicom": {
    "Content-MD5": "KPnqX2hT1xqxyvR6BvaPjA==",
    "Content-Disposition": "inline; filename=\"arquivo\"; filename*=UTF-8''arquivo",
    "Cache-Control": "public, max-age=31536000, immutable"
  },
  "signed_url_dicom": "https://storage.googleapis.com"
}

Cada campo contém as seguintes informações:

ID → Identificador único da imagem dentro do sistema.
headers → Cabeçalhos HTTP a serem utilizados ao enviar a imagem para o storage.
signed_id → Identificador criptograficamente assinado para um ActiveStorage::Blob ou ActiveStorage::Attachment, referente à imagem.
signed_url → URL para envio da imagem principal.
signed_url_medium → URL para envio da miniatura da imagem, com dimensões padrão de 270 x 270 pixels.

Quando enviada a imagem juntamente com o arquivo Dicom, são retornadas as seguintes informações na mesma requisição:

signed_id_dicom → Identificador criptograficamente assinado para um ActiveStorage::Blob ou ActiveStorage::Attachment, referente ao arquivo DICOM.
headers_dicom → Cabeçalhos HTTP a serem utilizados ao enviar os dados do arquivo DICOM para o storage.
signed_url_dicom → URL para envio do arquivo DICOM.

Com essas informações é preciso enviar os arquivos ao Storage:

Exemplo com CURL:

curl -X PUT "https://storage.googleapis.com/..." \
  -H "Content-MD5: 9giJyLnPMghVVGbUCSgTBw==" \
  -H "Content-Disposition: inline; filename=\"arquivo_1759429052156.jpg\"; filename*=UTF-8''arquivo_1759429052156.jpg" \
  -H "Cache-Control: public, max-age=31536000, immutable" \
  --upload-file "arquivo_1759429052156.jpg"

Após a conclusão do envio, é necessário atualizar o sistema associando o signed_id à respectiva imagem. Para permitir que o sistema encontre a imagem corretamente.

Exemplo com CURL:

curl -X put https://max.cfaz.net/api/v1/photos/{photo_id}\
  -H "Authorization: Bearer 62ed05b2bd52b20e5a1eff01a0b862e6" \
  -d "photo[id]={photo_id}" \
  -d "photo[image]=eyJfcmFpbHMiOnsiZGF0YSI6MTI4MTUyNywicHVyIjoiYmxvYl9pZCJ9fQ==--759c91540c706d7e6ed4afe69e034e69fdfeb224" \
  -d "photo[image_processing]=2"

Exemplo funcional com Python:

import os
import time
import subprocess
import hashlib
import base64
from datetime import datetime, timezone

import requests
from PIL import Image
import pydicom
import io

# ====== Configurações ======
CFAZ_URL = "https://max.cfaz.net/api/v1/"
TOKEN = ""
HEADERS = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {TOKEN}"
}

# ====== Utilitários ======

def calcular_checksum(caminho_arquivo: str) -> str:
    md5 = hashlib.md5()
    with open(caminho_arquivo, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            md5.update(chunk)
    md5_bytes = bytes.fromhex(md5.hexdigest())
    return base64.b64encode(md5_bytes).decode("ascii")

# Função para extrair informações de imagens (PNG, JPG, etc.) e do arquivo Dcm
def info_arquivo(caminho_arquivo: str, tipo="image") -> dict:
    tamanho = os.path.getsize(caminho_arquivo) # bytes
    checksum = calcular_checksum(caminho_arquivo)
    if tipo == "image":
        with Image.open(caminho_arquivo) as img:
            largura, altura = img.size
            content_type = Image.MIME.get(img.format) # exemplo: "image/png"
        return {
            "file_name": os.path.basename(caminho_arquivo),
            "file_size": tamanho,
            "checksum": checksum,
            "width_px": largura,
            "height_px": altura,
            "content_type": content_type
        }
    elif tipo == "dicom":
        return {
            "file_name_dicom": os.path.basename(caminho_arquivo),
            "file_size_dicom": tamanho,
            "checksum_dicom": checksum,
            "content_type_dicom": "application/dicom"
        }

def convert_dicom_para_jpg(caminho_arquivo: str) -> str:
    # Converter o arquivo Dicom em uma imagem utilizando DCMTK
    diretorio = os.path.dirname(caminho_arquivo)
    nome_arquivo = os.path.basename(caminho_arquivo)
    timestamp_ms = int(time.time() * 1000)
    jpg_file = f"{diretorio}/{nome_arquivo}_{timestamp_ms}.jpg"

    command = [
        "dcmj2pnm", "+Wm", "+Wn", "--write-jpeg", "--compr-quality", "100",
        caminho_arquivo, jpg_file
    ]
    try:
        subprocess.run(command, check=True)
        print(f"[OK] Arquivo convertido para JPG: {jpg_file}")
        return jpg_file
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Erro ao converter DICOM para JPG: {e}")

# Criar miniatura da imagem
def criar_miniatura_da_imagem(caminho_arquivo: str) -> bytes:
    try:
        img = Image.open(caminho_arquivo)
        nome_arquivo = os.path.basename(caminho_arquivo)
        largura, altura = img.size

        # Calcula nova largura/altura mantendo proporção
        if altura > largura:
            nova_altura = 270
            nova_largura = int(270 * largura / altura)
        else:
            nova_largura = 270
            nova_altura = int(270 * altura / largura)

        # Redimensiona
        img = img.resize((nova_largura, nova_altura), Image.LANCZOS)

        # Salva em buffer (com qualidade 50%)
        buffer = io.BytesIO()
        img.save(buffer, format="JPEG", quality=50, optimize=True)
        buffer.seek(0)

        return {
          "name": nome_arquivo,
          "data": buffer.getvalue()
        }

    except Exception as e:
        raise RuntimeError(f"Erro ao redimensionar imagem: {e}")

# ====== Extração de dados ======
def extrair_dados_do_arquivo(caminho_arquivo_dcm, callback):
  # Ler arquivo Dicom para extrair os dados do paciente e Solicitante
  ds = pydicom.dcmread(caminho_arquivo_dcm)

  patient_datum = {
    "name": str(ds.PatientName),
    "birthdate": str(ds.PatientBirthDate),
    "gender": "true" if str(ds.PatientSex).upper() == "F" else "false"
    }
  dentist_datum = {"name": str(ds.InstitutionName)}

  # Converte o arquivo Dicom em uma imagem, pois mandamos os 2 no caso de Dicom 2D
  caminho_arquivo_jpg = convert_dicom_para_jpg(caminho_arquivo_dcm)
  # Parametros necessários para enviar o arquivo
  file_params = info_arquivo(caminho_arquivo_jpg, tipo="image")
  dcm_file_params = info_arquivo(caminho_arquivo_dcm, tipo="dicom")

  data = {
      "patient_datum": patient_datum,
      "dentist_datum": dentist_datum,
      "file_params": file_params,
      "dcm_file_params": dcm_file_params,
      "file_path": caminho_arquivo_jpg,
      "dcm_file_path": caminho_arquivo_dcm
  }
  callback(data)

# ====== Envio para backend ======
def enviar_backend(dados):
  agora = datetime.now(timezone.utc)
  # Formatar no padrão ISO 8601 com milissegundos e 'Z'
  iso_format = agora.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" # Data do pedido

  clinic_id = 919 # Clínica onde será criado o pedido
  request_status_id = 2862 # ID do status do pedido

  # Cria um pedido para colocar os arquivos
  request_params = {
    "request": {
      "date": iso_format,
      "clinic_id": clinic_id,
      "request_status_id": request_status_id,
      "patient_datum": dados["patient_datum"],
      "dentist_datum": dados["dentist_datum"]
    }
  }
  create_request(request_params, dados)

def create_request(request_params, dados):
  response = requests.post(f"{CFAZ_URL}/requests",json=request_params, headers = HEADERS)
  if response.status_code != 200:
    raise RuntimeError(f"Erro ao criar pedido: {response.text}")
  # Criado o pedido e retornado o ID do mesmo
  request_id = response.json()["id"]
  print(f"[OK] Pedido criado: ID {request_id}")
  # Realizando o envio do arquivo e imagens para o Pedido.
  envia_imagens_para_backend(request_id, dados)

def envia_imagens_para_backend(request_id, dados):
  # Junta todas as informações para enviar
  params = { "request_id": request_id } | dados["file_params"] | dados["dcm_file_params"]
  print(params)
  response = requests.post(f"{CFAZ_URL}/photos/signed_url", json=params, headers = HEADERS)
  if response.status_code != 200:
    raise RuntimeError(f"Erro ao solicitar URLs assinadas: {response.text}")

  print(response.text)

  # Requisição bem sucedida,
  # Recebemos os dados para encaminhar ao storage
  print("Preparando para enviar as imagens")
  resp_json = response.json()
  arquivos = [
    {
        "id": resp_json["id"],
        "file_name": dados["file_params"]["file_name"],
        "signed_url": resp_json["signed_url"],
        "signed_url_medium": resp_json.get("signed_url_medium"),
        "headers": resp_json.get("headers", {}),
        "signed_id": resp_json["signed_id"],
        "file_path": dados["file_path"]
    },
    {
        "id": resp_json["id"],
        "file_name": dados["dcm_file_params"]["file_name_dicom"],
        "signed_url": resp_json["signed_url_dicom"],
        "headers": resp_json.get("headers_dicom", {}),
        "signed_id": resp_json["signed_id_dicom"],
        "file_path": dados["dcm_file_path"],
        "is_dicom": True
    }
  ]
  for arquivo in arquivos:
    enviar_para_storage(arquivo)

def enviar_para_storage(file):
  print(f"[UPLOAD] Enviando arquivo: {file['file_name']}")
  if "signed_url_medium" in file:
    enviar_miniatura_para_storage(file)
    url = file["signed_url"]
    headers = file.get("headers", {})
    with open(file["file_path"], "rb") as blob_file:
      response = requests.put(url, headers=headers, data=blob_file)
  else:
    url = file["signed_url"]
    headers = file.get("headers", {})
    with open(file["file_path"], "rb") as blob_file:
      response = requests.put(url, headers=headers, data=blob_file)

  if response.status_code != 200:
    raise RuntimeError(f"Falha ao enviar arquivo {file['file_name']}")

  payload = {"photo": {"id": file["id"]}}
  if "is_dicom" in file:
    payload["photo"]["dicom"] = file["signed_id"]
  else:
    payload["photo"]["image"] = file["signed_id"]
    payload["photo"]["image_processing"] = 2
  print(payload)
  requests.put(f"{CFAZ_URL}/photos/{file['id']}", json=payload, headers=HEADERS)
  print(f"[OK] Arquivo registrado no backend: {file['file_name']}")

def enviar_miniatura_para_storage(file):
  url = file["signed_url_medium"]
  headers = file.get("headers", {})
  print("Criando e enviando miniatura da imagem:", file["file_path"])
  miniatura = criar_miniatura_da_imagem(file["file_path"])
  response = requests.put(url, headers=headers, data=miniatura)

# ====== Função principal ======
def processar_arquivo(caminho):
  extrair_dados_do_arquivo(caminho, enviar_backend)

# ====== Início ======
processar_arquivo("arquivo")