MarkifyDoc 開発者向けAPI仕様書

import os
import time
import requests

MARKIFY_API_KEY = os.getenv("MARKIFY_API_KEY", "mkd_live_xxxxxx")
BASE_URL = "https://api.markifydoc.com/api/v1"
HEADERS = {"Authorization": f"Bearer {MARKIFY_API_KEY}"}

def parse_and_ingest_pdf(file_path: str):
    file_size = os.path.getsize(file_path)
    file_name = os.path.basename(file_path)

    # 1. 申请 S3/R2 直传临时链接
    presign_res = requests.post(
        f"{BASE_URL}/upload-url",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={"filename": file_name, "file_size": file_size}
    ).json()["data"]

    # 2. 客户端直接流式上传
    with open(file_path, "rb") as f:
        requests.put(presign_res["upload_url"], data=f, headers={"Content-Type": "application/pdf"})

    # 3. 提交多模态视觉解析任务
    task_res = requests.post(
        f"{BASE_URL}/parse",
        headers={**HEADERS, "Content-Type": "application/json"},
        json={
            "r2_source_key": presign_res["r2_source_key"],
            "original_filename": file_name,
            "pages_estimated": 10,
        }
    ).json()["data"]
    task_id = task_res["task_id"]

    # 4. 轮询状态（或由 Webhook 异步唤醒）
    while True:
        status_res = requests.get(f"{BASE_URL}/tasks/{task_id}", headers=HEADERS).json()["data"]
        status = status_res["status"]
        if status == "SUCCESS":
            md_url = status_res["download_urls"]["markdown"]
            raw_markdown = requests.get(md_url).text
            print(f"Parsed {len(raw_markdown)} characters of Markdown with KaTeX formulas.")
            return raw_markdown
        elif status == "FAILED":
            raise RuntimeError(f"Parse failed: {status_res.get('failed_pages_detail')}")
        time.sleep(3)

if __name__ == "__main__":
    markdown = parse_and_ingest_pdf("./nature_paper.pdf")
    # 下游直接对接 LangChain / LlamaIndex:
    # chunks = RecursiveCharacterTextSplitter().split_text(markdown)
    # vectorstore.add_texts(chunks)