Spaces:

jun2024
/

fc

Paused

fc / apps /oa_filing /api /case_import_api.py

jun2024

Upload 2241 files

b4f2238 verified 7 months ago

8.74 kB

"""案件导入 API。"""

from __future__ import annotations

import logging
import os
from typing import Any

from django.http import HttpRequest, HttpResponse, JsonResponse
from django.utils import timezone
from ninja import Router, UploadedFile

from apps.oa_filing.models import CaseImportSession
from apps.oa_filing.schemas.case_import_schemas import (
    CaseImportResponse,
    CaseImportResult,
    CaseImportSessionOut,
    CasePreviewItem,
    CasePreviewResponse,
)

logger = logging.getLogger("apps.oa_filing.api.case_import")
router = Router()

# Django-Q 默认 timeout=600 秒；OA 案件导入通常需要更长时间。
# 保持小于默认 retry(1200) 以降低重复执行风险。
CASE_IMPORT_TASK_TIMEOUT_SECONDS = int(os.environ.get("OA_CASE_IMPORT_TASK_TIMEOUT_SECONDS", "1100") or "1100")


@router.post("/case-import", response=CaseImportSessionOut)
def trigger_case_import(request: HttpRequest) -> Any:
    """触发从OA导入案件（预览模式）。
    接收上传的Excel文件，解析案件编号，预览匹配结果。
    """
    import json

    from django.db.models import Q

    from apps.organization.models import AccountCredential

    if not request.user.is_authenticated:
        return {"error": "未登录"}

    lawyer_id = getattr(request.user, "id", None)
    if lawyer_id is None:
        return {"error": "无效用户"}

    # 查找用户的 jtn.com 凭证
    credential = AccountCredential.objects.filter(
        Q(account__icontains="jtn.com") | Q(url__icontains="jtn.com"),
        lawyer_id=lawyer_id,
    ).first()

    if not credential:
        return {"error": "未找到金诚同达OA账号凭证"}

    # 获取上传的文件
    file: UploadedFile | None = request.FILES.get("file")
    if not file:
        return {"error": "未上传文件"}

    # 保存上传的文件
    import uuid
    from pathlib import Path

    from django.conf import settings

    upload_dir = Path(settings.BASE_DIR) / "media" / "oa_imports"
    upload_dir.mkdir(parents=True, exist_ok=True)

    filename = f"{uuid.uuid4().hex}_{file.name}"
    file_path = upload_dir / filename

    with open(file_path, "wb") as f:
        for chunk in file.chunks():
            f.write(chunk)

    # 创建导入会话
    from apps.organization.models import Lawyer

    lawyer = Lawyer.objects.get(pk=lawyer_id)
    session = CaseImportSession.objects.create(
        lawyer=lawyer,
        credential=credential,
        status="pending",
        uploaded_filename=file.name,
    )

    logger.info("创建案件导入会话: session_id=%d filename=%s", session.id, file.name)

    # 启动后台任务进行预览
    from django_q.tasks import async_task

    async_task(
        "apps.oa_filing.tasks.run_case_import_preview_task",
        session.id,
        str(file_path),
        timeout=CASE_IMPORT_TASK_TIMEOUT_SECONDS,
        task_name=f"oa_case_import_preview_{session.id}",
    )

    return session


@router.get("/case-import/{session_id}", response=CaseImportSessionOut)
def get_case_import_session(request: HttpRequest, session_id: int) -> Any:
    """查询案件导入会话状态。"""
    return CaseImportSession.objects.get(pk=session_id)


