대용량 문서 처리

# pip install pypdf requests
import requests
import io
import re
import json
import time
from pypdf import PdfReader, PdfWriter

MAX_RETRIES = 3
CHUNK_SIZE = 100

API_URL = "https://api.upstage.ai/v1/document-digitization"
API_KEY = "UPSTAGE_API_KEY_INJECT"
FILE_PATH = "./sample.pdf"
REQUEST_OPTIONS = {
    "model": "document-parse",
    "output_formats": "['html', 'markdown', 'text']",
}


def parse_large_pdf(file_path, api_key):
    headers = {"Authorization": f"Bearer {api_key}"}
    reader = PdfReader(file_path)
    total_pages = len(reader.pages)

    html_parts, markdown_parts, text_parts = [], [], []
    all_elements = []
    api_value, model_value = "", ""
    total_usage_pages = 0
    element_id_offset = 0
    failed_chunks = []

    for i in range(0, total_pages, CHUNK_SIZE):
        end_page = min(i + CHUNK_SIZE, total_pages)

        writer = PdfWriter()
        for page_num in range(i, end_page):
            writer.add_page(reader.pages[page_num])

        res_json = _request_with_retry(writer, i, end_page, headers)
        if res_json is None:
            failed_chunks.append((i + 1, end_page))
            continue

        if not api_value:
            api_value = res_json.get("api", "")
        if not model_value:
            model_value = res_json.get("model", "")

        chunk_elements = res_json.get("elements", [])
        # 정규식 기반 ID 재매핑: 단순한 숫자 ID를 가진 API 생성 HTML에 대해 동작합니다.
        # 출력 결과를 브라우저에서 렌더링한다면, 적절한 파서로 HTML을
        # 새니타이즈하는 것을 고려하세요.
        for el in chunk_elements:
            old_id = el["id"]
            new_id = old_id + element_id_offset
            el["id"] = new_id
            el["page"] += i
            if "content" in el and "html" in el["content"]:
                el["content"]["html"] = re.sub(
                    rf"\bid=['\"]{old_id}['\"]",
                    f"id='{new_id}'",
                    el["content"]["html"],
                )
            all_elements.append(el)

        chunk_html = res_json.get("content", {}).get("html", "")
        if chunk_html:

            def update_html_id(match):
                return f"id='{int(match.group(1)) + element_id_offset}'"

            html_parts.append(
                re.sub(r"\bid=['\"](\d+)['\"]", update_html_id, chunk_html)
            )

        chunk_md = res_json.get("content", {}).get("markdown", "")
        if chunk_md:
            markdown_parts.append(chunk_md)

        chunk_text = res_json.get("content", {}).get("text", "")
        if chunk_text:
            text_parts.append(chunk_text)

        total_usage_pages += res_json.get("usage", {}).get("pages", 0)
        element_id_offset += len(chunk_elements)

    if failed_chunks:
        print(f"Warning: {len(failed_chunks)} chunk(s) failed: {failed_chunks}")

    return {
        "api": api_value,
        "model": model_value,
        "content": {
            "html": "\n".join(html_parts),
            "markdown": "\n\n".join(markdown_parts),
            "text": "\n".join(text_parts),
        },
        "elements": all_elements,
        "usage": {"pages": total_usage_pages},
    }


def _request_with_retry(writer, chunk_start, chunk_end, headers):
    with io.BytesIO() as pdf_bytes:
        writer.write(pdf_bytes)

        for attempt in range(MAX_RETRIES):
            pdf_bytes.seek(0)
            files = {
                "document": (
                    f"chunk_{chunk_start}.pdf",
                    pdf_bytes,
                    "application/pdf",
                )
            }

            try:
                response = requests.post(
                    API_URL, headers=headers, files=files, data=REQUEST_OPTIONS
                )
                if response.status_code == 200:
                    return response.json()
                elif response.status_code == 429:
                    if attempt < MAX_RETRIES - 1:
                        wait = 5 * (2**attempt)
                        print(
                            f"Rate limit hit. Waiting {wait}s... "
                            f"({attempt + 1}/{MAX_RETRIES})"
                        )
                        time.sleep(wait)
                elif 400 <= response.status_code < 500:
                    raise ValueError(
                        f"Client error ({response.status_code}): {response.text}"
                    )
                else:
                    if attempt < MAX_RETRIES - 1:
                        wait = 3 * (2**attempt)
                        print(
                            f"Server error ({response.status_code}). "
                            f"Retrying in {wait}s... ({attempt + 1}/{MAX_RETRIES})"
                        )
                        time.sleep(wait)
            except requests.exceptions.RequestException as e:
                if attempt < MAX_RETRIES - 1:
                    wait = 3 * (2**attempt)
                    print(
                        f"Network error: {e}. Retrying in {wait}s... "
                        f"({attempt + 1}/{MAX_RETRIES})"
                    )
                    time.sleep(wait)

    print(
        f"Failed to process pages {chunk_start + 1}–{chunk_end}. Skipping chunk."
    )
    return None


if __name__ == "__main__":
    result = parse_large_pdf(FILE_PATH, API_KEY)
    with open("result.json", "w", encoding="utf-8") as f:
        json.dump(result, f, ensure_ascii=False, indent=2)
    print(f"Done. {result['usage']['pages']} pages processed. Saved to result.json")

import requests

api_key = "UPSTAGE_API_KEY_INJECT"
filename = "invoice.png"

url = "https://api.upstage.ai/v1/document-digitization/async"
headers = {"Authorization": f"Bearer {api_key}"}
files = {"document": open(filename, "rb")}
data = {"model": "document-parse"}
response = requests.post(url, headers=headers, files=files, data=data)
print(response.json())

import requests

api_key = "UPSTAGE_API_KEY"

url = "https://api.upstage.ai/v1/document-digitization/requests/REQUEST_ID"
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.get(url, headers=headers)
print(response.json())

import requests

api_key = "UPSTAGE_API_KEY"

url = "https://api.upstage.ai/v1/document-digitization/requests"
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.get(url, headers=headers)
print(response.json())

import requests

api_key = "UPSTAGE_API_KEY_INJECT"
request_id = "e7b1b3b0-1b3b-4b3b-8b3b-1b3b3b3b3b3b"

url = f"https://api.upstage.ai/v1/document-digitization/requests/{request_id}"
headers = {"Authorization": f"Bearer {api_key}"}
response = requests.delete(url, headers=headers)
print(response.json())

import requests

api_key = "UPSTAGE_API_KEY_INJECT"

url = "https://api.upstage.ai/v1/document-digitization/requests/delete"
headers = {"Authorization": f"Bearer {api_key}"}
payload = {"request_ids": [
    "e7b1b3b0-1b3b-4b3b-8b3b-1b3b3b3b3b3b",
    "a1c2e3f4-5678-4abc-9def-0123456789ab",
]}
response = requests.post(url, headers=headers, json=payload)
print(response.json())