"""合同批量绑定一级文件夹服务。""" from __future__ import annotations import logging import re from difflib import SequenceMatcher from pathlib import Path from typing import Any from django.db import transaction from django.db.models import Count from django.utils.translation import gettext_lazy as _ from apps.contracts.models import Contract, ContractTypeFolderRootPreset from apps.contracts.services.folder.folder_binding_service import FolderBindingService from apps.core.exceptions import ValidationException from apps.core.infrastructure.subprocess_runner import SubprocessRunner from apps.core.models.enums import CaseType logger = logging.getLogger(__name__) class ContractBatchFolderBindingService: """批量绑定合同文件夹（目标固定为根目录的一级子文件夹）。""" AUTO_SCORE_THRESHOLD = 0.72 AUTO_SCORE_GAP = 0.08 NON_CONTRACT_DIR_KEYWORDS = ("未成交", "归档", "模板", "示例") def __init__(self, *, folder_binding_service: FolderBindingService | None = None) -> None: self._folder_binding_service = folder_binding_service or FolderBindingService() self._case_type_label_map = dict(CaseType.choices) @property def folder_binding_service(self) -> FolderBindingService: return self._folder_binding_service def list_unbound_case_type_cards(self) -> list[dict[str, Any]]: unbound = ( Contract.objects.filter(folder_binding__isnull=True) .exclude(case_type__isnull=True) .exclude(case_type="") .values("case_type") .annotate(unbound_count=Count("id")) .order_by("case_type") ) preset_map = { item.case_type: item.root_path for item in ContractTypeFolderRootPreset.objects.filter( case_type__in=[str(row["case_type"]) for row in unbound] ) } cards: list[dict[str, Any]] = [] for row in unbound: case_type = str(row["case_type"] or "").strip() if not case_type: continue cards.append( { "case_type": case_type, "case_type_display": self._case_type_label_map.get(case_type, case_type), "unbound_count": int(row["unbound_count"] or 0), "root_path": str(preset_map.get(case_type, "") or ""), } ) return cards def preview(self, *, case_type_roots: list[dict[str, Any]]) -> dict[str, Any]: items: list[dict[str, Any]] = [] total_contracts = 0 auto_selected = 0 for item in case_type_roots: case_type = str(item.get("case_type") or "").strip() root_path = str(item.get("root_path") or "").strip() if not case_type: continue preview_item = self._preview_single_case_type(case_type=case_type, root_path=root_path) items.append(preview_item) total_contracts += int(preview_item.get("contract_count", 0) or 0) auto_selected += int(preview_item.get("auto_selected_count", 0) or 0) return { "items": items, "summary": { "total_contracts": total_contracts, "auto_selected": auto_selected, }, } def save( self, *, case_type_roots: list[dict[str, Any]], contract_selections: list[dict[str, Any]], ) -> dict[str, Any]: root_map: dict[str, str] = {} for item in case_type_roots: case_type = str(item.get("case_type") or "").strip() root_path = str(item.get("root_path") or "").strip() if not case_type: continue if root_path: self._ensure_accessible_directory(root_path) root_map[case_type] = root_path ContractTypeFolderRootPreset.objects.update_or_create( case_type=case_type, defaults={"root_path": root_path}, ) bound_count = 0 skipped_count = 0 errors: list[str] = [] target_ids: list[int] = [] for item in contract_selections: raw_contract_id = item.get("contract_id") if raw_contract_id is None: continue target_ids.append(int(raw_contract_id)) unbound_contract_map = { contract.id: contract for contract in Contract.objects.filter(id__in=target_ids, folder_binding__isnull=True).only( "id", "name", "case_type" ) } with transaction.atomic(): for item in contract_selections: apply_flag = bool(item.get("apply")) if not apply_flag: skipped_count += 1 continue contract_id = int(item.get("contract_id") or 0) selected_folder_path = str(item.get("selected_folder_path") or "").strip() case_type = str(item.get("case_type") or "").strip() if not contract_id or not selected_folder_path: skipped_count += 1 continue contract = unbound_contract_map.get(contract_id) if contract is None: skipped_count += 1 errors.append(str(_("合同 %(id)s 不存在或已绑定") % {"id": contract_id})) continue if contract.case_type != case_type: skipped_count += 1 errors.append(str(_("合同 %(id)s 的类型已变化，请刷新后重试") % {"id": contract_id})) continue root_path = str(root_map.get(case_type, "") or "").strip() if not root_path: skipped_count += 1 errors.append(str(_("合同类型 %(type)s 缺少根目录") % {"type": case_type})) continue try: self._validate_selected_folder(root_path=root_path, selected_folder_path=selected_folder_path) self.folder_binding_service.create_binding(owner_id=contract_id, folder_path=selected_folder_path) bound_count += 1 except Exception as exc: skipped_count += 1 errors.append(str(_("合同 %(id)s 绑定失败: %(error)s") % {"id": contract_id, "error": exc})) return { "bound_count": bound_count, "skipped_count": skipped_count, "error_count": len(errors), "errors": errors, } def open_folder(self, *, root_path: str, folder_path: str) -> None: target = self._validate_selected_folder(root_path=root_path, selected_folder_path=folder_path) SubprocessRunner(allowed_programs={"open"}).run( args=["open", target.as_posix()], timeout_seconds=5, check=False, ) def _preview_single_case_type(self, *, case_type: str, root_path: str) -> dict[str, Any]: item: dict[str, Any] = { "case_type": case_type, "case_type_display": self._case_type_label_map.get(case_type, case_type), "root_path": root_path, "contract_count": 0, "auto_selected_count": 0, "options": [], "rows": [], "error": "", } contracts = list( Contract.objects.filter(case_type=case_type, folder_binding__isnull=True) .prefetch_related("cases") .order_by("id") ) item["contract_count"] = len(contracts) if not root_path: item["error"] = str(_("请先填写根目录")) return item try: root = self._ensure_accessible_directory(root_path) candidates = self._list_first_level_dirs(root) except Exception as exc: item["error"] = str(exc) return item item["options"] = [ {"name": folder.name, "path": folder.as_posix(), "is_penalized": self._is_non_contract_dir(folder.name)} for folder in candidates ] rows: list[dict[str, Any]] = [] auto_selected_count = 0 for contract in contracts: case_names = [str(case.name or "").strip() for case in contract.cases.all() if str(case.name or "").strip()] case_filing_numbers = [ str(case.filing_number or "").strip() for case in contract.cases.all() if str(case.filing_number or "").strip() ] recommend = self._recommend_folder( contract_name=str(contract.name or ""), contract_filing_number=str(contract.filing_number or ""), oa_case_number=str(contract.law_firm_oa_case_number or ""), case_names=case_names, case_filing_numbers=case_filing_numbers, candidates=candidates, ) selected_folder_path = str(recommend.get("recommended_folder_path") or "") auto_selected_flag = bool(recommend.get("auto_selected")) and bool(selected_folder_path) if auto_selected_flag: auto_selected_count += 1 rows.append( { "contract_id": contract.id, "case_type": case_type, "contract_name": contract.name, "case_names": case_names, "filing_number": str(contract.filing_number or ""), "oa_case_number": str(contract.law_firm_oa_case_number or ""), "recommended_folder_path": selected_folder_path, "recommended_folder_name": str(recommend.get("recommended_folder_name") or ""), "confidence": float(recommend.get("confidence") or 0.0), "reason": str(recommend.get("reason") or ""), "auto_selected": auto_selected_flag, "selected_folder_path": selected_folder_path, "apply": auto_selected_flag, "candidates": recommend.get("candidates", []), } ) item["rows"] = rows item["auto_selected_count"] = auto_selected_count return item def _recommend_folder( self, *, contract_name: str, contract_filing_number: str, oa_case_number: str, case_names: list[str], case_filing_numbers: list[str], candidates: list[Path], ) -> dict[str, Any]: targets: list[tuple[str, str, float]] = [] if contract_name: targets.append((contract_name, str(_("合同名称")), 1.0)) for name in case_names: targets.append((name, str(_("案件名称")), 1.0)) if contract_filing_number: targets.append((contract_filing_number, str(_("合同建档编号")), 0.95)) if oa_case_number: targets.append((oa_case_number, str(_("律所OA案件编号")), 1.0)) for number in case_filing_numbers: targets.append((number, str(_("案件建档编号")), 0.95)) scored_candidates: list[dict[str, Any]] = [] for candidate in candidates: score_data = self._score_candidate(candidate_name=candidate.name, targets=targets) score = float(score_data.get("score") or 0.0) if self._is_non_contract_dir(candidate.name): score = max(0.0, score - 0.25) scored_candidates.append( { "path": candidate.as_posix(), "name": candidate.name, "score": round(score, 4), "reason": str(score_data.get("reason") or ""), } ) scored_candidates.sort(key=lambda row: (-float(row["score"]), str(row["name"]))) if not scored_candidates: return { "recommended_folder_path": "", "recommended_folder_name": "", "confidence": 0.0, "reason": str(_("未找到一级子文件夹")), "auto_selected": False, "candidates": [], } top = scored_candidates[0] second_score = float(scored_candidates[1]["score"]) if len(scored_candidates) > 1 else 0.0 top_score = float(top["score"]) auto_selected = top_score >= self.AUTO_SCORE_THRESHOLD and (top_score - second_score) >= self.AUTO_SCORE_GAP return { "recommended_folder_path": str(top["path"]) if auto_selected else "", "recommended_folder_name": str(top["name"]) if auto_selected else "", "confidence": top_score, "reason": str(top["reason"]), "auto_selected": auto_selected, "candidates": scored_candidates, } def _score_candidate(self, *, candidate_name: str, targets: list[tuple[str, str, float]]) -> dict[str, Any]: candidate_norm = self._normalize_text(candidate_name) candidate_alias = self._normalize_alias(candidate_name) best_score = 0.0 best_reason = "" for target_value, target_label, target_weight in targets: target_norm = self._normalize_text(target_value) target_alias = self._normalize_alias(target_value) if not target_norm and not target_alias: continue ratio_norm = self._sequence_ratio(candidate_norm, target_norm) ratio_alias = self._sequence_ratio(candidate_alias, target_alias) base_score = max(ratio_norm, ratio_alias) contains_bonus = 0.0 if candidate_norm and target_norm and (candidate_norm in target_norm or target_norm in candidate_norm): contains_bonus = 0.12 number_bonus = 0.0 if self._looks_like_number(target_value) and target_value in candidate_name: number_bonus = 0.3 score = min(1.0, (base_score + contains_bonus + number_bonus) * float(target_weight)) if score > best_score: best_score = score best_reason = str(_("命中%(label)s") % {"label": target_label}) return {"score": best_score, "reason": best_reason} def _validate_selected_folder(self, *, root_path: str, selected_folder_path: str) -> Path: root = self._ensure_accessible_directory(root_path) target = self._ensure_accessible_directory(selected_folder_path) try: target.relative_to(root) except ValueError as exc: raise ValidationException(message=_("选择目录不在根目录范围内")) from exc if target.parent != root: raise ValidationException(message=_("只能绑定根目录下的一级子文件夹")) return target def _ensure_accessible_directory(self, folder_path: str) -> Path: path = Path(folder_path).expanduser().resolve() if not path.exists() or not path.is_dir(): raise ValidationException(message=_("目录不可访问: %(path)s") % {"path": folder_path}) return path def _list_first_level_dirs(self, root: Path) -> list[Path]: dirs = [item.resolve() for item in root.iterdir() if item.is_dir() and not item.name.startswith(".")] dirs.sort(key=lambda folder: folder.name.lower()) return dirs def _normalize_text(self, value: str) -> str: text = str(value or "").strip() if not text: return "" text = self._strip_leading_date(text) text = self._strip_leading_labels(text) text = re.sub(r"[（）()【】\[\]{}]", " ", text) text = re.sub(r"[，。、“”‘’；：,.!?！？\-_/\\|]", " ", text) text = re.sub(r"\s+", " ", text).strip().lower() return text def _normalize_alias(self, value: str) -> str: text = self._normalize_text(value) for token in ( "股份有限公司", "有限责任公司", "有限公司", "有限合伙", "公司", "一案", "二案", "三案", "四案", "系列案", "案件", "案", ): text = text.replace(token.lower(), "") text = re.sub(r"\s+", " ", text).strip() return text def _strip_leading_date(self, text: str) -> str: return re.sub(r"^\d{4}[.\-_]\d{1,2}[.\-_]\d{1,2}\s*[-_－]?\s*", "", text) def _strip_leading_labels(self, text: str) -> str: value = text while True: updated = re.sub(r"^\[[^\]]+\]\s*", "", value) if updated == value: break value = updated return value def _looks_like_number(self, value: str) -> bool: raw = str(value or "").strip() if not raw: return False return any(char.isdigit() for char in raw) def _sequence_ratio(self, left: str, right: str) -> float: if not left or not right: return 0.0 return float(SequenceMatcher(None, left, right).ratio()) def _is_non_contract_dir(self, folder_name: str) -> bool: return any(keyword in str(folder_name or "") for keyword in self.NON_CONTRACT_DIR_KEYWORDS)