diff --git a/.gitignore b/.gitignore index 609b66b..72f03fd 100644 --- a/.gitignore +++ b/.gitignore @@ -44,6 +44,7 @@ pip-delete-this-directory.txt # Runtime data and logs runtime/ backend/runtime/ +backend/storage/ logs/ backend/logs/ *.db diff --git a/backend/app/api/v1/endpoints/data_process.py b/backend/app/api/v1/endpoints/data_process.py index b024ff6..96cd93f 100644 --- a/backend/app/api/v1/endpoints/data_process.py +++ b/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", ) diff --git a/backend/app/modules/data_process/algorithms.py b/backend/app/modules/data_process/algorithms.py index 9e4fb12..a293f9e 100644 --- a/backend/app/modules/data_process/algorithms.py +++ b/backend/app/modules/data_process/algorithms.py @@ -11,18 +11,53 @@ import csv import hashlib import io import json +import math import re import unicodedata +import xml.etree.ElementTree as ET +import zipfile from bisect import bisect_left +from collections import Counter from collections.abc import Iterable, Mapping, Sequence +from copy import deepcopy from dataclasses import dataclass -from pathlib import Path +from datetime import date, datetime, time +from pathlib import Path, PurePosixPath from typing import Any, Literal +from urllib.parse import unquote, urlsplit +from docx import Document +from docx.oxml.table import CT_Tbl +from docx.oxml.text.paragraph import CT_P +from docx.table import Table +from docx.text.paragraph import Paragraph +from llama_index.core.node_parser import SentenceSplitter, TokenTextSplitter +from openpyxl import load_workbook +from openpyxl.utils.cell import range_boundaries +from pptx import Presentation +from pypdf import PdfReader -TextFormat = Literal["json", "jsonl", "csv", "markdown", "txt"] -ChunkMethod = Literal["semantic", "heading", "fixed", "custom"] +TextFormat = Literal[ + "json", + "jsonl", + "csv", + "markdown", + "txt", + "pdf", + "docx", + "xlsx", + "pptx", +] +ChunkMethod = Literal["structure", "fixed", "custom"] DatasetSplit = Literal["train", "validation", "test"] +StructuredPreprocessOption = Literal[ + "clean_invalid", + "detect_structure", + "deduplicate", + "normalize_format", + "filter_anomaly", + "desensitize", +] SUPPORTED_TEXT_FORMATS: tuple[TextFormat, ...] = ( "json", @@ -30,6 +65,10 @@ SUPPORTED_TEXT_FORMATS: tuple[TextFormat, ...] = ( "csv", "markdown", "txt", + "pdf", + "docx", + "xlsx", + "pptx", ) _FORMAT_ALIASES: dict[str, TextFormat] = { @@ -42,6 +81,58 @@ _FORMAT_ALIASES: dict[str, TextFormat] = { "markdown": "markdown", "txt": "txt", "text": "txt", + "pdf": "pdf", + "docx": "docx", + "xlsx": "xlsx", + "pptx": "pptx", +} +_LEGACY_OFFICE_FORMATS: dict[str, str] = { + "doc": "docx", + "xls": "xlsx", + "ppt": "pptx", +} +_OFFICE_OPEN_XML_FORMATS = {"docx", "xlsx", "pptx"} +_MAX_ARCHIVE_ENTRIES = 10_000 +_MAX_ARCHIVE_UNCOMPRESSED_BYTES = 512 * 1024 * 1024 +_MAX_ARCHIVE_ENTRY_BYTES = 128 * 1024 * 1024 +_MAX_ARCHIVE_COMPRESSION_RATIO = 200 +_MAX_EXTRACTED_TEXT_CHARS = 20_000_000 +_MAX_PDF_PAGES = 2_000 +_MAX_PRESENTATION_SLIDES = 2_000 +_MAX_WORKBOOK_SHEETS = 100 +_MAX_WORKBOOK_ROWS = 100_000 +_MAX_WORKBOOK_SCANNED_ROWS = 200_000 +_MAX_WORKBOOK_COLUMNS = 256 +_MAX_WORKBOOK_CELLS = 2_000_000 +_MAX_WORKBOOK_HEADER_ROWS = 8 +_MAX_WORKBOOK_HEADER_SCAN_ROWS = 64 +_MAX_WORKBOOK_MERGED_RANGES = 100_000 +_MAX_STRUCTURED_FIELDS = 1_024 +_MAX_STRUCTURED_DEPTH = 16 +_MAX_ANOMALY_TEXT_CHARS = 1_000_000 +_STRUCTURED_OPTIONS = { + "clean_invalid", + "detect_structure", + "deduplicate", + "normalize_format", + "filter_anomaly", + "desensitize", +} +_IDENTITY_FIELD_PATTERN = re.compile(r"(?:^|[._])(?:id|uuid|key|code)$|(?:^|[._]).+_id$") +_MOJIBAKE_MARKERS = ("\ufffd", "锟斤拷", "烫烫烫", "屯屯屯", "Ã", "Â", "â€") +_NAME_FIELD_NAMES = { + "name", + "full_name", + "fullname", + "real_name", + "contact_name", + "customer_name", + "recipient_name", + "姓名", + "真实姓名", + "联系人", + "联系人姓名", + "收件人", } _EMAIL_PATTERN = re.compile( @@ -51,23 +142,39 @@ _EMAIL_PATTERN = re.compile( ) _PHONE_PATTERN = re.compile(r"(?姓名|真实姓名|联系人(?:姓名)?|收件人)" + r"(?P\s*(?:[::=]|为)\s*|\s+)" + r"(?P[\u3400-\u4dbf\u4e00-\u9fff·]{2,8})" ) +_ENGLISH_NAME_CONTEXT_PATTERN = re.compile( + r"(?im)(?P