大版本更新迁移说明

{
  "document": { "password": "..." },
  "capabilities": { "table_view": "html", "pages": true, "...": "..." },
  "scope": { "page_range": "1-10" },
  "config": {
    "force_engine": "textin",
    "engine_params": { 
      "parse_mode": "auto", 
      "formula_level": 0,
      "image_output_type": "url" 
    }
  }
}

{
  "element_id": "el_001",
  "type": "Title",
  "text": "检验结果",
  "page_number": 1,
  "coordinates": [0.1, 0.12, 0.32, 0.12, 0.32, 0.16, 0.1, 0.16],
  "metadata": {
    "category_depth": 0,
    "children_ids": ["el_002", "el_003"],
    "is_continuation": false
  }
}

# 旧版：绝对像素坐标（受 dpi 影响），字段名 position
# position = [x1, y1, x2, y2, x3, y3, x4, y4]  单位：像素

# 新版：归一化坐标（范围 0~1），字段名 coordinates
# 如需换算为像素，从 data.pages[] 取该页宽高：
page_width  = page["page_width"]
page_height = page["page_height"]
pixel_coords = [
    (coords[i] * page_width, coords[i + 1] * page_height)
    for i in range(0, 8, 2)
]

import requests, time, json

APP_ID  = "your_app_id"
SECRET  = "your_secret_code"
HEADERS = {"x-ti-app-id": APP_ID, "x-ti-secret-code": SECRET}


def submit_async_job(file_path: str, config: dict = None) -> str:
    """提交异步解析任务，返回 job_id"""
    url = "https://api.textin.com/api/v1/xparse/parse/async"
    with open(file_path, "rb") as f:
        files = {"file": (file_path, f, "application/pdf")}
        data  = {"config": json.dumps(config)} if config else {}
        resp  = requests.post(url, headers=HEADERS, files=files, data=data)
    resp.raise_for_status()                       # 先校验 HTTP 状态
    result = resp.json()
    if result.get("code") != 200:                 # 再校验业务 code，避免 KeyError
        raise RuntimeError(f"提交失败: {result.get('message')}")
    return result["data"]["job_id"]


def poll_result(job_id: str, interval: int = 3, max_wait: int = 300) -> dict:
    """轮询任务状态，完成后返回完整解析结果"""
    url     = f"https://api.textin.com/api/v1/xparse/parse/async/{job_id}"
    elapsed = 0
    while elapsed < max_wait:
        resp = requests.get(url, headers=HEADERS)
        resp.raise_for_status()
        job_data = resp.json()["data"]
        status   = job_data["status"]
        if status == "completed":
            # result_url 鉴权要求待确认，此处带 headers 以求稳妥
            return requests.get(job_data["result_url"], headers=HEADERS).json()
        elif status == "failed":
            raise RuntimeError(f"任务失败: {job_data.get('message')}")
        print(f"  {status}，已等待 {elapsed}s ...")
        time.sleep(interval)
        elapsed += interval
    raise TimeoutError(f"超时（{max_wait}s），job_id={job_id}")


config = {
    "capabilities": {
        "include_hierarchy":  True,
        "table_view":         "html",
        "include_image_data": True,   # ⚠️ 默认 false，需显式开启
        "pages":              True,   # ⚠️ 默认 false，需显式开启
        "title_tree":         True,
    }
}
job_id = submit_async_job("large_document.pdf", config)
result = poll_result(job_id)
print(result["data"]["markdown"])

# 注意:file_url 同样走 multipart/form-data 提交。
# requests 用 files= 触发 multipart;下方用 data= 会变成 urlencoded;
# 服务端通常也能解析表单字段,但若严格遵循 multipart 协议,建议统一用 files=。
fields = {
    "file_url": (None, "https://your-storage.com/document.pdf"),
    "webhook": (None, "https://your-server.com/callback/textin"),
    "config": (None, json.dumps({"capabilities": {"table_view": "html"}})),
}

resp = requests.post(
    "https://api.textin.com/api/v1/xparse/parse/async",
    headers=HEADERS,
    files=fields
)
resp.raise_for_status()
job_id = resp.json()["data"]["job_id"]

# 你的服务器收到的 Webhook Body（Method: POST, Content-Type: application/json）：
# {
#   "job_id": "xxx",
#   "status": "completed",
#   "result_url": "https://..."
# }