fix: 完善数据预处理与 JSON 上传链路

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

@@ -8,12 +8,14 @@ import os
 import re
 import socket
 import time
+from collections.abc import Iterator, Mapping
 from concurrent.futures import ThreadPoolExecutor, as_completed
 from contextlib import contextmanager
+from copy import deepcopy
 from dataclasses import asdict
 from pathlib import Path
 from threading import BoundedSemaphore, Lock
-from typing import Any, Iterator, Literal
+from typing import Any, Literal
 from urllib.parse import quote, urlsplit
 
 import httpx
@@ -45,9 +47,10 @@ from app.modules.data_process.algorithms import (
     is_near_duplicate,
     near_duplicate_fingerprint,
     parse_text_content,
-    preprocess_structured_records,
+    preprocess_structured_records_with_lineage,
     remove_document_noise,
     score_quality,
+    structured_json_dumps,
 )
 from app.modules.data_process.document_chunking import (
     DocumentChunk,
@@ -75,9 +78,11 @@ from app.modules.data_process.store import (
     NotFoundError,
     get_data_process_store,
     new_id,
+    repeat_task_id,
 )
 from app.schemas.data_process import (
     DataProcessRegenerateRequest,
+    DataProcessRepeatRequest,
     DataProcessStatus,
     DataProcessTaskCreate,
     DataProcessTaskUpdate,
@@ -261,7 +266,14 @@ 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 后入库。
+        raw_content = source.get("raw_content")
+        if isinstance(raw_content, bytes):
+            return parse_text_content(
+                raw_content,
+                filename=str(source.get("name") or "source.xlsx"),
+                file_format="xlsx",
+            )
+        # 兼容原始对象已缺失的历史文件：退化为上传阶段生成的 JSONL。
         return parse_text_content(content, file_format="jsonl")
     if file_format in {"pdf", "docx", "pptx"}:
         # 文档上传阶段已抽取文本，预览阶段只需要对正文切片。
@@ -381,11 +393,12 @@ def _build_preview_items(
     seen_near_duplicate_bands: dict[tuple[int, int], list[str]] = {}
     items: list[dict[str, Any]] = []
 
-    def append_item(item: dict[str, Any]) -> None:
+    def append_item(item: dict[str, Any], *, dedup_content: str) -> None:
         content = str(item.get("edited_content") or "").strip()
         if should_clean_invalid and not content:
             return
-        content_hash = hashlib.sha256(content.encode("utf-8")).hexdigest()
+        # 去重必须基于脱敏前内容，否则不同原文可能在替换 PII 后被错误合并。
+        content_hash = hashlib.sha256(dedup_content.strip().encode("utf-8")).hexdigest()
         if should_deduplicate and content_hash in seen_content_hashes:
             return
         seen_content_hashes.add(content_hash)
@@ -447,6 +460,7 @@ def _build_preview_items(
                         continue
                     for key in band_keys:
                         seen_near_duplicate_bands.setdefault(key, []).append(content)
+                dedup_content = content
                 pii_counts: dict[str, int] = {}
                 if should_desensitize:
                     content, pii_counts = desensitize_pii(content)
@@ -476,7 +490,8 @@ def _build_preview_items(
                             else "original"
                         ),
                         "quality_score": quality,
-                    }
+                    },
+                    dedup_content=dedup_content,
                 )
             continue
 
@@ -488,48 +503,76 @@ def _build_preview_items(
             "filter_anomaly",
         }
         source_records = list(parsed.records)
-        processed_records = preprocess_structured_records(
+        processed_records = preprocess_structured_records_with_lineage(
             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=(",", ":"),
+        for processed in processed_records:
+            source_index = processed.source_index
+            record = processed.record
+            original_record = (
+                source_records[source_index]
+                if source_index < len(source_records)
+                else record
             )
+            source_locator = (
+                deepcopy(parsed.record_locators[source_index])
+                if source_index < len(parsed.record_locators)
+                else None
+            )
+            original_content = structured_json_dumps(original_record)
             pii_counts: dict[str, int] = {}
             edited_record = record
+            dedup_content = (
+                canonical_record_json(record)
+                if "normalize_format" in preprocess_options
+                else structured_json_dumps(record)
+            )
             if should_desensitize:
                 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=(",", ":"),
-                )
+                else structured_json_dumps(edited_record)
             )
             quality = _preview_quality(content, config)
             quality["pii_replacements"] = pii_counts
+            if source_locator is not None:
+                quality["source_locator"] = source_locator
+            source_start = (
+                source_locator.get("source_start")
+                if source_locator is not None
+                else None
+            )
+            source_end = (
+                source_locator.get("source_end")
+                if source_locator is not None
+                else None
+            )
+            source_start_line = (
+                source_locator.get("start_line")
+                if source_locator is not None
+                else None
+            )
+            source_end_line = (
+                source_locator.get("end_line")
+                if source_locator is not None
+                else None
+            )
             append_item(
                 {
                     "source_file_id": source["id"],
                     "original_content": original_content,
                     "edited_content": content,
-                    "source_start": None,
-                    "source_end": None,
-                    "source_start_line": None,
-                    "source_end_line": None,
+                    "source_start": source_start,
+                    "source_end": source_end,
+                    "source_start_line": source_start_line,
+                    "source_end_line": source_end_line,
                     "token_count": estimate_token_count(content),
                     "status": "modified" if content != original_content else "original",
                     "quality_score": quality,
-                }
+                },
+                dedup_content=dedup_content,
             )
     return items
 
