파싱 파이프라인 패키지 구조 — 노드 규약과 상태 설계

DEFAULT_CONFIG = {
    "ocr": False,
    "coordinates": True,
    "output_formats": "['html', 'text', 'markdown']",
    "model": "document-parse",
    "base64_encoding": "['figure', 'chart', 'table']",
}

response = requests.post(
    "https://api.upstage.ai/v1/document-ai/document-parse",
    headers={"Authorization": f"Bearer {self.api_key}"},
    data=self.config,
    files={"document": open(input_file, "rb")},
)

{
  "api": "2.0",
  "model": "document-parse-240910",
  "usage": {"pages": 21},
  "content": {...},          # 문서 전체 통합 결과
  "elements": [...]          # 요소 배열 (핵심)
}

{
  "category": "paragraph",
  "content": {
    "html": "<p id='0' data-category='paragraph' style='font-size:20px'>Europe, Africa ...</p>",
    "markdown": "Europe, Africa, Middle East and Asia-Pacific prices and commentary",
    "text": "Europe, Africa, Middle East and Asia-Pacific prices and commentary"
  },
  "coordinates": [
    {"x": 0.3217, "y": 0.1249},
    {"x": 0.9273, "y": 0.1249},
    {"x": 0.9273, "y": 0.1560},
    {"x": 0.3217, "y": 0.1560}
  ],
  "id": 0,
  "page": 1
}

<table id='10' style='font-size:14px'>
  <tr><td colspan="4">Bitumen prices at key locations, 16-22 Mar</td><td rowspan="2">$/t ±</td></tr>
  <tr><td></td><td></td><td>Low</td><td>High</td></tr>
  <tr><td>Mediterranean</td><td></td><td>445.43</td><td>449.77</td><td>+29.05</td></tr>
</table>

pages_count = 0
for meta in state["metadata"]:
    for k, v in meta.items():
        if k == "usage":
            pages_count += int(v["pages"])

total_cost = pages_count * 0.01

from .state import ParseState
from abc import ABC, abstractmethod


class BaseNode(ABC):
    def __init__(self, verbose=False, **kwargs):
        self.name = self.__class__.__name__
        self.verbose = verbose

    @abstractmethod
    def execute(self, state: ParseState) -> ParseState:
        pass

    def log(self, message: str, **kwargs):
        if self.verbose:
            print(f"[{self.name}] {message}")
            for key, value in kwargs.items():
                print(f"  {key}: {value}")

    def __call__(self, state: ParseState) -> ParseState:
        return self.execute(state)

from typing import TypedDict, Annotated, List, Dict
import operator
from .element import Element
from langchain_core.documents import Document


class ParseState(TypedDict):
    filepath: Annotated[str, "filepath"]
    filetype: Annotated[str, "filetype"]
    split_filepaths: Annotated[List[str], "split_filepaths"]
    working_filepath: Annotated[str, "working_filepath"]

    metadata: Annotated[List[Dict], operator.add]     # api, model, usage
    total_cost: Annotated[float, "total_cost"]

    raw_elements: Annotated[List[Dict], operator.add]
    elements_from_parser: Annotated[List[Dict], "elements_from_parser"]

    elements: Annotated[List[Element], "elements"]
    reconstructed_elements: Annotated[List[Dict], "reconstructed_elements"]

    export: Annotated[List, operator.add]

    texts_by_page: Annotated[Dict[int, str], "texts_by_page"]
    images_by_page: Annotated[Dict[int, List[Element]], "images_by_page"]
    tables_by_page: Annotated[Dict[int, List[Element]], "tables_by_page"]

    extracted_image_entities: Annotated[List[Element], "extracted_image_entities"]
    extracted_table_entities: Annotated[List[Element], "extracted_table_entities"]

    documents: Annotated[List[Document], "documents"]
    language: Annotated[str, "language"]

from typing import Dict
from dataclasses import dataclass
from copy import deepcopy


