미디어 목록, 메타데이터와 텍스트 읽기

미디어 항목은 경로나 RID로 열 수 있으며 파일 유사 스트림으로 읽습니다. 항목 전체를 처리할 때는 먼저 목록을 만들고 각 행의 RID를 사용하는 방식이 적합합니다. 경로별 목록은 같은 경로의 여러 항목 중 가장 최근 항목만 포함합니다.

Spark 목록과 미디어 참조

list_media_items_by_path_with_media_reference(ctx)는 다음 열의 Spark DataFrame을 반환합니다.

참조 열에 kind="reference", name="media_reference" 타입 클래스를 설정하면 미디어 참조로 해석할 수 있습니다.

from transforms.api import Output, transform
from transforms.mediasets import MediaSetInput

REFERENCE_TYPES = {
    "mediaReference": [{"kind": "reference", "name": "media_reference"}]
}


@transform.spark.using(
    images=MediaSetInput("/Example Project/Inputs/images"),
    output=Output("/Example Project/Outputs/image_listing"),
)
def compute(ctx, images, output):
    listing = images.list_media_items_by_path_with_media_reference(ctx)
    output.write_dataframe(listing, column_typeclasses=REFERENCE_TYPES)

이 목록 메서드의 camelCase 열 이름과 fluent 변환 결과의 media_item_rid, media_reference 열 이름을 구분합니다.

다운로드 없이 메타데이터 읽기

get_media_item_metadata(rid)는 미디어 전체를 내려받지 않고 유형별 메타데이터를 반환합니다. 이미지 크기, 음성 길이와 문서 페이지 수 등을 확인할 수 있습니다. 문서의 페이지 수는 metadata.document.pages로 접근합니다.

다음 예제는 메타데이터를 유효한 JSON 문자열로 인코딩하여 표에 추가합니다.

import json

from conjure_python_client import ConjureEncoder
from pyspark.sql import functions as F
from pyspark.sql.types import StringType
from transforms.api import Output, transform
from transforms.mediasets import MediaSetInput

REFERENCE_TYPES = {
    "mediaReference": [{"kind": "reference", "name": "media_reference"}]
}


@transform.spark.using(
    images=MediaSetInput("/Example Project/Inputs/images"),
    output=Output("/Example Project/Outputs/image_metadata"),
)
def compute(ctx, images, output):
    def metadata_json(rid):
        return json.dumps(
            images.get_media_item_metadata(rid), cls=ConjureEncoder, ensure_ascii=False
        )

    listing = images.list_media_items_by_path_with_media_reference(ctx)
    result = listing.withColumn(
        "metadata", F.udf(metadata_json, StringType())(F.col("mediaItemRid"))
    )
    output.write_dataframe(result, column_typeclasses=REFERENCE_TYPES)

이미지 OCR의 텍스트 스트림

내장 미디어 변환도 파일 유사 스트림을 반환할 수 있습니다. 다음 예제는 이미지별 OCR 결과를 UTF-8 문자열로 읽습니다.

from pyspark.sql import functions as F
from pyspark.sql.types import StringType
from transforms.api import Output, transform
from transforms.mediasets import MediaSetInput

REFERENCE_TYPES = {
    "mediaReference": [{"kind": "reference", "name": "media_reference"}]
}


@transform.spark.using(
    images=MediaSetInput("/Example Project/Inputs/images"),
    output=Output("/Example Project/Outputs/image_text"),
)
def compute(ctx, images, output):
    def extract_text(rid):
        with images.transform_image_to_text_ocr_output_text(rid) as content:
            return content.read().decode("utf-8")

    listing = images.list_media_items_by_path_with_media_reference(ctx)
    result = listing.withColumn(
        "text", F.udf(extract_text, StringType())(F.col("mediaItemRid"))
    )
    output.write_dataframe(result, column_typeclasses=REFERENCE_TYPES)

전통 OCR과 시각 언어 모델을 사용하는 추출은 계산 경로가 다릅니다. Fluent API의 OCR와 문서 텍스트는 단일 항목과 집합 전체의 반환형 및 오류 정책을 구분합니다.

