From 28494e633ca99e9713751abb3f111ff042d1f9c0 Mon Sep 17 00:00:00 2001 From: myhloli Date: Fri, 23 Oct 2026 11:53:26 +0800 Subject: [PATCH] refactor(api): unify task file fields and generate optional client IDs --- docs/en/usage/http_api.md | 6 +- docs/next/api/changes.md | 5 +- docs/next/api/examples.md | 2 +- docs/next/api/files.md | 2 +- docs/next/api/tasks.md | 33 ++-- docs/next/sdk/api-parser.md | 2 +- docs/next/workflows.md | 2 +- docs/zh/usage/http_api.md | 4 +- mineru/kit/file_parse/client.py | 4 +- mineru/kit/router/tasks.py | 40 ++--- mineru/parser/api_client.py | 8 +- mineru/parser/api_server.py | 8 +- mineru/parser/task_api.py | 167 ++++++++++-------- scripts/http_task_example.py | 10 +- tests/unittest/task_test_utils.py | 4 +- tests/unittest/test_engine_dead_lifecycle.py | 4 +- tests/unittest/test_file_parse.py | 4 +- tests/unittest/test_flash_html.py | 2 +- tests/unittest/test_parser_api_contract.py | 6 +- tests/unittest/test_parser_api_ocr.py | 8 +- tests/unittest/test_pdf_page_id.py | 8 +- tests/unittest/test_resource_retention.py | 2 +- tests/unittest/test_router_balancing.py | 8 +- tests/unittest/test_router_cascade.py | 6 +- tests/unittest/test_task_api.py | 174 +++++++++++++++---- tests/unittest/test_task_http.py | 83 +++++++-- tests/unittest/test_task_outputs.py | 2 +- tests/unittest/test_tasks_sdk.py | 6 +- 28 files changed, 403 insertions(+), 207 deletions(-) diff --git a/docs/en/usage/http_api.md b/docs/en/usage/http_api.md index e32c90c2..6b5e796a 100644 --- a/docs/en/usage/http_api.md +++ b/docs/en/usage/http_api.md @@ -1,6 +1,6 @@ # V1 HTTP API -Tasks is the single asynchronous parsing API. API Server, Router, the SDK and WebUI use a JSON manifest, upload only missing files, poll task metadata and download each file's result. See the [Tasks contract](../../next/api/tasks.md) and [Python SDK](sdk_api.md). +Tasks is the single asynchronous parsing API. API Server, Router, the SDK and WebUI use a JSON `files` array, upload only missing files, poll task metadata and download each file's result. See the [Tasks contract](../../next/api/tasks.md) and [Python SDK](sdk_api.md). ```bash mineru-kit api-server --host 127.0.0.1 --port 8000 --tier standard @@ -10,7 +10,7 @@ OUTPUT_FORMATS=default,docx MINERU_API_URL=http://127.0.0.1:8000 bash scripts/ht ## Create and upload -Send application/json to POST `/v1/tasks`. Each manifest entry contains a unique client_file_id, filename, bytes for uploaded files, and optional sha256sum/page_range. URL entries may omit bytes. Options are tier, ocr_mode, page_range, output_formats and include_model_output. +Send application/json to POST `/v1/tasks`. Each `files` entry requires `file_name` and, for uploaded files, `bytes`; `sha256sum`, `page_range`, and `client_file_id` are optional. Omitted or null IDs are assigned the smallest available `fN`, starting at `f1`, in input order after reserving all explicit IDs. IDs remain stable in task responses and upload links. URL entries may omit bytes. Options are tier, ocr_mode, page_range, output_formats and include_model_output. The response contains task_id, status_url and upload links for missing files. Follow each returned PUT URL and its headers; no completion request is needed. Each ready file queues independently. Waiting for uploads consumes no parsing slot. @@ -34,4 +34,4 @@ Both shell examples delegate to the standard-library Python Tasks client. RESULT Creation is not automatically retried. MAX_POLLS/POLL_INTERVAL bound polling; CONNECT_TIMEOUT/REQUEST_TIMEOUT bound individual requests. Exit codes are 0 (success), 2 (partial success), 1 (failure or cancellation) and 124 (poll budget exhausted). Completed downloads remain on disk. -The standalone upload and Jobs APIs have been removed. Coordinate server, Router and SDK upgrades; see [migration notes](../../next/api/changes.md). The separate [file_parse wrapper](../../next/api/file-parse.md) provides recoverable synchronous single-file calls. +Tasks and Files JSON use `file_name`; creation accepts `files`. Old `manifest`/`filename` fields and mixed old/new fields are rejected with 400. The standalone upload and Jobs APIs have been removed. Coordinate server, Router and SDK upgrades; see [migration notes](../../next/api/changes.md). The separate [file_parse wrapper](../../next/api/file-parse.md) provides recoverable synchronous single-file calls. diff --git a/docs/next/api/changes.md b/docs/next/api/changes.md index 961d776c..5ccee42e 100644 --- a/docs/next/api/changes.md +++ b/docs/next/api/changes.md @@ -4,6 +4,9 @@ | 旧方式 | 新方式 | |---|---| +| Tasks 创建请求 `manifest` | `files`，仍为 1–100 项数组；旧字段或新旧混传返回 400。 | +| Tasks / Files JSON `filename` | `file_name`；下载附件命名规则保持不变。 | +| 创建请求必填 `client_file_id` | 可省略或为 `null`；服务端按输入顺序补全最小可用的 `fN`，避开所有显式 ID，响应仍必填。 | | 创建 Upload、PUT、complete | POST `/v1/tasks` 提交清单，只 PUT 缺失文件；上传成功自动排队。 | | 创建、查询、取消 Job | POST/GET/DELETE Tasks；GET `/v1/tasks` 列出父任务。 | | `files[].output_files` | `files[].result_url`，每个输入一个完整产物链接。 | @@ -15,4 +18,4 @@ `--allow-local-source` 和 `--max-inline-bytes` 同时删除。Files CRUD、Health、Tiers、Models、Usage 保留；同步 `file_parse` 继续作为独立包装服务运行。 SDK 保留 `MinerUApiParser` 方法和返回类型，状态类型改为 `ApiTaskStatus`，保留 `ApiJobStatus` Python 别名。直接本地 parser 与 HF CPU/ZeroGPU 内嵌解析不经过 HTTP。 -停收旧请求、等待在途任务完成，再协调升级 worker、Router、SDK 和应用。SaaS Go 服务需部署方按同一契约配套更新；进程内任务和缓存索引不保证跨重启恢复。 +停收旧请求、等待在途任务完成，再协调升级 worker、Router、SDK、file_parse 包装服务和应用。SaaS Go 服务需部署方按同一契约配套更新；进程内任务和缓存索引不保证跨重启恢复。 diff --git a/docs/next/api/examples.md b/docs/next/api/examples.md index 477586d9..ea1de28f 100644 --- a/docs/next/api/examples.md +++ b/docs/next/api/examples.md @@ -11,7 +11,7 @@ TASK_ID=job_... MINERU_API_URL=http://127.0.0.1:8000 bash scripts/http_task_exam 直接提供 URL 时只需提交 JSON： ```json -{"manifest":[{"client_file_id":"report","filename":"report.pdf","url":"https://example.com/report.pdf"}],"output_formats":["default"],"include_model_output":false} +{"files":[{"client_file_id":"report","file_name":"report.pdf","url":"https://example.com/report.pdf"}],"output_formats":["default"],"include_model_output":false} ``` 对 `/v1/tasks` 执行 POST，保存返回的 `task_id/status_url`。本地文件清单包含 `bytes` 和可选 `sha256sum`；仅对响应中的待上传项使用返回的 PUT 地址。 diff --git a/docs/next/api/files.md b/docs/next/api/files.md index 3afdb21f..560c22ca 100644 --- a/docs/next/api/files.md +++ b/docs/next/api/files.md @@ -9,7 +9,7 @@ Files 保存源文件视图与已生成的结果。异步解析统一使用 [Tas | GET | `/v1/files/{file_id}/content` | 流式下载已保存的解析产物；源文件不提供公共下载。 | | DELETE | `/v1/files/{file_id}` | 删除文件视图；正在执行和下载的引用继续受保护。 | -元数据包括 `id/object/filename/bytes/created_at/expires_at/purpose/sha256sum`。 +元数据包括 `id/object/file_name/bytes/created_at/expires_at/purpose/sha256sum`。 `purpose` 为 `parse`（源文件）或 `parse_output`（结果）；删除返回 `id/object/deleted`。 列表使用 `object: list`、`data`、`first_id`、`last_id`、`has_more`；默认 20 条，上限 100。 diff --git a/docs/next/api/tasks.md b/docs/next/api/tasks.md index 8b222adc..2ae2b367 100644 --- a/docs/next/api/tasks.md +++ b/docs/next/api/tasks.md @@ -21,7 +21,7 @@ Task ID 是父任务标识；内部逐文件执行 ID 不属于公共契约。 | JSON 字段 | 默认值 | 说明 | |---|---|---| -| `manifest` | 必填 | 1–100 个文件组成的 JSON 数组，不接受 JSON 字符串。 | +| `files` | 必填 | 1–100 个文件组成的 JSON 数组，不接受 JSON 字符串。 | | `tier` | 服务默认值 | `flash/basic/standard/advanced`，受部署能力约束。 | | `ocr_mode` | `auto` | `auto/txt/ocr`。 | | `page_range` | 整本 | 全局 PDF 页范围，支持 `1-5`、`r1`、`all` 等现有语法。 | @@ -33,14 +33,17 @@ Task ID 是父任务标识；内部逐文件执行 ID 不属于公共契约。 | 字段 | 必填 | 说明 | |---|---|---| -| `client_file_id` | 是 | 任务内唯一，1–64 个字母、数字、下划线或短横线。 | -| `filename` | 是 | 逻辑文件名，保留扩展名；不同条目允许同名。 | +| `client_file_id` | 否 | 任务内唯一，1–64 个字母、数字、下划线或短横线；省略或 `null` 时由服务端补全。 | +| `file_name` | 是 | 逻辑文件名，保留扩展名；不同条目允许同名。 | | `url` | 否 | HTTP(S) 文件下载地址，受部署来源策略约束；省略或 `null` 表示客户端上传。 | | `bytes` | 条件 | 上传项必填；URL 项可省略或为 `null`。提供时必须是严格正整数。 | | `sha256sum` | 否 | 64 位小写十六进制 SHA-256；用于完整性校验，上传项还用于源文件缓存预检。 | | `page_range` | 否 | 逐文件 PDF 页范围，未提供时继承全局值；非 PDF 仍只支持整本解析。 | -先从真实文件生成 JSON 请求： +服务端先预留全部显式 `client_file_id`，再按缺省项的输入顺序，从 `f1` 开始分配最小未占用的 `fN`。例如显式占用了 `f2`，两项缺省 ID 分别生成 `f1`、`f3`。空字符串、非法 ID 和重复显式 ID 返回 400。 +创建响应的 `files` 顺序与输入一致，`files[]` 和 `uploads[]` 始终返回非空 ID；生成后上传、查询及 Router 转发均保持不变。 + +先从真实文件生成 JSON 请求（本例省略 ID，由服务端补全）： ```bash python3 - a.pdf b.pdf > task.json <<'PY' @@ -49,20 +52,19 @@ import json from pathlib import Path import sys -manifest = [] -for index, value in enumerate(sys.argv[1:], start=1): +files = [] +for value in sys.argv[1:]: path = Path(value) hasher = hashlib.sha256() with path.open('rb') as source: for chunk in iter(lambda: source.read(1024 * 1024), b''): hasher.update(chunk) - manifest.append({ - 'client_file_id': f'f{index}', - 'filename': path.name, + files.append({ + 'file_name': path.name, 'bytes': path.stat().st_size, 'sha256sum': hasher.hexdigest(), }) -print(json.dumps({'manifest': manifest, 'tier': 'flash', 'ocr_mode': 'txt', +print(json.dumps({'files': files, 'tier': 'flash', 'ocr_mode': 'txt', 'output_formats': ['default', 'latex']}, ensure_ascii=False)) PY @@ -80,16 +82,16 @@ URL 输入同样先创建 JSON 任务： ```json { - "manifest": [ - {"client_file_id": "remote", "filename": "document.pdf", "url": "https://example.com/download?signature=example"}, - {"client_file_id": "local", "filename": "other.pdf", "bytes": 12345} + "files": [ + {"client_file_id": "remote", "file_name": "document.pdf", "url": "https://example.com/download?signature=example"}, + {"client_file_id": "local", "file_name": "other.pdf", "bytes": 12345} ], "output_formats": ["html"] } ``` - URL 项的 `upload_status` 为 `not_required`，不生成上传链接、不接受 PUT；源 `file_id` 为 `null`。创建后即进入执行队列，不等待其他上传。 -- URL 保留原始签名地址，使用逻辑 `filename` 识别类型；不要求 HEAD 或 `Content-Length`。下载后返回实际 `bytes` 和 `sha256sum`，提供的大小/hash 必须匹配。URL 每次下载，不用 hash 缓存跳过下载。 +- URL 保留原始签名地址，使用逻辑 `file_name` 识别类型；不要求 HEAD 或 `Content-Length`。下载后返回实际 `bytes` 和 `sha256sum`，提供的大小/hash 必须匹配。URL 每次下载，不用 hash 缓存跳过下载。 - 上传项 hash 命中时为 `ready`，不返回上传链接，立即排队；未命中或无 hash 时返回该文件的链接。上传后仍计算 hash、校验大小并对源字节去重。 - 缓存只复用源文件，不复用不匹配的解析结果；每个文件使用本次解析参数独立执行。 - 每个上传项校验完成即排队，不需要客户端发送完成确认。等待上传不占解析并发，也不预占 worker。 @@ -221,6 +223,9 @@ URL 交给 worker 下载，客户端上传字节按 worker 的缓存预检结果 不声明新 Tasks 能力的旧 worker 不参与 Tasks 调度，更新时应先更新 worker，再更新 Router 和客户端。 Router 使用有界、可续租的内部结果租约保护 worker 产物，上传窗口不受结果租约影响；Router 丢失后租约会自行结束。已完成任务按 worker 并行续租，每轮网络等待最多 10 秒，避免失联 worker 阻塞其他 worker 的结果保留。 +创建请求仅接受 `files` 和 `file_name`，不兼容 `manifest`、`filename`；新旧字段混传同样返回 400。Tasks 响应及 Files 列表、详情统一使用 `file_name`。 +这是同一 `/v1` 下的破坏性变更：先停收请求并排空在途任务，再协调更新 worker、Router、SDK、file_parse 包装服务及其他客户端；SaaS 由部署方同步对齐。 + 本次协议不保留旧表单、直传或 `markdown/middle_json/structured_content/zip` Tasks 格式别名。 客户端应改用 JSON、`files[].result_url` 及逐文件保存方式。 API Server 和 Router 不提供 `/v1/file_parse`、`/v1/tasks/{task_id}/result`。 diff --git a/docs/next/sdk/api-parser.md b/docs/next/sdk/api-parser.md index 2d6f9dfb..ed1cb978 100644 --- a/docs/next/sdk/api-parser.md +++ b/docs/next/sdk/api-parser.md @@ -18,7 +18,7 @@ print(result.markdown()) ## 请求与结果 1. 校验文件及页范围，计算大小和 SHA-256，包括同机 HTTP 请求。 -2. 向 `/v1/tasks` 提交一个 `client_file_id="file"` 的 manifest，固定请求 `output_formats=["default"]`。 +2. 向 `/v1/tasks` 提交一个 `client_file_id="file"` 的 files 数组，固定请求 `output_formats=["default"]`。 3. 缓存命中不上传；未命中按服务返回的 PUT 链接和 headers 流式上传，并按清单大小设置 `Content-Length`。无需完成确认。 4. 按返回的同源 `status_url` 每秒查询，轮询预算 3600 次；用 `client_file_id` 关联结果。 5. 从该文件的 `result_url` 流式下载并校验 ZIP，读取 `middle_json.json`。 diff --git a/docs/next/workflows.md b/docs/next/workflows.md index 1b3db1f5..9e7467a3 100644 --- a/docs/next/workflows.md +++ b/docs/next/workflows.md @@ -82,7 +82,7 @@ │ └─ v1 API 调用 -> Official API 或 Local Parse Server - -> Tasks manifest / hash preflight / optional upload + -> Tasks files / hash preflight / optional upload -> parse-server -> per-file result_url / requested formats ``` diff --git a/docs/zh/usage/http_api.md b/docs/zh/usage/http_api.md index 78d606ff..44b8f23b 100644 --- a/docs/zh/usage/http_api.md +++ b/docs/zh/usage/http_api.md @@ -10,7 +10,9 @@ OUTPUT_FORMATS=default,docx MINERU_API_URL=http://127.0.0.1:8000 bash scripts/ht ## 文件清单与上传 -向 `POST /v1/tasks` 提交 `application/json`：`manifest` 每项包含唯一 `client_file_id`、逻辑 `filename`、上传文件的 `bytes` 和可选 `sha256sum/page_range`。也可以提供 URL，此时 bytes 可省略。解析选项为 `tier/ocr_mode/page_range/output_formats/include_model_output`。 +Tasks 与 Files JSON 统一使用 `file_name`；旧字段 `manifest`、`filename` 及新旧混传返回 400。 + +向 `POST /v1/tasks` 提交 `application/json`：`files` 每项包含必填逻辑 `file_name`、上传文件的 `bytes` 和可选 `client_file_id/sha256sum/page_range`。`client_file_id` 省略或为 `null` 时，先预留全部显式 ID，再按输入顺序从 `f1` 开始分配最小未占用的 `fN`；创建后上传和查询均保持不变。也可以提供 URL，此时 bytes 可省略。解析选项为 `tier/ocr_mode/page_range/output_formats/include_model_output`。 创建返回稳定的 `task_id/status_url` 和待上传项的链接。客户端只上传缺失文件，使用响应指定的 PUT 地址及请求头，无需 complete。每个文件就绪即独立排队，其他文件等待上传不会阻塞它。 同机也使用相同流程。SDK 总是计算 hash；已完成上传的有效缓存可以免传。并发首次请求允许重复传输，但源字节去重。缓存默认保留 24 小时、每 5 分钟扫描，正在执行及传输的引用受保护。 diff --git a/mineru/kit/file_parse/client.py b/mineru/kit/file_parse/client.py index d1654bae..106c1ec4 100644 --- a/mineru/kit/file_parse/client.py +++ b/mineru/kit/file_parse/client.py @@ -175,10 +175,10 @@ class TaskClient: options = {key: value for key, value in record.options.items() if value is not None} payload = { **options, - "manifest": [ + "files": [ { "client_file_id": "file", - "filename": record.filename, + "file_name": record.filename, "bytes": record.size, "sha256sum": record.sha256sum, } diff --git a/mineru/kit/router/tasks.py b/mineru/kit/router/tasks.py index 4f6c390d..cd5435c2 100644 --- a/mineru/kit/router/tasks.py +++ b/mineru/kit/router/tasks.py @@ -23,7 +23,7 @@ from ...parser.api_server import ApiServerError, ErrorDetail, _JobInputMetadata, from ...parser.task_api import ( TERMINAL_TASK_STATUSES, TaskExecution, - TaskManifestItem, + TaskFileItem, TaskStatusResponse, TaskUpload, UploadTaskOptions, @@ -262,13 +262,13 @@ class RouterTaskBackend: result_route.metadata["payload"]["expires_at"] = route.metadata["task_result_retain_until"] result_route.metadata["task_result_retain_until"] = route.metadata["task_result_retain_until"] - async def validate(self, options: UploadTaskOptions, manifest: list[TaskManifestItem]) -> None: + async def validate(self, options: UploadTaskOptions, inputs: list[TaskFileItem]) -> None: """能力预检不推进轮询游标，也不为尚未上传的任务预占 worker。""" tiers = self.pool.available_tiers() - options.tier = resolve_task_tier(options, manifest, tiers, select_default_quality_tier(tiers)) + options.tier = resolve_task_tier(options, inputs, tiers, select_default_quality_tier(tiers)) checked_workers: set[str] = set() - for item in manifest: - lightweight = is_flash_only_parse_extension(item.filename) + for item in inputs: + lightweight = is_flash_only_parse_extension(item.file_name) selection_tier = "flash" if lightweight else options.tier candidates = self.pool.capable_workers( tier=selection_tier, @@ -324,7 +324,7 @@ class RouterTaskBackend: self._pins.close() async def _wait_for_input_files( - self, worker: WorkerState, upstream_id: str, generation: int, manifest: list[TaskManifestItem] + self, worker: WorkerState, upstream_id: str, generation: int, inputs: list[TaskFileItem] ) -> dict[str, Any]: """等待已上传项生成公共 File；级联 Router 的上传就绪可能早于下游引用交接。""" while True: @@ -338,11 +338,11 @@ class RouterTaskBackend: ) require_worker_generation(worker, generation) ready = TaskStatusResponse.model_validate(_checked_json(response)).model_dump() - if ready["task_id"] != upstream_id or len(ready["files"]) != len(manifest) or ready["status"] == "waiting_upload": + if ready["task_id"] != upstream_id or len(ready["files"]) != len(inputs) or ready["status"] == "waiting_upload": _raise_api_error( 502, error_type="api_error", code="invalid_upstream_response", message="Worker inputs are not ready" ) - if any(item["client_file_id"] != source.client_file_id for item, source in zip(ready["files"], manifest)): + if any(item["client_file_id"] != source.client_file_id for item, source in zip(ready["files"], inputs)): _raise_api_error( 502, error_type="api_error", code="invalid_upstream_response", message="Worker input order changed" ) @@ -351,7 +351,7 @@ class RouterTaskBackend: and item["upload_status"] == "ready" and item["file_id"] is None and item["status"] in {"queued", "running"} - for item, source in zip(ready["files"], manifest) + for item, source in zip(ready["files"], inputs) ) if not pending: return ready @@ -359,17 +359,17 @@ class RouterTaskBackend: await asyncio.sleep(1) async def submit( - self, task_id: str, options: UploadTaskOptions, manifest: list[TaskManifestItem], uploads: dict[int, TaskUpload] + self, task_id: str, options: UploadTaskOptions, inputs: list[TaskFileItem], uploads: dict[int, TaskUpload] ) -> TaskExecution: """预占后提交 worker 任务，只向 worker 上传它实际缺失的文件。""" - lightweight = all(is_flash_only_parse_extension(item.filename) for item in manifest) + lightweight = all(is_flash_only_parse_extension(item.file_name) for item in inputs) tier = "flash" if lightweight else options.tier reservation = self.pool.reserve_job( tier=tier, - required_sources={"url" if item.url is not None else "upload" for item in manifest}, + required_sources={"url" if item.url is not None else "upload" for item in inputs}, required_task_outputs=set(options.output_formats), require_model_output=options.include_model_output, - require_flash=any(is_flash_only_parse_extension(item.filename) for item in manifest), + require_flash=any(is_flash_only_parse_extension(item.file_name) for item in inputs), ) if reservation is None: _raise_api_error( @@ -381,11 +381,11 @@ class RouterTaskBackend: registered_inputs: list[str] = [] transferred = False try: - worker_manifest = [ + worker_inputs = [ item.model_dump(exclude_none=True) if item.url is not None else {**item.model_dump(exclude_none=True), "sha256sum": uploads[index].sha256sum} - for index, item in enumerate(manifest) + for index, item in enumerate(inputs) ] response = await _worker_request( self.pool, @@ -394,7 +394,7 @@ class RouterTaskBackend: "/v1/tasks", request=self.request, json_body={ - "manifest": worker_manifest, + "files": worker_inputs, "tier": options.tier, "ocr_mode": options.ocr_mode, "output_formats": options.output_formats, @@ -420,7 +420,7 @@ class RouterTaskBackend: upstream_headers={"authorization": authorization} if authorization else {}, payload={"job_id": task_id, "status": "queued", "files": []}, ) - by_client_id = {item.client_file_id: index for index, item in enumerate(manifest)} + by_client_id = {item.client_file_id: index for index, item in enumerate(inputs)} for link in payload["uploads"]: index = by_client_id.get(link["client_file_id"]) if index not in uploads or link["upload_method"] != "PUT": @@ -449,10 +449,10 @@ class RouterTaskBackend: if response.status_code >= 400: _checked_json(response) require_worker_generation(worker, reservation.generation) - ready = await self._wait_for_input_files(worker, upstream_id, reservation.generation, manifest) + ready = await self._wait_for_input_files(worker, upstream_id, reservation.generation, inputs) aliases: dict[str, str] = {} for index, item in enumerate(ready["files"]): - if manifest[index].url is not None: + if inputs[index].url is not None: if item["upload_status"] != "not_required": _raise_api_error( 502, error_type="api_error", code="invalid_upstream_response", message="Unexpected URL upload" @@ -468,7 +468,7 @@ class RouterTaskBackend: { "id": upstream_file_id, "object": "file", - "filename": upload.filename, + "file_name": upload.filename, "bytes": upload.bytes, "sha256sum": upload.sha256sum, "purpose": "parse", diff --git a/mineru/parser/api_client.py b/mineru/parser/api_client.py index 6ca2b203..519a37ac 100644 --- a/mineru/parser/api_client.py +++ b/mineru/parser/api_client.py @@ -162,10 +162,10 @@ class MinerUApiParser(DocumentParser): if not path.is_file(): raise FileNotFoundError(path) payload: dict[str, Any] = { - "manifest": [ + "files": [ { "client_file_id": "file", - "filename": path.name, + "file_name": path.name, "bytes": path.stat().st_size, "sha256sum": _sha256_file(path), "page_range": page_range, @@ -237,7 +237,7 @@ class MinerUApiParser(DocumentParser): ) task = self._snapshot(response) _notify_job_status(task, status_callback) - task = self._upload_input(cli, task, file_path, status_callback, size=payload["manifest"][0]["bytes"]) + task = self._upload_input(cli, task, file_path, status_callback, size=payload["files"][0]["bytes"]) for _ in range(_POLL_MAX_ATTEMPTS): if task["status"] in _TERMINAL_JOB_STATUSES: break @@ -394,7 +394,7 @@ class MinerUApiParser(DocumentParser): ) task = self._snapshot(response) _notify_job_status(task, status_callback) - task = await self._async_upload_input(cli, task, file_path, status_callback, size=payload["manifest"][0]["bytes"]) + task = await self._async_upload_input(cli, task, file_path, status_callback, size=payload["files"][0]["bytes"]) for _ in range(_POLL_MAX_ATTEMPTS): if task["status"] in _TERMINAL_JOB_STATUSES: break diff --git a/mineru/parser/api_server.py b/mineru/parser/api_server.py index 17b445b5..9c2a76a4 100644 --- a/mineru/parser/api_server.py +++ b/mineru/parser/api_server.py @@ -308,7 +308,7 @@ class FileObjectModel(BaseModel): bytes: int created_at: int expires_at: int | None = None - filename: str + file_name: str purpose: FilePurpose sha256sum: str | None = None @@ -704,7 +704,7 @@ class FileStore: bytes=rec.bytes, created_at=rec.created_at, expires_at=rec.expires_at, - filename=rec.filename, + file_name=rec.filename, purpose=rec.purpose, sha256sum=rec.sha256sum, ) @@ -1414,9 +1414,9 @@ _OPENAPI_TAGS: list[dict[str, str]] = [ _FASTAPI_DESCRIPTION = """\ Parse documents into structured Middle JSON via self-hosted or remote capacity. -Submit a JSON manifest to `/v1/tasks`, upload missing files to their one-hour URLs, +Submit a JSON files array to `/v1/tasks`, upload missing files to their one-hour URLs, then poll the task and download each completed file's `result_url`. Each file runs -as soon as it is ready; other files may still be waiting for upload. Manifest +as soon as it is ready; other files may still be waiting for upload. File entries can instead provide a source `url`, with optional `bytes`; workers download these inputs in the background. Choose one or more of default/docx/html/latex in `output_formats`. Task status contains metadata only, never inline result bytes. diff --git a/mineru/parser/task_api.py b/mineru/parser/task_api.py index f5273e7d..26920935 100644 --- a/mineru/parser/task_api.py +++ b/mineru/parser/task_api.py @@ -54,19 +54,19 @@ TaskFileState = Literal["waiting_upload", "queued", "running", "completed", "fai logger = logging.getLogger(__name__) -class TaskManifestItem(BaseModel): - """按客户端标识关联上传或 URL 输入，不依赖文件名唯一性。""" +class TaskFileInput(BaseModel): + """声明上传或 URL 输入，客户端可以省略任务内文件标识。""" model_config = ConfigDict(extra="forbid") - client_file_id: str = Field(pattern=r"^[A-Za-z0-9_-]{1,64}$") - filename: str = Field(min_length=1, max_length=1024) + client_file_id: str | None = Field(default=None, pattern=r"^[A-Za-z0-9_-]{1,64}$") + file_name: str = Field(min_length=1, max_length=1024) bytes: int | None = Field(default=None, gt=0, strict=True) url: str | None = None sha256sum: str | None = Field(default=None, pattern=r"^[a-f0-9]{64}$") page_range: str | None = Field(default=None, description=PAGE_RANGE_DESCRIPTION) @model_validator(mode="after") - def validate_source(self) -> TaskManifestItem: + def validate_source(self) -> TaskFileInput: """上传项必须声明大小；URL 只校验合法性，保留签名地址的原始字符串。""" if self.url is None: if self.bytes is None: @@ -76,8 +76,14 @@ class TaskManifestItem(BaseModel): return self +class TaskFileItem(TaskFileInput): + """接单前补全文件标识，内部执行和状态响应只消费完整文件条目。""" + + client_file_id: str = Field(pattern=r"^[A-Za-z0-9_-]{1,64}$") + + class UploadTaskOptions(BaseModel): - """解析选项仅来自 JSON 主体，文件元数据独立保存在 manifest 中。""" + """解析选项仅来自 JSON 主体，文件元数据独立保存在 files 中。""" model_config = ConfigDict(extra="forbid") tier: Tier | None = None @@ -97,20 +103,20 @@ class UploadTaskOptions(BaseModel): class CreateTaskRequest(UploadTaskOptions): """严格 JSON 创建请求，清单是数组而不是编码后的字符串。""" - manifest: list[TaskManifestItem] = Field(min_length=1, max_length=MAX_TASK_FILES) + files: list[TaskFileInput] = Field(min_length=1, max_length=MAX_TASK_FILES) class TaskUploadLink(BaseModel): """描述一个文件的上传目标，客户端原样使用服务端给出的地址和请求头。""" - client_file_id: str + client_file_id: str = Field(pattern=r"^[A-Za-z0-9_-]{1,64}$") upload_url: str upload_method: Literal["PUT"] = "PUT" upload_headers: dict[str, str] expires_at: int -class TaskFileStatus(TaskManifestItem): +class TaskFileStatus(TaskFileItem): """同时提供上传准备情况与逐文件解析状态，禁止内联解析正文。""" upload_status: Literal["pending", "ready", "not_required", "expired"] @@ -194,7 +200,7 @@ class TaskBackend(Protocol): """父任务收尾后解除子结果租约，并开始正常保留期。""" ... - async def validate(self, options: UploadTaskOptions, manifest: list[TaskManifestItem]) -> None: + async def validate(self, options: UploadTaskOptions, inputs: list[TaskFileItem]) -> None: """在接单之前校验部署能力，不能预占解析名额。""" ... @@ -211,7 +217,7 @@ class TaskBackend(Protocol): ... async def submit( - self, task_id: str, options: UploadTaskOptions, manifest: list[TaskManifestItem], uploads: dict[int, TaskUpload] + self, task_id: str, options: UploadTaskOptions, inputs: list[TaskFileItem], uploads: dict[int, TaskUpload] ) -> TaskExecution: """使用预先分配的公开 ID 提交一次后台解析。""" ... @@ -236,13 +242,11 @@ def _invalid(message: str, *, code: str = "invalid_request", status_code: int = ) -def resolve_task_tier( - options: UploadTaskOptions, manifest: list[TaskManifestItem], tiers: set[Tier], default: Tier | None -) -> Tier: +def resolve_task_tier(options: UploadTaskOptions, inputs: list[TaskFileItem], tiers: set[Tier], default: Tier | None) -> Tier: """按现有轻量格式回退规则解析 tier，不触发 worker 选择或模型执行。""" - lightweight = [is_flash_only_parse_extension(item.filename) for item in manifest] + lightweight = [is_flash_only_parse_extension(item.file_name) for item in inputs] if any(lightweight) and "flash" not in tiers: - _invalid("Flash parsing is disabled, but an input requires it", param="manifest") + _invalid("Flash parsing is disabled, but an input requires it", param="files") tier = options.tier or default if tier is None and all(lightweight) and "flash" in tiers: tier = "flash" @@ -266,7 +270,7 @@ async def _save_upload( filename: str, mime_type: str, *, - expected: TaskManifestItem | None = None, + expected: TaskFileItem | None = None, canceled: Callable[[], bool] | None = None, ) -> TaskUpload: """流式落盘并校验完整性；声明大小用于一致性检查，不是上传大小上限。""" @@ -295,7 +299,31 @@ async def _save_upload( raise -async def _read_submission(request: Request) -> tuple[UploadTaskOptions, list[TaskManifestItem]]: +def _normalize_task_files(inputs: list[TaskFileInput]) -> list[TaskFileItem]: + """先预留全部显式标识，再按输入顺序给缺省项分配最小可用的 fN。""" + used_ids: set[str] = set() + for index, item in enumerate(inputs): + if item.client_file_id is None: + continue + if item.client_file_id in used_ids: + _invalid("client_file_id must be unique within a task", param=f"files.{index}.client_file_id") + used_ids.add(item.client_file_id) + + normalized: list[TaskFileItem] = [] + next_id = 1 + for item in inputs: + client_file_id = item.client_file_id + if client_file_id is None: + while f"f{next_id}" in used_ids: + next_id += 1 + client_file_id = f"f{next_id}" + used_ids.add(client_file_id) + next_id += 1 + normalized.append(TaskFileItem.model_validate({**item.model_dump(), "client_file_id": client_file_id})) + return normalized + + +async def _read_submission(request: Request) -> tuple[UploadTaskOptions, list[TaskFileItem]]: """只读取 JSON 清单，严格拒绝表单、直传字节和 URL 查询参数。""" if request.query_params: _invalid("POST parameters must be supplied in the JSON body, not the URL") @@ -304,19 +332,18 @@ async def _read_submission(request: Request) -> tuple[UploadTaskOptions, list[Ta _invalid("Use application/json", status_code=415) try: body = CreateTaskRequest.model_validate(await request.json()) - if len({item.client_file_id for item in body.manifest}) != len(body.manifest): - _invalid("client_file_id must be unique within a task", param="manifest") - options = UploadTaskOptions.model_validate(body.model_dump(exclude={"manifest"})) + inputs = _normalize_task_files(body.files) + options = UploadTaskOptions.model_validate(body.model_dump(exclude={"files"})) options.page_range = normalize_page_range_input(options.page_range) options.output_formats = list(dict.fromkeys(options.output_formats)) - for index, item in enumerate(body.manifest): + for index, item in enumerate(inputs): try: item.page_range = normalize_page_range_input( item.page_range if item.page_range is not None else options.page_range ) except MineruError as exc: - _invalid(str(exc), code=exc.code, param=f"manifest.{index}.page_range") - return options, body.manifest + _invalid(str(exc), code=exc.code, param=f"files.{index}.page_range") + return options, inputs except ClientDisconnect: _invalid("Request interrupted", code="request_interrupted") except MineruError as exc: @@ -368,10 +395,10 @@ class _TaskRecord: id: str backend: TaskBackend options: UploadTaskOptions - manifest: list[TaskManifestItem] + inputs: list[TaskFileItem] directory: Path expires_at: int - files: list[_TaskFileRecord] + file_states: list[_TaskFileRecord] created_at: str = field(default_factory=JobStore._now) status: TaskStatus = "waiting_upload" refresh_lock: asyncio.Lock = field(default_factory=asyncio.Lock) @@ -413,7 +440,7 @@ class TaskStore: if record.status == "canceled": return previous = record.status - states = [file.status for file in record.files] + states = [file.status for file in record.file_states] if "running" in states: record.status = "running" elif "queued" in states: @@ -433,7 +460,7 @@ class TaskStore: """只终结尚未受理上传的文件；截止前已经受理的流允许继续完成。""" if record.expires_at > time.time() or record.status == "canceled": return - for file in record.files: + for file in record.file_states: if file.status == "waiting_upload" and not file.upload_active: file.expired = True file.status = "failed" @@ -508,7 +535,10 @@ class TaskStore: ] page = records[:limit] return TaskListResponse( - data=[TaskListItem(task_id=r.id, status=r.status, created_at=r.created_at, file_count=len(r.files)) for r in page], + data=[ + TaskListItem(task_id=r.id, status=r.status, created_at=r.created_at, file_count=len(r.file_states)) + for r in page + ], first_id=page[0].id if page else None, last_id=page[-1].id if page else None, has_more=len(records) > limit, @@ -524,47 +554,47 @@ class TaskStore: record.changed.set() async def create( - self, backend: TaskBackend, options: UploadTaskOptions, manifest: list[TaskManifestItem], *, lease_seconds: int = 0 + self, backend: TaskBackend, options: UploadTaskOptions, inputs: list[TaskFileItem], *, lease_seconds: int = 0 ) -> _TaskRecord: """先完整校验清单并固定缓存，再为各个就绪文件独立提交执行。""" if self._closing: _invalid("Server is shutting down", status_code=503, code="server_shutting_down") - await backend.validate(options, manifest) + await backend.validate(options, inputs) task_id = JobStore._new_job_id() record = _TaskRecord( task_id, backend, options, - manifest, + inputs, self._root / task_id, int(time.time()) + TASK_UPLOAD_TTL, - [_TaskFileRecord(backend.fork()) for _ in manifest], + [_TaskFileRecord(backend.fork()) for _ in inputs], ) self.update_lease(record, lease_seconds) try: - for index, item in enumerate(manifest): - file = record.files[index] + for index, item in enumerate(inputs): + file = record.file_states[index] if item.url is None and item.sha256sum: - upload = await file.backend.find_upload(item.sha256sum, item.filename) + upload = await file.backend.find_upload(item.sha256sum, item.file_name) if upload is not None: if upload.bytes != item.bytes: _invalid("Declared size does not match cached bytes", code="upload_size_mismatch") file.upload = await file.backend.retain_upload(upload) self._records[task_id] = record self.start() - for index, item in enumerate(manifest): - if item.url is not None or record.files[index].upload is not None: + for index, item in enumerate(inputs): + if item.url is not None or record.file_states[index].upload is not None: self._start_file(record, index) record.operation = asyncio.create_task(self._monitor(record), name=f"mineru-task-{task_id}") return record except BaseException: - for file in record.files: + for file in record.file_states: file.backend.release_uploads(rollback=True) raise def _start_file(self, record: _TaskRecord, index: int) -> None: """事件循环内同步登记提交句柄，重复 PUT 不能创建第二个子作业。""" - file = record.files[index] + file = record.file_states[index] if file.operation is not None or file.status != "waiting_upload" or record.status == "canceled": return file.status = "queued" @@ -601,7 +631,7 @@ class TaskStore: return targets = [ file - for file in record.files + for file in record.file_states if file.submitted and file.status not in {"failed", "canceled"} and (file.status != "completed" or file.lease_refresh_at <= time.monotonic()) @@ -639,10 +669,10 @@ class TaskStore: async def _dispatch_file(self, record: _TaskRecord, index: int) -> None: """每个文件独立提交，失败或取消只清理自己的作业与源文件租约。""" - file = record.files[index] + file = record.file_states[index] try: execution = await file.backend.submit( - file.execution_id, record.options, [record.manifest[index]], {0: file.upload} if file.upload is not None else {} + file.execution_id, record.options, [record.inputs[index]], {0: file.upload} if file.upload is not None else {} ) file.submitted = True self._apply_execution(record, file, execution) @@ -682,21 +712,21 @@ class TaskStore: await self.refresh(record) if record.status not in TERMINAL_TASK_STATUSES: await self._wait_for_change(record) - operations = [file.operation for file in record.files if file.operation is not None] + operations = [file.operation for file in record.file_states if file.operation is not None] if operations: await asyncio.gather(*operations, return_exceptions=True) - while any(file.upload_active for file in record.files) or record.lease_expires_at > time.time(): + while any(file.upload_active for file in record.file_states) or record.lease_expires_at > time.time(): record.changed.clear() await self._wait_for_change(record) finished = time.time() - for file in record.files: + for file in record.file_states: await file.backend.release_result(file.execution_id, finished) file.backend.release_uploads(rollback=not file.submitted) record.retention_epoch = finished except Exception: logger.exception("Task result cleanup failed: %s", record.id) finally: - if not any(file.upload_active for file in record.files): + if not any(file.upload_active for file in record.file_states): shutil.rmtree(record.directory, ignore_errors=True) async def _wait_for_change(self, record: _TaskRecord) -> None: @@ -708,13 +738,13 @@ class TaskStore: async def upload(self, record: _TaskRecord, client_file_id: str, request: Request) -> bool: """在接收主体前检查固定期限，已受理流不因父任务状态变化中止。""" - index = next((i for i, item in enumerate(record.manifest) if item.client_file_id == client_file_id), None) + index = next((i for i, item in enumerate(record.inputs) if item.client_file_id == client_file_id), None) if index is None: _invalid("Unknown client_file_id", status_code=404, code="upload_not_found") - expected = record.manifest[index] + expected = record.inputs[index] if expected.url is not None: _invalid("URL inputs do not accept uploads") - file = record.files[index] + file = record.file_states[index] async with file.upload_lock: if record.expires_at <= time.time(): self._expire_uploads(record) @@ -732,7 +762,7 @@ class TaskStore: except ValueError: _invalid("Invalid Content-Length", code="upload_size_mismatch") if declared != expected.bytes: - _invalid("Content-Length does not match manifest bytes", code="upload_size_mismatch") + _invalid("Content-Length does not match declared file size", code="upload_size_mismatch") file.upload_active = True record.directory.mkdir(parents=True, exist_ok=True) path = record.directory / secrets.token_hex(16) @@ -745,7 +775,7 @@ class TaskStore: upload = await _save_upload( request.stream(), path, - expected.filename, + expected.file_name, request.headers.get("content-type", "application/octet-stream"), expected=expected, canceled=was_canceled, @@ -775,7 +805,7 @@ class TaskStore: record.finished_at = JobStore._now() record.finished_epoch = time.time() record.changed.set() - for file in record.files: + for file in record.file_states: if file.status not in TERMINAL_TASK_STATUSES: file.status = "canceled" if file.operation is not None: @@ -811,11 +841,11 @@ def task_payload(record: _TaskRecord, request: Request) -> dict[str, Any]: files: list[dict[str, Any]] = [] links: list[dict[str, Any]] = [] starts: list[str] = [] - for item, state in zip(record.manifest, record.files): + for item, state in zip(record.inputs, record.file_states): file = item.model_dump() upload = state.upload file.update( - name=item.filename, + name=item.file_name, upload_status="not_required" if item.url is not None else ("ready" if upload is not None else ("expired" if state.expired else "pending")), @@ -863,8 +893,8 @@ def task_payload(record: _TaskRecord, request: Request) -> dict[str, Any]: "output_formats": record.options.output_formats, "include_model_output": record.options.include_model_output, "progress": { - "completed": sum(file.status == "completed" for file in record.files), - "failed": sum(file.status == "failed" for file in record.files), + "completed": sum(file.status == "completed" for file in record.file_states), + "failed": sum(file.status == "failed" for file in record.file_states), "total": len(files), }, "files": files, @@ -922,8 +952,8 @@ def build_task_router(factory: Callable[[Request], TaskBackend]) -> APIRouter: backend = factory(request) store: TaskStore = request.app.state.task_store try: - options, manifest = await _read_submission(request) - record = await store.create(backend, options, manifest, lease_seconds=_retention_lease(request) or 0) + options, inputs = await _read_submission(request) + record = await store.create(backend, options, inputs, lease_seconds=_retention_lease(request) or 0) except BaseException: backend.release_uploads(rollback=True) raise @@ -994,13 +1024,13 @@ class ApiTaskBackend: """由父 Task 统一开始子作业与结果文件的保留期。""" self.jobs.release_task_result(task_id, finished) - async def validate(self, options: UploadTaskOptions, manifest: list[TaskManifestItem]) -> None: + async def validate(self, options: UploadTaskOptions, inputs: list[TaskFileItem]) -> None: """复用引擎就绪状态，提前校验显式 tier 与轻量格式能力。""" _require_model_preload_ready(self.request) if self.jobs._closing: _invalid("Server is shutting down", code="server_shutting_down", status_code=503) state = self.request.app.state - for index, item in enumerate(manifest): + for index, item in enumerate(inputs): if item.url is not None: try: _validate_source_policy( @@ -1008,8 +1038,8 @@ class ApiTaskBackend: allow_http_source=state.allow_http_source, ) except ValueError as exc: - _invalid(str(exc), param=f"manifest.{index}.url") - options.tier = resolve_task_tier(options, manifest, set(state.tier_runtime_options), state.default_tier) + _invalid(str(exc), param=f"files.{index}.url") + options.tier = resolve_task_tier(options, inputs, set(state.tier_runtime_options), state.default_tier) async def find_upload(self, sha256sum: str, filename: str) -> TaskUpload | None: """原子查找并取得缓存租约，准备阶段不会因原文件视图到期丢失字节。""" @@ -1046,7 +1076,7 @@ class ApiTaskBackend: self._pins.clear() async def submit( - self, task_id: str, options: UploadTaskOptions, manifest: list[TaskManifestItem], uploads: dict[int, TaskUpload] + self, task_id: str, options: UploadTaskOptions, inputs: list[TaskFileItem], uploads: dict[int, TaskUpload] ) -> TaskExecution: """提交单文件子作业，在同一次解析中生成专用产物并持有结果租约。""" # 接单时已持有源字节；用户删除文件视图也不能破坏尚在等待其他文件的任务。 @@ -1060,7 +1090,7 @@ class ApiTaskBackend: source=_UrlSource(url=item.url) if item.url is not None else _StoredSource(file_id=uploads[index].file_id), page_range=item.page_range or None, ) - for index, item in enumerate(manifest) + for index, item in enumerate(inputs) ], tier=options.tier, ocr_mode=options.ocr_mode, @@ -1074,9 +1104,7 @@ class ApiTaskBackend: self.files, self.jobs, job_id=task_id, - input_specs={ - index: _JobInputSpec(item.filename, item.bytes, item.sha256sum) for index, item in enumerate(manifest) - }, + input_specs={index: _JobInputSpec(item.file_name, item.bytes, item.sha256sum) for index, item in enumerate(inputs)}, ) return await self.get(task_id) @@ -1105,7 +1133,8 @@ __all__ = [ "CreateTaskRequest", "TaskBackend", "TaskExecution", - "TaskManifestItem", + "TaskFileItem", + "TaskFileInput", "TaskListItem", "TaskListResponse", "TaskStatusResponse", diff --git a/scripts/http_task_example.py b/scripts/http_task_example.py index e4f4a91c..1ce26e93 100644 --- a/scripts/http_task_example.py +++ b/scripts/http_task_example.py @@ -47,9 +47,9 @@ class InputFile: size: int sha256sum: str | None - def manifest(self) -> dict[str, Any]: + def task_file(self) -> dict[str, Any]: """生成 JSON 清单条目；关闭 hash 时完全省略该可选字段。""" - item: dict[str, Any] = {"client_file_id": self.client_file_id, "filename": self.path.name, "bytes": self.size} + item: dict[str, Any] = {"client_file_id": self.client_file_id, "file_name": self.path.name, "bytes": self.size} if self.sha256sum is not None: item["sha256sum"] = self.sha256sum return item @@ -254,8 +254,8 @@ class TaskClient: raise ClientError("Invalid or duplicate client_file_id") seen.add(client_id) local = self.by_id[client_id] - if item.get("filename") != local.path.name or item.get("bytes") != local.size: - raise ClientError("Task manifest differs from local inputs; use the original file list and order") + if item.get("file_name") != local.path.name or item.get("bytes") != local.size: + raise ClientError("Task files differ from local inputs; use the original file list and order") if local.sha256sum is not None and item.get("sha256sum") not in {None, local.sha256sum}: raise ClientError("Task hash differs from the local file") if item.get("upload_status") not in {"pending", "ready", "expired"}: @@ -305,7 +305,7 @@ class TaskClient: if not formats or any(value not in {"default", "docx", "html", "latex"} for value in formats): raise ClientError("OUTPUT_FORMATS must contain default/docx/html/latex") body: dict[str, Any] = { - "manifest": [file.manifest() for file in self.files], + "files": [file.task_file() for file in self.files], "ocr_mode": os.environ.get("OCR_MODE", "auto"), "page_range": os.environ.get("PAGE_RANGE", ""), "output_formats": list(dict.fromkeys(formats)), diff --git a/tests/unittest/task_test_utils.py b/tests/unittest/task_test_utils.py index 4163c291..5a1f977b 100644 --- a/tests/unittest/task_test_utils.py +++ b/tests/unittest/task_test_utils.py @@ -39,10 +39,10 @@ def submit_task(client: TestClient, path: Path, **options: Any) -> dict[str, Any response = client.post( "/v1/tasks", json={ - "manifest": [ + "files": [ { "client_file_id": "file", - "filename": path.name, + "file_name": path.name, "bytes": len(data), "sha256sum": hashlib.sha256(data).hexdigest(), } diff --git a/tests/unittest/test_engine_dead_lifecycle.py b/tests/unittest/test_engine_dead_lifecycle.py index 453161e6..ba129dbc 100644 --- a/tests/unittest/test_engine_dead_lifecycle.py +++ b/tests/unittest/test_engine_dead_lifecycle.py @@ -75,7 +75,7 @@ def test_api_engine_failure_drains_running_and_queued_jobs(tmp_path: Path, monke *( app.state.job_store.wait_for_terminal(file.execution_id, 2) for record in app.state.task_store._records.values() - for file in record.files + for file in record.file_states ) ) await asyncio.gather(*(record.operation for record in app.state.task_store._records.values())) @@ -93,7 +93,7 @@ def test_api_engine_failure_drains_running_and_queued_jobs(tmp_path: Path, monke response = getattr(client, method)( path, **( - {"json": {"tier": "flash", "manifest": [{"client_file_id": "file", "filename": "new.pdf", "bytes": 3}]}} + {"json": {"tier": "flash", "files": [{"client_file_id": "file", "file_name": "new.pdf", "bytes": 3}]}} if method == "post" else {} ), diff --git a/tests/unittest/test_file_parse.py b/tests/unittest/test_file_parse.py index 2397ed6e..30107468 100644 --- a/tests/unittest/test_file_parse.py +++ b/tests/unittest/test_file_parse.py @@ -114,7 +114,7 @@ class Upstream: if self.create_failure == "auth": return httpx.Response(401, json={"error": {"code": "invalid_api_key", "message": "Bad upstream key"}}) body = json.loads(await request.aread()) - assert len(body["manifest"]) == 1 + assert len(body["files"]) == 1 task_id = f"task_{len(self.tasks) + 1}" self.tasks[task_id] = { "options": body, @@ -135,7 +135,7 @@ class Upstream: self.uploads += 1 assert request.headers["x-upload-token"] == "storage-token" content = await request.aread() - assert len(content) == task["options"]["manifest"][0]["bytes"] + assert len(content) == task["options"]["files"][0]["bytes"] if self.upload_failures: self.upload_failures -= 1 return httpx.Response(503) diff --git a/tests/unittest/test_flash_html.py b/tests/unittest/test_flash_html.py index 6e2f65e5..03587a94 100644 --- a/tests/unittest/test_flash_html.py +++ b/tests/unittest/test_flash_html.py @@ -227,7 +227,7 @@ def test_html_parse_server_flash_only_admits_extensionless_url( with TestClient(app) as client: response = client.post( "/v1/tasks", - json={"manifest": [{"client_file_id": "file", "filename": "article.html", "url": "https://example.com/article"}]}, + json={"files": [{"client_file_id": "file", "file_name": "article.html", "url": "https://example.com/article"}]}, ) assert response.status_code == 202 diff --git a/tests/unittest/test_parser_api_contract.py b/tests/unittest/test_parser_api_contract.py index 2d5c98ed..27aab577 100644 --- a/tests/unittest/test_parser_api_contract.py +++ b/tests/unittest/test_parser_api_contract.py @@ -1008,7 +1008,7 @@ def test_api_server_basic_rejects_advanced_request(tmp_path: Path, monkeypatch: response = client.post( "/v1/tasks", json={ - "manifest": [{"client_file_id": "file", "filename": source.name, "bytes": source.stat().st_size}], + "files": [{"client_file_id": "file", "file_name": source.name, "bytes": source.stat().st_size}], "tier": "advanced", }, ) @@ -1032,7 +1032,7 @@ def test_api_server_no_advanced_rejects_advanced_request(tmp_path: Path, monkeyp response = client.post( "/v1/tasks", json={ - "manifest": [{"client_file_id": "file", "filename": source.name, "bytes": source.stat().st_size}], + "files": [{"client_file_id": "file", "file_name": source.name, "bytes": source.stat().st_size}], "tier": "advanced", }, ) @@ -1054,7 +1054,7 @@ def test_api_server_no_flash_rejects_flash_request(tmp_path: Path, monkeypatch: response = TestClient(app).post( "/v1/tasks", json={ - "manifest": [{"client_file_id": "file", "filename": source.name, "bytes": source.stat().st_size}], + "files": [{"client_file_id": "file", "file_name": source.name, "bytes": source.stat().st_size}], "tier": "flash", }, ) diff --git a/tests/unittest/test_parser_api_ocr.py b/tests/unittest/test_parser_api_ocr.py index 846821fd..4fd5e93b 100644 --- a/tests/unittest/test_parser_api_ocr.py +++ b/tests/unittest/test_parser_api_ocr.py @@ -61,10 +61,10 @@ def test_api_parser_serializes_request_ocr_mode( parser.parse(source, page_range="1") expected: dict[str, Any] = { - "manifest": [ + "files": [ { "client_file_id": "file", - "filename": source.name, + "file_name": source.name, "bytes": source.stat().st_size, "sha256sum": api_client._sha256_file(source), "page_range": "1", @@ -94,7 +94,7 @@ def test_api_server_rejects_invalid_request_ocr_mode(tmp_path: Path, value: obje response = client.post( "/v1/tasks", json={ - "manifest": [{"client_file_id": "file", "filename": "demo.pdf", "bytes": 9}], + "files": [{"client_file_id": "file", "file_name": "demo.pdf", "bytes": 9}], "ocr_mode": value, }, ) @@ -106,7 +106,7 @@ def test_api_server_rejects_invalid_request_ocr_mode(tmp_path: Path, value: obje def test_api_server_ocr_defaults_to_request_auto_without_startup_configuration(tmp_path: Path) -> None: """请求缺省值固定为 auto，应用生命周期不保存全局 OCR 模式。""" - request = CreateTaskRequest.model_validate({"manifest": [{"client_file_id": "file", "filename": "demo.pdf", "bytes": 9}]}) + request = CreateTaskRequest.model_validate({"files": [{"client_file_id": "file", "file_name": "demo.pdf", "bytes": 9}]}) assert request.ocr_mode == "auto" app = api_server.create_app(upload_dir=str(tmp_path), tier="flash") assert not hasattr(app.state, "ocr_mode") diff --git a/tests/unittest/test_pdf_page_id.py b/tests/unittest/test_pdf_page_id.py index a67a59c3..3f1b2619 100644 --- a/tests/unittest/test_pdf_page_id.py +++ b/tests/unittest/test_pdf_page_id.py @@ -176,16 +176,16 @@ def test_api_rejects_bad_ranges_before_creating_jobs(tmp_path: Path) -> None: "/v1/tasks", json={ "tier": "flash", - "manifest": [ - {"client_file_id": "first", "filename": "a.pdf", "bytes": 1}, - {"client_file_id": "second", "filename": "b.pdf", "bytes": 1, "page_range": raw}, + "files": [ + {"client_file_id": "first", "file_name": "a.pdf", "bytes": 1}, + {"client_file_id": "second", "file_name": "b.pdf", "bytes": 1, "page_range": raw}, ], }, ) assert response.status_code == 400 error = response.json()["error"] assert error["code"] == "page_range_invalid" - assert error["param"] == "manifest.1.page_range" + assert error["param"] == "files.1.page_range" schema = client.get("/openapi.json").json() assert "r1" in schema["components"]["schemas"]["TaskFileStatus"]["properties"]["page_range"]["description"] diff --git a/tests/unittest/test_resource_retention.py b/tests/unittest/test_resource_retention.py index 2ca93d37..225695ad 100644 --- a/tests/unittest/test_resource_retention.py +++ b/tests/unittest/test_resource_retention.py @@ -242,7 +242,7 @@ def test_hash_preflight_rejection_releases_request_lease(tmp_path: Path, monkeyp "/v1/tasks", json={ "tier": "flash", - "manifest": [{"client_file_id": "f1", "filename": "input.pdf", "bytes": 3, "sha256sum": sha}], + "files": [{"client_file_id": "f1", "file_name": "input.pdf", "bytes": 3, "sha256sum": sha}], }, ) assert response.status_code == 400 diff --git a/tests/unittest/test_router_balancing.py b/tests/unittest/test_router_balancing.py index 05fc881e..df0769c8 100644 --- a/tests/unittest/test_router_balancing.py +++ b/tests/unittest/test_router_balancing.py @@ -18,7 +18,7 @@ from mineru.parser import api_server from mineru.kit.router import RouterSettings, create_app from mineru.kit.router.workers import JobReservation -BODY = {"tier": "standard", "manifest": [{"client_file_id": "file", "filename": "source.pdf", "bytes": 5}]} +BODY = {"tier": "standard", "files": [{"client_file_id": "file", "file_name": "source.pdf", "bytes": 5}]} @pytest.mark.parametrize("formats", [["unknown"], "markdown", None, 1, {}, [{}], [1]]) @@ -77,10 +77,10 @@ def test_task_auth_preflight_uses_a_capable_worker_without_reservation() -> None ) form = { "tier": "standard", - "manifest": [ + "files": [ { "client_file_id": "f1", - "filename": "source.pdf", + "file_name": "source.pdf", "bytes": 3, } ], @@ -127,7 +127,7 @@ def test_cached_tasks_still_balance(tmp_path: Path, monkeypatch: pytest.MonkeyPa assert task["uploads"] == [] tasks.append(task) deadline = time.monotonic() + 5 - while not app.state.task_store._records[task["task_id"]].files[0].submitted: + while not app.state.task_store._records[task["task_id"]].file_states[0].submitted: assert time.monotonic() < deadline time.sleep(0.01) assert [worker.active_jobs for worker in app.state.worker_pool.workers] == [5, 5] diff --git a/tests/unittest/test_router_cascade.py b/tests/unittest/test_router_cascade.py index aa2b9089..285f132d 100644 --- a/tests/unittest/test_router_cascade.py +++ b/tests/unittest/test_router_cascade.py @@ -84,10 +84,10 @@ async def _submit(client: httpx.AsyncClient, *, filename: str = "input.html", ur "/v1/tasks", json={ "tier": "flash", - "manifest": [ + "files": [ { "client_file_id": "file", - "filename": filename, + "file_name": filename, "bytes": len(HTML), "sha256sum": hashlib.sha256(HTML).hexdigest(), **({"url": url} if url is not None else {}), @@ -171,7 +171,7 @@ def test_cascade_can_cancel_while_waiting_for_source_file(tmp_path: Path) -> Non await outer.state.task_store._records[task["task_id"]].operation for record in inner.state.task_store._records.values(): # 下游仍有结果租约，取消清理只需等待实际子文件退出。 - await asyncio.gather(*(file.operation for file in record.files if file.operation is not None)) + await asyncio.gather(*(file.operation for file in record.file_states if file.operation is not None)) assert not worker.state.job_store._jobs for app in (outer, inner): assert all(item.active_jobs == 0 for item in app.state.worker_pool.workers) diff --git a/tests/unittest/test_task_api.py b/tests/unittest/test_task_api.py index 0b34eb27..7bf8476c 100644 --- a/tests/unittest/test_task_api.py +++ b/tests/unittest/test_task_api.py @@ -60,7 +60,7 @@ def _url_manifest(**metadata: Any) -> dict[str, Any]: """构造与下载地址名称无关的逻辑文件条目，默认不声明大小或 hash。""" return { "client_file_id": "remote", - "filename": "logical.html", + "file_name": "logical.html", "url": "https://source.example/opaque?signature=a%2Fb&part=1", **metadata, } @@ -120,13 +120,13 @@ def _submit( manifest = [ { "client_file_id": f"f{index + 1}", - "filename": name, + "file_name": name, "bytes": len(data), **({"sha256sum": hashlib.sha256(data).hexdigest()} if hashes else {}), } for index, (name, data) in enumerate(inputs) ] - response = client.post("/v1/tasks", json={"manifest": manifest, "tier": "flash", **options}, headers=headers) + response = client.post("/v1/tasks", json={"files": manifest, "tier": "flash", **options}, headers=headers) assert response.status_code == 202, response.text return response.json() @@ -274,7 +274,7 @@ def test_no_hash_still_deduplicates_source_storage(service: tuple[TestClient, Fa _upload(client, second, [("second.html", HTML)]) assert _terminal(client, second)["status"] == "completed" records = app.state.task_store._records - assert records[first["task_id"]].files[0].upload.path == records[second["task_id"]].files[0].upload.path + assert records[first["task_id"]].file_states[0].upload.path == records[second["task_id"]].file_states[0].upload.path def test_waiting_upload_retains_deleted_source_views(service: tuple[TestClient, FastAPI]) -> None: @@ -362,7 +362,7 @@ def test_cancel_and_expire_waiting_upload_release_inputs(service: tuple[TestClie client.portal.call(wait_cleanup) assert not record.directory.exists() - assert all(file.backend._owned_files == [] for file in record.files) + assert all(file.backend._owned_files == [] for file in record.file_states) def test_cancel_running_task_releases_execution_slot( @@ -465,7 +465,7 @@ def test_partial_and_packaging_failure_have_honest_results( def test_input_validation_and_openapi(service: tuple[TestClient, FastAPI]) -> None: """请求与 OpenAPI 都使用严格 JSON 数组，旧格式与额外字段在接单前拒绝。""" client, app = service - valid = {"client_file_id": "f1", "filename": "a.html", "bytes": len(HTML)} + valid = {"client_file_id": "f1", "file_name": "a.html", "bytes": len(HTML)} for manifest in ( [], [valid] * 2, @@ -475,12 +475,20 @@ def test_input_validation_and_openapi(service: tuple[TestClient, FastAPI]) -> No [{**valid, "page_range": "invalid"}], json.dumps([valid]), ): - assert client.post("/v1/tasks", json={"manifest": manifest}).status_code == 400 + assert client.post("/v1/tasks", json={"files": manifest}).status_code == 400 for formats in ([], "default", ["markdown"], ["zip"], ["pdf"], [None]): - assert client.post("/v1/tasks", json={"manifest": [valid], "output_formats": formats}).status_code == 400 - assert client.post("/v1/tasks?tier=flash", json={"manifest": [valid]}).status_code == 400 - assert client.post("/v1/tasks", json={"manifest": [valid], "wait_timeout": 300}).status_code == 400 + assert client.post("/v1/tasks", json={"files": [valid], "output_formats": formats}).status_code == 400 + assert client.post("/v1/tasks?tier=flash", json={"files": [valid]}).status_code == 400 + assert client.post("/v1/tasks", json={"files": [valid], "wait_timeout": 300}).status_code == 400 assert client.post("/v1/tasks", json={"files": []}).status_code == 400 + for body in ( + {"manifest": [valid]}, + {"files": [valid], "manifest": [valid]}, + {"files": [{"filename": "a.html", "bytes": len(HTML)}]}, + {"files": [{**valid, "filename": "a.html"}]}, + {"files": [{"url": "https://source.example/a.html"}]}, + ): + assert client.post("/v1/tasks", json=body).status_code == 400 assert client.post("/v1/tasks", content=b"{", headers={"content-type": "application/json"}).status_code == 400 assert _job_count(app) == 0 spec = client.get("/openapi.json").json() @@ -488,7 +496,23 @@ def test_input_validation_and_openapi(service: tuple[TestClient, FastAPI]) -> No creation = spec["paths"]["/v1/tasks"]["post"] assert not creation.get("parameters") assert set(creation["requestBody"]["content"]) == {"application/json"} - assert creation["requestBody"]["content"]["application/json"]["schema"]["properties"]["manifest"]["type"] == "array" + request_schema = creation["requestBody"]["content"]["application/json"]["schema"] + assert request_schema["required"] == ["files"] + assert "manifest" not in request_schema["properties"] + files_schema = request_schema["properties"]["files"] + assert files_schema["type"] == "array" + assert files_schema["minItems"] == 1 and files_schema["maxItems"] == 100 + assert "file_name" in files_schema["items"]["required"] + assert "client_file_id" not in files_schema["items"]["required"] + assert "filename" not in files_schema["items"]["properties"] + for model in ("TaskFileStatus", "TaskUploadLink"): + id_schema = spec["components"]["schemas"][model] + assert "client_file_id" in id_schema["required"] + assert id_schema["properties"]["client_file_id"]["type"] == "string" + if hasattr(app.state, "job_store"): + metadata_schema = spec["components"]["schemas"]["FileObjectModel"] + assert "file_name" in metadata_schema["required"] + assert "filename" not in metadata_schema["properties"] assert "result_url" not in spec["components"]["schemas"]["TaskStatusResponse"]["properties"] assert "result_url" in spec["components"]["schemas"]["TaskFileStatus"]["properties"] assert "output_files" not in spec["components"]["schemas"]["TaskFileStatus"]["properties"] @@ -497,6 +521,84 @@ def test_input_validation_and_openapi(service: tuple[TestClient, FastAPI]) -> No assert limits["max_files_per_job"] == 100 +@pytest.mark.parametrize( + ("identities", "expected_ids"), + [ + ([{}, {}], ["f1", "f2"]), + ([{"client_file_id": None}, {}], ["f1", "f2"]), + ([{}, {"client_file_id": "f2"}, {}], ["f1", "f2", "f3"]), + ([{}, {"client_file_id": "f1"}, {}], ["f2", "f1", "f3"]), + ([{"client_file_id": "custom"}, {"client_file_id": "other"}], ["custom", "other"]), + ], +) +def test_optional_file_ids_survive_upload_and_router_forwarding( + service: tuple[TestClient, FastAPI], identities: list[dict[str, str | None]], expected_ids: list[str] +) -> None: + """缺省标识避让后续显式项，同名输入在重复查询、上传重试和 Router 转发后仍保持关联。""" + client, app = service + response = client.post( + "/v1/tasks", + json={"files": [{"file_name": "same.html", "bytes": len(HTML), **identity} for identity in identities]}, + ) + assert response.status_code == 202, response.text + task = response.json() + for _ in range(2): + assert [item["client_file_id"] for item in task["files"]] == expected_ids + assert [item["client_file_id"] for item in task["uploads"]] == expected_ids + assert all(item["file_name"] == "same.html" and "filename" not in item for item in task["files"]) + task = client.get(task["status_url"]).json() + for link in task["uploads"]: + for _ in range(2): + assert client.put(link["upload_url"], content=HTML, headers=link["upload_headers"]).status_code == 200 + completed = _terminal(client, task) + assert completed["status"] == "completed", completed + assert [item["client_file_id"] for item in completed["files"]] == expected_ids + assert _job_count(app) == len(expected_ids) + for index in range(len(expected_ids)): + assert b"Preserved content" in _archive(client, completed, index)["markdown.md"] + if hasattr(app.state, "test_transport"): + forwarded = [ + item.client_file_id + for worker in _execution_apps(app) + for record in worker.state.task_store._records.values() + for item in record.inputs + ] + assert sorted(forwarded) == sorted(expected_ids) + + +@pytest.mark.parametrize("client_file_id", ["", "with space", "bad/id", "x" * 65, 123, False]) +def test_invalid_explicit_file_ids_fail_before_creation(service: tuple[TestClient, FastAPI], client_file_id: object) -> None: + """可选标识不放宽显式值校验，错误仍定位到新 files 字段且不创建任务。""" + client, app = service + response = client.post( + "/v1/tasks", json={"files": [{"file_name": "a.html", "bytes": len(HTML), "client_file_id": client_file_id}]} + ) + assert response.status_code == 400 + assert response.json()["error"]["param"] == "files.0.client_file_id" + assert not app.state.task_store._records and _job_count(app) == 0 + + +def test_files_metadata_and_attachment_keep_logical_names(service: tuple[TestClient, FastAPI]) -> None: + """Tasks 和 Files 只返回新文件名字段，源文件与结果附件仍使用各自逻辑名称。""" + client, _ = service + inputs = [("report.v2.html", HTML)] + task = _submit(client, inputs) + _upload(client, task, inputs) + completed = _terminal(client, task) + item = completed["files"][0] + assert item["file_name"] == item["name"] == "report.v2.html" and "filename" not in item + source = client.get(f"/v1/files/{item['file_id']}").json() + result = client.get(item["result_url"].removesuffix("/content")).json() + assert source["file_name"] == "report.v2.html" and "filename" not in source + assert result["file_name"] == "report.v2.zip" and "filename" not in result + listed = client.get("/v1/files").json()["data"] + assert {entry["file_name"] for entry in listed} == {"report.v2.html", "report.v2.zip"} + assert all("filename" not in entry for entry in listed) + download = client.get(item["result_url"]) + assert download.status_code == 200 + assert download.headers["content-disposition"] == 'attachment; filename="report.v2.zip"' + + @pytest.mark.parametrize( "metadata", [ @@ -506,6 +608,7 @@ def test_input_validation_and_openapi(service: tuple[TestClient, FastAPI]) -> No {"sha256sum": hashlib.sha256(HTML).hexdigest()}, {"bytes": len(HTML), "sha256sum": hashlib.sha256(HTML).hexdigest()}, {"url": "https://source.example/opaque?signature=" + "a" * 3000}, + {"client_file_id": None}, ], ) def test_url_inputs_download_in_worker_and_report_actual_metadata( @@ -516,23 +619,26 @@ def test_url_inputs_download_in_worker_and_report_actual_metadata( item = _url_manifest(**metadata) if hasattr(app.state, "worker_pool"): app.state.worker_pool.workers[0].features["sources"] = ["file_id"] - response = client.post("/v1/tasks", json={"manifest": [item]}) + response = client.post("/v1/tasks", json={"files": [item]}) assert response.status_code == 202, response.text task = response.json() + expected_id = item["client_file_id"] or "f1" + assert task["files"][0]["client_file_id"] == expected_id assert task["status"] == "queued" and task["uploads"] == [] assert task["files"][0]["bytes"] == metadata.get("bytes") assert task["files"][0]["upload_status"] == "not_required" result = _terminal(client, task) assert result["status"] == "completed" file = result["files"][0] - assert file["filename"] == file["name"] == "logical.html" + assert file["client_file_id"] == expected_id + assert file["file_name"] == file["name"] == "logical.html" assert file["bytes"] == len(HTML) and file["sha256sum"] == hashlib.sha256(HTML).hexdigest() assert file["upload_status"] == "not_required" and file["file_id"] is None assert file["url"] == item["url"] assert b"Preserved content" in _archive(client, result)["markdown.md"] assert len(url_downloads) == 1 and str(url_downloads[0].url) == item["url"] assert url_downloads[0].method == "GET" and "authorization" not in url_downloads[0].headers - assert client.put(f"/v1/tasks/{task['task_id']}/uploads/remote", content=HTML).status_code == 400 + assert client.put(f"/v1/tasks/{task['task_id']}/uploads/{expected_id}", content=HTML).status_code == 400 if hasattr(app.state, "test_transport"): assert not any(method == "PUT" for _, method, _ in app.state.test_transport.calls) assert all(host == "worker-b" for host, method, path in app.state.test_transport.calls if path == "/v1/tasks") @@ -542,13 +648,13 @@ def test_url_manifest_validation(service: tuple[TestClient, FastAPI], url_downlo """大小只对 URL 可选，非法 URL、非正整数字节数及混合直传在接单前拒绝。""" client, app = service invalid = [ - {"client_file_id": "f1", "filename": "a.html"}, - {"client_file_id": "f1", "filename": "a.html", "bytes": None}, + {"client_file_id": "f1", "file_name": "a.html"}, + {"client_file_id": "f1", "file_name": "a.html", "bytes": None}, *[_url_manifest(url=value) for value in ("", "file:///tmp/a.html", "https://", "ftp://source/a.html")], *[_url_manifest(bytes=value) for value in (0, -1, True, "10", 10.5)], ] for item in invalid: - response = client.post("/v1/tasks", json={"manifest": [item]}) + response = client.post("/v1/tasks", json={"files": [item]}) assert response.status_code == 400, response.text response = client.post("/v1/tasks", files={"files": ("a.html", HTML)}) assert response.status_code == 415 @@ -563,11 +669,11 @@ def test_mixed_url_upload_partial_and_no_early_reservation( """URL 文件不等待客户端上传，错误和完成结果都按清单顺序独立返回。""" client, app = service manifest = [ - _url_manifest(filename="same.html"), - {"client_file_id": "local", "filename": "same.html", "bytes": len(SECOND)}, + _url_manifest(file_name="same.html"), + {"client_file_id": "local", "file_name": "same.html", "bytes": len(SECOND)}, _url_manifest(client_file_id="bad", url="https://source.example/missing"), ] - response = client.post("/v1/tasks", json={"manifest": manifest}) + response = client.post("/v1/tasks", json={"files": manifest}) assert response.status_code == 202, response.text task = response.json() assert task["status"] == "queued" @@ -599,7 +705,7 @@ def test_url_download_failures_are_per_file( ) -> None: """下载及完整性错误进入文件错误；全失败不生成结果链接或实际元数据。""" client, _ = service - response = client.post("/v1/tasks", json={"manifest": [_url_manifest(**metadata)]}) + response = client.post("/v1/tasks", json={"files": [_url_manifest(**metadata)]}) assert response.status_code == 202 result = _terminal(client, response.json()) assert result["status"] == "failed" and result["files"][0]["result_url"] is None @@ -616,7 +722,7 @@ def test_url_hash_does_not_skip_download(service: tuple[TestClient, FastAPI], ur _upload(client, cached, inputs) assert _terminal(client, cached)["status"] == "completed" for _ in range(2): - response = client.post("/v1/tasks", json={"manifest": [_url_manifest(sha256sum=hashlib.sha256(HTML).hexdigest())]}) + response = client.post("/v1/tasks", json={"files": [_url_manifest(sha256sum=hashlib.sha256(HTML).hexdigest())]}) assert response.status_code == 202 assert _terminal(client, response.json())["status"] == "completed" assert len(url_downloads) == 2 @@ -628,7 +734,7 @@ def test_url_download_obeys_worker_policy_and_limit( """HTTP 开关和 URL 下载预算继续由真正执行下载的服务控制。""" client, app = service item = _url_manifest(url="http://source.example/opaque") - response = client.post("/v1/tasks", json={"manifest": [item]}) + response = client.post("/v1/tasks", json={"files": [item]}) if hasattr(app.state, "worker_pool"): assert response.status_code == 202 assert _terminal(client, response.json())["status"] == "failed" @@ -639,14 +745,14 @@ def test_url_download_obeys_worker_policy_and_limit( worker.state.allow_http_source = True worker.state.max_url_bytes = len(HTML) - 1 worker.state.url_timeout = 7 - response = client.post("/v1/tasks", json={"manifest": [item]}) + response = client.post("/v1/tasks", json={"files": [item]}) result = _terminal(client, response.json()) assert result["status"] == "failed" assert "max_url_bytes" in result["files"][0]["error"]["message"] assert url_downloads[0].extensions["timeout"]["read"] == 7 for worker in _execution_apps(app): worker.state.max_url_bytes = len(HTML) - response = client.post("/v1/tasks", json={"manifest": [item]}) + response = client.post("/v1/tasks", json={"files": [item]}) assert _terminal(client, response.json())["status"] == "completed" @@ -664,7 +770,7 @@ def test_url_html_preserves_source_context( return await original(*args, **kwargs) monkeypatch.setattr(api_server, "parse_async", inspect_context) - response = client.post("/v1/tasks", json={"manifest": [_url_manifest()]}) + response = client.post("/v1/tasks", json={"files": [_url_manifest()]}) assert _terminal(client, response.json())["status"] == "completed" assert len(contexts) == 1 and len(url_downloads) == 1 assert contexts[0].source_uri == _url_manifest()["url"] @@ -680,7 +786,7 @@ def test_url_logical_filename_drives_flash_fallback( response = client.post( "/v1/tasks", json={ - "manifest": [_url_manifest(url="https://source.example/document.pdf")], + "files": [_url_manifest(url="https://source.example/document.pdf")], "tier": "standard", }, ) @@ -724,7 +830,7 @@ def test_cancel_url_download_closes_stream_and_releases_worker( return httpx.AsyncClient(transport=httpx.MockTransport(respond), **kwargs) monkeypatch.setattr(api_server, "httpx", SimpleNamespace(AsyncClient=download_client)) - response = client.post("/v1/tasks", json={"manifest": [_url_manifest()]}) + response = client.post("/v1/tasks", json={"files": [_url_manifest()]}) assert response.status_code == 202 task = response.json() assert started.wait(5) @@ -773,8 +879,8 @@ def test_task_links_respect_proxy_root(tmp_path: Path, root: str) -> None: """状态、上传与结果链接全部携带部署前缀，客户端不用重建任何地址。""" app = api_server.create_app(upload_dir=str(tmp_path), tier="flash", api_key="secret") with TestClient(app, root_path=root, headers={"Authorization": "Bearer secret"}) as client: - manifest = [{"client_file_id": "f1", "filename": "a.html", "bytes": len(HTML)}] - response = client.post(root + "/v1/tasks", json={"manifest": manifest, "tier": "flash"}) + manifest = [{"client_file_id": "f1", "file_name": "a.html", "bytes": len(HTML)}] + response = client.post(root + "/v1/tasks", json={"files": manifest, "tier": "flash"}) assert response.status_code == 202 task = response.json() assert task["status_url"].startswith("http://testserver" + root + "/v1/tasks/") @@ -803,7 +909,7 @@ def test_inflight_upload_may_finish_after_deadline(service: tuple[TestClient, Fa async def body() -> AsyncIterator[bytes]: """第一块已经被接收后模拟到期，仍在传输的文件不能被清理。""" yield HTML[:10] - assert record.files[0].upload_active + assert record.file_states[0].upload_active record.expires_at = int(time.time()) - 1 state = (await transport.get(task["status_url"])).json() assert state["files"][0]["status"] == "waiting_upload" @@ -899,7 +1005,7 @@ def test_router_batch_uses_independent_workers_and_task_capabilities(tmp_path: P for worker in app.state.worker_pool.workers: worker.features.pop("task_output_formats") response = client.post( - "/v1/tasks", json={"manifest": [{"client_file_id": "f1", "filename": "a.html", "bytes": len(HTML)}]} + "/v1/tasks", json={"files": [{"client_file_id": "f1", "file_name": "a.html", "bytes": len(HTML)}]} ) assert response.status_code == 503 assert all(worker.active_jobs == 0 for worker in app.state.worker_pool.workers) @@ -1061,7 +1167,7 @@ def test_model_output_option_reaches_worker_and_result( response = client.post( "/v1/tasks", json={ - "manifest": [{"client_file_id": "file", "filename": "input.html", "bytes": len(HTML)}], + "files": [{"client_file_id": "file", "file_name": "input.html", "bytes": len(HTML)}], "tier": "flash", "include_model_output": enabled, }, @@ -1077,7 +1183,7 @@ def test_model_output_option_reaches_worker_and_result( rejected = client.post( "/v1/tasks", json={ - "manifest": [{"client_file_id": "file", "filename": "input.html", "bytes": len(HTML)}], + "files": [{"client_file_id": "file", "file_name": "input.html", "bytes": len(HTML)}], "output_formats": ["html"], "include_model_output": True, }, diff --git a/tests/unittest/test_task_http.py b/tests/unittest/test_task_http.py index 78297915..2a6d8173 100644 --- a/tests/unittest/test_task_http.py +++ b/tests/unittest/test_task_http.py @@ -134,13 +134,13 @@ def _submit(client: httpx.Client, files: list[tuple[str, bytes]], *, hashes: boo manifest = [ { "client_file_id": f"f{index + 1}", - "filename": name, + "file_name": name, "bytes": len(data), **({"sha256sum": hashlib.sha256(data).hexdigest()} if hashes else {}), } for index, (name, data) in enumerate(files) ] - response = client.post("/v1/tasks", json={"manifest": manifest, "tier": "flash", "ocr_mode": "txt", **options}) + response = client.post("/v1/tasks", json={"files": manifest, "tier": "flash", "ocr_mode": "txt", **options}) assert response.status_code == 202, response.text return response.json() @@ -165,6 +165,57 @@ def _members(data: bytes) -> dict[str, bytes]: return {name: archive.read(name) for name in archive.namelist()} +def test_pdf_without_client_id_keeps_metadata_and_cached_results(live_service: _LiveService, tmp_path: Path) -> None: + """真实 TCP 上传 PDF，确认自动标识、Files 元数据、附件命名及缓存命中后的正文均正确。""" + from reportlab.pdfgen.canvas import Canvas + + buffer = io.BytesIO() + canvas = Canvas(buffer, pagesize=(595, 842)) + canvas.drawString(72, 770, "Tasks field migration") + canvas.drawString(72, 740, "Generated file IDs stay stable across uploads and queries.") + canvas.save() + data = buffer.getvalue() + (tmp_path / "report.v2.pdf").write_bytes(data) + payload = { + "files": [{"file_name": "report.v2.pdf", "bytes": len(data), "sha256sum": hashlib.sha256(data).hexdigest()}], + "tier": "flash", + "ocr_mode": "txt", + } + with _client(live_service) as client: + response = client.post("/v1/tasks", json=payload) + assert response.status_code == 202, response.text + task = response.json() + assert task["files"][0]["client_file_id"] == task["uploads"][0]["client_file_id"] == "f1" + link = task["uploads"][0] + for _ in range(2): + upload = client.put(link["upload_url"], content=data, headers=link["upload_headers"]) + assert upload.status_code == 200, upload.text + completed = _wait(client, task) + assert completed["status"] == "completed", completed + item = completed["files"][0] + assert item["client_file_id"] == "f1" and item["file_name"] == "report.v2.pdf" and "filename" not in item + result = client.get(item["result_url"].removesuffix("/content")) + assert result.status_code == 200 and result.json()["file_name"] == "report.v2.zip" + download = client.get(item["result_url"]) + assert download.status_code == 200 + assert download.headers["content-disposition"] == 'attachment; filename="report.v2.zip"' + (tmp_path / "report.v2.zip").write_bytes(download.content) + members = _members(download.content) + assert b"Tasks field migration" in members["markdown.md"] + assert b"Generated file IDs stay stable" in members["markdown.md"] + assert len(json.loads(members["middle_json.json"])["pages"]) == 1 + cached = client.post("/v1/tasks", json=payload) + assert cached.status_code == 202, cached.text + cached_task = cached.json() + assert cached_task["uploads"] == [] and cached_task["files"][0]["client_file_id"] == "f1" + assert cached_task["files"][0]["upload_status"] == "ready" + cached_result = _wait(client, cached_task) + assert cached_result["status"] == "completed", cached_result + cached_download = client.get(cached_result["files"][0]["result_url"]) + assert cached_download.status_code == 200 + assert _members(cached_download.content)["markdown.md"] == members["markdown.md"] + + def test_router_keeps_tls_proxy_headers_on_the_public_hop_only(tmp_path: Path) -> None: """TLS 终止代理后的上传、hash 复用及下载保持可用，worker 不继承客户端来源头。""" calls: list[tuple[str, str, dict[str, str]]] = [] @@ -309,8 +360,8 @@ def test_hash_preflight_sends_only_missing_file_bytes(live_service: _LiveService "/v1/tasks", headers={"Authorization": "Bearer wrong"}, json={ - "manifest": [ - {"client_file_id": "f1", "filename": "a.html", "bytes": len(HTML)}, + "files": [ + {"client_file_id": "f1", "file_name": "a.html", "bytes": len(HTML)}, ] }, ) @@ -401,7 +452,7 @@ def test_upload_over_200_mib_is_accepted(live_service: _LiveService, tmp_path: P with _client(live_service) as client: response = client.post( "/v1/tasks", - json={"tier": "flash", "manifest": [{"client_file_id": "large", "filename": source.name, "bytes": size}]}, + json={"tier": "flash", "files": [{"client_file_id": "large", "file_name": source.name, "bytes": size}]}, ) assert response.status_code == 202, response.text task = response.json() @@ -507,7 +558,7 @@ def test_url_task_over_http_downloads_chunked_source(live_service: _LiveService, worker.state.allow_http_source = True item = { "client_file_id": "url-1", - "filename": f"logical.{document}", + "file_name": f"logical.{document}", "url": f"{source_server.url}/opaque?signature=a%2Fb&part=1", **({"page_range": "2"} if document == "pdf" else {}), } @@ -515,7 +566,7 @@ def test_url_task_over_http_downloads_chunked_source(live_service: _LiveService, response = client.post( "/v1/tasks", json={ - "manifest": [item], + "files": [item], "tier": "flash", "ocr_mode": "txt", "page_range": "1" if document == "pdf" else "", @@ -535,13 +586,13 @@ def test_url_task_over_http_downloads_chunked_source(live_service: _LiveService, assert "authorization" not in requests[0].headers if document == "html": # 同一原始 URL 保留 HTML 来源上下文，重复执行的产物应完全一致。 - response = client.post("/v1/tasks", json={"tier": "flash", "ocr_mode": "txt", "manifest": [item]}) + response = client.post("/v1/tasks", json={"tier": "flash", "ocr_mode": "txt", "files": [item]}) assert response.status_code == 202, response.text baseline_result = _wait(client, response.json()) assert baseline_result["status"] == "completed" baseline_url = baseline_result["files"][0]["result_url"] else: - baseline = _submit(client, [(item["filename"], data)], page_range="2") + baseline = _submit(client, [(item["file_name"], data)], page_range="2") assert client.put(baseline["uploads"][0]["upload_url"], content=data).status_code == 200 baseline_result = _wait(client, baseline) assert baseline_result["status"] == "completed" @@ -581,14 +632,14 @@ class _CloudState: record = self.tasks[task_id] ready, expired = record["ready"], record["expired"] files = [] - for index, item in enumerate(record["manifest"]): + for index, item in enumerate(record["files"]): client_id = item["client_file_id"] status = "completed" if client_id in ready else "waiting_upload" if client_id in expired: status = "failed" if client_id in ready and self.forced_status in {"failed", "running"}: status = self.forced_status - if self.forced_status == "partial" and index == len(record["manifest"]) - 1 and client_id in ready: + if self.forced_status == "partial" and index == len(record["files"]) - 1 and client_id in ready: status = "failed" result_url = None if status == "completed": @@ -598,7 +649,7 @@ class _CloudState: files.append( { **item, - "name": item["filename"], + "name": item["file_name"], "upload_status": "ready" if client_id in ready else ("expired" if client_id in expired else "pending"), "status": status, "result_url": result_url, @@ -623,7 +674,7 @@ class _CloudState: "upload_headers": {"Content-Type": "application/octet-stream", "X-Upload-Token": "storage-token"}, "expires_at": record["expires_at"], } - for item in record["manifest"] + for item in record["files"] if item["client_file_id"] not in ready | expired ] payload = { @@ -663,10 +714,10 @@ def _cloud_apps(state: _CloudState) -> tuple[FastAPI, FastAPI]: state.gateway_body_sizes.append(len(body)) assert len(body) < 4096 submitted = await request.json() - manifest = submitted["manifest"] + manifest = submitted["files"] task_id = f"cloud_{len(state.tasks) + 1}" state.tasks[task_id] = { - "manifest": manifest, + "files": manifest, "ready": {item["client_file_id"] for item in manifest if item.get("sha256sum") in state.cache}, "upload_version": 0, "download_version": 0, @@ -711,7 +762,7 @@ def _cloud_apps(state: _CloudState) -> tuple[FastAPI, FastAPI]: if state.transient_upload and state.upload_attempts[key] == 1: return Response(status_code=503) data = await request.body() - item = next(item for item in record["manifest"] if item["client_file_id"] == client_file_id) + item = next(item for item in record["files"] if item["client_file_id"] == client_file_id) assert len(data) == item["bytes"] actual = hashlib.sha256(data).hexdigest() assert item.get("sha256sum") in {None, actual} diff --git a/tests/unittest/test_task_outputs.py b/tests/unittest/test_task_outputs.py index 486fab54..355a9915 100644 --- a/tests/unittest/test_task_outputs.py +++ b/tests/unittest/test_task_outputs.py @@ -172,7 +172,7 @@ def test_task_formats_use_only_service_authentication(tmp_path: Path, formats: l """启用 Key 时四种格式都需要服务认证，通过认证后都可执行。""" app = api_server.create_app(upload_dir=str(tmp_path), tier="flash", api_key="secret") with TestClient(app) as client: - body = {"manifest": [{"client_file_id": "f1", "filename": "a.html", "bytes": len(HTML)}], "output_formats": formats} + body = {"files": [{"client_file_id": "f1", "file_name": "a.html", "bytes": len(HTML)}], "output_formats": formats} assert client.post("/v1/tasks", json=body).status_code == 401 headers = {"Authorization": "Bearer secret"} task = _submit(client, [("a.html", HTML)], output_formats=formats, headers=headers) diff --git a/tests/unittest/test_tasks_sdk.py b/tests/unittest/test_tasks_sdk.py index 8633858c..7b34db72 100644 --- a/tests/unittest/test_tasks_sdk.py +++ b/tests/unittest/test_tasks_sdk.py @@ -108,8 +108,8 @@ def test_tasks_sdk_upload_cache_and_lost_response( if request.method == "POST": assert request.url.path == "/v1/tasks" body = json.loads(request.content) - assert body["manifest"][0]["sha256sum"] == hashlib.sha256(source.read_bytes()).hexdigest() - assert body["manifest"][0]["page_range"] == "1" + assert body["files"][0]["sha256sum"] == hashlib.sha256(source.read_bytes()).hexdigest() + assert body["files"][0]["page_range"] == "1" assert body["ocr_mode"] == "txt" and body["tier"] == "flash" return httpx.Response(202, json=copy.deepcopy(state)) if request.method == "PUT": @@ -270,7 +270,7 @@ def test_task_model_output_requires_default() -> None: """不把原本单文件的 HTML/DOCX 请求隐式改成 ZIP。""" with pytest.raises(ValueError, match="requires default"): CreateTaskRequest( - manifest=[{"client_file_id": "file", "filename": "a.pdf", "bytes": 1}], + files=[{"client_file_id": "file", "file_name": "a.pdf", "bytes": 1}], output_formats=["html"], include_model_output=True, )