문서-텍스트 미디어 변환

documentToText는 문서 내용을 텍스트로 추출하고 필요하면 문단, 제목, 표 등의 레이아웃을 보존합니다. extractTextV2는 추출 문자열 목록을 반환하고 extractLayoutAwareTextV2는 여러 페이지의 레이아웃 블록을 반환합니다. 이전 extractLayoutAwareContent와 ocrOnPage를 대체하는 V2 연산이 권장됩니다.

요청 구성

다음은 한국어 스캔 문서의 첫 다섯 페이지를 레이아웃 정보와 함께 TEXT로 추출하는 요청 예시입니다. JSON 주석이나 자리표시자 없이 특정 연산을 선택한 형태입니다.

{
  "type": "documentToText",
  "documentToText": {
    "operation": {
      "type": "extractLayoutAwareTextV2",
      "extractLayoutAwareTextV2": {
        "pageRange": {"startPageInclusive": 0, "endPageExclusive": 5},
        "config": {
          "mode": "SCAN",
          "format": "TEXT",
          "languages": [{"type": "language", "language": "KOR"}]
        }
      }
    }
  }
}

전자 PDF나 영어 문서에는 언어 지정이 필요하지 않습니다. 영어가 아닌 스캔 문서에는 해당 OCR 언어를 지정합니다. 큰 문서는 페이지마다 별도 요청을 보내는 대신 5~10페이지 정도를 한 범위로 시작하여 문서 크기, 모델 및 처리량 한도에 맞게 조정할 수 있습니다.

Python 변환 예시

다음 예시는 전자 PDF를 페이지별로 처리하는 경량 변환입니다. 미디어 항목이 문서가 아니거나 페이지 수가 없으면 오류를 발생시킵니다. THREAD_NUMBER=20으로 페이지 작업을 병렬 실행하고 결과를 JSON 문자열로 저장합니다. 두 RID는 가상 값이므로 실제 입력과 출력으로 바꿉니다. 증분 처리가 필요하면 표시된 데코레이터를 활성화합니다.

import json
import polars as pl
from concurrent.futures import ThreadPoolExecutor
from transforms.api import Output, incremental, transform
from transforms.mediasets import MediaSetInput
from transforms.mediasets.utils._constants import MEDIA_ITEM_RID, MEDIA_REFERENCE, PATH


THREAD_NUMBER = 20
OUTPUT_DATASET_RID = "ri.foundry.main.dataset.example"
INPUT_MEDIA_SET_RID = "ri.mio.main.media-set.example"


# @incremental(v2_semantics=True)  # Uncomment this line if incremental is needed.
@transform.using(
    output=Output(OUTPUT_DATASET_RID),
    media_input=MediaSetInput(INPUT_MEDIA_SET_RID),
)
def extract(media_input, output):
    media_refs = pl.from_pandas(
        media_input.list_media_items_by_path_with_media_reference().pandas(),
        schema_overrides={
            MEDIA_ITEM_RID: pl.String,
            MEDIA_REFERENCE: pl.String,
            PATH: pl.String,
        },
    )

    def process_batch(batch_df: pl.DataFrame) -> pl.DataFrame:
        def create_page_tasks(row):
            media_item_rid = row[MEDIA_ITEM_RID]
            metadata = media_input.get_media_item_metadata(media_item_rid).document

            if metadata is None:
                raise ValueError(f"Media item {media_item_rid} is not a document")

            if metadata.pages is None:
                raise ValueError(f"Media item {media_item_rid} has no page count")

            return [(row, page_num) for page_num in range(metadata.pages)]

        def process_single_page(task):
            row, page_num = task
            media_item_rid = row[MEDIA_ITEM_RID]
            media_reference = row[MEDIA_REFERENCE]

            extraction_result = media_input.transform_media_item(
                media_item_rid,
                str(page_num),
                {
                    "type": "documentToText",
                    "documentToText": {
                        "operation": {
                            "type": "extractLayoutAwareTextV2",
                            "extractLayoutAwareTextV2": {
                                "pageRange": {
                                    "startPageInclusive": page_num,
                                    "endPageExclusive": page_num + 1,
                                },
                                "config": {
                                    "mode": "ELECTRONIC",
                                    "format": "TEXT",
                                },
                            },
                        },
                    },
                },
            )

            return {
                "media_item_rid": media_item_rid,
                "media_reference": media_reference,
                "page_num": page_num,
                "extraction_result": json.dumps(extraction_result.json()),
            }

        all_tasks = []
        for row in batch_df.iter_rows(named=True):
            all_tasks.extend(create_page_tasks(row))

        with ThreadPoolExecutor(max_workers=THREAD_NUMBER) as executor:
            results = list(executor.map(process_single_page, all_tasks))

        return pl.DataFrame(results)

    extracted_data = media_refs.lazy().map_batches(
        process_batch,
        schema={
            "media_item_rid": pl.String,
            "media_reference": pl.String,
            "page_num": pl.Int64,
            "extraction_result": pl.String,
        },
        streamable=True,
    )

    output.write_dataframe(extracted_data)