@router.post("/case-import/{session_id}/execute")
def execute_case_import(request: HttpRequest, session_id: int) -> HttpResponse:
    """执行案件导入。
    对预览阶段标记为 unmatched 的案件，从OA提取数据并创建/更新合同。
    """
    import json

    if not request.user.is_authenticated:
        return JsonResponse({"error": "未登录"}, status=401)

    lawyer_id = getattr(request.user, "id", None)
    if lawyer_id is None:
        return JsonResponse({"error": "无效用户"}, status=400)

    # 获取会话
    try:
        session = CaseImportSession.objects.get(pk=session_id)
    except CaseImportSession.DoesNotExist:
        return JsonResponse({"error": "会话不存在"}, status=404)

    # 解析请求体
    try:
        body = json.loads(request.body)
        case_nos = body.get("case_nos", [])
        matched_case_nos = body.get("matched_case_nos", [])
    except Exception:
        return JsonResponse({"error": "无效的请求数据"}, status=400)

    if not case_nos:
        return JsonResponse({"error": "案件编号列表为空"}, status=400)

    # 启动后台任务执行导入
    from django_q.tasks import async_task

    async_task(
        "apps.oa_filing.tasks.run_case_import_task",
        session.id,
        case_nos,
        matched_case_nos=matched_case_nos,
        timeout=CASE_IMPORT_TASK_TIMEOUT_SECONDS,
        task_name=f"oa_case_import_{session.id}",
    )

    logger.info("启动案件导入任务: session_id=%d case_nos=%d", session_id, len(case_nos))

    return JsonResponse(
        {
            "message": "导入任务已启动",
            "session_id": session_id,
        }
    )


@router.get("/case-import/{session_id}/preview")
def get_case_import_preview(request: HttpRequest, session_id: int) -> JsonResponse:
    """获取案件导入预览结果。"""
    try:
        session = CaseImportSession.objects.get(pk=session_id)
    except CaseImportSession.DoesNotExist:
        return JsonResponse({"error": "会话不存在"}, status=404)

    result_data = session.result_data or {}
    preview_list = result_data.get("preview", [])

    preview_items = [
        CasePreviewItem(
            case_no=item.get("case_no", ""),
            status=item.get("status", "error"),
            existing_contract_id=item.get("existing_contract_id"),
            customer_names=item.get("customer_names", []),
            error_message=item.get("error_message", ""),
        )
        for item in preview_list
    ]

    total = len(preview_items)
    matched = sum(1 for item in preview_items if item.status == "matched")
    unmatched = sum(1 for item in preview_items if item.status == "unmatched")

    response = CasePreviewResponse(
        total_cases=total,
        matched=matched,
        unmatched=unmatched,
        preview=preview_items,
    )

    return JsonResponse(response.model_dump())


