Enviar imágenes directamente al Storage

Creada por Wenderson Cotta Sansão, Modificado el Jue., 11 Jun. a las 5:04 P. M. por Wenderson Cotta Sansão

Para grandes volúmenes de imágenes y archivos, es posible realizar el envío directamente al storage.

Antes del envío, es necesario que exista un pedido registrado en el sistema. Para ello, puede utilizar un ID de pedido ya existente o crear uno nuevo mediante la API.

Al crear un pedido, es obligatorio informar:

ID de la clínica;
ID del estado del pedido.

El sistema no aceptará solicitudes sin estos parámetros o si los IDs informados no corresponden, respectivamente, a una clínica válida y a un estado de pedido perteneciente al usuario.

Con el ID del pedido y la información de metadatos de los archivos, es posible solicitar al servidor los datos necesarios para el envío al storage.

La solicitud debe realizarse a la URL:

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

El cuerpo de la solicitud debe contener los siguientes parámetros:

request_id → ID del pedido al que se vincularán las imágenes.
file_name → El nombre del archivo.
file_size → Tamaño del archivo en bytes.
content_type → Tipo del archivo (ej.: image/jpeg).
checksum → Suma de verificación del archivo (MD5 codificado en Base64).
width_px → Ancho de la imagen, en píxeles.
height_px → Alto de la imagen, en píxeles.

Es posible almacenar una imagen asociada a un archivo DCM 2D. Para ello, basta informar en la misma solicitud los siguientes datos del archivo DCM:

file_name_dicom → El nombre del archivo.
file_size_dicom → Tamaño del archivo en bytes.
content_type_dicom → Tipo del archivo (ej.: application/dicom).
checksum_dicom → Suma de verificación del archivo (MD5 codificado en Base64).

Cálculo del checksum:

El cálculo del checksum debe realizarse de la siguiente manera:

Leer el contenido binario del archivo (no en texto, sino los bytes)
Calcular el hash MD5 de ese contenido.
Codificar el resultado en Base64 (no en hexadecimal).
Enviar ese valor como checksum.

Ejemplo con 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"

El retorno esperado es:

{
  "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 contiene la siguiente información:

ID → Identificador único de la imagen dentro del sistema.
headers → Encabezados HTTP a utilizar al enviar la imagen al storage.
signed_id → Identificador criptográficamente firmado para un ActiveStorage::Blob o ActiveStorage::Attachment, referente a la imagen.
signed_url → URL para el envío de la imagen principal.
signed_url_medium → URL para el envío de la miniatura de la imagen, con dimensiones estándar de 270 x 270 píxeles.

Cuando se envía la imagen junto con el archivo DICOM, se retorna la siguiente información en la misma solicitud:

signed_id_dicom → Identificador criptográficamente firmado para un ActiveStorage::Blob o ActiveStorage::Attachment, referente al archivo DICOM.
headers_dicom → Encabezados HTTP a utilizar al enviar los datos del archivo DICOM al storage.
signed_url_dicom → URL para el envío del archivo DICOM.

Con esta información es necesario enviar los archivos al Storage:

Ejemplo con 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"

Tras completar el envío, es necesario actualizar el sistema asociando el signed_id a la imagen correspondiente, para permitir que el sistema la encuentre correctamente.

Ejemplo con 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"

Ejemplo funcional con 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

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

# ====== Utilidades ======

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")

# Función para extraer información de imágenes (PNG, JPG, etc.) y del archivo 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) # ejemplo: "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:
    # Convertir el archivo Dicom en una imagen 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] Archivo convertido a JPG: {jpg_file}")
        return jpg_file
    except subprocess.CalledProcessError as e:
        raise RuntimeError(f"Error al convertir DICOM a JPG: {e}")

# Crear miniatura de la imagen
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 nuevo ancho/alto manteniendo proporción
        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)

        # Guarda en buffer (con calidad 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"Error al redimensionar imagen: {e}")

# ====== Extracción de datos ======
def extrair_dados_do_arquivo(caminho_arquivo_dcm, callback):
  # Leer archivo Dicom para extraer los datos del paciente y 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)}

  # Convierte el archivo Dicom en una imagen, ya que enviamos los 2 en el caso de Dicom 2D
  caminho_arquivo_jpg = convert_dicom_para_jpg(caminho_arquivo_dcm)
  # Parámetros necesarios para enviar el archivo
  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)

# ====== Envío al backend ======
def enviar_backend(dados):
  agora = datetime.now(timezone.utc)
  # Formatear en el estándar ISO 8601 con milisegundos y 'Z'
  iso_format = agora.strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" # Fecha del pedido

  clinic_id = 919 # Clínica donde se creará el pedido
  request_status_id = 2862 # ID del estado del pedido

  # Crea un pedido para colocar los archivos
  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"Error al crear pedido: {response.text}")
  # Creado el pedido y retornado el ID del mismo
  request_id = response.json()["id"]
  print(f"[OK] Pedido creado: ID {request_id}")
  # Realizando el envío del archivo e imágenes al Pedido.
  envia_imagens_para_backend(request_id, dados)

def envia_imagens_para_backend(request_id, dados):
  # Une toda la información 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"Error al solicitar URLs firmadas: {response.text}")

  print(response.text)

  # Solicitud exitosa,
  # Recibimos los datos para enviar al storage
  print("Preparando para enviar las imágenes")
  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 archivo: {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"Falla al enviar archivo {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] Archivo registrado en el backend: {file['file_name']}")

def enviar_miniatura_para_storage(file):
  url = file["signed_url_medium"]
  headers = file.get("headers", {})
  print("Creando y enviando miniatura de la imagen:", file["file_path"])
  miniatura = criar_miniatura_da_imagem(file["file_path"])
  response = requests.put(url, headers=headers, data=miniatura)

# ====== Función principal ======
def processar_arquivo(caminho):
  extrair_dados_do_arquivo(caminho, enviar_backend)

# ====== Inicio ======
processar_arquivo("arquivo")