Python 함수 예시

함수는 변환 작업을 생성하고 완료 상태를 기다린 뒤 결과를 읽습니다. preview=True는 사용하는 미디어 변환 API의 미리보기 기능 지정이며 Python 변환 템플릿의 미리보기 지원 여부와 별개입니다. 여러 페이지를 다룰 때는 제한 시간을 고려하여 5~10페이지 배치로 나누는 방식이 권장됩니다.

첫 함수는 미디어 집합 RID와 항목 RID를 직접 받습니다. 객체 입력 버전은 가져온 객체의 media_reference에서 미디어 집합 뷰 항목을 읽습니다. <YOUR_OBJECT_TYPE>은 사용하는 온톨로지 객체의 실제 가져온 Python 클래스명으로 두 곳 모두 치환해야 하는 템플릿 표기이며 그대로 실행하는 식별자가 아닙니다. 객체의 미디어 속성 이름도 실제 스키마에 맞춰야 합니다. 생성된 함수 이름에 vlm이 있어도 이 예제의 실행 연산은 전통적 레이아웃 텍스트 V2입니다.

from time import sleep

from foundry_sdk import FoundryClient
from foundry_sdk.v2.media_sets import models
from functions.api import function

#### Helper functions

def _create_transform_job(
    media_set_rid: str, media_item_rid: str, transformation: models.DocumentToTextTransformation
) -> str:
    fc = FoundryClient()
    job_initiation_resp = fc.media_sets.MediaSet.transform(
        media_set_rid=media_set_rid,
        media_item_rid=media_item_rid,
        transformation=transformation,
        preview=True,
    )
    job_id = job_initiation_resp.job_id
    return job_id


def _is_transform_finished(media_set_rid: str, media_item_rid: str, job_id: str) -> bool:
    fc = FoundryClient()
    status = fc.media_sets.MediaSet.get_status(media_set_rid, media_item_rid, job_id, preview=True)
    return status.status in ("SUCCESSFUL", "FAILED")


def _get_transform_result(media_set_rid: str, media_item_rid: str, job_id: str) -> str:
    fc = FoundryClient()
    result = fc.media_sets.MediaSet.get_result(media_set_rid, media_item_rid, job_id, preview=True)
    return result.decode("utf-8")


def _run_transform_blocking(
    media_set_rid: str, media_item_rid: str, transformation: models.DocumentToTextTransformation
) -> str:
    job_id = _create_transform_job(media_set_rid, media_item_rid, transformation)
    while not _is_transform_finished(media_set_rid, media_item_rid, job_id):
        sleep(0.5)
    return _get_transform_result(media_set_rid, media_item_rid, job_id)

# Use this function if your input is a media set
# We suggest running the function across batches of five to ten pages to avoid timeout
@function(beta=True)
def transform_vlm(media_set_rid: str, media_item_rid: str, start_page_inclusive: int, end_page_exclusive: int) -> str:
    LANGUAGES: list[models.OcrLanguageOrScript] = [models.OcrLanguageWrapper(language="ENG")]

    return _run_transform_blocking(
        media_set_rid,
        media_item_rid,
        models.DocumentToTextTransformation(
            operation=models.ExtractDocumentLayoutAwareTextV2Operation(
                page_range=models.PageRange(start_page_inclusive=start_page_inclusive, end_page_exclusive=end_page_exclusive),
                config=models.ExtractDocumentLayoutAwareTextV2Config(languages=LANGUAGES),
            )
        ),
    )

# Use this function if your input is an object
# We suggest running the function across batches of five to ten pages to avoid timeout
@function(beta=True, edits=[<YOUR_OBJECT_TYPE>])
def transform_vlm_object(myObject: <YOUR_OBJECT_TYPE>, start_page_inclusive: int, end_page_exclusive: int) -> str:
    reference_view = myObject.media_reference.get_media_reference().reference.media_set_view_item

    media_set_rid = reference_view.media_set_rid
    media_item_rid = reference_view.media_item_rid
    LANGUAGES: list[models.OcrLanguageOrScript] = [models.OcrLanguageWrapper(language="ENG")]

    return _run_transform_blocking(
        media_set_rid,
        media_item_rid,
        models.DocumentToTextTransformation(
            operation=models.ExtractDocumentLayoutAwareTextV2Operation(
                page_range=models.PageRange(start_page_inclusive=start_page_inclusive, end_page_exclusive=end_page_exclusive),
                config=models.ExtractDocumentLayoutAwareTextV2Config(languages=LANGUAGES),
            )
        ),
    )

일시적 오류와 재시도 가능 여부를 반환값으로 처리하려면 Python 함수 배포의 오류 처리를 적용합니다.