@dataclass
class Element:
    category: str  # table, figure, chart, heading1, header, footer, caption,
                   # paragraph, equation, list, index, footnote
    content: str = ""
    html: str = ""
    markdown: str = ""
    base64_encoding: str = None
    image_filename: str = None
    page: int = None
    id: int = None
    coordinates: list[Dict] = None
    entity: str = ""

    def copy(self):
        return deepcopy(self)

class WorkingQueueNode(BaseNode):
    def execute(self, state: ParseState):
        working_filepath = state.get("working_filepath", None)
        # 비어 있으면 첫 번째 파일 선택
        if (
            "working_filepath" not in state
            or state["working_filepath"] is None
            or state["working_filepath"] == ""
        ):
            if len(state["split_filepaths"]) > 0:
                working_filepath = state["split_filepaths"][0]
            else:
                working_filepath = "<<FINISHED>>"
        else:
            if working_filepath == "<<FINISHED>>":
                return {"working_filepath": "<<FINISHED>>"}

            current_index = state["split_filepaths"].index(working_filepath)
            if current_index + 1 < len(state["split_filepaths"]):
                working_filepath = state["split_filepaths"][current_index + 1]
            else:
                working_filepath = "<<FINISHED>>"
        return {"working_filepath": working_filepath}


def continue_parse(state: ParseState):
    return state["working_filepath"] != "<<FINISHED>>"

workflow.add_conditional_edges(
    "working_queue_node",
    continue_parse,
    {True: "document_parse_node", False: "post_document_parse_node"},
)
workflow.add_edge("document_parse_node", "working_queue_node")

class DocumentParseNode(BaseNode):
    def __init__(self, api_key, use_ocr=False, verbose=False, **kwargs):
        super().__init__(verbose=verbose, **kwargs)
        self.api_key = api_key
        self.config = DEFAULT_CONFIG
        if use_ocr:
            self.config["ocr"] = True

    def parse_start_end_page(self, filepath):
        """파일명 끝 9글자(예: 0040_0049)에서 페이지 범위를 읽는다"""
        filename = os.path.basename(filepath)
        name_without_ext = filename.rsplit(".", 1)[0]

        try:
            if len(name_without_ext) < 9:
                return (-1, -1)

            page_numbers = name_without_ext[-9:]

            if not (
                page_numbers[4] == "_"
                and page_numbers[:4].isdigit()
                and page_numbers[5:].isdigit()
            ):
                return (-1, -1)

            start_page = int(page_numbers[:4])
            end_page = int(page_numbers[5:])

            if start_page > end_page:
                return (-1, -1)

            return (start_page, end_page)

        except (IndexError, ValueError):
            return (-1, -1)

    def execute(self, state: ParseState):
        filepath = state["working_filepath"]
        parsed_json = self._upstage_layout_analysis(filepath)

        start_page, _ = self.parse_start_end_page(filepath)
        page_offset = start_page - 1 if start_page != -1 else 0

        with open(parsed_json, "r") as f:
            data = json.load(f)
            for element in data["elements"]:
                element["page"] += page_offset

        metadata = {
            "api": data.pop("api"),
            "model": data.pop("model"),
            "usage": data.pop("usage"),
        }

        return {"metadata": [metadata], "raw_elements": [data["elements"]]}

class PostDocumentParseNode(BaseNode):
    def execute(self, state: ParseState):
        elements_list = state["raw_elements"]
        id_counter = 0
        post_processed_elements = []

        for elements in elements_list:
            for element in elements:
                elem = element.copy()
                elem["id"] = id_counter
                id_counter += 1
                post_processed_elements.append(elem)

        pages_count = 0
        for meta in state["metadata"]:
            for k, v in meta.items():
                if k == "usage":
                    pages_count += int(v["pages"])

        total_cost = pages_count * 0.01
        self.log(f"Total Cost: ${total_cost:.2f}")

        return {
            "elements_from_parser": post_processed_elements,
            "total_cost": total_cost,
        }