feat(data-process): 完善文件解析与切分存储链路

backend/app/api/v1/endpoints/data_process.py

@@ -3,13 +3,15 @@ from __future__ import annotations
 import hashlib
 import ipaddress
 import json
+import logging
 import os
+import re
 import socket
 from contextlib import contextmanager
 from dataclasses import asdict
 from pathlib import Path
 from typing import Any, Iterator, Literal
-from urllib.parse import urlsplit
+from urllib.parse import quote, urlsplit
 
 import psycopg
 from fastapi import (
@@ -18,22 +20,38 @@ from fastapi import (
     Body,
     Depends,
     File,
+    Header,
     HTTPException,
     Query,
     UploadFile,
 )
+from fastapi.responses import StreamingResponse
 from psycopg.rows import dict_row
 
 from app.modules.data_process.algorithms import (
+    ParsedText,
+    TextChunk,
+    canonical_record_json,
     chunk_unstructured,
-    decode_utf8,
+    content_quality_flags,
     desensitize_pii,
+    desensitize_structured_record,
+    detect_document_structure,
     estimate_token_count,
+    extract_pdf_page_texts,
     generate_standard_records,
+    is_near_duplicate,
+    near_duplicate_fingerprint,
     parse_text_content,
+    preprocess_structured_records,
     score_quality,
 )
 from app.modules.data_process.generation import generate_model_records
+from app.modules.data_process.storage import (
+    LocalDataProcessStorage,
+    StagedSourceObject,
+    get_data_process_storage,
+)
 from app.modules.data_process.store import (
     ConflictError,
     DataProcessStore,
@@ -41,11 +59,12 @@ from app.modules.data_process.store import (
     InvalidStateError,
     NotFoundError,
     get_data_process_store,
+    new_id,
 )
 from app.schemas.data_process import (
+    DataProcessStatus,
     DataProcessTaskCreate,
     DataProcessTaskUpdate,
-    DataProcessStatus,
     ExternalPullRequest,
     ExternalSourceRequest,
     GenerateRequest,
@@ -57,12 +76,34 @@ from app.schemas.data_process import (
     ResultUpdate,
 )
 
-
 router = APIRouter(prefix="/data-process")
+logger = logging.getLogger(__name__)
 MAX_SOURCE_FILE_BYTES = 200 * 1024 * 1024
 MAX_SOURCE_FILE_COUNT = 20
 MAX_SOURCE_BATCH_BYTES = 500 * 1024 * 1024
 MAX_EXTERNAL_PULL_BYTES = 50 * 1024 * 1024
+STRUCTURED_SOURCE_SUFFIXES = frozenset(
+    {".json", ".jsonl", ".ndjson", ".csv", ".tsv", ".xlsx"}
+)
+UNSTRUCTURED_SOURCE_SUFFIXES = frozenset(
+    {
+        ".txt",
+        ".md",
+        ".markdown",
+        ".pdf",
+        ".docx",
+        ".pptx",
+        ".json",
+        ".jsonl",
+        ".ndjson",
+    }
+)
+SUPPORTED_SOURCE_SUFFIXES = STRUCTURED_SOURCE_SUFFIXES | UNSTRUCTURED_SOURCE_SUFFIXES
+LEGACY_OFFICE_CONVERSIONS = {
+    ".doc": ".docx",
+    ".xls": ".xlsx",
+    ".ppt": ".pptx",
+}
 
 
 def ok(data: Any = None, message: str = "ok") -> dict[str, Any]:
@@ -99,6 +140,57 @@ def _safe_file_name(value: str | None, fallback: str) -> str:
     return name if name not in {"", ".", ".."} else fallback
 
 
+def _range_not_satisfiable(size: int) -> HTTPException:
+    return HTTPException(
+        status_code=416,
+        detail={"code": 416, "message": "invalid source byte range", "data": None},
+        headers={"Content-Range": f"bytes */{size}"},
+    )
+
+
+def _source_byte_range(value: str | None, size: int) -> tuple[int, int] | None:
+    if value is None:
+        return None
+    match = re.fullmatch(r"bytes=(\d*)-(\d*)", value.strip())
+    if match is None or size <= 0:
+        raise _range_not_satisfiable(size)
+    start_text, end_text = match.groups()
+    if not start_text and not end_text:
+        raise _range_not_satisfiable(size)
+    if start_text:
+        start = int(start_text)
+        end = int(end_text) if end_text else size - 1
+        if start >= size or end < start:
+            raise _range_not_satisfiable(size)
+    else:
+        suffix_length = int(end_text)
+        if suffix_length <= 0:
+            raise _range_not_satisfiable(size)
+        start = max(0, size - suffix_length)
+        end = size - 1
+    return start, min(end, size - 1)
+
+
+def _commit_source_batch(
+    store: DataProcessStore,
+    storage: LocalDataProcessStorage,
+    task_id: str,
+    prepared: list[dict[str, Any]],
+    staged: list[StagedSourceObject],
+) -> list[dict[str, Any]]:
+    storage.publish(staged)
+    try:
+        return store.add_source_files(task_id, prepared)
+    except Exception:
+        for item in staged:
+            try:
+                storage.delete(item.reference)
+            except Exception:
+                # 文件系统回滚失败不能覆盖数据库抛出的根因，并继续清理其余对象。
+                pass
+        raise
+
+
 def _value(config: dict[str, Any], snake_name: str, camel_name: str, default: Any) -> Any:
     if snake_name in config:
         return config[snake_name]
@@ -126,6 +218,204 @@ def _preview_quality(content: str, config: dict[str, Any]) -> dict[str, Any]:
     )
 
 
+def _parse_stored_source(source: dict[str, Any]) -> ParsedText:
+    """重新读取已入库的规范化正文，避免把二进制格式当二进制重复解析。"""
+
+    content = str(source.get("content") or "")
+    file_format = str(source.get("file_format") or "").lower()
+    if file_format == "xlsx":
+        # XLSX 上传阶段已安全解析为 JSONL 后入库。
+        return parse_text_content(content, file_format="jsonl")
+    if file_format in {"pdf", "docx", "pptx"}:
+        # 文档上传阶段已抽取文本，预览阶段只需要对正文切片。
+        return ParsedText(format=file_format, text=content, records=())
+    return parse_text_content(
+        content,
+        filename=source.get("name"),
+        file_format=file_format or None,
+    )
+
+
+def _shift_chunk(chunk: TextChunk, offset: int, source_text: str) -> TextChunk:
+    start = offset + chunk.start
+    end = offset + chunk.end
+    return TextChunk(
+        content=chunk.content,
+        start=start,
+        end=end,
+        start_line=source_text.count("\n", 0, start) + 1,
+        end_line=source_text.count("\n", 0, max(start, end - 1)) + 1,
+        token_count=chunk.token_count,
+    )
+
+
+def _merge_short_chunks(
+    chunks: list[TextChunk],
+    source_text: str,
+    *,
+    min_token_count: int,
+    chunk_size: int,
+) -> list[TextChunk]:
+    """在同一章节内合并短块，同时严格遵守切片大小上限。"""
+
+    merged: list[TextChunk] = []
+    index = 0
+    while index < len(chunks):
+        current = chunks[index]
+        if current.token_count >= min_token_count:
+            merged.append(current)
+            index += 1
+            continue
+
+        candidates: list[tuple[int, int, int]] = []
+        if merged:
+            candidates.append((merged[-1].start, current.end, -1))
+        if index + 1 < len(chunks):
+            candidates.append((current.start, chunks[index + 1].end, 1))
+        selected = next(
+            (
+                (start, end, direction)
+                for start, end, direction in candidates
+                if estimate_token_count(source_text[start:end]) <= chunk_size
+            ),
+            None,
+        )
+        if selected is None:
+            merged.append(current)
+            index += 1
+            continue
+
+        start, end, direction = selected
+        combined = TextChunk(
+            content=source_text[start:end],
+            start=start,
+            end=end,
+            start_line=source_text.count("\n", 0, start) + 1,
+            end_line=source_text.count("\n", 0, max(start, end - 1)) + 1,
+            token_count=estimate_token_count(source_text[start:end]),
+        )
+        if direction < 0:
+            merged[-1] = combined
+            index += 1
+        else:
+            merged.append(combined)
+            index += 2
+    return merged
+
+
+def _chunk_source_text(
+    text: str,
+    config: dict[str, Any],
+    preprocess_options: set[str],
+) -> list[tuple[TextChunk, tuple[str, ...]]]:
+    """按可选文档结构分段后切片，章节之间绝不共享 overlap。"""
+
+    method = str(_value(config, "chunk_method", "chunkMethod", "structure"))
+    detect_structure = (
+        method == "structure" or "detect_document_structure" in preprocess_options
+    )
+    preserve_context = "preserve_context" in preprocess_options
+    merge_short = "merge_short_content" in preprocess_options
+    chunk_size = int(_value(config, "chunk_size", "chunkSize", 800))
+    configured_minimum = int(_value(config, "min_chunk_size", "minChunkSize", 100))
+    minimum = configured_minimum if merge_short else 1
+    overlap = (
+        int(_value(config, "chunk_overlap", "chunkOverlap", 100))
+        if preserve_context
+        else 0
+    )
+    common = {
+        "method": method,
+        "chunk_size": chunk_size,
+        "chunk_overlap": overlap,
+        "min_chunk_size": minimum,
+        "custom_delimiter": str(
+            _value(config, "custom_delimiter", "customDelimiter", "") or ""
+        ),
+        "preserve_code_blocks": bool(
+            _value(config, "preserve_code_blocks", "preserveCodeBlocks", False)
+        ),
+        "preserve_tables": bool(
+            _value(config, "preserve_tables", "preserveTables", False)
+        ),
+        "preserve_lists": bool(
+            _value(config, "preserve_lists", "preserveLists", False)
+        ),
+    }
+
+    sections: list[tuple[int, int, tuple[str, ...]]] = [(0, len(text), ())]
+    if detect_structure:
+        structure = detect_document_structure(text)
+        if structure.headings:
+            sections = []
+            first_start = structure.headings[0].start
+            if first_start > 0 and text[:first_start].strip():
+                sections.append((0, first_start, ()))
+            stack: list[tuple[int, str]] = []
+            for index, heading in enumerate(structure.headings):
+                while stack and stack[-1][0] >= heading.level:
+                    stack.pop()
+                stack.append((heading.level, heading.title))
+                end = (
+                    structure.headings[index + 1].start
+                    if index + 1 < len(structure.headings)
+                    else len(text)
+                )
+                sections.append((heading.start, end, tuple(title for _, title in stack)))
+
+    result: list[tuple[TextChunk, tuple[str, ...]]] = []
+    for start, end, heading_path in sections:
+        section_text = text[start:end]
+        local_chunks = chunk_unstructured(section_text, **common)
+        shifted = [_shift_chunk(chunk, start, text) for chunk in local_chunks]
+        if merge_short:
+            shifted = _merge_short_chunks(
+                shifted,
+                text,
+                min_token_count=configured_minimum,
+                chunk_size=chunk_size,
+            )
+        result.extend((chunk, heading_path) for chunk in shifted)
+    return result
+
+
+_NEGATION_MARKERS = frozenset({"不", "无", "未", "否", "没有", "并非", "not", "no", "never"})
+
+
+def _safe_near_duplicate(left: str, right: str) -> bool:
+    """保守判断近重复，数字或否定含义变化时始终保留。"""
+
+    if min(estimate_token_count(left), estimate_token_count(right)) < 20:
+        return False
+    if re.findall(r"\d+(?:\.\d+)?", left) != re.findall(r"\d+(?:\.\d+)?", right):
+        return False
+    left_lower = left.casefold()
+    right_lower = right.casefold()
+    left_negations = {marker for marker in _NEGATION_MARKERS if marker in left_lower}
+    right_negations = {marker for marker in _NEGATION_MARKERS if marker in right_lower}
+    if left_negations != right_negations:
+        return False
+    return is_near_duplicate(
+        left,
+        right,
+        similarity_threshold=0.92,
+        max_hamming_distance=2,
+    )
+
+
+def _near_duplicate_band_keys(content: str) -> tuple[tuple[int, int], ...]:
+    """将 64 位 SimHash 分为三段，汉明距离不超过 2 时至少命中一段。"""
+
+    fingerprint = int(near_duplicate_fingerprint(content), 16)
+    widths = (22, 21, 21)
+    shift = 0
+    keys: list[tuple[int, int]] = []
+    for index, width in enumerate(widths):
+        keys.append((index, (fingerprint >> shift) & ((1 << width) - 1)))
+        shift += width
+    return tuple(keys)
+
+
 def _build_preview_items(
     task: dict[str, Any], source_files: list[dict[str, Any]]
 ) -> list[dict[str, Any]]:
@@ -140,6 +430,7 @@ def _build_preview_items(
         preprocess_options & {"deduplicate", "deduplicate_content"}
     )
     seen_content_hashes: set[str] = set()
+    seen_near_duplicate_bands: dict[tuple[int, int], list[str]] = {}
     items: list[dict[str, Any]] = []
 
     def append_item(item: dict[str, Any]) -> None:
@@ -155,38 +446,58 @@ def _build_preview_items(
         items.append(item)
 
     for source in source_files:
-        parsed = parse_text_content(
-            source.get("content") or "",
-            filename=source.get("name"),
-            file_format=source.get("file_format"),
-        )
+        parsed = _parse_stored_source(source)
         if process_type == "unstructured":
-            chunks = chunk_unstructured(
-                parsed.text,
-                method=_value(config, "chunk_method", "chunkMethod", "semantic"),
-                chunk_size=int(_value(config, "chunk_size", "chunkSize", 800)),
-                chunk_overlap=int(_value(config, "chunk_overlap", "chunkOverlap", 100)),
-                min_chunk_size=int(_value(config, "min_chunk_size", "minChunkSize", 100)),
-                custom_delimiter=str(
-                    _value(config, "custom_delimiter", "customDelimiter", "") or ""
-                ),
-                preserve_code_blocks=bool(
-                    _value(config, "preserve_code_blocks", "preserveCodeBlocks", False)
-                ),
-                preserve_tables=bool(
-                    _value(config, "preserve_tables", "preserveTables", False)
-                ),
-                preserve_lists=bool(
-                    _value(config, "preserve_lists", "preserveLists", False)
-                ),
-            )
-            for chunk in chunks:
+            chunks = _chunk_source_text(parsed.text, config, preprocess_options)
+            for chunk, heading_path in chunks:
+                preprocess_flags = content_quality_flags(
+                    chunk.content,
+                    min_chars=0,
+                    min_tokens=0,
+                )
+                flag_set = set(preprocess_flags)
+                if "clean_invalid_content" in preprocess_options and flag_set & {
+                    "empty_content",
+                    "low_printable_ratio",
+                    "repetitive_content",
+                }:
+                    continue
+                if "filter_low_quality" in preprocess_options and flag_set & {
+                    "content_too_long",
+                    "mojibake",
+                    "low_printable_ratio",
+                    "repetitive_content",
+                }:
+                    continue
+                if "deduplicate_content" in preprocess_options:
+                    band_keys = _near_duplicate_band_keys(chunk.content)
+                    candidates = {
+                        previous
+                        for key in band_keys
+                        for previous in seen_near_duplicate_bands.get(key, ())
+                    }
+                    if any(
+                        _safe_near_duplicate(chunk.content, previous)
+                        for previous in candidates
+                    ):
+                        continue
+                    for key in band_keys:
+                        seen_near_duplicate_bands.setdefault(key, []).append(chunk.content)
                 content = chunk.content
                 pii_counts: dict[str, int] = {}
                 if should_desensitize:
                     content, pii_counts = desensitize_pii(content)
                 quality = _preview_quality(content, config)
                 quality["pii_replacements"] = pii_counts
+                quality["preprocess_flags"] = list(preprocess_flags)
+                chunk_method = str(
+                    _value(config, "chunk_method", "chunkMethod", "structure")
+                )
+                if (
+                    chunk_method == "structure"
+                    or "detect_document_structure" in preprocess_options
+                ):
+                    quality["heading_path"] = list(heading_path)
                 append_item(
                     {
                         "source_file_id": source["id"],
@@ -203,19 +514,41 @@ def _build_preview_items(
                 )
             continue
 
-        record_contents = [
-            json.dumps(record, ensure_ascii=False, separators=(",", ":"))
-            for record in parsed.records
-            if not should_clean_invalid
-            or any(value not in (None, "", [], {}) for value in record.values())
-        ]
-        if not record_contents and parsed.text:
-            record_contents = [parsed.text]
-        for content in record_contents:
-            original_content = content
-            pii_counts = {}
+        structured_options = preprocess_options & {
+            "clean_invalid",
+            "detect_structure",
+            "deduplicate",
+            "normalize_format",
+            "filter_anomaly",
+        }
+        source_records = list(parsed.records)
+        processed_records = preprocess_structured_records(
+            source_records,
+            structured_options,
+        )
+        if not processed_records and parsed.text and not source_records:
+            processed_records = [{"value": parsed.text}]
+        same_cardinality = len(processed_records) == len(source_records)
+        for index, record in enumerate(processed_records):
+            original_record = source_records[index] if same_cardinality else record
+            original_content = json.dumps(
+                original_record,
+                ensure_ascii=False,
+                separators=(",", ":"),
+            )
+            pii_counts: dict[str, int] = {}
+            edited_record = record
             if should_desensitize:
-                content, pii_counts = desensitize_pii(content)
+                edited_record, pii_counts = desensitize_structured_record(record)
+            content = (
+                canonical_record_json(edited_record)
+                if "normalize_format" in preprocess_options
+                else json.dumps(
+                    edited_record,
+                    ensure_ascii=False,
+                    separators=(",", ":"),
+                )
+            )
             quality = _preview_quality(content, config)
             quality["pii_replacements"] = pii_counts
             append_item(
@@ -488,58 +821,91 @@ async def upload_source_files(
     task_id: str,
     files: list[UploadFile] = File(...),
     store: DataProcessStore = Depends(get_data_process_store),
+    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
 ) -> dict[str, Any]:
     if not files:
         raise fail(400, "at least one source file is required")
     if len(files) > MAX_SOURCE_FILE_COUNT:
         raise fail(413, f"a source batch may contain at most {MAX_SOURCE_FILE_COUNT} files")
     prepared: list[dict[str, Any]] = []
+    staged: list[StagedSourceObject] = []
+    batch_id = storage.new_batch_id()
     batch_size = 0
-    with api_errors():
-        store.get_task(task_id)
-        for upload in files:
-            raw = await upload.read(MAX_SOURCE_FILE_BYTES + 1)
-            if len(raw) > MAX_SOURCE_FILE_BYTES:
-                raise fail(413, f"source file exceeds {MAX_SOURCE_FILE_BYTES} bytes")
-            content = decode_utf8(raw)
-            name = _safe_file_name(upload.filename, "source.txt")
-            suffix = Path(name).suffix.lower()
-            if suffix not in {
-                ".txt",
-                ".md",
-                ".markdown",
-                ".csv",
-                ".tsv",
-                ".json",
-                ".jsonl",
-                ".ndjson",
-            }:
-                raise fail(415, f"unsupported source file format: {suffix or 'none'}")
-            parsed = parse_text_content(content, filename=name)
-            if not parsed.text:
-                raise fail(400, f"source file is empty: {name}")
-            normalized_raw = parsed.text.encode("utf-8")
-            batch_size += len(normalized_raw)
-            if batch_size > MAX_SOURCE_BATCH_BYTES:
-                raise fail(413, f"source batch exceeds {MAX_SOURCE_BATCH_BYTES} bytes")
-            record_count = len(parsed.records) or (1 if parsed.text else 0)
-            prepared.append(
-                {
-                    "name": name,
-                    "content": parsed.text,
-                    "raw_size": len(normalized_raw),
-                    "checksum_sha256": hashlib.sha256(normalized_raw).hexdigest(),
-                    "file_format": parsed.format,
-                    "record_count": record_count,
-                    "metadata": {
-                        "content_type": upload.content_type or "text/plain",
-                        "original_size_bytes": len(raw),
-                        "original_checksum_sha256": hashlib.sha256(raw).hexdigest(),
-                    },
-                    "created_by": None,
-                }
+    commit_attempted = False
+    try:
+        with api_errors():
+            task = store.get_task(task_id)
+            process_type = str(task["process_type"])
+            if process_type == "external":
+                raise InvalidStateError(
+                    "external tasks must import data through the external source endpoint"
+                )
+            allowed_suffixes = (
+                UNSTRUCTURED_SOURCE_SUFFIXES
+                if process_type == "unstructured"
+                else STRUCTURED_SOURCE_SUFFIXES
             )
-        created = store.add_source_files(task_id, prepared)
+            for upload in files:
+                raw = await upload.read(MAX_SOURCE_FILE_BYTES + 1)
+                if len(raw) > MAX_SOURCE_FILE_BYTES:
+                    raise fail(413, f"source file exceeds {MAX_SOURCE_FILE_BYTES} bytes")
+                name = _safe_file_name(upload.filename, "source.txt")
+                suffix = Path(name).suffix.lower()
+                if suffix in LEGACY_OFFICE_CONVERSIONS:
+                    replacement = LEGACY_OFFICE_CONVERSIONS[suffix]
+                    raise fail(
+                        415,
+                        f"legacy {suffix} format is not supported; "
+                        f"convert the file to {replacement} and upload again",
+                    )
+                if suffix not in SUPPORTED_SOURCE_SUFFIXES:
+                    raise fail(415, f"unsupported source file format: {suffix or 'none'}")
+                if suffix not in allowed_suffixes:
+                    raise fail(
+                        415,
+                        f"{suffix} is not supported for {process_type} data processing",
+                    )
+                parsed = parse_text_content(raw, filename=name)
+                if not parsed.text:
+                    raise fail(400, f"source file is empty: {name}")
+                batch_size += len(raw)
+                if batch_size > MAX_SOURCE_BATCH_BYTES:
+                    raise fail(413, f"source batch exceeds {MAX_SOURCE_BATCH_BYTES} bytes")
+                source_file_id = new_id("dpsf")
+                staged_object = storage.stage_bytes(
+                    batch_id=batch_id,
+                    task_id=task_id,
+                    source_file_id=source_file_id,
+                    version=1,
+                    name=name,
+                    content=raw,
+                )
+                staged.append(staged_object)
+                record_count = len(parsed.records) or (1 if parsed.text else 0)
+                prepared.append(
+                    {
+                        "id": source_file_id,
+                        "storage_object_id": staged_object.reference,
+                        "name": name,
+                        "content": parsed.text,
+                        "raw_size": len(raw),
+                        "checksum_sha256": hashlib.sha256(raw).hexdigest(),
+                        "file_format": parsed.format,
+                        "record_count": record_count,
+                        "metadata": {
+                            "content_type": upload.content_type or "text/plain",
+                            "original_size_bytes": len(raw),
+                            "original_checksum_sha256": hashlib.sha256(raw).hexdigest(),
+                        },
+                        "created_by": None,
+                    }
+                )
+            # publish 无论成功或失败都会消费并清理暂存对象，外层不能再次 discard。
+            commit_attempted = True
+            created = _commit_source_batch(store, storage, task_id, prepared, staged)
+    finally:
+        if not commit_attempted:
+            storage.discard(staged)
     return ok({"files": created}, "source files uploaded")
 
 
@@ -559,15 +925,145 @@ def source_file_content(
         return ok(store.source_content_window(task_id, file_id, offset, limit))
 
 
+@router.get("/{task_id}/source-files/{file_id}/raw")
+def source_file_raw(
+    task_id: str,
+    file_id: str,
+    range_header: str | None = Header(default=None, alias="Range"),
+    store: DataProcessStore = Depends(get_data_process_store),
+    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
+) -> StreamingResponse:
+    with api_errors():
+        source = store.get_source_file(task_id, file_id, include_content=False)
+        if str(source.get("file_format") or "").lower() != "pdf":
+            raise fail(415, "raw inline preview is only available for PDF source files")
+        storage_object_id = str(source.get("storage_object_id") or "")
+        actual_size = storage.file_size(
+            storage_object_id,
+            expected_task_id=task_id,
+            expected_source_file_id=file_id,
+        )
+        if actual_size is None:
+            raise fail(410, "the original PDF is unavailable for this legacy source file")
+        expected_size = int(source.get("size_bytes") or 0)
+        if actual_size != expected_size:
+            raise ValueError("source object size does not match metadata")
+        selected_range = _source_byte_range(range_header, actual_size)
+        start, end = selected_range or (0, actual_size - 1)
+        length = end - start + 1
+        name = _safe_file_name(str(source.get("name") or "source.pdf"), "source.pdf")
+        headers = {
+            "Accept-Ranges": "bytes",
+            "Cache-Control": "private, no-store",
+            "Content-Disposition": f"inline; filename*=UTF-8''{quote(name, safe='')}",
+            "Content-Length": str(length),
+            "X-Accel-Buffering": "no",
+            "X-Content-Type-Options": "nosniff",
+        }
+        checksum = str(source.get("checksum_sha256") or "")
+        if checksum:
+            headers["ETag"] = f'"{checksum}"'
+        if selected_range is not None:
+            headers["Content-Range"] = f"bytes {start}-{end}/{actual_size}"
+        body = storage.iter_bytes(
+            storage_object_id,
+            expected_task_id=task_id,
+            expected_source_file_id=file_id,
+            expected_size=actual_size,
+            start=start,
+            length=length,
+        )
+        return StreamingResponse(
+            body,
+            status_code=206 if selected_range is not None else 200,
+            media_type="application/pdf",
+            headers=headers,
+        )
+
+
+@router.get("/{task_id}/source-files/{file_id}/pdf-pages")
+def source_file_pdf_pages(
+    task_id: str,
+    file_id: str,
+    store: DataProcessStore = Depends(get_data_process_store),
+    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
+) -> dict[str, Any]:
+    """返回切片全文偏移对应的 PDF 物理页范围。"""
+
+    with api_errors():
+        source = store.get_source_file(task_id, file_id, include_content=False)
+        if str(source.get("file_format") or "").lower() != "pdf":
+            raise fail(415, "PDF page mapping is only available for PDF source files")
+        storage_object_id = str(source.get("storage_object_id") or "")
+        actual_size = storage.file_size(
+            storage_object_id,
+            expected_task_id=task_id,
+            expected_source_file_id=file_id,
+        )
+        if actual_size is None:
+            raise fail(410, "the original PDF is unavailable for this legacy source file")
+        expected_size = int(source.get("size_bytes") or 0)
+        if actual_size != expected_size:
+            raise ValueError("source object size does not match metadata")
+        raw = b"".join(
+            storage.iter_bytes(
+                storage_object_id,
+                expected_task_id=task_id,
+                expected_source_file_id=file_id,
+                expected_size=actual_size,
+            )
+        )
+        pages = extract_pdf_page_texts(raw)
+        return ok(
+            {
+                "page_count": len(pages),
+                "pages": [
+                    {
+                        "page_number": page.page_number,
+                        "source_start": page.source_start,
+                        "source_end": page.source_end,
+                    }
+                    for page in pages
+                ],
+            }
+        )
+
+
 @router.delete("/{task_id}/source-files/{file_id}")
 def delete_source_file(
     task_id: str,
     file_id: str,
     store: DataProcessStore = Depends(get_data_process_store),
+    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
 ) -> dict[str, Any]:
     with api_errors():
+        source = store.get_source_file(task_id, file_id, include_content=False)
+        storage_object_id = str(source.get("storage_object_id") or "")
+        storage.validate_owner(
+            storage_object_id,
+            expected_task_id=task_id,
+            expected_source_file_id=file_id,
+        )
         store.delete_source_file(task_id, file_id)
-        return ok({"deleted": file_id}, "source file removed")
+        cleanup_pending = False
+        try:
+            storage.delete(
+                storage_object_id,
+                expected_task_id=task_id,
+                expected_source_file_id=file_id,
+            )
+        except Exception:
+            # 数据库软删除已经提交，不能再向客户端返回可重试的失败；保留逻辑引用，
+            # 由后续存储清理任务重试物理删除。
+            cleanup_pending = True
+            logger.exception(
+                "failed to remove data process source object after soft deletion",
+                extra={"task_id": task_id, "source_file_id": file_id},
+            )
+        return ok(
+            {"deleted": file_id, "storage_cleanup_pending": cleanup_pending},
+            "source file removed",
+        )
 
 
 def _external_postgres_connection(payload: ExternalSourceRequest) -> psycopg.Connection[Any]:
@@ -635,7 +1131,11 @@ def test_external_source(
     store: DataProcessStore = Depends(get_data_process_store),
 ) -> dict[str, Any]:
     with api_errors():
-        store.get_task(task_id)
+        task = store.get_task(task_id)
+        if str(task.get("process_type")) != "external":
+            raise InvalidStateError(
+                "external source access requires an external data processing task"
+            )
         try:
             with _external_postgres_connection(payload) as conn:
                 conn.execute("SELECT 1 AS ok").fetchone()
@@ -649,6 +1149,7 @@ def pull_external_source(
     task_id: str,
     payload: ExternalPullRequest,
     store: DataProcessStore = Depends(get_data_process_store),
+    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
 ) -> dict[str, Any]:
     query = (payload.query or "").strip()
     if query.endswith(";"):
@@ -659,7 +1160,9 @@ def pull_external_source(
     if first_token not in {"select", "with"}:
         raise fail(400, "a read-only SELECT or WITH query is required for external pull")
     with api_errors():
-        store.get_task(task_id)
+        task = store.get_task(task_id)
+        if str(task.get("process_type")) != "external":
+            raise InvalidStateError("external pull requires an external data processing task")
         try:
             with _external_postgres_connection(payload) as conn:
                 conn.execute("SET TRANSACTION READ ONLY")
@@ -686,20 +1189,40 @@ def pull_external_source(
             raise fail(400, "external query returned no rows")
         content = "".join(content_parts)
         raw = content.encode("utf-8")
-        source = store.add_source_file(
-            task_id,
-            name=_safe_file_name(payload.file_name, "external-data.jsonl"),
-            content=content,
-            raw_size=len(raw),
-            checksum_sha256=hashlib.sha256(raw).hexdigest(),
-            file_format="jsonl",
-            record_count=len(rows),
-            metadata={
-                "external_type": payload.type,
-                "external_host": urlsplit(payload.url).hostname,
-                "external_limit": payload.limit,
-            },
+        name = _safe_file_name(payload.file_name, "external-data.jsonl")
+        source_file_id = new_id("dpsf")
+        staged = storage.stage_bytes(
+            batch_id=storage.new_batch_id(),
+            task_id=task_id,
+            source_file_id=source_file_id,
+            version=1,
+            name=name,
+            content=raw,
         )
+        sources = _commit_source_batch(
+            store,
+            storage,
+            task_id,
+            [
+                {
+                    "id": source_file_id,
+                    "storage_object_id": staged.reference,
+                    "name": name,
+                    "content": content,
+                    "raw_size": len(raw),
+                    "checksum_sha256": hashlib.sha256(raw).hexdigest(),
+                    "file_format": "jsonl",
+                    "record_count": len(rows),
+                    "metadata": {
+                        "external_type": payload.type,
+                        "external_host": urlsplit(payload.url).hostname,
+                        "external_limit": payload.limit,
+                    },
+                }
+            ],
+            [staged],
+        )
+        source = sources[0]
         return ok({"files": [source]}, "external source pulled")
 
 
@@ -724,7 +1247,7 @@ def _prepare_preview_items(
     if not sources:
         raise InvalidStateError("at least one source file is required")
     items = _build_preview_items(task, sources)
-    if not items:
+    if not items and source_file_ids is None:
         raise InvalidStateError("source files did not produce preview items")
     return items
 
@@ -736,10 +1259,38 @@ def build_preview(
     store: DataProcessStore = Depends(get_data_process_store),
 ) -> dict[str, Any]:
     with api_errors():
-        items = _prepare_preview_items(task_id, store, payload.source_file_ids)
-        created = store.replace_preview_items(task_id, items)
+        selected_ids = payload.source_file_ids
+        items = _prepare_preview_items(task_id, store, selected_ids)
+        created = store.replace_preview_items(
+            task_id,
+            items,
+            source_file_ids=selected_ids,
+        )
+        target_ids = selected_ids or [
+            str(source["id"]) for source in store.list_source_files(task_id)
+        ]
+        file_counts = dict.fromkeys(target_ids, 0)
+        for item in created:
+            source_file_id = str(item.get("source_file_id") or "")
+            if source_file_id in file_counts:
+                file_counts[source_file_id] += 1
+        files = [
+            {
+                "source_file_id": source_file_id,
+                "preview_count": count,
+                "status": "completed" if count else "empty",
+            }
+            for source_file_id, count in file_counts.items()
+        ]
         return ok(
-            {"items": created, "total": len(created), "page": 1, "page_size": len(created)},
+            {
+                "items": created,
+                "total": len(created),
+                "page": 1,
+                "page_size": len(created),
+                "file_counts": file_counts,
+                "files": files,
+            },
             "preview built",
         )