단일 노드의 목록

Lightweight 목록은 ctx 없이 호출하고 .pandas()로 가져옵니다. 이 예제는 미디어 참조 문자열을 포함한 표를 일반 데이터셋에 씁니다.

from transforms.api import Output, transform
from transforms.mediasets import MediaSetInput


@transform.using(
    images=MediaSetInput("/Example Project/Inputs/images"),
    output=Output("/Example Project/Outputs/lightweight_listing"),
)
def compute(images, output):
    listing = images.list_media_items_by_path_with_media_reference().pandas()
    output.write_table(listing)

Layout-aware V2로 PDF의 구조 보존하기

Layout-aware 추출은 문단, 제목과 표의 내용에 블록 유형, 신뢰도, ID, 경계 상자와 페이지 정보를 함께 제공합니다. PDF와 이미지의 구조적 내용 추출에 사용하며, 아래 구현은 PDF 문서의 페이지 수를 이용하는 경로입니다. 복잡한 문서를 시각 모델로 추가 처리할 때 경계 상자 정보도 활용할 수 있습니다.

extractLayoutAwareTextV2 요청의 설정은 다음과 같습니다.

전자 PDF에서는 언어를 지정할 필요가 없습니다. 영어 OCR도 언어 생략이 가능하며, 다른 언어의 스캔 문서는 해당 언어를 지정합니다. AUTO는 자동 판정을 위한 작은 계산 오버헤드가 있으므로 문서 종류를 아는 경우에 맞는 모드를 선택합니다.

PDF의 V2 응답은 pages 배열을 가진 객체이며 각 페이지에는 블록 배열이 있습니다. TEXT 형식의 블록 텍스트는 {"type":"text","text":"추출된 내용"} 같은 tagged 값입니다. 블록의 페이지 번호는 0부터 시작합니다.

다음은 PDF 한 페이지씩 OCR한 뒤 문서당 한 행의 text 열에 전체 V2 JSON을 저장하는 변환입니다.

import json

from pyspark.sql import functions as F
from pyspark.sql.types import StringType
from transforms.api import Output, transform
from transforms.mediasets import MediaSetInput

REFERENCE_TYPES = {
    "mediaReference": [{"kind": "reference", "name": "media_reference"}]
}


@transform.spark.using(
    pdfs=MediaSetInput("/Example Project/Inputs/pdfs"),
    output=Output("/Example Project/Outputs/pdf_layout_text"),
)
def compute(ctx, pdfs, output):
    def extract_document(rid):
        document = pdfs.get_media_item_metadata(rid).document
        if document is None or document.pages is None:
            raise ValueError(f"문서의 페이지 수를 확인할 수 없습니다: {rid}")
        pages = []
        for page in range(document.pages):
            response = pdfs.transform_media_item(rid, str(page), {
                "type": "documentToText",
                "documentToText": {
                    "operation": {
                        "type": "extractLayoutAwareTextV2",
                        "extractLayoutAwareTextV2": {
                            "pageRange": {
                                "startPageInclusive": page,
                                "endPageExclusive": page + 1,
                            },
                            "config": {
                                "mode": "SCAN",
                                "format": "TEXT",
                                "languages": [{"type": "language", "language": "ENG"}],
                            },
                        },
                    },
                },
            })
            pages.extend(response.json()["pages"])
        return json.dumps({"pages": pages}, ensure_ascii=False)

    listing = pdfs.list_media_items_by_path_with_media_reference(ctx)
    result = listing.withColumn(
        "text", F.udf(extract_document, StringType())(F.col("mediaItemRid"))
    )
    output.write_dataframe(result, column_typeclasses=REFERENCE_TYPES)

문서가 아니거나 페이지 수가 없으면 오류를 발생시킵니다. 문서 수가 많으면 병렬 실행을 활용하고, 페이지가 많은 문서는 한 요청에 여러 페이지를 묶을 수 있습니다. 페이지 범위 5~10개를 시작점으로 문서 크기와 모델/호출량 제한에 맞춰 조정합니다.