feat: refine page range handling for PDF and non-PDF inputs, ensuring only PDF supports explicit ranges

2026-08-30 17:12:39 +08:00 · 2026-08-26 19:50:40 +08:00
parent dac6aa757d
commit b39de7d54d
24 changed files with 523 additions and 117 deletions
@@ -873,7 +873,7 @@ Use this table for common error codes:
 | `file_type_unsupported` | Format unsupported | Report unsupported type |
 | `file_encrypted` | Password-protected file | Ask user for an unlocked copy |
 | `file_corrupted` | File cannot be read | Ask for a valid copy |
-| `page_range_invalid` | Bad PDF/image `--pages` value, or `--pages` used with a non-PDF Flash input | Correct the range or omit it for full-document parsing |
+| `page_range_invalid` | Bad PDF `--pages` value, or `--pages` used with any non-PDF input | Correct the PDF range or omit it for full-document parsing |
 | `parse_not_required` | The input is a directly readable text file | Read the source file directly; do not retry `mineru parse` |
 | `not_cached` / `cache_miss` | Requested cached content does not exist | Run `mineru parse` |

@@ -330,7 +330,7 @@ If your device does not meet the GPU acceleration requirements, you can explicit
 mineru parse "<input_path>" --tier flash -o <output_path>
 ```

-`mineru` currently supports local `PDF`, `EPUB`, image, `CSV`, `RTF`, `DOCX`, `PPTX`, `XLSX`, `ODT`, `ODS`, and `ODP` file or directory inputs through the CLI, API, and `mineru-router`. EPUB, RTF, and OpenDocument use local semantic Flash parsing without external office-suite conversion. Non-PDF Flash inputs are always parsed as full documents; EPUB pages follow OPF spine order and preserve resolvable internal links from authored contents, including list and table layouts. The legacy Gradio WebUI keeps its existing advertised upload formats. For detailed instructions, please refer to the [Usage Guide](https://opendatalab.github.io/MinerU/usage/).
+`mineru` currently supports local `PDF`, `EPUB`, image, `CSV`, `RTF`, `DOCX`, `PPTX`, `XLSX`, `ODT`, `ODS`, and `ODP` file or directory inputs through the CLI, API, and `mineru-router`. EPUB, RTF, and OpenDocument use local semantic Flash parsing without external office-suite conversion. Only PDF accepts page-range parsing; every non-PDF input is parsed and cached as one complete file, while slides, sheets, EPUB spine items, and explicit page breaks remain logical pages in structured output. EPUB pages follow OPF spine order and preserve resolvable internal links from authored contents, including list and table layouts. The legacy Gradio WebUI keeps its existing advertised upload formats. For detailed instructions, please refer to the [Usage Guide](https://opendatalab.github.io/MinerU/usage/).

 # FAQ

@@ -336,7 +336,7 @@ mineru -p <input_path> -o <output_path>
 mineru parse "<input_path>" --tier flash -o <output_path>
 ```

