| """ |
| 外部模板分析服务 |
| |
| 负责模板上传校验、文件保存、版本管理。 |
| 结构提取和 LLM 分析方法将在后续任务 (6.2, 6.3) 中实现。 |
| |
| Requirements: 1.1, 1.2, 1.3, 1.6, 1.7, 1.8, 9.1, 9.2, 9.3, 9.4 |
| """ |
|
|
| from __future__ import annotations |
|
|
| import json |
| import logging |
| import re |
| import uuid |
| from pathlib import Path |
| from typing import TYPE_CHECKING, Any, ClassVar |
| from xml.etree import ElementTree as ET |
|
|
| from django.conf import settings |
| from django.core.exceptions import ValidationError |
| from django.db import transaction |
| from django.utils.translation import gettext_lazy as _ |
|
|
| from .fingerprint_service import FingerprintService |
|
|
| if TYPE_CHECKING: |
| from django.core.files.uploadedfile import UploadedFile |
|
|
| from apps.documents.models.external_template import ExternalTemplate |
| from apps.documents.services.placeholders.registry import PlaceholderRegistry |
|
|
| logger: logging.Logger = logging.getLogger(__name__) |
|
|
|
|
| class AnalysisService: |
| """外部模板分析服务：上传校验 + 结构提取 + LLM 字段映射""" |
|
|
| MAX_FILE_SIZE: ClassVar[int] = 20 * 1024 * 1024 |
|
|
| def __init__( |
| self, |
| fingerprint_service: FingerprintService, |
| llm_service: Any, |
| placeholder_registry: PlaceholderRegistry, |
| ) -> None: |
| self._fingerprint_service = fingerprint_service |
| self._llm_service = llm_service |
| self._placeholder_registry = placeholder_registry |
|
|
| |
| |
| |
|
|
| def upload_template( |
| self, |
| file: UploadedFile, |
| name: str, |
| source_name: str, |
| uploaded_by: Any, |
| ) -> ExternalTemplate: |
| """ |
| 上传外部模板： |
| 1. 校验 .docx 格式、文件大小 ≤ 20MB |
| 2. 保存文件（UUID 重命名） |
| 3. 验证 python-docx 可解析 |
| 4. 处理版本管理（同机构 + 类别组合自增版本号） |
| 5. 创建 ExternalTemplate 记录 |
| """ |
| from apps.documents.models.external_template import ExternalTemplate |
|
|
| self._validate_file(file) |
|
|
| law_firm_id: int = uploaded_by.law_firm_id |
| abs_path, rel_path = self._save_file(file, law_firm_id) |
|
|
| try: |
| self._validate_parseable(abs_path) |
| except ValidationError: |
| |
| if abs_path.exists(): |
| abs_path.unlink() |
| logger.info("已删除无法解析的文件: %s", abs_path) |
| raise |
|
|
| file_size: int = file.size if file.size is not None else 0 |
| original_filename: str = file.name if file.name else "" |
|
|
| with transaction.atomic(): |
| version, deactivated = self._handle_versioning( |
| law_firm_id=law_firm_id, |
| source_name=source_name, |
| ) |
|
|
| template: ExternalTemplate = ExternalTemplate.objects.create( |
| name=name, |
| source_name=source_name, |
| file_path=rel_path, |
| original_filename=original_filename, |
| file_size=file_size, |
| version=version, |
| is_active=True, |
| uploaded_by=uploaded_by, |
| law_firm_id=law_firm_id, |
| ) |
|
|
| logger.info( |
| "模板上传成功: id=%d, name=%s, version=%d, deactivated=%d", |
| template.id, |
| template.name, |
| version, |
| deactivated, |
| ) |
| return template |
|
|
| def _validate_file(self, file: UploadedFile) -> None: |
| """校验 .docx 格式和文件大小""" |
| filename: str = file.name if file.name else "" |
| if not filename.lower().endswith(".docx"): |
| logger.info("文件格式校验失败: %s", filename) |
| raise ValidationError(_("仅支持 .docx 格式")) |
|
|
| file_size: int = file.size if file.size is not None else 0 |
| if file_size > self.MAX_FILE_SIZE: |
| logger.info( |
| "文件大小超出限制: %s, size=%d", |
| filename, |
| file_size, |
| ) |
| raise ValidationError(_("文件大小超出限制")) |
|
|
| def _validate_parseable(self, file_path: Path) -> None: |
| """尝试 python-docx 打开文件，验证可解析性""" |
| try: |
| from docx import Document as DocxDocument |
|
|
| DocxDocument(str(file_path)) |
| except Exception as exc: |
| logger.info( |
| "文件无法解析: %s, error=%s", |
| file_path.name, |
| str(exc), |
| ) |
| raise ValidationError(_("文件无法解析，请检查文件是否损坏或加密")) from exc |
|
|
| def _save_file(self, file: UploadedFile, law_firm_id: int) -> tuple[Path, str]: |
| """ |
| UUID 重命名保存文件 |
| |
| Returns: |
| (绝对路径, 相对于 MEDIA_ROOT 的路径) |
| """ |
| media_root = Path(settings.MEDIA_ROOT) |
| rel_dir = Path("documents") / "external_templates" / str(law_firm_id) |
| abs_dir = media_root / rel_dir |
| abs_dir.mkdir(parents=True, exist_ok=True) |
|
|
| new_filename = f"{uuid.uuid4()}.docx" |
| rel_path = rel_dir / new_filename |
| abs_path = media_root / rel_path |
|
|
| with abs_path.open("wb") as dest: |
| for chunk in file.chunks(): |
| dest.write(chunk) |
|
|
| logger.info("文件已保存: %s", abs_path) |
| return abs_path, str(rel_path) |
|
|
| |
| |
| |
|
|
| def _handle_versioning( |
| self, |
| *, |
| law_firm_id: int, |
| source_name: str, |
| ) -> tuple[int, int]: |
| """ |
| 处理版本管理：同一来源自增版本号，旧版本 is_active=False |
| |
| Returns: |
| (新版本号, 被停用的旧版本数量) |
| """ |
| from apps.documents.models.external_template import ExternalTemplate |
|
|
| existing = ExternalTemplate.objects.filter( |
| law_firm_id=law_firm_id, |
| source_name=source_name, |
| ) |
|
|
| max_version: int | None = existing.order_by("-version").values_list("version", flat=True).first() |
|
|
| new_version: int = (max_version or 0) + 1 |
|
|
| deactivated: int = existing.filter(is_active=True).update(is_active=False) |
|
|
| return new_version, deactivated |
|
|
| |
| |
| |
|
|
| def extract_structure(self, template_id: int) -> dict[str, Any]: |
| """ |
| 从 .docx 提取文档结构：段落、表格（含嵌套）、复选框、删除不适用项。 |
| |
| 1. 从数据库加载模板记录，获取 file_path |
| 2. 使用 python-docx 打开文件 |
| 3. 调用 _extract_paragraphs / _extract_tables / _extract_checkboxes |
| 4. 将结构 JSON 保存到模板记录的 structure_json 字段 |
| 5. 返回结构 dict |
| """ |
| from apps.documents.models.external_template import ExternalTemplate |
|
|
| template: ExternalTemplate = ExternalTemplate.objects.get(pk=template_id) |
| abs_path = Path(settings.MEDIA_ROOT) / template.file_path |
|
|
| from docx import Document as DocxDocument |
|
|
| doc: Any = DocxDocument(str(abs_path)) |
|
|
| paragraphs = self._extract_paragraphs(doc) |
| tables = self._extract_tables(doc) |
| checkboxes = self._extract_checkboxes(doc) |
|
|
| structure: dict[str, Any] = { |
| "paragraphs": paragraphs, |
| "tables": tables, |
| "checkboxes": checkboxes, |
| } |
|
|
| template.structure_json = structure |
| template.save(update_fields=["structure_json", "updated_at"]) |
|
|
| logger.info( |
| "结构提取完成: template_id=%d, paragraphs=%d, tables=%d, checkboxes=%d", |
| template_id, |
| len(paragraphs), |
| len(tables), |
| len(checkboxes), |
| ) |
| return structure |
|
|
| def _extract_paragraphs(self, doc: Any) -> list[dict[str, Any]]: |
| """ |
| 提取段落文本和位置索引。 |
| |
| 对每个段落记录：paragraph_index、text、position_locator。 |
| 同时检测"删除不适用项"模式。 |
| """ |
| paragraphs: list[dict[str, Any]] = [] |
|
|
| for idx, para in enumerate(doc.paragraphs): |
| text: str = para.text.strip() |
| if not text: |
| continue |
|
|
| entry: dict[str, Any] = { |
| "paragraph_index": idx, |
| "text": text, |
| "position_locator": { |
| "type": "paragraph", |
| "paragraph_index": idx, |
| }, |
| } |
|
|
| options = self._detect_delete_inapplicable(text) |
| if options is not None: |
| entry["delete_inapplicable"] = options |
| entry["position_locator"]["type"] = "delete_inapplicable" |
|
|
| paragraphs.append(entry) |
|
|
| return paragraphs |
|
|
| def _extract_tables(self, doc: Any) -> list[dict[str, Any]]: |
| """ |
| 提取表格结构（递归嵌套、合并单元格）。 |
| |
| 对每个表格遍历行和单元格，处理合并单元格（通过 _tc 检测）， |
| 递归处理嵌套表格。 |
| """ |
| tables: list[dict[str, Any]] = [] |
|
|
| for table_idx, table in enumerate(doc.tables): |
| table_data = self._extract_single_table(table, table_idx, []) |
| tables.append(table_data) |
|
|
| return tables |
|
|
| def _extract_single_table( |
| self, |
| table: Any, |
| table_index: int, |
| nested_path: list[int], |
| ) -> dict[str, Any]: |
| """递归提取单个表格的结构，通过 XML 正确处理合并单元格。""" |
| ns = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"} |
| rows_data: list[dict[str, Any]] = [] |
| tbl = table._tbl |
| tr_elements = tbl.findall("w:tr", ns) |
|
|
| for row_idx, tr in enumerate(tr_elements): |
| cells_data: list[dict[str, Any]] = [] |
| col_cursor = 0 |
|
|
| for tc in tr.findall("w:tc", ns): |
| tc_pr = tc.find("w:tcPr", ns) |
| grid_span = 1 |
| is_v_merge_continue = False |
|
|
| if tc_pr is not None: |
| gs = tc_pr.find("w:gridSpan", ns) |
| if gs is not None: |
| grid_span = int(gs.get(f"{{{ns['w']}}}val", "1")) |
| vm = tc_pr.find("w:vMerge", ns) |
| if vm is not None: |
| val = vm.get(f"{{{ns['w']}}}val", "continue") |
| is_v_merge_continue = val != "restart" |
|
|
| if is_v_merge_continue: |
| col_cursor += grid_span |
| continue |
|
|
| texts = [t.text or "" for t in tc.findall(".//w:t", ns)] |
| text = "".join(texts).strip() |
|
|
| cell_entry: dict[str, Any] = { |
| "row": row_idx, |
| "col": col_cursor, |
| "text": text, |
| "position_locator": { |
| "type": "table_cell", |
| "table_index": table_index, |
| "row": row_idx, |
| "col": col_cursor, |
| "nested_table_path": nested_path, |
| }, |
| } |
| if grid_span > 1: |
| cell_entry["col_span"] = grid_span |
|
|
| options = self._detect_delete_inapplicable(text) |
| if options is not None: |
| cell_entry["delete_inapplicable"] = options |
| cell_entry["position_locator"]["type"] = "delete_inapplicable" |
|
|
| |
| |
| if col_cursor < len(table.rows[row_idx].cells): |
| py_cell = table.rows[row_idx].cells[col_cursor] |
| nested_tables: list[dict[str, Any]] = [] |
| for nested_idx, nested_table in enumerate(py_cell.tables): |
| nested_data = self._extract_single_table( |
| nested_table, |
| nested_idx, |
| [*nested_path, table_index], |
| ) |
| nested_tables.append(nested_data) |
| if nested_tables: |
| cell_entry["nested_tables"] = nested_tables |
|
|
| cells_data.append(cell_entry) |
| col_cursor += grid_span |
|
|
| rows_data.append({"row_index": row_idx, "cells": cells_data}) |
|
|
| return { |
| "table_index": table_index, |
| "nested_table_path": nested_path, |
| "rows": rows_data, |
| } |
|
|
| def _extract_checkboxes(self, doc: Any) -> list[dict[str, Any]]: |
| """ |
| 识别复选框控件。 |
| |
| 解析文档 XML，查找 w14:checkbox 或 w:sdt 中包含复选框的结构化文档标签。 |
| """ |
| checkboxes: list[dict[str, Any]] = [] |
| ns: dict[str, str] = { |
| "w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main", |
| "w14": "http://schemas.microsoft.com/office/word/2010/wordml", |
| } |
|
|
| body_xml: str = doc.element.xml |
| try: |
| root: ET.Element = ET.fromstring(body_xml) |
| except ET.ParseError: |
| logger.info("复选框提取: XML 解析失败，跳过") |
| return checkboxes |
|
|
| |
| sdt_elements = root.iter(f"{{{ns['w']}}}sdt") |
| checkbox_index = 0 |
|
|
| for sdt in sdt_elements: |
| sdt_pr = sdt.find(f"{{{ns['w']}}}sdtPr", ns) |
| if sdt_pr is None: |
| continue |
|
|
| |
| is_checkbox = ( |
| sdt_pr.find(f"{{{ns['w14']}}}checkbox", ns) is not None |
| or sdt_pr.find(f"{{{ns['w']}}}checkbox", ns) is not None |
| ) |
| if not is_checkbox: |
| continue |
|
|
| |
| sdt_content = sdt.find(f"{{{ns['w']}}}sdtContent", ns) |
| label_text = "" |
| if sdt_content is not None: |
| runs = sdt_content.iter(f"{{{ns['w']}}}t") |
| label_parts: list[str] = [r.text for r in runs if r.text] |
| label_text = "".join(label_parts) |
|
|
| |
| checked = False |
| checkbox_el = sdt_pr.find(f"{{{ns['w14']}}}checkbox", ns) |
| if checkbox_el is None: |
| checkbox_el = sdt_pr.find(f"{{{ns['w']}}}checkbox", ns) |
| if checkbox_el is not None: |
| checked_el = checkbox_el.find(f"{{{ns['w14']}}}checked", ns) |
| if checked_el is None: |
| checked_el = checkbox_el.find(f"{{{ns['w']}}}checked", ns) |
| if checked_el is not None: |
| val = checked_el.get( |
| f"{{{ns['w14']}}}val", |
| checked_el.get(f"{{{ns['w']}}}val", "0"), |
| ) |
| checked = val in ("1", "true") |
|
|
| checkboxes.append( |
| { |
| "checkbox_index": checkbox_index, |
| "label": label_text, |
| "checked": checked, |
| "position_locator": { |
| "type": "checkbox", |
| "checkbox_index": checkbox_index, |
| }, |
| } |
| ) |
| checkbox_index += 1 |
|
|
| logger.info("复选框提取完成: 共 %d 个", len(checkboxes)) |
| return checkboxes |
|
|
| def _detect_delete_inapplicable(self, text: str) -> list[str] | None: |
| """ |
| 检测"删除不适用项"格式（如"自然人/法人/非法人组织"）。 |
| |
| 使用正则匹配以 / 或 ／ 分隔的选项模式。 |
| 返回选项列表，若未检测到则返回 None。 |
| """ |
| |
| pattern = r"([\u4e00-\u9fff\w]+(?:[/／][\u4e00-\u9fff\w]+){1,})" |
| match = re.search(pattern, text) |
| if match is None: |
| return None |
|
|
| matched_text: str = match.group(1) |
| options: list[str] = re.split(r"[/／]", matched_text) |
|
|
| |
| has_chinese = all(re.search(r"[\u4e00-\u9fff]", opt) for opt in options) |
| if not has_chinese: |
| return None |
|
|
| return options |
|
|
| |
| |
| |
|
|
| def analyze_template(self, template_id: int) -> list[Any]: |
| """ |
| 分析模板并生成字段映射： |
| 1. 提取结构 → 计算指纹 |
| 2. 指纹匹配 → 命中则复用映射（标注 mapping_source） |
| 3. 未命中 → 构建 LLM 提示词 → 调用 LLM → 解析结果 |
| 4. 创建 FieldMapping 记录 |
| 5. 更新模板状态为 ready |
| """ |
| from apps.documents.models.choices import TemplateStatus |
| from apps.documents.models.external_template import ExternalTemplate, ExternalTemplateFieldMapping |
|
|
| template: ExternalTemplate = ExternalTemplate.objects.get(pk=template_id) |
|
|
| |
| template.status = TemplateStatus.ANALYZING |
| template.save(update_fields=["status", "status_changed_at", "updated_at"]) |
|
|
| try: |
| |
| structure_json: dict[str, Any] = self.extract_structure(template_id) |
|
|
| |
| abs_path = Path(settings.MEDIA_ROOT) / template.file_path |
| fingerprint: str = self._fingerprint_service.compute_fingerprint(abs_path) |
| template.structure_fingerprint = fingerprint |
| template.save(update_fields=["structure_fingerprint", "updated_at"]) |
|
|
| |
| matched: ExternalTemplate | None = self._fingerprint_service.find_matching_template( |
| fingerprint, template.law_firm_id |
| ) |
|
|
| |
| if matched is not None and matched.pk == template.pk: |
| matched = None |
|
|
| created_mappings: list[ExternalTemplateFieldMapping] |
|
|
| if matched is not None: |
| |
| created_mappings = self._copy_mappings_from( |
| source_template=matched, |
| target_template=template, |
| ) |
| template.mapping_source = matched |
| logger.info( |
| "复用映射: template_id=%d, source_id=%d, count=%d", |
| template_id, |
| matched.pk, |
| len(created_mappings), |
| ) |
| else: |
| |
| prompt: str = self._build_llm_prompt(structure_json) |
| response = self._llm_service.complete( |
| prompt=prompt, |
| system_prompt=("你是一个法律文书模板分析专家。请分析模板结构并返回字段映射的 JSON 数组。"), |
| temperature=0.1, |
| max_tokens=4096, |
| ) |
| raw_mappings: list[dict[str, Any]] = self._parse_llm_response(response.content) |
| created_mappings = self._create_field_mappings(template, raw_mappings) |
| logger.info( |
| "LLM 分析完成: template_id=%d, mappings=%d", |
| template_id, |
| len(created_mappings), |
| ) |
|
|
| |
| template.status = TemplateStatus.READY |
| template.save( |
| update_fields=[ |
| "status", |
| "mapping_source", |
| "status_changed_at", |
| "updated_at", |
| ] |
| ) |
|
|
| return created_mappings |
|
|
| except Exception: |
| |
| logger.exception("模板分析失败: template_id=%d", template_id) |
| template.refresh_from_db() |
| template.status = TemplateStatus.ANALYSIS_FAILED |
| template.save(update_fields=["status", "status_changed_at", "updated_at"]) |
| raise |
|
|
| def _copy_mappings_from( |
| self, |
| source_template: Any, |
| target_template: Any, |
| ) -> list[Any]: |
| """从源模板复制映射到目标模板""" |
| from apps.documents.models.external_template import ExternalTemplateFieldMapping |
|
|
| source_mappings = ExternalTemplateFieldMapping.objects.filter(template=source_template) |
| created: list[ExternalTemplateFieldMapping] = [] |
| for m in source_mappings: |
| new_mapping = ExternalTemplateFieldMapping.objects.create( |
| template=target_template, |
| position_locator=m.position_locator, |
| position_description=m.position_description, |
| semantic_label=m.semantic_label, |
| fill_type=m.fill_type, |
| sort_order=m.sort_order, |
| ) |
| created.append(new_mapping) |
| return created |
|
|
| def _build_llm_prompt(self, structure_json: dict[str, Any]) -> str: |
| """ |
| 构建 LLM 提示词：结构 JSON + fill_type 说明 |
| """ |
| structure_text: str = json.dumps(structure_json, ensure_ascii=False, indent=2) |
|
|
| prompt: str = ( |
| f"请分析以下法律文书模板的结构，识别所有可填充位置。\n\n" |
| f"## 文档结构 JSON\n```json\n{structure_text}\n```\n\n" |
| f"## 填充类型说明\n" |
| f"- text: 文本替换（替换段落或单元格中的文本）\n" |
| f"- checkbox: 勾选复选框\n" |
| f"- delete_inapplicable: 删除不适用项（保留匹配项，删除其余选项）\n\n" |
| f"## 输出要求\n" |
| f"请返回一个 JSON 数组，每个元素包含以下字段：\n" |
| f"- position_locator: 位置定位器（直接使用结构 JSON 中的 position_locator）\n" |
| f"- semantic_label: 语义标签（中文描述该位置应填写的内容，如'被申请人姓名'、'住所地'）\n" |
| f"- fill_type: 填充类型（text / checkbox / delete_inapplicable）\n\n" |
| f"仅返回 JSON 数组，不要包含其他文字说明。" |
| ) |
| return prompt |
|
|
| def _parse_llm_response(self, response: str) -> list[dict[str, Any]]: |
| """ |
| 解析 LLM 返回的 JSON 映射结果。 |
| 处理 markdown 代码块包裹的情况。 |
| """ |
| text: str = response.strip() |
|
|
| |
| if text.startswith("```"): |
| |
| lines: list[str] = text.split("\n") |
| |
| start_idx = 1 |
| end_idx = len(lines) |
| for i in range(len(lines) - 1, 0, -1): |
| if lines[i].strip().startswith("```"): |
| end_idx = i |
| break |
| text = "\n".join(lines[start_idx:end_idx]) |
|
|
| try: |
| parsed: Any = json.loads(text) |
| except json.JSONDecodeError: |
| logger.error("LLM 返回的 JSON 解析失败: %s", text[:500]) |
| raise |
|
|
| if not isinstance(parsed, list): |
| logger.error("LLM 返回的不是 JSON 数组: type=%s", type(parsed).__name__) |
| raise ValueError(_("LLM 返回格式异常，期望 JSON 数组")) |
|
|
| result: list[dict[str, Any]] = [] |
| for item in parsed: |
| if not isinstance(item, dict): |
| continue |
| result.append( |
| { |
| "position_locator": item.get("position_locator", {}), |
| "semantic_label": str(item.get("semantic_label", "")), |
| "fill_type": str(item.get("fill_type", "text")), |
| } |
| ) |
|
|
| return result |
|
|
| def _create_field_mappings( |
| self, |
| template: Any, |
| mappings: list[dict[str, Any]], |
| ) -> list[Any]: |
| """根据解析后的映射数据创建 FieldMapping 记录""" |
| from apps.documents.models.choices import FillType |
| from apps.documents.models.external_template import ExternalTemplateFieldMapping |
|
|
| valid_fill_types: set[str] = {ft.value for ft in FillType} |
| created: list[ExternalTemplateFieldMapping] = [] |
|
|
| for idx, m in enumerate(mappings): |
| fill_type: str = m.get("fill_type", FillType.TEXT) |
| if fill_type not in valid_fill_types: |
| fill_type = FillType.TEXT |
|
|
| position_locator: dict[str, Any] = m.get("position_locator", {}) |
| pos_type: str = str(position_locator.get("type", "")) |
| position_description: str = "" |
| if pos_type == "paragraph": |
| p_idx = position_locator.get("paragraph_index", "") |
| position_description = f"段落 {p_idx}" |
| elif pos_type == "table_cell": |
| t_idx = position_locator.get("table_index", "") |
| row = position_locator.get("row", "") |
| col = position_locator.get("col", "") |
| position_description = f"表格{t_idx} 行{row} 列{col}" |
| elif pos_type == "checkbox": |
| cb_idx = position_locator.get("checkbox_index", "") |
| position_description = f"复选框 {cb_idx}" |
| elif pos_type == "delete_inapplicable": |
| position_description = "删除不适用项" |
|
|
| new_mapping = ExternalTemplateFieldMapping.objects.create( |
| template=template, |
| position_locator=position_locator, |
| position_description=position_description, |
| semantic_label=m.get("semantic_label", ""), |
| fill_type=fill_type, |
| sort_order=idx, |
| ) |
| created.append(new_mapping) |
|
|
| return created |
|
|
| |
| |
| |
|
|
| def retry_analysis(self, template_id: int) -> list[Any]: |
| """重新触发 LLM 分析：删除旧映射后重新分析。""" |
| from apps.documents.models.external_template import ExternalTemplateFieldMapping |
|
|
| deleted_count, _ = ExternalTemplateFieldMapping.objects.filter(template_id=template_id).delete() |
| logger.info( |
| "重新分析: template_id=%d, 已删除旧映射 %d 条", |
| template_id, |
| deleted_count, |
| ) |
| return self.analyze_template(template_id) |
|
|
| def create_manual_mapping( |
| self, |
| *, |
| template_id: int, |
| position_locator: dict[str, Any], |
| position_description: str, |
| semantic_label: str, |
| fill_type: str = "text", |
| ) -> Any: |
| """手动新增单条映射。""" |
| from apps.documents.models.external_template import ExternalTemplate, ExternalTemplateFieldMapping |
|
|
| template = ExternalTemplate.objects.get(pk=template_id) |
| max_order: int = ( |
| ExternalTemplateFieldMapping.objects.filter(template=template) |
| .order_by("-sort_order") |
| .values_list("sort_order", flat=True) |
| .first() |
| or 0 |
| ) |
| return ExternalTemplateFieldMapping.objects.create( |
| template=template, |
| position_locator=position_locator, |
| position_description=position_description, |
| semantic_label=semantic_label, |
| fill_type=fill_type, |
| sort_order=max_order + 1, |
| ) |
|
|
| def delete_mapping(self, mapping_id: int) -> None: |
| """删除单条映射。""" |
| from apps.documents.models.external_template import ExternalTemplateFieldMapping |
|
|
| ExternalTemplateFieldMapping.objects.filter(pk=mapping_id).delete() |
|
|