@router.post("/case-import/{session_id}/batch-create")
def batch_create_cases(request: HttpRequest, session_id: int) -> Any:
    """批量创建案件（通过API调用避免异步上下文问题）。
    接收案件数据列表，在新的HTTP请求中处理，不受Django-Q异步上下文影响。
    """
    import json

    from apps.oa_filing.models import CaseImportSession

    # 获取会话
    try:
        session = CaseImportSession.objects.get(pk=session_id)
    except CaseImportSession.DoesNotExist:
        return {"error": "会话不存在"}

    # 解析请求体
    try:
        body = json.loads(request.body)
        cases = body.get("cases", [])
    except Exception:
        return {"error": "无效的请求数据"}

    from apps.oa_filing.services.case_import_service import CaseImportService

    service = CaseImportService(session)

    results: list[CaseImportResult] = []
    success_count = 0
    skip_count = 0
    error_count = 0

    for case_data in cases:
        case_no = case_data.get("case_no", "")
        try:
            from apps.oa_filing.services.oa_scripts.jtn_case_import import OACaseData

            # 构造 OACaseData
            oa_data = OACaseData(case_no=case_no, keyid="")

            # 创建/更新
            contract_id = service._create_or_update_case(oa_data)

            if contract_id:
                results.append(
                    CaseImportResult(
                        case_no=case_no,
                        status="created",
                        contract_id=contract_id,
                    )
                )
                success_count += 1
            else:
                results.append(
                    CaseImportResult(
                        case_no=case_no,
                        status="error",
                        message="创建合同失败",
                    )
                )
                error_count += 1

        except Exception as exc:
            logger.warning("批量创建案件异常 %s: %s", case_no, exc)
            results.append(
                CaseImportResult(
                    case_no=case_no,
                    status="error",
                    message=str(exc),
                )
            )
            error_count += 1

    # 更新会话
    session.success_count = success_count
    session.skip_count = skip_count
    session.error_count = error_count
    session.save()

    return CaseImportResponse(
        success=success_count,
        failed=error_count,
        skipped=skip_count,
        total=len(cases),
        details=results,
    ).model_dump()

	"""案件导入 API。"""

	from __future__ import annotations

	import logging
	import os
	from typing import Any

	from django.http import HttpRequest, HttpResponse, JsonResponse
	from django.utils import timezone
	from ninja import Router, UploadedFile

	from apps.oa_filing.models import CaseImportSession
	from apps.oa_filing.schemas.case_import_schemas import (
	CaseImportResponse,
	CaseImportResult,
	CaseImportSessionOut,
	CasePreviewItem,
	CasePreviewResponse,
	)

	logger = logging.getLogger("apps.oa_filing.api.case_import")
	router = Router()

	# Django-Q 默认 timeout=600 秒；OA 案件导入通常需要更长时间。
	# 保持小于默认 retry(1200) 以降低重复执行风险。
	CASE_IMPORT_TASK_TIMEOUT_SECONDS = int(os.environ.get("OA_CASE_IMPORT_TASK_TIMEOUT_SECONDS", "1100") or "1100")


	@router.post("/case-import", response=CaseImportSessionOut)
	def trigger_case_import(request: HttpRequest) -> Any:
	"""触发从OA导入案件（预览模式）。

	接收上传的Excel文件，解析案件编号，预览匹配结果。
	"""
	import json

	from django.db.models import Q

	from apps.organization.models import AccountCredential

	if not request.user.is_authenticated:
	return {"error": "未登录"}

	lawyer_id = getattr(request.user, "id", None)
	if lawyer_id is None:
	return {"error": "无效用户"}

	# 查找用户的 jtn.com 凭证
	credential = AccountCredential.objects.filter(
	Q(account__icontains="jtn.com") \| Q(url__icontains="jtn.com"),
	lawyer_id=lawyer_id,
	).first()

	if not credential:
	return {"error": "未找到金诚同达OA账号凭证"}

	# 获取上传的文件
	file: UploadedFile \| None = request.FILES.get("file")
	if not file:
	return {"error": "未上传文件"}

	# 保存上传的文件
	import uuid
	from pathlib import Path

	from django.conf import settings

	upload_dir = Path(settings.BASE_DIR) / "media" / "oa_imports"
	upload_dir.mkdir(parents=True, exist_ok=True)

	filename = f"{uuid.uuid4().hex}_{file.name}"
	file_path = upload_dir / filename

	with open(file_path, "wb") as f:
	for chunk in file.chunks():
	f.write(chunk)

	# 创建导入会话
	from apps.organization.models import Lawyer

	lawyer = Lawyer.objects.get(pk=lawyer_id)
	session = CaseImportSession.objects.create(
	lawyer=lawyer,
	credential=credential,
	status="pending",
	uploaded_filename=file.name,
	)

	logger.info("创建案件导入会话: session_id=%d filename=%s", session.id, file.name)

	# 启动后台任务进行预览
	from django_q.tasks import async_task

	async_task(
	"apps.oa_filing.tasks.run_case_import_preview_task",
	session.id,
	str(file_path),
	timeout=CASE_IMPORT_TASK_TIMEOUT_SECONDS,
	task_name=f"oa_case_import_preview_{session.id}",
	)

	return session


	@router.get("/case-import/{session_id}", response=CaseImportSessionOut)
	def get_case_import_session(request: HttpRequest, session_id: int) -> Any:
	"""查询案件导入会话状态。"""
	return CaseImportSession.objects.get(pk=session_id)


	@router.post("/case-import/{session_id}/execute")
	def execute_case_import(request: HttpRequest, session_id: int) -> HttpResponse:
	"""执行案件导入。

	对预览阶段标记为 unmatched 的案件，从OA提取数据并创建/更新合同。
	"""
	import json

	if not request.user.is_authenticated:
	return JsonResponse({"error": "未登录"}, status=401)

	lawyer_id = getattr(request.user, "id", None)
	if lawyer_id is None:
	return JsonResponse({"error": "无效用户"}, status=400)

	# 获取会话
	try:
	session = CaseImportSession.objects.get(pk=session_id)
	except CaseImportSession.DoesNotExist:
	return JsonResponse({"error": "会话不存在"}, status=404)

	# 解析请求体
	try:
	body = json.loads(request.body)
	case_nos = body.get("case_nos", [])
	matched_case_nos = body.get("matched_case_nos", [])
	except Exception:
	return JsonResponse({"error": "无效的请求数据"}, status=400)

	if not case_nos:
	return JsonResponse({"error": "案件编号列表为空"}, status=400)

	# 启动后台任务执行导入
	from django_q.tasks import async_task

	async_task(
	"apps.oa_filing.tasks.run_case_import_task",
	session.id,
	case_nos,
	matched_case_nos=matched_case_nos,
	timeout=CASE_IMPORT_TASK_TIMEOUT_SECONDS,
	task_name=f"oa_case_import_{session.id}",
	)

	logger.info("启动案件导入任务: session_id=%d case_nos=%d", session_id, len(case_nos))

	return JsonResponse(
	{
	"message": "导入任务已启动",
	"session_id": session_id,
	}
	)


	@router.get("/case-import/{session_id}/preview")
	def get_case_import_preview(request: HttpRequest, session_id: int) -> JsonResponse:
	"""获取案件导入预览结果。"""
	try:
	session = CaseImportSession.objects.get(pk=session_id)
	except CaseImportSession.DoesNotExist:
	return JsonResponse({"error": "会话不存在"}, status=404)

	result_data = session.result_data or {}
	preview_list = result_data.get("preview", [])

	preview_items = [
	CasePreviewItem(
	case_no=item.get("case_no", ""),
	status=item.get("status", "error"),
	existing_contract_id=item.get("existing_contract_id"),
	customer_names=item.get("customer_names", []),
	error_message=item.get("error_message", ""),
	)
	for item in preview_list
	]

	total = len(preview_items)
	matched = sum(1 for item in preview_items if item.status == "matched")
	unmatched = sum(1 for item in preview_items if item.status == "unmatched")

	response = CasePreviewResponse(
	total_cases=total,
	matched=matched,
	unmatched=unmatched,
	preview=preview_items,
	)

	return JsonResponse(response.model_dump())


	@router.post("/case-import/{session_id}/batch-create")
	def batch_create_cases(request: HttpRequest, session_id: int) -> Any:
	"""批量创建案件（通过API调用避免异步上下文问题）。

	接收案件数据列表，在新的HTTP请求中处理，不受Django-Q异步上下文影响。
	"""
	import json

	from apps.oa_filing.models import CaseImportSession

	# 获取会话
	try:
	session = CaseImportSession.objects.get(pk=session_id)
	except CaseImportSession.DoesNotExist:
	return {"error": "会话不存在"}

	# 解析请求体
	try:
	body = json.loads(request.body)
	cases = body.get("cases", [])
	except Exception:
	return {"error": "无效的请求数据"}

	from apps.oa_filing.services.case_import_service import CaseImportService

	service = CaseImportService(session)

	results: list[CaseImportResult] = []
	success_count = 0
	skip_count = 0
	error_count = 0

	for case_data in cases:
	case_no = case_data.get("case_no", "")
	try:
	from apps.oa_filing.services.oa_scripts.jtn_case_import import OACaseData

	# 构造 OACaseData
	oa_data = OACaseData(case_no=case_no, keyid="")

	# 创建/更新
	contract_id = service._create_or_update_case(oa_data)

	if contract_id:
	results.append(
	CaseImportResult(
	case_no=case_no,
	status="created",
	contract_id=contract_id,
	)
	)
	success_count += 1
	else:
	results.append(
	CaseImportResult(
	case_no=case_no,
	status="error",
	message="创建合同失败",
	)
	)
	error_count += 1

	except Exception as exc:
	logger.warning("批量创建案件异常 %s: %s", case_no, exc)
	results.append(
	CaseImportResult(
	case_no=case_no,
	status="error",
	message=str(exc),
	)
	)
	error_count += 1

	# 更新会话
	session.success_count = success_count
	session.skip_count = skip_count
	session.error_count = error_count
	session.save()

	return CaseImportResponse(
	success=success_count,
	failed=error_count,
	skipped=skip_count,
	total=len(cases),
	details=results,
	).model_dump()