File size: 9,671 Bytes
b4f2238 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 | """OA立案 / OA导入 任务入口。"""
from __future__ import annotations
import logging
from pathlib import Path
from django.utils import timezone
from django_q.exceptions import TimeoutException
logger = logging.getLogger("apps.oa_filing.tasks")
def run_client_import_task(session_id: int, headless: bool = True, limit: int | None = None) -> None:
"""Django-Q 任务入口：执行 OA 客户导入。
通过字符串路径 ``apps.oa_filing.tasks.run_client_import_task`` 调用。
"""
from apps.oa_filing.models import ClientImportPhase, ClientImportSession, ClientImportStatus
from apps.oa_filing.services.client_import_service import ClientImportService
try:
session = ClientImportSession.objects.select_related("credential", "lawyer").get(pk=session_id)
except ClientImportSession.DoesNotExist:
logger.error("客户导入会话不存在: session_id=%s", session_id)
return
if session.status in {ClientImportStatus.COMPLETED, ClientImportStatus.CANCELLED}:
logger.info("会话已结束，跳过执行: session_id=%s status=%s", session_id, session.status)
return
# 若还未标记开始，先记录开始时间，避免前端长时间显示 pending。
if session.started_at is None:
session.started_at = timezone.now()
session.status = ClientImportStatus.IN_PROGRESS
session.phase = ClientImportPhase.DISCOVERING
session.progress_message = "正在启动导入任务"
session.error_message = ""
session.save(update_fields=["started_at", "status", "phase", "progress_message", "error_message", "updated_at"])
try:
ClientImportService(session).run_import(headless=headless, limit=limit)
except TimeoutException as exc:
logger.exception("客户导入任务超时: session_id=%s error=%s", session_id, exc)
session.status = ClientImportStatus.FAILED
session.phase = ClientImportPhase.FAILED
session.error_message = str(exc)
session.progress_message = "导入超时"
session.completed_at = timezone.now()
session.save(
update_fields=["status", "phase", "error_message", "progress_message", "completed_at", "updated_at"]
)
raise
except Exception as exc:
logger.exception("客户导入任务执行失败: session_id=%s error=%s", session_id, exc)
session.status = ClientImportStatus.FAILED
session.phase = ClientImportPhase.FAILED
session.error_message = str(exc)
session.progress_message = "导入失败"
session.completed_at = timezone.now()
session.save(
update_fields=["status", "phase", "error_message", "progress_message", "completed_at", "updated_at"]
)
def run_case_import_preview_task(session_id: int, file_path: str) -> None:
"""Django-Q 任务入口：预览 OA 案件导入。
解析Excel文件，预览匹配结果。
"""
from apps.oa_filing.models import CaseImportPhase, CaseImportSession, CaseImportStatus
from apps.oa_filing.services.case_import_service import CaseImportService
try:
session = CaseImportSession.objects.select_related("credential", "lawyer").get(pk=session_id)
except CaseImportSession.DoesNotExist:
logger.error("案件导入会话不存在: session_id=%s", session_id)
return
if session.status in {CaseImportStatus.COMPLETED, CaseImportStatus.CANCELLED}:
logger.info("会话已结束，跳过执行: session_id=%s status=%s", session_id, session.status)
return
try:
# 更新状态为解析中
session.started_at = timezone.now()
session.status = CaseImportStatus.IN_PROGRESS
session.phase = CaseImportPhase.PARSING
session.progress_message = "正在解析Excel文件"
session.save()
# 解析Excel
service = CaseImportService(session)
case_nos = service.parse_excel(file_path)
if not case_nos:
session.status = CaseImportStatus.COMPLETED
session.phase = CaseImportPhase.COMPLETED
session.progress_message = "未从Excel中解析出案件编号"
session.total_count = 0
session.completed_at = timezone.now()
session.save()
return
# 更新状态为预览
session.phase = CaseImportPhase.PREVIEW
session.total_count = len(case_nos)
session.progress_message = f"解析完成，共 {len(case_nos)} 个案件，正在匹配"
session.save()
# 预览匹配
preview_results = service.preview_cases(case_nos)
# 统计
matched_count = sum(1 for r in preview_results if r.status == "matched")
unmatched_count = sum(1 for r in preview_results if r.status == "unmatched")
error_count = sum(1 for r in preview_results if r.status == "error")
# 保存预览结果
session.matched_count = matched_count
session.unmatched_count = unmatched_count
session.error_count = error_count
# 预览完成后保持pending状态，等待用户确认执行
session.status = CaseImportStatus.PENDING
session.phase = CaseImportPhase.PREVIEW
session.progress_message = "预览完成，可开始导入"
session.result_data = {
"preview": [
{
"case_no": r.case_no,
"status": r.status,
"existing_contract_id": r.existing_contract_id,
"customer_names": r.customer_names or [],
"error_message": r.error_message,
}
for r in preview_results
]
}
session.save()
# 清理临时文件
Path(file_path).unlink(missing_ok=True)
logger.info(
"案件预览完成: session_id=%d total=%d matched=%d unmatched=%d error=%d",
session_id,
len(case_nos),
matched_count,
unmatched_count,
error_count,
)
except TimeoutException as exc:
logger.exception("案件预览任务超时: session_id=%s error=%s", session_id, exc)
session.status = CaseImportStatus.FAILED
session.phase = CaseImportPhase.FAILED
session.error_message = str(exc)
session.progress_message = "预览超时"
session.completed_at = timezone.now()
session.save()
raise
except Exception as exc:
logger.exception("案件预览任务执行失败: session_id=%s error=%s", session_id, exc)
session.status = CaseImportStatus.FAILED
session.phase = CaseImportPhase.FAILED
session.error_message = str(exc)
session.progress_message = "预览失败"
session.completed_at = timezone.now()
session.save()
def run_case_import_task(
session_id: int,
case_nos: list[str],
matched_case_nos: list[str] | None = None,
headless: bool = True,
) -> None:
"""Django-Q 任务入口：执行 OA 案件导入。
对预览阶段标记为 unmatched 的案件，从OA提取数据并创建/更新合同。
"""
from apps.oa_filing.models import CaseImportPhase, CaseImportSession, CaseImportStatus
from apps.oa_filing.services.case_import_service import CaseImportService
try:
session = CaseImportSession.objects.select_related("credential", "lawyer").get(pk=session_id)
except CaseImportSession.DoesNotExist:
logger.error("案件导入会话不存在: session_id=%s", session_id)
return
if session.status in {CaseImportStatus.COMPLETED, CaseImportStatus.CANCELLED}:
logger.info("会话已结束，跳过执行: session_id=%s status=%s", session_id, session.status)
return
# 若还未标记开始，先记录开始时间
if session.started_at is None:
session.started_at = timezone.now()
session.status = CaseImportStatus.IN_PROGRESS
session.phase = CaseImportPhase.DISCOVERING
session.progress_message = "正在启动导入任务"
session.error_message = ""
session.save(update_fields=["started_at", "status", "phase", "progress_message", "error_message", "updated_at"])
try:
service = CaseImportService(session)
results = service.run_import(
case_nos=case_nos,
matched_case_nos=matched_case_nos,
headless=headless,
)
# 结果已在 service.run_import 中保存到 session
logger.info("案件导入任务完成: session_id=%d", session_id)
except TimeoutException as exc:
logger.exception("案件导入任务超时: session_id=%s error=%s", session_id, exc)
session.status = CaseImportStatus.FAILED
session.phase = CaseImportPhase.FAILED
session.error_message = str(exc)
session.progress_message = "导入超时"
session.completed_at = timezone.now()
session.save(
update_fields=["status", "phase", "error_message", "progress_message", "completed_at", "updated_at"]
)
raise
except Exception as exc:
logger.exception("案件导入任务执行失败: session_id=%s error=%s", session_id, exc)
session.status = CaseImportStatus.FAILED
session.phase = CaseImportPhase.FAILED
session.error_message = str(exc)
session.progress_message = "导入失败"
session.completed_at = timezone.now()
session.save(
update_fields=["status", "phase", "error_message", "progress_message", "completed_at", "updated_at"]
)
|