File size: 3,848 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 | """Admin 层适配器实现。
实现对 Admin 端口的具体适配，封装对外部模块的调用。
"""
from __future__ import annotations
import logging
from importlib import import_module
from typing import Any
from django.apps import apps as django_apps
from django.core.files.base import ContentFile
logger = logging.getLogger(__name__)
def _get_gsxt_report_task_model() -> Any:
return django_apps.get_model("automation", "GsxtReportTask")
def _get_gsxt_report_status_enum() -> Any:
module = import_module("apps.automation.models.gsxt_report")
return module.GsxtReportStatus
def _get_account_credential_model() -> Any:
return django_apps.get_model("organization", "AccountCredential")
class GsxtReportAdapter:
"""企业信用报告适配器。"""
def create_report_task(
self,
client_id: int,
company_name: str,
credit_code: str,
) -> int:
"""创建报告下载任务。"""
gsxt_report_task_model = _get_gsxt_report_task_model()
gsxt_report_status = _get_gsxt_report_status_enum()
task = gsxt_report_task_model.objects.create(
client_id=client_id,
company_name=company_name,
credit_code=credit_code,
status=gsxt_report_status.WAITING_CAPTCHA,
)
logger.info(f"创建企业信用报告任务: client_id={client_id}, task_id={task.id}")
return task.id
def start_login(self, credential_id: int, task_id: int) -> None:
"""启动登录流程。"""
from apps.automation.services.gsxt.gsxt_login_service import start_login_gsxt
account_credential_model = _get_account_credential_model()
credential = account_credential_model.objects.get(pk=credential_id)
start_login_gsxt(credential, task_id)
logger.info(f"启动国家企业信用信息公示系统登录: task_id={task_id}")
def get_waiting_email_task(self, client_id: int) -> Any | None:
"""获取等待邮件的任务。"""
gsxt_report_task_model = _get_gsxt_report_task_model()
gsxt_report_status = _get_gsxt_report_status_enum()
return gsxt_report_task_model.objects.filter(
client_id=client_id,
status=gsxt_report_status.WAITING_EMAIL,
).first()
def upload_report(self, task_id: int, file_content: bytes, file_name: str) -> bool:
"""上传报告文件。"""
gsxt_report_task_model = _get_gsxt_report_task_model()
gsxt_report_status = _get_gsxt_report_status_enum()
try:
task = gsxt_report_task_model.objects.get(id=task_id)
task.report_file.save(file_name, ContentFile(file_content), save=True)
task.status = gsxt_report_status.SUCCESS
task.save(update_fields=["report_file", "status", "updated_at"])
logger.info(f"上传企业信用报告成功: task_id={task_id}")
return True
except Exception as e:
logger.error(f"上传企业信用报告失败: task_id={task_id}, error={e}")
return False
def get_task_status_choices(self) -> list[tuple[str, str]]:
"""获取任务状态选项。"""
gsxt_report_status = _get_gsxt_report_status_enum()
return gsxt_report_status.choices
class CredentialAdapter:
"""账号凭证适配器。"""
def get_gsxt_credential(self) -> Any | None:
"""获取国家企业信用信息公示系统的登录凭证。
返回成功率最高且最近成功登录的凭证。
"""
account_credential_model = _get_account_credential_model()
return (
account_credential_model.objects.filter(site_name__icontains="国家企业信用")
.order_by("-last_login_success_at", "-login_success_count")
.first()
)
|