-当前 `mineru` 支持本地 `PDF / EPUB / 图片 / CSV / RTF / DOCX / PPTX / XLSX / ODT / ODS / ODP` 文件或目录输入，并可通过命令行、API、`mineru-router` 进行解析。EPUB、RTF 与 OpenDocument 使用本地 Flash 语义解析，不依赖外部 Office 套件转换；所有非 PDF Flash 输入只支持整本解析，EPUB 页面严格遵循 OPF spine 顺序，并保留作者目录中可解析的列表、表格等内部跳转。旧 Gradio WebUI 仍保持原有对外说明。具体使用方法请参考[使用指南](https://opendatalab.github.io/MinerU/zh/usage/)。
+当前 `mineru` 支持本地 `PDF / EPUB / 图片 / CSV / RTF / DOCX / PPTX / XLSX / ODT / ODS / ODP` 文件或目录输入，并可通过命令行、API、`mineru-router` 进行解析。EPUB、RTF 与 OpenDocument 使用本地 Flash 语义解析，不依赖外部 Office 套件转换；只有 PDF 接受页范围解析，其他输入均按一个完整文件解析和缓存，但 slide、sheet、EPUB spine 项及显式分页仍作为结构化输出中的逻辑页保留。EPUB 页面严格遵循 OPF spine 顺序，并保留作者目录中可解析的列表、表格等内部跳转。旧 Gradio WebUI 仍保持原有对外说明。具体使用方法请参考[使用指南](https://opendatalab.github.io/MinerU/zh/usage/)。


 # FAQ
@@ -148,6 +148,6 @@ If your device does not meet the GPU acceleration requirements, you can explicit
 mineru parse "<input_path>" --tier flash -o <output_path>
 ```

-`mineru` currently supports local `PDF`, `EPUB`, image, `CSV`, `RTF`, `DOCX`, `PPTX`, `XLSX`, `ODT`, `ODS`, and `ODP` file or directory inputs through modern CLI and API entry points. EPUB, RTF, and OpenDocument are parsed locally with Flash semantics and do not require an external office suite. Non-PDF Flash inputs only support full-document parsing; EPUB pages follow OPF spine order and preserve resolvable internal links from authored contents.
+`mineru` currently supports local `PDF`, `EPUB`, image, `CSV`, `RTF`, `DOCX`, `PPTX`, `XLSX`, `ODT`, `ODS`, and `ODP` file or directory inputs through modern CLI and API entry points. EPUB, RTF, and OpenDocument are parsed locally with Flash semantics and do not require an external office suite. Only PDF accepts page ranges; every non-PDF input is parsed and cached as one complete file, while format-native boundaries remain logical pages in structured output. EPUB pages follow OPF spine order and preserve resolvable internal links from authored contents.

 The legacy Gradio WebUI keeps its existing upload formats and does not yet accept CSV. For detailed instructions, please refer to the [Usage Guide](../usage/index.md).
@@ -31,7 +31,7 @@ Options:
  --help                          Show help information
 ```
 > [!TIP]
-> `mineru` currently supports local `PDF`, `EPUB`, image, `CSV`, `RTF`, `DOCX`, `PPTX`, `XLSX`, `ODT`, `ODS`, and `ODP` file or directory inputs. Non-PDF Flash formats only support full-document parsing; EPUB pages follow OPF spine order and preserve resolvable authored internal links.
+> `mineru` currently supports local `PDF`, `EPUB`, image, `CSV`, `RTF`, `DOCX`, `PPTX`, `XLSX`, `ODT`, `ODS`, and `ODP` file or directory inputs. Only PDF accepts page ranges; every non-PDF input is parsed and cached as one complete file, while format-native boundaries remain logical pages in structured output. EPUB pages follow OPF spine order and preserve resolvable authored internal links.

 ```bash
 mineru-api --help
@@ -60,7 +60,7 @@ Job 状态:
 | 字段 | 类型 | 必填 | 说明 |
 |------|------|:--:|------|
 | `source` | object | 是 | 文件来源。 |
-| `page_range` | string | 否 | 针对单个 PDF/图片文件的页码范围；非 PDF Flash 输入显式传入时返回 `page_range_invalid`。 |
+| `page_range` | string | 否 | 仅针对单个 PDF 文件的页码范围；图片及其他非 PDF 输入显式传入时返回 `page_range_invalid`。 |

 ### Source 类型

@@ -122,7 +122,7 @@ Local Parse Server 的 source 策略由启动参数决定:

 | 字段 | 类型 | 默认 | 说明 |
 |------|------|------|------|
-| `page_range` | string | `null` | PDF/图片页码范围。省略或传 `null` 表示解析整个文件；非 PDF Flash 输入只允许省略或传 `null`。推荐 `~` 分隔，如 `1~10`、`1,3,5~7`、`-5~-1`。 |
+| `page_range` | string | `null` | PDF 页码范围。省略或传 `null` 表示解析整个文件；图片及其他非 PDF 输入只允许省略或传 `null`。推荐 `~` 分隔，如 `1~10`、`1,3,5~7`、`-5~-1`。 |

 OCR 策略和图片分析能力由 `tier` 与服务端实际引擎自动决定，客户端不能通过文件级参数单独关闭或开启。

@@ -256,7 +256,7 @@ watch 流程也调用同一发现/刷新步骤:
 1. CLI 或 SDK 调用 doclib `POST /parses`。
 2. `ParseService` 先执行文件发现/刷新，确认 path 当前状态。
 3. 如文件需要重新入库，则等待或同步完成入库，获得当前 `sha256`。
-4. 查询 `(sha256, tier)` 下已完成批次，判断 page_range 是否覆盖请求范围。
+4. 查询 `(sha256, tier)` 下已完成批次；PDF 判断 page_range 覆盖，非 PDF 只接受完整整文件 batch。
 5. 缓存命中则返回 `cache_hit=true` 和空 `wait_parse_ids`。
 6. 已有 active parse 覆盖的页复用并提升 `priority`。
 7. 未覆盖页插入新的 `parses` 记录，Agent 请求使用更高 `priority`。
@@ -412,8 +412,9 @@ Watch、parsing-rules 和 exclude 规则也由 CLI 写入 SQLite。Parsing-rules
 | 类型 | 策略 |
 |------|------|
 | 纯文本 | 直接读取，无需模型解析 |
+| Image | 按 tier 路由到默认选择 / flash / basic / standard / advanced，但整文件解析并拒绝显式 page_range |
 | EPUB / Office / HTML / CSV | 本地 CPU Flash 整本解析；显式 page_range 拒绝，EPUB 可增加首页目录 |
-| PDF / Image | 按 tier 路由到默认选择 / flash / basic / standard / advanced |
+| PDF | 按 tier 路由，并支持增量 page_range 解析和缓存 |

 ## 9. 错误与恢复

@@ -30,7 +30,7 @@ mineru-kit parse <input...> -o <output> [flags]
 | 多文件 | 一次提交多个文件 |
 | 目录 | 展开目录中的可解析文件 |

-当前支持的扩展名包括 PDF、EPUB、常见图片、CSV、RTF、DOC/DOCX、PPT/PPTX、XLS/XLSX、ODT/ODS/ODP、HTML/HTM。目录输入只展开一层，不递归；非 PDF Flash 输入只支持整本解析，显式 `--pages` 返回 `page_range_invalid`。EPUB 页面严格遵循 OPF spine 顺序，并保留作者目录中可解析的内部跳转。
+当前支持的扩展名包括 PDF、EPUB、常见图片、CSV、RTF、DOC/DOCX、PPT/PPTX、XLS/XLSX、ODT/ODS/ODP、HTML/HTM。目录输入只展开一层，不递归；只有 PDF 支持 `--pages`，图片及其他非 PDF 输入均整文件解析，显式 `--pages` 返回 `page_range_invalid`。EPUB 页面严格遵循 OPF spine 顺序，并保留作者目录中可解析的内部跳转。

 当前不支持：

@@ -340,7 +340,7 @@ mineru config parsing-rules add "*/论文/*" --tier basic --pages all
 mineru config parsing-rules add "*/合同/*" --tier standard --remote
 ```

-parsing-rule 的 `tier`、`page_range` 和 `remote` 只适用于 PDF 和 image。EPUB、Office、HTML 和 CSV 命中规则时忽略这些字段，按 `flash` 整本解析，并且 parse row 记录为 `flash`；其它 text 只入库和索引，不创建 parse row。
+parsing-rule 的 `tier` 和 `remote` 适用于 PDF 和 image，但 `page_range` 只适用于 PDF。image 忽略 page_range 并整文件解析；EPUB、Office、HTML 和 CSV 命中规则时忽略 tier、page_range 和 remote，按 `flash` 整本解析，并且 parse row 记录为 `flash`；其它 text 只入库和索引，不创建 parse row。

 PDF/image 命中规则时，系统必须检查本地或远端能力是否支持对应 tier。rule 未指定 tier 时，按 `standard` -> `advanced` -> `basic` -> `flash` 选择可用 tier，并记录实际 tier。完整文件类型归一规则见 [ADR-0024](../decisions/0024-file-type-tier-normalization.md)。

@@ -49,7 +49,7 @@ mineru parse <file> [flags]
 | Flag | 类型 | 默认 | 说明 |
 |------|------|------|------|
 | `--tier` | `flash` / `basic` / `standard` / `advanced` | 不传 | 解析 tier；省略时由服务端决定；语义见 [解析 Tier](../tiers.md) |
-| `-p, --pages` | range | 不传 | PDF/图片的页码范围；非 PDF Flash 输入不支持该参数 |
+| `-p, --pages` | range | 不传 | 仅 PDF 支持的页码范围；图片及其他非 PDF 输入不支持该参数 |
 | `--after` | cursor | 不传 | 从服务端返回的 cursor 继续读取 |
 | `--limit` | int | `30000` | STDOUT 内容软字符上限 |
 | `--force` | bool | false | 跳过 done 缓存；复用 active parse 或为未覆盖页创建新 parse；不删除或作废旧缓存 |
@@ -95,7 +95,7 @@ PDF/image 的默认选择策略通过当前目标 parse-server 的能力发现

 默认 STDOUT 输出应适合 Agent context。长文档不一次性输出全文，而是输出有限范围，并通过 marker 指示如何继续。

-当前 CLI 不在命令层硬编码默认页码范围；PDF/图片不传 `--pages` 时由 doclib 的内容读取计划决定首次返回范围。非 PDF Flash 输入始终整本解析，显式传入 `--pages` 返回 `page_range_invalid`。
+当前 CLI 不在命令层硬编码默认页码范围；PDF 不传 `--pages` 时由 doclib 的内容读取计划决定首次返回范围。图片及其他非 PDF 输入始终整文件解析，显式传入 `--pages` 返回 `page_range_invalid`。

 分页文档使用物理页码：

@@ -121,7 +121,7 @@ Tier 语义见 [解析 Tier](tiers.md)。本节定义 `flash`、`basic`、`stand
 | `invalid_request_error` | `invalid_request` | 400 | 否 | 出错参数 | 参数格式或组合非法 |
 | `invalid_request_error` | `unsupported_output_format` | 400 | 否 | `output_formats` | 输出格式不支持 |
 | `invalid_request_error` | `unsupported_source` | 400 | 否 | `source` | 当前部署不支持该 source 类型或 source 策略 |
-| `invalid_request_error` | `page_range_invalid` | 400 | 否 | `page_range` | 页码范围格式非法、超出文档页数，或对仅支持整本解析的非 PDF Flash 输入指定范围 |
+| `invalid_request_error` | `page_range_invalid` | 400 | 否 | `page_range` | PDF 页码范围格式非法或超出页数，或对图片及其他非 PDF 输入指定范围 |
 | `invalid_request_error` | `file_type_unsupported` | 400 | 否 | `file` | 文件类型不支持 |
 | `invalid_request_error` | `file_encrypted` | 400 | 否 | `file` | 文件加密或受密码保护 |
 | `invalid_request_error` | `file_corrupted` | 400 | 否 | `file` | 文件损坏无法读取 |
@@ -160,7 +160,7 @@ MinerU 的 tier 策略同时遵守隐私优先和质量优先。
 | watch 自动发现文件 | 本地 `flash` |
 | `mineru-kit api-server` 未传 `--tier` | 以 `standard` 作为服务默认 tier |
 | `mineru-kit parse` local 模式未传 `--tier` / `--backend` | PDF/image 直接按 `standard` 解析；EPUB/Office/HTML/CSV 按 `flash` 语义处理；其它 text 不作为解析输入 |
-| watch 命中 parsing-rule 且 rule 指定 tier | PDF/image 按 rule 中的 tier、页码范围和 remote 配置执行；EPUB/Office/HTML/CSV 忽略 rule tier、page_range 和 remote，按 `flash` 整本解析；其它 text 只入库和索引 |
+| watch 命中 parsing-rule 且 rule 指定 tier | PDF 按 rule 中的 tier、页码范围和 remote 配置执行；image 按 tier 和 remote 整文件执行并忽略 page_range；EPUB/Office/HTML/CSV 忽略 rule tier、page_range 和 remote，按 `flash` 整本解析；其它 text 只入库和索引 |
 | watch 命中 parsing-rule 但 rule 未指定 tier | PDF/image 按 `standard` -> `advanced` -> `basic` -> `flash` 选择；EPUB/Office/HTML/CSV 按 `flash`；其它 text 只入库和索引 |
 | 用户主动 parse，未指定 tier；HTTP API 传 JSON `null`；Python SDK 传 `None` | PDF/image 有能力发现上下文时按 `standard` -> `advanced` -> `basic` 选择；EPUB/Office/HTML/CSV 按 [ADR-0024](decisions/0024-file-type-tier-normalization.md) 与 [ADR-0028](decisions/0028-csv-structured-flash-parsing.md) 归一为 `flash`；其它 text 不进入解析 |
 | 用户主动指定 `--tier flash` | 本地 `flash` |
@@ -521,7 +521,7 @@ PDF 和 image block 图片通常在读取时从源页面按 bbox 裁剪，不写
 (sha256, tier)
 ```

-页码范围由 `parses.page_range` 和对应 JSON 文件共同表达。请求某个页码范围时，doclib 会:
+页码范围由 `parses.page_range` 和对应 JSON 文件共同表达。只有 PDF 使用增量页范围缓存；图片及其他非 PDF 输入仍用实际逻辑页范围命名产物，但整个文件始终对应一个原子 batch。请求 PDF 页码范围时，doclib 会:

 1. 查询同一 `sha256 + tier` 下已完成批次。
 2. 忽略已经 invalidate 的批次。
@@ -530,7 +530,7 @@ PDF 和 image block 图片通常在读取时从源页面按 bbox 裁剪，不写
 5. 对未覆盖的页创建新的 parse 任务。
 6. 如果页码已经被 pending / parsing 批次覆盖，则提升优先级，而不是重复创建任务。

-这意味着同一个文档可以多轮解析不同页，最终由多个 JSON 批次共同覆盖用户需要的页码集合。
+这意味着同一个 PDF 可以多轮解析不同页，最终由多个 JSON 批次共同覆盖用户需要的页码集合。非 PDF 的历史 partial batch 不参与缺页差集计算；缓存不完整时重新排一个整文件任务。

 `--force` 会跳过第 4 步的 done 缓存命中判断，但仍可复用已经覆盖请求页码的 active parse。它不会删除或作废旧 done 批次。若 force 关联的 wait parse 失败，旧批次仍可继续用于读取和搜索，但本次 force 请求应显示为失败。

@@ -543,10 +543,11 @@ compaction 负责:
 1. 扫描同一 `sha256 + tier` 下多个有效 done 批次。
 2. 合并相邻或重叠的页码范围。
 3. 读取旧批次 JSON。
-4. 按 `page_idx` 收集页面内容；如果同一页出现多次，`done_at` 较新的有效批次覆盖较旧批次。
-5. 删除旧 JSON 文件。
-6. 为合并后的页码范围写出新的 JSON 文件。
+4. 用 JSON 中的实际 `page_idx` 扩充历史记录不足的范围，完整写入新缓存后再删除旧文件。
+5. 按 `page_idx` 收集页面内容；如果同一页出现多次，`done_at` 较新的有效批次覆盖较旧批次。
+6. 为合并后的页码范围预写并原子提升新的 JSON 文件。
 7. 用较少的 done parse row 替代旧 row。
+8. 删除未被新 row 引用的旧 JSON 文件。

 已 invalidate 的批次不参与 compaction 的页面选择；其 JSON 文件可以由 cleanup 或 compaction 的清理阶段删除。

@@ -159,6 +159,6 @@ mineru -p <input_path> -o <output_path>
 mineru parse "<input_path>" --tier flash -o <output_path>
 ```

-当前 `mineru` 的现代 CLI 和 API 入口支持本地 `PDF / EPUB / 图片 / CSV / RTF / DOCX / PPTX / XLSX / ODT / ODS / ODP` 文件或目录输入。EPUB、RTF 与 OpenDocument 采用本地 Flash 语义解析，不依赖外部 Office 套件。非 PDF Flash 输入只支持整本解析；EPUB 页面严格遵循 OPF spine 顺序，并保留作者目录中可解析的内部跳转。
+当前 `mineru` 的现代 CLI 和 API 入口支持本地 `PDF / EPUB / 图片 / CSV / RTF / DOCX / PPTX / XLSX / ODT / ODS / ODP` 文件或目录输入。EPUB、RTF 与 OpenDocument 采用本地 Flash 语义解析，不依赖外部 Office 套件。只有 PDF 接受页范围；其他输入均按一个完整文件解析和缓存，但各格式自身的边界仍作为结构化输出中的逻辑页保留。EPUB 页面严格遵循 OPF spine 顺序，并保留作者目录中可解析的内部跳转。

 旧 Gradio WebUI 保持原有上传格式，暂不接收 CSV。具体使用方法请参考[使用指南](../usage/index.md)。
@@ -28,7 +28,7 @@ Options:
  --help                          显示帮助信息
 ```
 > [!TIP]
-> `mineru` 当前支持本地 `PDF`、`EPUB`、图片、`CSV`、`RTF`、`DOCX`、`PPTX`、`XLSX`、`ODT`、`ODS`、`ODP` 文件或目录输入。非 PDF Flash 格式只支持整本解析；EPUB 页面严格遵循 OPF spine 顺序，并保留作者目录中可解析的内部跳转。
+> `mineru` 当前支持本地 `PDF`、`EPUB`、图片、`CSV`、`RTF`、`DOCX`、`PPTX`、`XLSX`、`ODT`、`ODS`、`ODP` 文件或目录输入。只有 PDF 接受页范围；其他输入均按一个完整文件解析和缓存，但各格式自身的边界仍作为结构化输出中的逻辑页保留。EPUB 页面严格遵循 OPF spine 顺序，并保留作者目录中可解析的内部跳转。

 ```bash
 mineru-api --help
@@ -106,8 +106,13 @@ class Compaction:
        if len(rows) <= 1:
            return 0

+        loaded = self._load_batch_payloads(sha256, tier, rows)
+        if loaded is None:
+            return 0
+        pages_by_page_idx, envelope = loaded
+
        # collect all done page numbers
-        all_page_numbers: set[int] = set()
+        all_page_numbers = {page_idx + 1 for page_idx in pages_by_page_idx}
        max_done_at = 0
        for r in rows:
            all_page_numbers |= parse_page_range_set(r["page_range"])
@@ -132,6 +137,17 @@ class Compaction:
        if len(merged_ranges) >= len(rows):
            return 0  # no benefit

+        compacted_paths = self._write_compacted_json_files(
+            sha256,
+            tier,
+            merged_ranges,
+            max_done_at,
+            pages_by_page_idx,
+            envelope,
+        )
+        if compacted_paths is None:
+            return 0
+
        # atomic replace
        now = int(time.time() * 1000)
        await self.db.execute(
@@ -149,11 +165,98 @@ class Compaction:
                (sha256, tier, page_range, PARSE_STATUS_DONE, max_done_at, now, now),
            )

-        # compact JSON files (only from done batches, not superseded)
-        await self._compact_json(sha256, tier, merged_ranges, rows, max_done_at)
+        self._delete_obsolete_json_files(sha256, tier, compacted_paths)
+
+        actual_page_count = max(pages_by_page_idx) + 1
+        await self.db.execute(
+            "UPDATE docs SET page_count=?, updated_at=? WHERE sha256=? AND (file_type IS NULL OR file_type<>?)",
+            (actual_page_count, now, sha256, "pdf"),
+        )

        return len(rows) - len(merged_ranges)

+    def _load_batch_payloads(
+        self,
+        sha256: str,
+        tier: Tier,
+        done_rows: Sequence[ParseBatchRow],
+    ) -> tuple[dict[int, dict[str, Any]], dict[str, Any]] | None:
+        """完整读取 done batch；任一源文件损坏时放弃本轮压缩。"""
+        pages_by_page_idx: dict[int, dict[str, Any]] = {}
+        envelope: dict[str, Any] = {}
+        for row in reversed(done_rows):
+            fpath = parse_batch_json_path(self.data_dir, sha256, tier, row["page_range"], row["done_at"])
+            if not os.path.isfile(fpath):
+                return None
+            try:
+                with open(fpath, encoding="utf-8") as f:
+                    batch_payload = json.load(f)
+                batch_pages = _normalize_batch_pages(batch_payload)
+            except Exception:
+                return None
+            if not batch_pages:
+                return None
+            for page in batch_pages:
+                page_idx = page.get("page_idx")
+                if type(page_idx) is not int or page_idx < 0:
+                    return None
+                pages_by_page_idx[page_idx] = page
+            if not envelope:
+                envelope = {
+                    key: batch_payload[key]
+                    for key in ("is_full_document", "file_suffix", "effort", "parse_mode", "mineru_version")
+                    if key in batch_payload
+                }
+        return (pages_by_page_idx, envelope) if pages_by_page_idx else None
+
+    def _write_compacted_json_files(
+        self,
+        sha256: str,
+        tier: Tier,
+        merged_ranges: Sequence[str],
+        max_done_at: int,
+        pages_by_page_idx: dict[int, dict[str, Any]],
+        envelope: dict[str, Any],
+    ) -> set[str] | None:
+        """先完整写入临时 JSON，再原子提升所有目标文件。"""
+        prepared: list[tuple[str, str]] = []
+        try:
+            for page_range in merged_ranges:
+                page_numbers = parse_page_range_set(page_range)
+                json_pages = [
+                    pages_by_page_idx[page_no - 1] for page_no in sorted(page_numbers) if page_no - 1 in pages_by_page_idx
+                ]
+                if not json_pages:
+                    raise ValueError(f"Compacted page range has no source pages: {page_range}")
+                final_path = parse_batch_json_path(self.data_dir, sha256, tier, page_range, max_done_at)
+                temp_path = f"{final_path}.tmp-{time.time_ns()}"
+                payload: dict[str, Any] = {"schema_version": MIDDLE_JSON_SCHEMA_VERSION, "pages": json_pages}
+                payload.update(envelope)
+                with open(temp_path, "w", encoding="utf-8") as f:
+                    json.dump(payload, f, ensure_ascii=False, indent=4)
+                prepared.append((temp_path, final_path))
+            for temp_path, final_path in prepared:
+                os.replace(temp_path, final_path)
+            return {final_path for _, final_path in prepared}
+        except Exception:
+            for temp_path, _ in prepared:
+                try:
+                    os.unlink(temp_path)
+                except OSError:
+                    pass
+            return None
+
+    def _delete_obsolete_json_files(self, sha256: str, tier: Tier, keep_paths: set[str]) -> None:
+        """仅在新缓存和数据库均就绪后删除已被替代的 JSON。"""
+        tier_dir = os.path.join(self.data_dir, "parsed", sha256[:2], sha256, tier)
+        for fname in os.listdir(tier_dir):
+            path = os.path.join(tier_dir, fname)
+            if fname.endswith(".json") and path not in keep_paths:
+                try:
+                    os.unlink(path)
+                except OSError:
+                    pass
+
    async def _compact_json(
        self,
        sha256: str,
@@ -164,57 +267,18 @@ class Compaction:
    ) -> None:
        """Merge per-batch JSON files to match compacted parses rows.
        Only reads files belonging to *done_rows* — ignores superseded files."""
-        tier_dir = os.path.join(self.data_dir, "parsed", sha256[:2], sha256, tier)
-        if not os.path.isdir(tier_dir):
+        loaded = self._load_batch_payloads(sha256, tier, done_rows)
+        if loaded is None:
            return
-
-        # only read files from done batches (exclude superseded)
-        # process oldest first → newest overwrites (done_rows is sorted by done_at DESC)
-        pages_by_page_idx: dict[int, dict] = {}
-        envelope: dict[str, Any] = {}
-        for row in reversed(done_rows):
-            fpath = parse_batch_json_path(self.data_dir, sha256, tier, row["page_range"], row["done_at"])
-            if not os.path.isfile(fpath):
-                continue
-            try:
-                with open(fpath, encoding="utf-8") as f:
-                    batch_payload = json.load(f)
-            except Exception:
-                continue
-            batch_pages = _normalize_batch_pages(batch_payload)
-            for p in batch_pages:
-                pages_by_page_idx[p["page_idx"]] = p
-            # 继承源 batch JSON 的 envelope 元数据（2.0 schema 字段）
-            if not envelope:
-                envelope = {
-                    k: batch_payload[k] for k in ("file_suffix", "effort", "parse_mode", "mineru_version") if k in batch_payload
-                }
-
-        if not pages_by_page_idx:
+        pages_by_page_idx, envelope = loaded
+        compacted_paths = self._write_compacted_json_files(
+            sha256,
+            tier,
+            merged_ranges,
+            max_done_at,
+            pages_by_page_idx,
+            envelope,
+        )
+        if compacted_paths is None:
            return
-
-        # delete old files
-        for fname in os.listdir(tier_dir):
-            if fname.endswith(".json"):
-                try:
-                    os.unlink(os.path.join(tier_dir, fname))
-                except OSError:
-                    pass
-
-        # write one compacted JSON per merged range
-        for page_range in merged_ranges:
-            page_numbers = parse_page_range_set(page_range)
-            json_pages = [
-                pages_by_page_idx[page_no - 1] for page_no in sorted(page_numbers) if page_no - 1 in pages_by_page_idx
-            ]
-            if not json_pages:
-                continue
-            json_path = parse_batch_json_path(self.data_dir, sha256, tier, page_range, max_done_at)
-            try:
-                with open(json_path, "w", encoding="utf-8") as f:
-                    # compacted payload 用 2.0 schema，继承源 batch 的 envelope 元数据
-                    payload: dict[str, Any] = {"schema_version": MIDDLE_JSON_SCHEMA_VERSION, "pages": json_pages}
-                    payload.update(envelope)
-                    json.dump(payload, f, ensure_ascii=False, indent=4)
-            except Exception:
-                pass
+        self._delete_obsolete_json_files(sha256, tier, compacted_paths)
@@ -19,6 +19,7 @@ from ...filetypes import (
    IMAGE_EXTENSIONS,
    INGESTIBLE_EXTENSIONS,
    OFFICE_EXTENSIONS,
+    PAGE_RANGE_PARSE_EXTENSIONS,
    TEXT_EXTENSIONS,
    TIERED_PARSE_EXTENSIONS,
    file_type_for_extension,
@@ -654,7 +655,9 @@ class ParseService:
        # determine tier and page_range for initial parse
        tier: Tier = "flash"
        privacy = "local"
-        initial_page_range = default_parse_range(page_count)
+        initial_page_range = (
+            default_parse_range(page_count) if ext in PAGE_RANGE_PARSE_EXTENSIONS else expand_page_range(None, page_count or 1)
+        )

        # check parsing-rules
        matched = await self.config_svc.match_rules(path, RULE_TYPE_PARSING_RULE)
@@ -669,7 +672,7 @@ class ParseService:
                tier = "flash"
                privacy = "local"
            rule_page_range = rule.get("page_range")
-            if rule_page_range and ext in TIERED_PARSE_EXTENSIONS:
+            if rule_page_range and ext in PAGE_RANGE_PARSE_EXTENSIONS:
                initial_page_range = expand_page_range(rule_page_range, page_count or 1)

        # insert parse batch
@@ -734,10 +737,10 @@ class ParseService:
                "Text files do not require MinerU parsing. Read the file directly.",
                "path",
            )
-        if page_range and ext not in TIERED_PARSE_EXTENSIONS:
+        if page_range and ext not in PAGE_RANGE_PARSE_EXTENSIONS:
            raise InvalidRequestError(
                "page_range_invalid",
-                f"Page range is only supported for PDF and image files; '{ext}' uses full-document parsing.",
+                f"Page range is only supported for PDF files; '{ext}' uses full-document parsing.",
                "page_range",
            )
        if remote and ext not in TIERED_PARSE_EXTENSIONS:
@@ -764,8 +767,8 @@ class ParseService:
            requested_tier = "flash"

        # ── expand page range ──
-        default_page_range = default_parse_range(page_count)
-        requested_page_range_input = page_range or default_page_range
+        supports_page_range = ext in PAGE_RANGE_PARSE_EXTENSIONS
+        requested_page_range_input = page_range or (default_parse_range(page_count) if supports_page_range else None)
        request_page_range = expand_page_range(requested_page_range_input, page_count or 1)
        needed_page_numbers = parse_page_range_set(request_page_range)

@@ -778,13 +781,15 @@ class ParseService:
                    (sha256, requested_tier, PARSE_STATUS_DONE),
                ),
            )
-            for batch in done_batches:
-                if not _json_file_exists_by_batch(self.data_dir, sha256, requested_tier, batch):
-                    continue  # JSON gone → cache invalid
-                covered_page_numbers = parse_page_range_set(batch["page_range"])
-                needed_page_numbers -= covered_page_numbers
-
-            if not needed_page_numbers:
+            valid_done_batches = [
+                batch for batch in done_batches if _json_file_exists_by_batch(self.data_dir, sha256, requested_tier, batch)
+            ]
+            if supports_page_range:
+                for batch in valid_done_batches:
+                    needed_page_numbers -= parse_page_range_set(batch["page_range"])
+                if not needed_page_numbers:
+                    return _done_response(sha256, short_id, requested_tier, request_page_range)
+            elif any(needed_page_numbers <= parse_page_range_set(batch["page_range"]) for batch in valid_done_batches):
                return _done_response(sha256, short_id, requested_tier, request_page_range)

        # ── step 2: remove page numbers covered by pending/parsing batches ──
@@ -797,13 +802,17 @@ class ParseService:
            ),
        )
        if active_batches:
-            active_covered_page_numbers: set[int] = set()
-            for batch in active_batches:
-                covered_page_numbers = parse_page_range_set(batch["page_range"])
-                if needed_page_numbers & covered_page_numbers:
-                    reused_parse_ids.append(batch["id"])
-                active_covered_page_numbers |= covered_page_numbers
-            needed_page_numbers -= active_covered_page_numbers
+            if supports_page_range:
+                active_covered_page_numbers: set[int] = set()
+                for batch in active_batches:
+                    covered_page_numbers = parse_page_range_set(batch["page_range"])
+                    if needed_page_numbers & covered_page_numbers:
+                        reused_parse_ids.append(batch["id"])
+                    active_covered_page_numbers |= covered_page_numbers
+                needed_page_numbers -= active_covered_page_numbers
+            else:
+                reused_parse_ids = [batch["id"] for batch in active_batches]
+                needed_page_numbers.clear()

            # bump priority for reused in-progress batches
            now = _now_ms()
@@ -829,7 +838,7 @@ class ParseService:
                )

        # ── step 3: enqueue remaining uncovered page numbers ──
-        uncovered_page_range = _page_numbers_to_range_str(needed_page_numbers)
+        uncovered_page_range = _page_numbers_to_range_str(needed_page_numbers) if supports_page_range else request_page_range
        now = _now_ms()
        parse_id = await self.db.execute_insert(
            "INSERT INTO parses (sha256, tier, page_range, status, privacy, priority, created_at, updated_at) "
@@ -918,6 +927,7 @@ class ParseService:
                error_code="file_type_unsupported",
            )
            return False
+        file_ext = file_row.get("ext") or Path(file_row["path"]).suffix.lower().lstrip(".")

        output_dir = os.path.join(self.data_dir, "parsed", sha256[:2], sha256, tier)

@@ -993,9 +1003,16 @@ class ParseService:
            await self._record_parse_task_finished(task_start_ms, tier=tier, status="failed", error_code="parse_failed")
            return False

+        persisted_page_range = page_range
+        full_document_page_count: int | None = None
+        if file_ext not in PAGE_RANGE_PARSE_EXTENSIONS:
+            actual_page_numbers = {page.page_idx + 1 for page in result.pages}
+            persisted_page_range = _page_numbers_to_range_str(actual_page_numbers)
+            full_document_page_count = max(actual_page_numbers)
+
        # save per-batch JSON (markdown is generated on read from /docs/{doc_ref}/content)
        done_at_ms = _now_ms()
-        json_path = os.path.join(output_dir, _safe_filename(page_range, done_at_ms))
+        json_path = os.path.join(output_dir, _safe_filename(persisted_page_range, done_at_ms))
        write_start_ms = _now_ms()
        try:
            os.makedirs(output_dir, exist_ok=True)
@@ -1008,7 +1025,7 @@ class ParseService:
                task["id"],
                file_row["path"],
                tier,
-                page_range,
+                persisted_page_range,
                exc,
                exc_info=(type(exc), exc, exc.__traceback__),
            )
@@ -1038,11 +1055,16 @@ class ParseService:

            # update docs metadata (tier-gated)
            await self._maybe_update_docs_meta(sha256, tier)
+            if full_document_page_count is not None:
+                await self.db.execute(
+                    "UPDATE docs SET page_count=?, updated_at=? WHERE sha256=?",
+                    (full_document_page_count, done_at_ms, sha256),
+                )

            # mark done
            await self.db.execute(
-                "UPDATE parses SET status=?, done_at=?, locked_at=NULL, via=?, updated_at=? WHERE id=?",
-                (PARSE_STATUS_DONE, done_at_ms, via, done_at_ms, task["id"]),
+                "UPDATE parses SET status=?, page_range=?, done_at=?, locked_at=NULL, via=?, updated_at=? WHERE id=?",
+                (PARSE_STATUS_DONE, persisted_page_range, done_at_ms, via, done_at_ms, task["id"]),
            )
        except Exception as exc:
            await self._fail_task(task["id"], "parse_json_write_failed", str(exc)[:500])
@@ -1077,7 +1099,8 @@ class ParseService:
        """Parse via local library call."""
        from ...parser import parse

-        parser_page_range = page_range if file_row["ext"] in TIERED_PARSE_EXTENSIONS else ""
+        file_ext = file_row.get("ext") or Path(file_row["path"]).suffix.lower().lstrip(".")
+        parser_page_range = page_range if file_ext in PAGE_RANGE_PARSE_EXTENSIONS else ""
        result = await asyncio.to_thread(
            parse,
            file_row["path"],
@@ -1108,7 +1131,9 @@ class ParseService:
            include_images=Path(file_row["path"]).suffix.lower().lstrip(".") in OFFICE_EXTENSIONS,
        )
        try:
-            result = await parser.parse_async(file_row["path"], page_range=page_range)
+            file_ext = file_row.get("ext") or Path(file_row["path"]).suffix.lower().lstrip(".")
+            parser_page_range = page_range if file_ext in PAGE_RANGE_PARSE_EXTENSIONS else ""
+            result = await parser.parse_async(file_row["path"], page_range=parser_page_range)
        except _APITransportError as exc:
            if via == "remote":
                code = "remote_timeout" if exc.timed_out else "remote_unreachable"
@@ -1120,7 +1145,8 @@ class ParseService:
                code,
                f"{target} transport failed during {exc.stage} after {exc.attempts} attempt(s) ({type(exc.cause).__name__}).",
            ) from exc
-        _remap_api_result_pages_to_page_range(result, page_range)
+        if file_ext in PAGE_RANGE_PARSE_EXTENSIONS:
+            _remap_api_result_pages_to_page_range(result, page_range)
        return result, via

    @staticmethod
@@ -36,6 +36,8 @@ TEXT_EXTENSIONS: frozenset[str] = frozenset({"txt", "md", "markdown", "rst", "te

 TIERED_PARSE_EXTENSIONS: frozenset[str] = PDF_EXTENSIONS | IMAGE_EXTENSIONS

+PAGE_RANGE_PARSE_EXTENSIONS: frozenset[str] = PDF_EXTENSIONS
+
 FLASH_ONLY_PARSE_EXTENSIONS: frozenset[str] = OFFICE_EXTENSIONS | HTML_EXTENSIONS | CSV_EXTENSIONS | EPUB_EXTENSIONS

 PARSEABLE_EXTENSIONS: frozenset[str] = TIERED_PARSE_EXTENSIONS | FLASH_ONLY_PARSE_EXTENSIONS
@@ -125,6 +127,11 @@ def is_tiered_parse_extension(path_or_ext: str | Path) -> bool:
    return normalize_parse_extension(path_or_ext) in TIERED_PARSE_EXTENSIONS


+def is_page_range_parse_extension(path_or_ext: str | Path) -> bool:
+    """仅允许 PDF 输入使用局部页范围解析。"""
+    return normalize_parse_extension(path_or_ext) in PAGE_RANGE_PARSE_EXTENSIONS
+
+
 def is_flash_only_parse_extension(path_or_ext: str | Path) -> bool:
    return normalize_parse_extension(path_or_ext) in FLASH_ONLY_PARSE_EXTENSIONS

@@ -25,6 +25,7 @@ VECTOR_IMAGE_CONTENT_TYPES = frozenset(
    }
 )
 PIL_IMAGE_LOAD_ERRORS = (UnidentifiedImageError, OSError, SyntaxError)
+VECTOR_IMAGE_RENDER_DPI: Final = 144
 STANDARD_VECTOR_PLACEHOLDER_SIZE: Final = (320, 180)
 STANDARD_VECTOR_PLACEHOLDER_LINES: Final = (
    "WMF/EMF placeholder",
@@ -172,7 +173,7 @@ def serialize_vector_image_with_placeholder(pil_image: Image.Image, image_format

    if is_windows_environment():
        try:
-            pil_image.load()
+            pil_image.load(dpi=VECTOR_IMAGE_RENDER_DPI)
            return image_to_b64str(pil_image, image_format="PNG")
        except PIL_IMAGE_LOAD_ERRORS as e:
            logger.warning(f"Failed to render {image_format} image: {e}, size: {pil_image.size}. Using placeholder instead.")
@@ -227,7 +227,7 @@ class DocumentParser(ABC):
        path:
            Path to the document file.
        page_range:
-            1-based page range string (``"1~5,-3~-1"``).  Empty means all pages.
+            PDF-only 1-based page range string (``"1~5,-3~-1"``).  Empty means all pages.
        """

    async def parse_async(self, path: str | Path, *, page_range: str = "") -> ParseResult:
@@ -1,5 +1,6 @@
 # Copyright (c) Opendatalab. All rights reserved.
 """统一文档解析器，委托 backend.analyze 处理 PDF、EPUB、图片、CSV 与 Office/RTF/ODF。"""
+
 from __future__ import annotations

 import asyncio
@@ -11,7 +12,7 @@ from typing import Any, Literal, cast

 from ..backend.analyze import aio_doc_analyze, doc_analyze
 from ..errors import InvalidRequestError
-from ..filetypes import IMAGE_EXTENSIONS
+from ..filetypes import IMAGE_EXTENSIONS, PAGE_RANGE_PARSE_EXTENSIONS
 from ..types import FILE_SUFFIXES, FileSuffix, MiddleJson, ModelJson, PageInfo, Tier
 from .tier import effort_for_tier
 from .base import DocumentParser, ParseResult
@@ -20,6 +21,8 @@ logger = logging.getLogger(__name__)

 _Effort = Literal["flash", "medium", "high", "xhigh"]
 _ParseMode = Literal["auto", "txt", "ocr"]
+
+
@dataclass
 class _PreparedInput:
    """记录文档输入准备结果，避免跨文档复用 parser 实例状态。"""
@@ -116,6 +119,7 @@ class MinerUParser(DocumentParser):
        file_name = path.stem
        file_bytes = path.read_bytes()
        suffix = guess_suffix_by_path(path)
+        source_suffix = suffix

        if suffix in IMAGE_EXTENSIONS:
            from ..model.flash.pdf.document import PDFDocument
@@ -141,10 +145,10 @@ class MinerUParser(DocumentParser):

        if suffix not in FILE_SUFFIXES:
            raise ValueError(f"Unsupported file type: {suffix or path.suffix or 'unknown'}")
-        if suffix != "pdf" and page_range.strip():
+        if source_suffix not in PAGE_RANGE_PARSE_EXTENSIONS and page_range.strip():
            raise InvalidRequestError(
                "page_range_invalid",
-                f"Page range is only supported for PDF files; '{suffix}' uses full-document parsing.",
+                f"Page range is only supported for PDF files; '{source_suffix}' uses full-document parsing.",
                "page_range",
            )
        file_bytes, retained_page_indices, broken_page_indices = self._maybe_adjust_pdf_bytes(
@@ -69,7 +69,6 @@ from mineru.parser.tier import resolve_tier_and_backend
 from mineru.parser.api_client import _APITransportError, _V1APIError
 from mineru.parser.base import ParseResult
 from mineru.parser import MIDDLE_JSON_SCHEMA_VERSION
-from mineru.parser.base import _LEGACY_SCHEMA_VERSION
 from mineru.types import (
    BlockBase,
    BlockType,
@@ -149,6 +148,23 @@ class _FakeDB:
                    row["locked_at"] = None
                    row["updated_at"] = updated_at
                    return _Cursor(1)
+        if sql.startswith("UPDATE parses SET status=?, page_range=?, done_at=?"):
+            status, page_range, done_at, via, updated_at, parse_id = params
+            for row in self.parses:
+                if row["id"] == parse_id:
+                    row["status"] = status
+                    row["page_range"] = page_range
+                    row["done_at"] = done_at
+                    row["via"] = via
+                    row["locked_at"] = None
+                    row["updated_at"] = updated_at
+                    return _Cursor(1)
+        if sql.startswith("UPDATE docs SET page_count=?"):
+            page_count, updated_at, sha256 = params
+            if self.doc_row and self.doc_row["sha256"] == sha256:
+                self.doc_row["page_count"] = page_count
+                self.doc_row["updated_at"] = updated_at
+                return _Cursor(1)
        return _Cursor(0)

    async def execute_insert(self, sql: str, params: tuple[Any, ...]) -> int:
@@ -260,12 +276,14 @@ def _write_batch(
    json_pages: list[dict],
    *,
    file_suffix: str = "pdf",
+    is_full_document: bool = True,
 ) -> None:
    path = Path(parse_batch_json_path(str(data_dir), sha256, tier, page_range, done_at))
    path.parent.mkdir(parents=True, exist_ok=True)
    payload = {
        "schema_version": MIDDLE_JSON_SCHEMA_VERSION,
        "pages": json_pages,
+        "is_full_document": is_full_document,
        "file_suffix": file_suffix,
        "effort": "medium",
        "parse_mode": "txt",
@@ -1221,11 +1239,100 @@ def test_compaction_uses_configured_data_dir(tmp_path: Path) -> None:
    compacted_path = Path(parse_batch_json_path(str(tmp_path), sha256, tier, "1~2", 2000))
    compacted = json.loads(compacted_path.read_text(encoding="utf-8"))

-    assert compacted["schema_version"] == _LEGACY_SCHEMA_VERSION
+    assert compacted["schema_version"] == MIDDLE_JSON_SCHEMA_VERSION
+    assert compacted["is_full_document"] is True
    assert compacted["pages"] == [older_page, newer_duplicate]
    assert sorted(path.name for path in compacted_path.parent.glob("*.json")) == ["1~2_2000.json"]


+def test_compaction_repairs_underreported_full_document_ranges(tmp_path: Path) -> None:
+    """验证历史整本 JSON 超出 recorded range 时按实际 page_idx 修复且不丢页。"""
+
+    async def _run() -> None:
+        """创建真实缓存和数据库行并执行一次文档级压缩。"""
+        sha256 = "d" * 64
+        tier = "flash"
+        db = DatabaseManager(str(tmp_path / "doclib.db"))
+        await db.initialize()
+        await db.execute(
+            "INSERT INTO docs (sha256, short_id, size_bytes, file_type, page_count, first_seen_at, updated_at) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?)",
+            (sha256, "ddddddd", 1, "epub", 10, 1, 1),
+        )
+        for done_at in (1000, 2000):
+            await db.execute(
+                "INSERT INTO parses (sha256, tier, page_range, status, done_at, created_at, updated_at) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (sha256, tier, "1~10", "done", done_at, done_at, done_at),
+            )
+            _write_batch(
+                tmp_path,
+                sha256,
+                tier,
+                "1~10",
+                done_at,
+                [{"page_idx": page_idx, "blocks": []} for page_idx in range(12)],
+                file_suffix="epub",
+            )
+
+        compaction = Compaction(db=db, interval_sec=600, data_dir=str(tmp_path))
+        assert await compaction._compact_doc_tier(sha256, tier) == 1
+
+        rows = await db.fetchall("SELECT page_range, done_at FROM parses WHERE sha256=? AND tier=?", (sha256, tier))
+        doc = await db.fetchone("SELECT page_count FROM docs WHERE sha256=?", (sha256,))
+        compacted_path = Path(parse_batch_json_path(str(tmp_path), sha256, tier, "1~12", 2000))
+        compacted = json.loads(compacted_path.read_text(encoding="utf-8"))
+        assert rows == [{"page_range": "1~12", "done_at": 2000}]
+        assert doc == {"page_count": 12}
+        assert compacted["is_full_document"] is True
+        assert [page["page_idx"] for page in compacted["pages"]] == list(range(12))
+        assert sorted(path.name for path in compacted_path.parent.glob("*.json")) == ["1~12_2000.json"]
+
+    asyncio.run(_run())
+
+
+def test_compaction_keeps_rows_when_any_source_batch_is_missing(tmp_path: Path) -> None:
+    """验证源 batch 不完整时 compaction 不修改数据库或现存 JSON。"""
+
+    async def _run() -> None:
+        """构造一条缺失 JSON 的 done row 并确认压缩被跳过。"""
+        sha256 = "e" * 64
+        tier = "flash"
+        db = DatabaseManager(str(tmp_path / "doclib.db"))
+        await db.initialize()
+        await db.execute(
+            "INSERT INTO docs (sha256, short_id, size_bytes, file_type, page_count, first_seen_at, updated_at) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?)",
+            (sha256, "eeeeeee", 1, "epub", 12, 1, 1),
+        )
+        for page_range, done_at in (("1~10", 1000), ("11~12", 2000)):
+            await db.execute(
+                "INSERT INTO parses (sha256, tier, page_range, status, done_at, created_at, updated_at) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (sha256, tier, page_range, "done", done_at, done_at, done_at),
+            )
+        _write_batch(
+            tmp_path,
+            sha256,
+            tier,
+            "1~10",
+            1000,
+            [{"page_idx": page_idx, "blocks": []} for page_idx in range(10)],
+            file_suffix="epub",
+        )
+
+        compaction = Compaction(db=db, interval_sec=600, data_dir=str(tmp_path))
+        assert await compaction._compact_doc_tier(sha256, tier) == 0
+        rows = await db.fetchall(
+            "SELECT page_range, done_at FROM parses WHERE sha256=? AND tier=? ORDER BY done_at",
+            (sha256, tier),
+        )
+        assert rows == [{"page_range": "1~10", "done_at": 1000}, {"page_range": "11~12", "done_at": 2000}]
+        assert Path(parse_batch_json_path(str(tmp_path), sha256, tier, "1~10", 1000)).is_file()
+
+    asyncio.run(_run())
+
+
 def test_invalidate_deletes_fts_when_no_done_batches_remain(tmp_path: Path) -> None:
    sha256 = "c" * 64
    parses = [{"sha256": sha256, "tier": "standard", "page_range": "1", "status": "done", "done_at": 1000}]
@@ -1336,6 +1443,8 @@ def test_request_parse_explicit_image_ingests_and_queues_parse(tmp_path: Path, m
        monkeypatch.setattr(parse_svc_module, "extract_metadata", _metadata)

        result = await service.request_parse(str(source), tier="flash")
+        with pytest.raises(InvalidRequestError) as range_exc:
+            await service.request_parse(str(source), tier="flash", page_range="1")
        file_row = await db.fetchone("SELECT path, ext, sha256, status FROM files WHERE path=?", (str(source),))
        doc_row = await db.fetchone(
            "SELECT short_id, file_type, page_count, is_image_based FROM docs WHERE sha256=?",
@@ -1345,6 +1454,8 @@ def test_request_parse_explicit_image_ingests_and_queues_parse(tmp_path: Path, m

        assert result.status == "pending"
        assert result.tier == "flash"
+        assert range_exc.value.code == "page_range_invalid"
+        assert range_exc.value.param == "page_range"
        assert doc_row is not None
        assert result.short_id == doc_row["short_id"]
        assert file_row is not None
@@ -1826,6 +1937,84 @@ def test_force_request_reuses_active_and_creates_only_uncovered_parse(tmp_path:
    assert parses[-1]["page_range"] == "1~5,9~10"


+def test_non_pdf_requests_use_one_full_document_batch(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """验证非 PDF 初始、历史 partial 和 active 路径始终使用一个整本 batch。"""
+
+    class _NoRulesConfig:
+        """关闭解析规则，让测试只观察默认整本调度。"""
+
+        async def match_rules(self, path: str, rule_type: str) -> list[dict[str, Any]]:
+            """返回空规则列表。"""
+            return []
+
+    async def _run() -> None:
+        """依次验证初始任务、历史 partial 修复和 active 复用。"""
+        db = DatabaseManager(str(tmp_path / "doclib.db"))
+        await db.initialize()
+        service = ParseService(
+            db=db,
+            fts=FTSManager(db),
+            config_svc=_NoRulesConfig(),
+            data_dir=str(tmp_path / "data"),
+            parse_lock_timeout_sec=1800,
+        )
+        source = tmp_path / "book.epub"
+        source.write_bytes(b"epub-placeholder")
+
+        async def _metadata(path: str) -> dict[str, Any]:
+            """返回十二个逻辑 spine 页的确定性 metadata。"""
+            return {
+                "page_count": 12,
+                "title": "Book",
+                "author": None,
+                "subject": None,
+                "keywords": None,
+                "is_image_based": 0,
+            }
+
+        monkeypatch.setattr(parse_svc_module, "extract_metadata", _metadata)
+
+        initial = await service.request_parse(str(source))
+        assert initial.page_range == "1~12"
+        assert initial.created_parse_ids == []
+        assert len(initial.reused_parse_ids) == 1
+        initial_id = initial.reused_parse_ids[0]
+        file_row = await db.fetchone("SELECT sha256 FROM files WHERE path=?", (str(source),))
+        assert file_row is not None
+        sha256 = file_row["sha256"]
+
+        await db.execute(
+            "UPDATE parses SET page_range=?, status=?, done_at=?, updated_at=? WHERE id=?",
+            ("1~10", "done", 1000, 1000, initial_id),
+        )
+        _write_batch(
+            tmp_path / "data",
+            sha256,
+            "flash",
+            "1~10",
+            1000,
+            [{"page_idx": page_idx, "blocks": []} for page_idx in range(12)],
+            file_suffix="epub",
+        )
+
+        repaired = await service.request_parse(str(source))
+        assert repaired.page_range == "1~12"
+        assert len(repaired.created_parse_ids) == 1
+        repaired_id = repaired.created_parse_ids[0]
+        repaired_row = await db.fetchone("SELECT page_range, status FROM parses WHERE id=?", (repaired_id,))
+        assert repaired_row == {"page_range": "1~12", "status": "pending"}
+
+        reused = await service.request_parse(str(source), force=True)
+        assert reused.created_parse_ids == []
+        assert reused.reused_parse_ids == [repaired_id]
+        assert reused.page_range == "1~12"
+
+    asyncio.run(_run())
+
+
 def test_list_parse_records_by_ids_returns_precise_status(tmp_path: Path) -> None:
    parses = [
        {"id": 1, "sha256": "f" * 64, "tier": "standard", "page_range": "1~5", "status": "done", "done_at": 1000},
@@ -4228,10 +4417,11 @@ def test_parse_via_api_maps_transport_errors_by_type(


@pytest.mark.parametrize(
-    ("path", "expected_include_images"),
+    ("path", "expected_include_images", "expected_page_range"),
    [
-        ("/tmp/doc.docx", True),
-        ("/tmp/doc.pdf", False),
+        ("/tmp/doc.docx", True, ""),
+        ("/tmp/doc.pdf", False, "1"),
+        ("/tmp/image.png", False, ""),
    ],
 )
 def test_parse_via_api_requests_image_cache_only_for_office(
@@ -4239,6 +4429,7 @@ def test_parse_via_api_requests_image_cache_only_for_office(
    tmp_path: Path,
    path: str,
    expected_include_images: bool,
+    expected_page_range: str,
 ) -> None:
    calls: list[bool] = []
    service = ParseService(
@@ -4268,8 +4459,8 @@ def test_parse_via_api_requests_image_cache_only_for_office(

        async def parse_async(self, _path: str, *, page_range: str = "") -> ParseResult:
            assert _path == path
-            assert page_range == "1"
-            return ParseResult(middle_json=MiddleJson(pages=[], file_suffix="pdf", effort="medium", parse_mode="txt", mineru_version=__version__))
+            assert page_range == expected_page_range
+            return ParseResult(middle_json=MiddleJson(pages=[], is_full_document=True, file_suffix="pdf", effort="medium", parse_mode="txt", mineru_version=__version__))

    service._resolve_api_target = _resolve_api_target  # type: ignore[method-assign]
    service._resolve_tier = lambda tier, _via: tier  # type: ignore[method-assign]
@@ -4340,6 +4531,75 @@ def test_process_doc_fails_when_batch_json_cannot_be_written(tmp_path: Path) ->
    assert fts.replaced == []


+def test_process_doc_normalizes_full_document_range_from_actual_pages(tmp_path: Path) -> None:
+    """验证整本结果按实际逻辑页重命名 JSON，并同步 parse row 与 docs.page_count。"""
+    sha256 = "f" * 64
+    task = {
+        "id": 1,
+        "sha256": sha256,
+        "tier": "flash",
+        "page_range": "1~10",
+        "status": "parsing",
+        "privacy": "local",
+    }
+    parses = [
+        {
+            **task,
+            "error_code": None,
+            "error_msg": None,
+            "done_at": None,
+            "locked_at": 123,
+            "updated_at": 123,
+        }
+    ]
+    doc_row = {"sha256": sha256, "short_id": "fffffff", "page_count": 10, "meta_tier": None}
+    db = _FakeDB(
+        parses=parses,
+        file_row={
+            "path": "/tmp/book.epub",
+            "ext": "epub",
+            "sha256": sha256,
+            "status": "active",
+            "filename": "book.epub",
+            "title": "",
+            "author": "",
+        },
+        doc_row=doc_row,
+    )
+    service = ParseService(db=db, fts=_FakeFTS(), config_svc=None, data_dir=str(tmp_path), parse_lock_timeout_sec=1800)
+
+    async def _parse(file_row: dict, tier: Tier, page_range: str) -> ParseResult:
+        """返回十二页整本 EPUB 结果。"""
+        assert page_range == "1~10"
+        return ParseResult(
+            middle_json=MiddleJson(
+                pages=[PageInfo(page_idx=page_idx) for page_idx in range(12)],
+                is_full_document=True,
+                file_suffix="epub",
+                effort="flash",
+                parse_mode="txt",
+                mineru_version=__version__,
+            )
+        )
+
+    async def _skip(*args: object, **kwargs: object) -> None:
+        """跳过与范围归一无关的 FTS 和 metadata tier 更新。"""
+        return None
+
+    service._parse_via_local = _parse  # type: ignore[method-assign]
+    service._maybe_update_fts = _skip  # type: ignore[method-assign]
+    service._maybe_update_docs_meta = _skip  # type: ignore[method-assign]
+
+    assert asyncio.run(service.process_doc(task)) is True
+    assert parses[0]["status"] == "done"
+    assert parses[0]["page_range"] == "1~12"
+    assert doc_row["page_count"] == 12
+    batch_path = Path(parse_batch_json_path(str(tmp_path), sha256, "flash", "1~12", parses[0]["done_at"]))
+    assert batch_path.is_file()
+    assert not Path(parse_batch_json_path(str(tmp_path), sha256, "flash", "1~10", parses[0]["done_at"])).exists()
+    assert len(json.loads(batch_path.read_text(encoding="utf-8"))["pages"]) == 12
+
+
 def test_process_doc_writes_cached_image_sidecars(tmp_path: Path) -> None:
    sha256 = "c" * 64
    task = {
@@ -2,6 +2,10 @@ from __future__ import annotations

 from pathlib import Path

+import pytest
+from PIL import Image
+
+from mineru.errors import InvalidRequestError
 from mineru.filetypes import (
    CSV_EXTENSIONS,
    EPUB_EXTENSIONS,
@@ -11,17 +15,40 @@ from mineru.filetypes import (
    MIME_TYPE_BY_EXTENSION,
    ODF_EXTENSIONS,
    OFFICE_EXTENSIONS,
+    PAGE_RANGE_PARSE_EXTENSIONS,
    PARSEABLE_EXTENSIONS,
    TEXT_EXTENSIONS,
    is_flash_only_parse_extension,
+    is_page_range_parse_extension,
 )
 from mineru.kit.common import ensure_supported_inputs, expand_input_paths
+from mineru.parser import parse


 def test_office_extensions_includes_legacy_binary_formats() -> None:
    assert {"doc", "docx", "ppt", "pptx", "xls", "xlsx", "rtf"} <= OFFICE_EXTENSIONS


+def test_only_pdf_supports_page_range_parsing() -> None:
+    """验证分页能力与图片的质量 tier/remote 能力相互独立。"""
+    assert PAGE_RANGE_PARSE_EXTENSIONS == frozenset({"pdf"})
+    assert is_page_range_parse_extension("sample.pdf")
+    assert not is_page_range_parse_extension("sample.png")
+    assert not is_page_range_parse_extension("sample.epub")
+
+
+def test_image_parser_rejects_explicit_page_range(tmp_path: Path) -> None:
+    """验证图片在转为内部 PDF 之前按整文件契约拒绝 page_range。"""
+    source = tmp_path / "sample.png"
+    Image.new("RGB", (8, 8), "white").save(source)
+
+    with pytest.raises(InvalidRequestError) as exc_info:
+        parse(source, tier="flash", page_range="1")
+
+    assert exc_info.value.code == "page_range_invalid"
+    assert exc_info.value.param == "page_range"
+
+
 def test_legacy_office_extensions_are_parseable_and_ingestible() -> None:
    for ext in ("doc", "ppt", "xls"):
        assert ext in PARSEABLE_EXTENSIONS
@@ -1,4 +1,5 @@
 from pathlib import Path
+from unittest.mock import Mock

 from pytest import MonkeyPatch

@@ -45,6 +46,20 @@ def test_vector_image_part_skip_log_is_debug(monkeypatch: MonkeyPatch) -> None:
    assert fake_logger.warning_messages == []


+def test_vector_image_windows_render_uses_144_dpi(monkeypatch: MonkeyPatch) -> None:
+    """验证 Windows 下 WMF/EMF 统一按 144 DPI 栅格化。"""
+
+    fake_image = Mock(format="WMF", size=(640, 360))
+    fake_encoder = Mock(return_value="data:image/png;base64,rendered")
+    monkeypatch.setattr(office_image, "is_windows_environment", lambda: True)
+    monkeypatch.setattr(office_image, "image_to_b64str", fake_encoder)
+
+    assert office_image.serialize_vector_image_with_placeholder(fake_image) == "data:image/png;base64,rendered"
+    fake_image.load.assert_called_once_with(dpi=office_image.VECTOR_IMAGE_RENDER_DPI)
+    fake_encoder.assert_called_once_with(fake_image, image_format="PNG")
+    assert office_image.VECTOR_IMAGE_RENDER_DPI == 144
+
+
 def test_docx_nested_ordered_lists_render_with_local_markers() -> None:
    """验证真实 DOCX 的多级有序列表使用当前层编号，并由 Markdown 缩进表达层级。"""
    file_bytes = (_PROJECT_ROOT / "demo" / "office_docs" / "docx_01.docx").read_bytes()