@@ -850,6 +893,135 @@ def prepare_regeneration(
         )
 
 
+def _repeat_file_copies(
+    store: DataProcessStore,
+    storage: LocalDataProcessStorage,
+    source_task_id: str,
+    request_id: str,
+) -> tuple[dict[str, dict[str, str]], list[StagedSourceObject]]:
+    """为新任务创建独立的源文件引用，避免删除任一任务时互相影响。"""
+
+    target_task_id = repeat_task_id(source_task_id, request_id)
+    copies: dict[str, dict[str, str]] = {}
+    staged: list[StagedSourceObject] = []
+    batch_id = storage.new_batch_id()
+    for summary in store.list_source_files(source_task_id):
+        old_file_id = str(summary["id"])
+        source = store.get_source_file(source_task_id, old_file_id, include_content=True)
+        new_file_id = new_id("dpsf")
+        old_reference = str(source.get("storage_object_id") or "")
+        if old_reference.startswith("local://data-process/"):
+            staged_object = storage.stage_copy(
+                batch_id=batch_id,
+                source_reference=old_reference,
+                expected_source_task_id=source_task_id,
+                expected_source_file_id=old_file_id,
+                task_id=target_task_id,
+                source_file_id=new_file_id,
+                version=1,
+                name=str(source["name"]),
+            )
+            staged.append(staged_object)
+            new_reference = staged_object.reference
+        elif old_reference.startswith("db://data-process/") or not old_reference:
+            new_reference = f"db://data-process/{target_task_id}/{new_file_id}/v1"
+        else:
+            raise ValueError("源任务包含不受支持的文件存储引用")
+        copies[old_file_id] = {
+            "id": new_file_id,
+            "storage_object_id": new_reference,
+        }
+    return copies, staged
+
+
+def _remove_repeated_storage_objects(
+    storage: LocalDataProcessStorage,
+    task_id: str,
+    staged: list[StagedSourceObject],
+    copies: dict[str, dict[str, str]],
+) -> None:
+    source_file_ids = {
+        str(copy["storage_object_id"]): str(copy["id"])
+        for copy in copies.values()
+    }
+    for item in staged:
+        try:
+            storage.delete(
+                item.reference,
+                expected_task_id=task_id,
+                expected_source_file_id=source_file_ids[item.reference],
+            )
+        except Exception:
+            logger.exception(
+                "failed to roll back repeated data process source object task_id=%s",
+                task_id,
+            )
+
+
+@router.post("/{task_id}/repeat", status_code=202)
+def repeat_generation(
+    task_id: str,
+    payload: DataProcessRepeatRequest,
+    background_tasks: BackgroundTasks,
+    store: DataProcessStore = Depends(get_data_process_store),
+    storage: LocalDataProcessStorage = Depends(get_data_process_storage),
+) -> dict[str, Any]:
+    """按原任务快照创建独立任务，并立即在后台开始新一批生成。"""
+
+    with api_errors():
+        repeated = store.find_repeated_task(task_id, payload.request_id)
+        staged: list[StagedSourceObject] = []
+        target_task_id = repeat_task_id(task_id, payload.request_id)
+        if repeated is None:
+            copies, staged = _repeat_file_copies(
+                store,
+                storage,
+                task_id,
+                payload.request_id,
+            )
+            storage.publish(staged)
+            try:
+                repeated = store.repeat_task(
+                    task_id,
+                    expected_updated_at=payload.expected_updated_at,
+                    request_id=payload.request_id,
+                    file_copies=copies,
+                )
+            except Exception:
+                _remove_repeated_storage_objects(
+                    storage,
+                    target_task_id,
+                    staged,
+                    copies,
+                )
+                raise
+            if not repeated["created"]:
+                _remove_repeated_storage_objects(
+                    storage,
+                    target_task_id,
+                    staged,
+                    copies,
+                )
+
+        repeated_task = repeated["task"]
+        if repeated_task.get("status") == "pending":
+            try:
+                started = store.start_generation(target_task_id, replace_existing=True)
+                background_tasks.add_task(
+                    _run_generation,
+                    store,
+                    target_task_id,
+                    str(started["generation_run_id"]),
+                )
+            except ConflictError:
+                latest = store.get_task(target_task_id)
+                if latest.get("status") != "running":
+                    raise
+        repeated["task"] = store.get_task(target_task_id)
+        repeated["progress"] = store.progress(target_task_id)
+        return ok(repeated, "已按原配置创建新任务并开始后台生成")
+
+
 @router.delete("/{task_id}")
 def delete_task(
     task_id: str,
@@ -919,7 +1091,7 @@ async def upload_source_files(
                         f"{suffix} is not supported for {process_type} data processing",
                     )
                 parsed = parse_text_content(raw, filename=name)
-                if not parsed.text:
+                if not parsed.text.strip():
                     raise fail(400, f"source file is empty: {name}")
                 batch_size += len(raw)
                 if batch_size > MAX_SOURCE_BATCH_BYTES:
@@ -934,7 +1106,11 @@ async def upload_source_files(
                     content=raw,
                 )
                 staged.append(staged_object)
-                record_count = len(parsed.records) or (1 if parsed.text else 0)
+                record_count = (
+                    len(parsed.records)
+                    if process_type == "structured"
+                    else (1 if parsed.text else 0)
+                )
                 prepared.append(
                     {
                         "id": source_file_id,
@@ -1362,15 +1538,28 @@ def _prepare_preview_items(
         _value(config, "chunk_method", "chunkMethod", "layout_hybrid")
     )
     is_unstructured = task.get("process_type") == "unstructured"
-    if is_unstructured and (
+    needs_unstructured_raw = is_unstructured and (
         chunk_method == "layout_hybrid"
         or preprocess_options & {"clean_invalid", "clean_invalid_content"}
-    ):
+    )
+    has_structured_xlsx = not is_unstructured and any(
+        str(source.get("file_format") or "").lower() == "xlsx"
+        for source in sources
+    )
+    if needs_unstructured_raw or has_structured_xlsx:
         for index, source in enumerate(sources):
-            if (
-                chunk_method != "layout_hybrid"
-                and str(source.get("file_format") or "").lower() != "pdf"
-            ):
+            source_format = str(source.get("file_format") or "").lower()
+            needs_structured_xlsx = not is_unstructured and source_format == "xlsx"
+            needs_layout_raw = is_unstructured and chunk_method == "layout_hybrid"
+            needs_pdf_noise = (
+                is_unstructured
+                and not needs_layout_raw
+                and source_format == "pdf"
+                and bool(
+                    preprocess_options & {"clean_invalid", "clean_invalid_content"}
+                )
+            )
+            if not (needs_structured_xlsx or needs_layout_raw or needs_pdf_noise):
                 continue
             storage_object_id = str(source.get("storage_object_id") or "")
             actual_size = storage.file_size(
@@ -1379,7 +1568,7 @@ def _prepare_preview_items(
                 expected_source_file_id=str(source["id"]),
             )
             if actual_size is None:
-                if chunk_method == "layout_hybrid":
+                if needs_layout_raw:
                     raise InvalidStateError(
                         "版面结构混合切分无法读取原始文件，请重新上传后再处理"
                     )
@@ -1396,12 +1585,10 @@ def _prepare_preview_items(
                 )
             )
             enriched = dict(source)
-            if chunk_method == "layout_hybrid":
+            if needs_structured_xlsx or needs_layout_raw:
                 enriched["raw_content"] = raw
                 sources[index] = enriched
                 continue
-            if str(source.get("file_format") or "").lower() != "pdf":
-                continue
             pages = extract_pdf_page_texts(raw)
             extracted_text = "\n\n".join(page.text for page in pages if page.text)
             if extracted_text != str(source.get("content") or ""):
@@ -1413,7 +1600,7 @@ def _prepare_preview_items(
             enriched["document_noise_spans"] = detect_pdf_document_noise(pages)
             sources[index] = enriched
     items = _build_preview_items(task, sources)
-    if not items and source_file_ids is None:
+    if not items and source_file_ids is None and is_unstructured:
         raise InvalidStateError("source files did not produce preview items")
     return items
 
@@ -1435,6 +1622,7 @@ def _run_preview(
         len(source_file_ids),
     )
     try:
+        is_unstructured = store.get_task(task_id).get("process_type") == "unstructured"
         if not store.mark_preview_running(task_id, preview_run_id):
             logger.info(
                 "data process preview skipped inactive run task_id=%s preview_run_id=%s",
@@ -1461,7 +1649,7 @@ def _run_preview(
                 storage,
                 [source_file_id],
             )
-            if not items:
+            if not items and is_unstructured:
                 raise InvalidStateError(
                     f"source file did not produce preview items: {source_file_id}"
                 )
@@ -1657,8 +1845,13 @@ def update_preview_item(
 ) -> dict[str, Any]:
     with api_errors():
         task = store.get_task(task_id)
+        existing = store.get_preview_item(task_id, preview_id)
         update = payload.model_dump(exclude_unset=True, mode="json")
-        update["quality_score"] = _preview_quality(payload.edited_content, task.get("config") or {})
+        quality = _preview_quality(payload.edited_content, task.get("config") or {})
+        source_locator = (existing.get("quality_score") or {}).get("source_locator")
+        if isinstance(source_locator, Mapping):
+            quality["source_locator"] = deepcopy(dict(source_locator))
+        update["quality_score"] = quality
         item = store.update_preview_item(
             task_id,
             preview_id,