711 lines
33 KiB
Python
711 lines
33 KiB
Python
"""场外基金申购赎回业务编排服务。"""
|
|
|
|
import asyncio
|
|
from collections.abc import Sequence
|
|
from datetime import UTC, datetime
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from typing import Literal, cast
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.core.config import get_settings
|
|
from app.core.contracts import RequestContext
|
|
from app.core.offsite_fund_contracts import (
|
|
DocumentType,
|
|
OffsiteDocumentSummary,
|
|
OperationDecision,
|
|
ReceiveRecognizedMailRequest,
|
|
RecognizedAttachment,
|
|
)
|
|
from app.model.audit import InteractionAudit
|
|
from app.model.offsite_fund import (
|
|
OffsiteExecutionPlanTask,
|
|
OffsiteFundAttachment,
|
|
OffsiteFundDocument,
|
|
OffsiteFundMail,
|
|
OffsiteNotification,
|
|
OffsiteQueryRecord,
|
|
OffsiteRuleResult,
|
|
)
|
|
from app.service.offsite_fund_rules import (
|
|
OffsiteFundRuleEngine,
|
|
decimal_from,
|
|
normalize_amount_yuan,
|
|
parse_application_date,
|
|
)
|
|
from app.service.offsite_nl2sql_adapter import OffsiteNl2SqlAdapter
|
|
from app.service.offsite_smtp_adapter import (
|
|
OffsiteMailReplyRequest,
|
|
OffsiteSmtpSender,
|
|
SmtpAttachment,
|
|
SmtpSendResult,
|
|
)
|
|
|
|
SHANGHAI = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
class OffsiteFundService:
|
|
def __init__(
|
|
self, session: AsyncSession, smtp_sender: OffsiteSmtpSender | None = None
|
|
) -> None:
|
|
self.session = session
|
|
self.rules = OffsiteFundRuleEngine()
|
|
self.nl2sql = OffsiteNl2SqlAdapter()
|
|
self.smtp_sender = smtp_sender or OffsiteSmtpSender(get_settings())
|
|
|
|
async def receive_recognized_mail(
|
|
self, payload: ReceiveRecognizedMailRequest, context: RequestContext
|
|
) -> dict[str, object]:
|
|
denied = self._permission_error(context, ("offsite:write",))
|
|
if denied is not None:
|
|
return denied
|
|
settings = get_settings()
|
|
if payload.sender not in settings.offsite_allowed_senders:
|
|
return {"code": 403, "message": "发件人不在场外业务白名单", "data": {}}
|
|
business = [item for item in payload.attachments if item.document_type != "other"]
|
|
if not business:
|
|
return {"code": 0, "message": "ok", "data": {"business": False}}
|
|
async with self.session.begin():
|
|
existing = await self.session.scalar(select(OffsiteFundMail).where(
|
|
OffsiteFundMail.imap_uid == payload.imap_uid,
|
|
OffsiteFundMail.message_id == payload.message_id,
|
|
))
|
|
if existing is not None:
|
|
return {"code": 0, "message": "ok", "data": {"mail_id": existing.mail_id}}
|
|
mail_id = await self._next_mail_id()
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
mail = OffsiteFundMail(
|
|
mail_id=mail_id, imap_uid=payload.imap_uid, message_id=payload.message_id,
|
|
received_date=datetime.now(SHANGHAI).date(), sender=payload.sender,
|
|
return_path=payload.return_path, auth_result=payload.auth_result,
|
|
original_eml_path=payload.eml_path, status="recognized",
|
|
created_at=now, updated_at=now,
|
|
)
|
|
self.session.add(mail)
|
|
summaries: list[OffsiteDocumentSummary] = []
|
|
for index, item in enumerate(payload.attachments, start=1):
|
|
attachment_id = f"{mail_id}-A{index:02d}"
|
|
self.session.add(OffsiteFundAttachment(
|
|
attachment_id=attachment_id, mail_id=mail_id, filename=item.filename,
|
|
file_hash=item.file_hash, media_type=item.media_type,
|
|
size_bytes=item.size_bytes, document_type=item.document_type,
|
|
original_file_path=item.original_file_path, ocr_text=item.ocr_text,
|
|
extracted_fields=item.extracted_fields,
|
|
field_confidence={
|
|
key: str(value) for key, value in item.field_confidence.items()
|
|
},
|
|
page_evidence=item.page_evidence, status="recognized", created_at=now,
|
|
))
|
|
if item.document_type == "other":
|
|
continue
|
|
if item.document_type == "summary":
|
|
summaries.append(await self._create_document(
|
|
mail_id, f"{attachment_id}-S", attachment_id, item, "subscription", now))
|
|
summaries.append(await self._create_document(
|
|
mail_id, f"{attachment_id}-R", attachment_id, item, "redemption", now))
|
|
else:
|
|
summaries.append(await self._create_document(
|
|
mail_id, attachment_id, attachment_id, item, item.document_type, now))
|
|
self._add_audit(context, "offsite.mail_recognized", {
|
|
"mail_id": mail_id,
|
|
"imap_uid": payload.imap_uid,
|
|
"message_id": payload.message_id,
|
|
"business_attachment_count": len(business),
|
|
})
|
|
return {"code": 0, "message": "ok", "data": {
|
|
"mail_id": mail_id,
|
|
"documents": [item.model_dump(mode="json") for item in summaries],
|
|
}}
|
|
|
|
async def confirm_document(
|
|
self, task_id: str, decision: OperationDecision, operator_id: str,
|
|
context: RequestContext,
|
|
) -> dict[str, object]:
|
|
denied = self._permission_error(context, ("offsite:confirm", "offsite:write"))
|
|
if denied is not None:
|
|
return denied
|
|
async with self.session.begin():
|
|
document = await self.session.scalar(select(OffsiteFundDocument).where(
|
|
OffsiteFundDocument.task_id == task_id).with_for_update())
|
|
if document is None:
|
|
return {"code": 404, "message": "单据不存在", "data": {}}
|
|
document.operator_decision = decision
|
|
document.status = "operator_confirmed"
|
|
document.updated_at = datetime.now(UTC).replace(tzinfo=None)
|
|
self._add_audit(context, "offsite.document_confirmed", {
|
|
"task_id": task_id,
|
|
"operator_id": operator_id,
|
|
"decision": decision,
|
|
})
|
|
return {"code": 0, "message": "ok", "data": {"task_id": task_id, "decision": decision}}
|
|
|
|
async def recalculate_statistics(
|
|
self, fund_code: str, application_date: str, context: RequestContext
|
|
) -> dict[str, object]:
|
|
denied = self._permission_error(context, ("offsite:read", "offsite:write"))
|
|
if denied is not None:
|
|
return denied
|
|
target_date = parse_application_date(application_date)
|
|
if target_date is None:
|
|
return {"code": 422, "message": "申请日期格式不正确", "data": {}}
|
|
rows = (await self.session.execute(select(OffsiteFundDocument).where(
|
|
OffsiteFundDocument.fund_code == fund_code,
|
|
OffsiteFundDocument.application_date == target_date,
|
|
OffsiteFundDocument.operator_decision == "确认正常",
|
|
))).scalars().all()
|
|
subscription_total = sum(
|
|
((row.subscription_amount_yuan or Decimal("0")) for row in rows),
|
|
Decimal("0"),
|
|
)
|
|
redemption_total = sum(
|
|
((row.redemption_shares or Decimal("0")) for row in rows),
|
|
Decimal("0"),
|
|
)
|
|
agency_breakdown = self._agency_breakdown(rows)
|
|
redemption_amount_yuan: Decimal | None = None
|
|
net_flow_amount_yuan: Decimal | None = subscription_total if redemption_total == 0 else None
|
|
return {"code": 0, "message": "ok", "data": {
|
|
"fund_code": fund_code, "application_date": target_date.isoformat(),
|
|
"fund_name": next((row.fund_name for row in rows if row.fund_name), None),
|
|
"subscription_amount_yuan": str(subscription_total),
|
|
"subscription_count": sum(1 for row in rows if row.document_type == "subscription"),
|
|
"redemption_shares": str(redemption_total),
|
|
"redemption_count": sum(1 for row in rows if row.document_type == "redemption"),
|
|
"latest_nav": None,
|
|
"redemption_amount_yuan": (
|
|
str(redemption_amount_yuan) if redemption_amount_yuan is not None else None
|
|
),
|
|
"net_flow_amount_yuan": (
|
|
str(net_flow_amount_yuan) if net_flow_amount_yuan is not None else None
|
|
),
|
|
"agency_breakdown": agency_breakdown,
|
|
}}
|
|
|
|
async def trigger_agent_nl2sql(
|
|
self, task_id: str, operator_id: str, manual_confirmed: bool,
|
|
context: RequestContext,
|
|
) -> dict[str, object]:
|
|
denied = self._permission_error(
|
|
context, ("offsite:nl2sql", "offsite:write", "financial:nl2sql:read")
|
|
)
|
|
if denied is not None:
|
|
return denied
|
|
if not manual_confirmed:
|
|
return {"code": 422, "message": "必须传递人工已确认原始文件内容状态", "data": {}}
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
records: list[dict[str, object]] = []
|
|
async with self.session.begin():
|
|
document = await self.session.scalar(select(OffsiteFundDocument).where(
|
|
OffsiteFundDocument.task_id == task_id))
|
|
if document is None:
|
|
return {"code": 404, "message": "单据不存在", "data": {}}
|
|
questions = self._nl2sql_questions(document)
|
|
if not questions:
|
|
return {"code": 422, "message": "查询条件不足", "data": {}}
|
|
for rule_code, question in questions:
|
|
result = self.nl2sql.query(question, context)
|
|
self.session.add(OffsiteQueryRecord(
|
|
task_id=task_id, rule_code=rule_code,
|
|
natural_language_request=question, script_path=self.nl2sql.script_path,
|
|
result_summary=result, status=str(result.get("status", "error")),
|
|
error_message=(
|
|
result.get("message") if isinstance(result.get("message"), str) else None
|
|
),
|
|
created_at=now,
|
|
))
|
|
await self._update_query_plan_status(task_id, rule_code, result, now)
|
|
records.append({"rule_code": rule_code, "status": result.get("status")})
|
|
self._add_audit(context, "offsite.nl2sql_triggered", {
|
|
"task_id": task_id,
|
|
"operator_id": operator_id,
|
|
"query_count": len(records),
|
|
})
|
|
return {"code": 0, "message": "ok", "data": {"task_id": task_id, "queries": records}}
|
|
|
|
async def create_notification(
|
|
self, task_id: str, notification_type: str, operator_id: str,
|
|
context: RequestContext,
|
|
) -> dict[str, object]:
|
|
denied = self._permission_error(context, ("offsite:notify", "offsite:write"))
|
|
if denied is not None:
|
|
return denied
|
|
settings = get_settings()
|
|
receiver = {
|
|
"risk": settings.offsite_risk_receiver_id,
|
|
"settlement": settings.offsite_settlement_receiver_id,
|
|
"mail_return": settings.offsite_mail_return_receiver,
|
|
}[notification_type]
|
|
draft = f"{task_id} 待发送{notification_type}通知"
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with self.session.begin():
|
|
document = await self.session.scalar(select(OffsiteFundDocument).where(
|
|
OffsiteFundDocument.task_id == task_id))
|
|
if document is None:
|
|
return {"code": 404, "message": "单据不存在", "data": {}}
|
|
validation = self._validate_notification(document, notification_type)
|
|
if validation is not None:
|
|
return validation
|
|
rule_results = (await self.session.execute(select(OffsiteRuleResult).where(
|
|
OffsiteRuleResult.task_id == task_id
|
|
))).scalars().all()
|
|
payload = self._notification_payload(document, rule_results, operator_id)
|
|
notice = OffsiteNotification(
|
|
notification_type=notification_type, business_key=task_id,
|
|
receiver_id=receiver, operator_id=operator_id, agent_draft=draft,
|
|
final_content=draft, payload=payload, status="待发送",
|
|
created_at=now, updated_at=now,
|
|
)
|
|
self.session.add(notice)
|
|
await self.session.flush()
|
|
notice_id = notice.id
|
|
self._add_audit(context, "offsite.notification_created", {
|
|
"task_id": task_id,
|
|
"notification_type": notification_type,
|
|
"operator_id": operator_id,
|
|
"receiver_id": receiver,
|
|
})
|
|
return {"code": 0, "message": "ok", "data": {"notification_id": str(notice_id)}}
|
|
|
|
async def send_notification(
|
|
self,
|
|
notification_id: int,
|
|
operator_id: str,
|
|
operator_confirmed: bool,
|
|
final_content: str | None,
|
|
context: RequestContext,
|
|
) -> dict[str, object]:
|
|
denied = self._permission_error(context, ("offsite:notify", "offsite:write"))
|
|
if denied is not None:
|
|
return denied
|
|
if not operator_confirmed:
|
|
return {"code": 422, "message": "邮件发送前必须完成运营确认", "data": {}}
|
|
prepared = await self._prepare_notification_send(
|
|
notification_id, operator_id, final_content, context
|
|
)
|
|
if isinstance(prepared, dict):
|
|
return prepared
|
|
notice, mail, attachment, receiver = prepared
|
|
try:
|
|
request = self._build_mail_reply_request(
|
|
notice, mail, attachment, receiver, operator_id, final_content
|
|
)
|
|
result = await asyncio.to_thread(self.smtp_sender.send_reply, request)
|
|
except OSError as exc:
|
|
result = SmtpSendResult(
|
|
status="发送失败",
|
|
dry_run=False,
|
|
provider_message_id=None,
|
|
failure_reason=type(exc).__name__,
|
|
retry_count=notice.retry_count + 1,
|
|
request_summary={"notification_id": str(notification_id)},
|
|
)
|
|
await self._finish_notification(notification_id, result, context)
|
|
return {
|
|
"code": 0,
|
|
"message": "ok",
|
|
"data": {
|
|
"notification_id": str(notification_id),
|
|
"status": result.status,
|
|
"dry_run": result.dry_run,
|
|
"provider_message_id": result.provider_message_id,
|
|
"failure_reason": result.failure_reason,
|
|
"retry_count": result.retry_count,
|
|
},
|
|
}
|
|
|
|
async def _prepare_notification_send(
|
|
self,
|
|
notification_id: int,
|
|
operator_id: str,
|
|
final_content: str | None,
|
|
context: RequestContext,
|
|
) -> tuple[
|
|
OffsiteNotification,
|
|
OffsiteFundMail,
|
|
OffsiteFundAttachment,
|
|
str,
|
|
] | dict[str, object]:
|
|
settings = get_settings()
|
|
async with self.session.begin():
|
|
notice = await self.session.scalar(select(OffsiteNotification).where(
|
|
OffsiteNotification.id == notification_id
|
|
).with_for_update())
|
|
if notice is None:
|
|
return {"code": 404, "message": "通知不存在", "data": {}}
|
|
if notice.status == "发送成功":
|
|
return {
|
|
"code": 0,
|
|
"message": "ok",
|
|
"data": {
|
|
"notification_id": str(notification_id),
|
|
"status": notice.status,
|
|
"provider_message_id": notice.provider_message_id,
|
|
},
|
|
}
|
|
if notice.status == "发送中":
|
|
return {"code": 409, "message": "通知正在发送中", "data": {}}
|
|
if notice.retry_count > 0 and notice.retry_count >= settings.offsite_max_retry_count:
|
|
return {"code": 422, "message": "通知已达到最大重试次数", "data": {}}
|
|
document = await self.session.scalar(select(OffsiteFundDocument).where(
|
|
OffsiteFundDocument.task_id == notice.business_key))
|
|
if document is None:
|
|
return {"code": 404, "message": "通知关联的单据或邮件不存在", "data": {}}
|
|
mail = await self.session.scalar(select(OffsiteFundMail).where(
|
|
OffsiteFundMail.mail_id == document.mail_id))
|
|
if mail is None:
|
|
return {"code": 404, "message": "通知关联的单据或邮件不存在", "data": {}}
|
|
if notice.notification_type == "mail_return":
|
|
receiver = settings.offsite_mail_return_receiver
|
|
else:
|
|
receiver = notice.receiver_id if "@" in notice.receiver_id else ""
|
|
if not receiver:
|
|
return {"code": 422, "message": "通知对象未配置可发送的邮箱地址", "data": {}}
|
|
attachment = await self.session.scalar(select(OffsiteFundAttachment).where(
|
|
OffsiteFundAttachment.attachment_id == document.attachment_id))
|
|
if attachment is None:
|
|
return {"code": 404, "message": "通知关联的原始附件不存在", "data": {}}
|
|
if final_content is not None and final_content != notice.final_content:
|
|
self._add_audit(context, "offsite.notification_content_changed", {
|
|
"notification_id": str(notification_id),
|
|
"operator_id": operator_id,
|
|
"changed": True,
|
|
})
|
|
notice.final_content = final_content
|
|
notice.operator_id = operator_id
|
|
notice.status = "发送中"
|
|
return notice, mail, attachment, receiver
|
|
|
|
def _build_mail_reply_request(
|
|
self,
|
|
notice: OffsiteNotification,
|
|
mail: OffsiteFundMail,
|
|
attachment: OffsiteFundAttachment,
|
|
receiver: str,
|
|
operator_id: str,
|
|
final_content: str | None,
|
|
) -> OffsiteMailReplyRequest:
|
|
path = Path(attachment.original_file_path)
|
|
payload = path.read_bytes()
|
|
return OffsiteMailReplyRequest(
|
|
to_address=receiver,
|
|
subject="场外基金申购赎回处理结果",
|
|
body=final_content if final_content is not None else notice.final_content,
|
|
operator_id=operator_id,
|
|
operator_confirmed=True,
|
|
reply_to_message_id=mail.message_id,
|
|
attachments=(
|
|
SmtpAttachment(
|
|
filename=attachment.filename,
|
|
media_type=attachment.media_type,
|
|
payload=payload,
|
|
),
|
|
),
|
|
retry_count=notice.retry_count,
|
|
)
|
|
|
|
async def _finish_notification(
|
|
self, notification_id: int, result: SmtpSendResult, context: RequestContext
|
|
) -> None:
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with self.session.begin():
|
|
notice = await self.session.scalar(select(OffsiteNotification).where(
|
|
OffsiteNotification.id == notification_id
|
|
).with_for_update())
|
|
if notice is None:
|
|
return
|
|
notice.status = result.status
|
|
notice.provider_message_id = result.provider_message_id
|
|
notice.failure_reason = result.failure_reason
|
|
notice.retry_count = result.retry_count
|
|
notice.sent_at = now if result.status == "发送成功" else None
|
|
notice.updated_at = now
|
|
if result.status == "发送成功" and notice.notification_type == "mail_return":
|
|
document = await self.session.scalar(select(OffsiteFundDocument).where(
|
|
OffsiteFundDocument.task_id == notice.business_key))
|
|
if document is not None:
|
|
mail = await self.session.scalar(select(OffsiteFundMail).where(
|
|
OffsiteFundMail.mail_id == document.mail_id).with_for_update())
|
|
if mail is not None:
|
|
mail.status = "normal_return_sent"
|
|
self._add_audit(context, "offsite.notification_send_finished", {
|
|
"notification_id": str(notification_id),
|
|
"status": result.status,
|
|
"dry_run": result.dry_run,
|
|
"retry_count": result.retry_count,
|
|
})
|
|
|
|
async def _next_mail_id(self) -> str:
|
|
today = datetime.now(SHANGHAI).date()
|
|
prefix = today.strftime("%Y%m%d")
|
|
current = await self.session.scalar(select(func.count()).select_from(OffsiteFundMail).where(
|
|
OffsiteFundMail.received_date == today))
|
|
return f"{prefix}-{int(current or 0) + 1:03d}"
|
|
|
|
async def _create_document(
|
|
self,
|
|
mail_id: str,
|
|
task_id: str,
|
|
attachment_id: str,
|
|
item: RecognizedAttachment,
|
|
document_type: Literal["subscription", "redemption"],
|
|
now: datetime,
|
|
) -> OffsiteDocumentSummary:
|
|
fields = item.extracted_fields
|
|
raw_date = fields.get("申请日期")
|
|
amount_yuan = normalize_amount_yuan(fields.get("申购金额"), fields.get("金额单位"))
|
|
redemption_shares = decimal_from(fields.get("赎回份额"))
|
|
status = self._recognition_status(item, document_type)
|
|
document = OffsiteFundDocument(
|
|
task_id=task_id, mail_id=mail_id, attachment_id=attachment_id,
|
|
document_type=document_type, fund_code=self._text(fields.get("基金代码")),
|
|
fund_name=self._text(fields.get("基金名称")),
|
|
account_identifier=self._text(fields.get("账户标识")),
|
|
investor_name=self._text(fields.get("投资者名称") or fields.get("客户标识")),
|
|
application_no=self._text(fields.get("申请编号")),
|
|
application_date=parse_application_date(raw_date),
|
|
raw_application_date=self._text(raw_date), agency=self._text(fields.get("代销机构")),
|
|
subscription_amount_yuan=amount_yuan, redemption_shares=redemption_shares,
|
|
status=status, created_at=now, updated_at=now,
|
|
)
|
|
self.session.add(document)
|
|
self._append_plan_tasks(task_id, document_type, fields, now)
|
|
decisions = self.rules.check_subscription(
|
|
amount_yuan=amount_yuan, nav=decimal_from(fields.get("最新净值")),
|
|
total_fund_shares=decimal_from(fields.get("基金最新总份额")),
|
|
before_holding_shares=decimal_from(fields.get("申请前持有份额")),
|
|
) if document_type == "subscription" else self.rules.check_redemption(
|
|
redemption_shares=redemption_shares,
|
|
total_fund_shares=decimal_from(fields.get("基金最新总份额")),
|
|
available_quantity=decimal_from(fields.get("当前最新可用份额")),
|
|
)
|
|
results: dict[str, Literal["正常", "异常", "无法判断"]] = {}
|
|
for decision in decisions:
|
|
results[decision.rule_code] = decision.result
|
|
self.session.add(OffsiteRuleResult(
|
|
task_id=task_id, rule_code=decision.rule_code, rule_name=decision.rule_name,
|
|
result=decision.result, document_value=decision.document_value,
|
|
database_value=decision.database_value, calculation=decision.calculation,
|
|
created_at=now,
|
|
))
|
|
return OffsiteDocumentSummary(
|
|
task_id=task_id, document_type=cast(DocumentType, document_type), status=status,
|
|
rule_results=results)
|
|
|
|
def _append_plan_tasks(
|
|
self, task_id: str, document_type: str, fields: dict[str, object], now: datetime
|
|
) -> None:
|
|
rules: tuple[tuple[str, str], ...] = (
|
|
("subscription_minimum_amount", "申购最低金额"),
|
|
("subscription_holding_ratio", "申购后单一投资者持有比例"),
|
|
("subscription_single_share_limit", "申购单笔份额上限"),
|
|
)
|
|
if document_type == "redemption":
|
|
rules = (("redemption_large_ratio", "赎回巨额比例"),
|
|
("redemption_available_quantity", "账户可用份额"))
|
|
for rule_code, title in rules:
|
|
for stage in ("查询", "计算", "核对"):
|
|
if rule_code == "subscription_minimum_amount" and stage == "查询":
|
|
status = "不适用"
|
|
else:
|
|
status = "已完成" if stage != "查询" else "待执行"
|
|
self.session.add(OffsiteExecutionPlanTask(
|
|
task_id=task_id, stage=stage, rule_code=rule_code, title=title,
|
|
depends_on=[], input_json=fields, output_json=None,
|
|
status=status,
|
|
error_message=None, created_at=now, updated_at=now,
|
|
))
|
|
|
|
@staticmethod
|
|
def _nl2sql_questions(document: OffsiteFundDocument) -> list[tuple[str, str]]:
|
|
if not document.fund_code:
|
|
return []
|
|
base = f"基金代码为{document.fund_code}"
|
|
if document.account_identifier:
|
|
base += f"，账户标识为{document.account_identifier}"
|
|
if document.document_type == "subscription":
|
|
return [
|
|
(
|
|
"subscription_holding_ratio",
|
|
base + "，查询基金最新总份额、最新净值和申请前持有份额",
|
|
),
|
|
("subscription_single_share_limit", base + "，查询基金最新总份额和最新净值"),
|
|
]
|
|
return [
|
|
("redemption_large_ratio", base + "，查询产品最新总份额"),
|
|
("redemption_available_quantity", base + "，查询账户当前最新可用份额"),
|
|
]
|
|
|
|
@staticmethod
|
|
def _text(value: object) -> str | None:
|
|
if value is None:
|
|
return None
|
|
text = str(value).strip()
|
|
return text or None
|
|
|
|
async def _update_query_plan_status(
|
|
self, task_id: str, rule_code: str, result: dict[str, object], now: datetime
|
|
) -> None:
|
|
task = await self.session.scalar(select(OffsiteExecutionPlanTask).where(
|
|
OffsiteExecutionPlanTask.task_id == task_id,
|
|
OffsiteExecutionPlanTask.rule_code == rule_code,
|
|
OffsiteExecutionPlanTask.stage == "查询",
|
|
).with_for_update())
|
|
if task is None:
|
|
return
|
|
status = str(result.get("status", "error"))
|
|
task.status = "已完成" if status in {"ready", "success"} else "查询失败"
|
|
if status == "need_confirmation":
|
|
task.status = "无法判断"
|
|
task.output_json = result
|
|
message = result.get("message")
|
|
task.error_message = message if isinstance(message, str) else None
|
|
task.updated_at = now
|
|
|
|
@staticmethod
|
|
def _permission_error(
|
|
context: RequestContext, permissions: tuple[str, ...]
|
|
) -> dict[str, object] | None:
|
|
roles = {"operator", "risk_operator", "admin", "super_admin"}
|
|
if not roles.intersection(context.roles):
|
|
return {"code": 403, "message": "当前角色不能操作场外基金流程", "data": {}}
|
|
if not set(permissions).intersection(context.permissions):
|
|
return {"code": 403, "message": "缺少场外基金操作权限", "data": {}}
|
|
return None
|
|
|
|
def _add_audit(
|
|
self, context: RequestContext, action_type: str, detail: dict[str, object]
|
|
) -> None:
|
|
self.session.add(InteractionAudit(
|
|
actor_type="user",
|
|
actor_id=int(context.user_id) if context.user_id.isdigit() else None,
|
|
target_customer_id=None,
|
|
session_id=None,
|
|
portal=context.portal,
|
|
action_type=action_type,
|
|
detail={**detail, "trace_id": context.trace_id},
|
|
created_at=datetime.now(UTC).replace(tzinfo=None),
|
|
))
|
|
|
|
@staticmethod
|
|
def _recognition_status(
|
|
item: RecognizedAttachment, document_type: Literal["subscription", "redemption"]
|
|
) -> str:
|
|
fields = item.extracted_fields
|
|
required = ["基金代码", "基金名称", "账户标识", "申请编号", "申请日期", "代销机构"]
|
|
if not (fields.get("投资者名称") or fields.get("客户标识")):
|
|
return "recognition_exception"
|
|
if document_type == "subscription":
|
|
required.extend(["申购金额", "金额单位"])
|
|
else:
|
|
required.append("赎回份额")
|
|
if any(not str(fields.get(name) or "").strip() for name in required):
|
|
return "recognition_exception"
|
|
if not item.field_confidence:
|
|
return "recognition_review"
|
|
confidence_values = [
|
|
value for value in (decimal_from(value) for value in item.field_confidence.values())
|
|
if value is not None
|
|
]
|
|
if not confidence_values:
|
|
return "recognition_review"
|
|
min_confidence = min(confidence_values)
|
|
if min_confidence < Decimal("0.80"):
|
|
return "recognition_exception"
|
|
if min_confidence < Decimal("0.95"):
|
|
return "recognition_review"
|
|
return "planned"
|
|
|
|
@staticmethod
|
|
def _agency_breakdown(rows: Sequence[OffsiteFundDocument]) -> list[dict[str, object]]:
|
|
grouped: dict[str, dict[str, object]] = {}
|
|
for row in rows:
|
|
agency = row.agency or "未识别代销机构"
|
|
item = grouped.setdefault(agency, {
|
|
"agency": agency,
|
|
"subscription_amount_yuan": Decimal("0"),
|
|
"subscription_count": 0,
|
|
"redemption_shares": Decimal("0"),
|
|
"redemption_count": 0,
|
|
})
|
|
if row.document_type == "subscription":
|
|
item["subscription_amount_yuan"] = cast(
|
|
Decimal, item["subscription_amount_yuan"]
|
|
) + (row.subscription_amount_yuan or Decimal("0"))
|
|
subscription_count = item["subscription_count"]
|
|
item["subscription_count"] = (
|
|
subscription_count + 1 if isinstance(subscription_count, int) else 1
|
|
)
|
|
elif row.document_type == "redemption":
|
|
item["redemption_shares"] = cast(
|
|
Decimal, item["redemption_shares"]
|
|
) + (row.redemption_shares or Decimal("0"))
|
|
redemption_count = item["redemption_count"]
|
|
item["redemption_count"] = (
|
|
redemption_count + 1 if isinstance(redemption_count, int) else 1
|
|
)
|
|
return [
|
|
{
|
|
**item,
|
|
"subscription_amount_yuan": str(item["subscription_amount_yuan"]),
|
|
"redemption_shares": str(item["redemption_shares"]),
|
|
}
|
|
for item in grouped.values()
|
|
]
|
|
|
|
@staticmethod
|
|
def _validate_notification(
|
|
document: OffsiteFundDocument, notification_type: str
|
|
) -> dict[str, object] | None:
|
|
if notification_type == "risk" and document.operator_decision != "确认异常":
|
|
return {"code": 422, "message": "风控通知只允许发送已确认异常单据", "data": {}}
|
|
if notification_type == "settlement" and document.operator_decision != "确认正常":
|
|
return {"code": 422, "message": "资金清算通知只允许发送已确认正常单据", "data": {}}
|
|
if notification_type == "mail_return" and document.operator_decision == "未处理":
|
|
return {"code": 422, "message": "邮件返回前必须先完成人工确认", "data": {}}
|
|
return None
|
|
|
|
@staticmethod
|
|
def _notification_payload(
|
|
document: OffsiteFundDocument,
|
|
rule_results: Sequence[OffsiteRuleResult],
|
|
operator_id: str,
|
|
) -> dict[str, object]:
|
|
anomalies = [
|
|
{
|
|
"rule_code": result.rule_code,
|
|
"rule_name": result.rule_name,
|
|
"result": result.result,
|
|
"document_value": result.document_value,
|
|
"database_value": result.database_value,
|
|
"calculation": result.calculation,
|
|
}
|
|
for result in rule_results if result.result == "异常"
|
|
]
|
|
return {
|
|
"task_id": document.task_id,
|
|
"mail_id": document.mail_id,
|
|
"attachment_id": document.attachment_id,
|
|
"document_type": document.document_type,
|
|
"fund_code": document.fund_code,
|
|
"fund_name": document.fund_name,
|
|
"account_identifier": OffsiteFundService._mask_account(document.account_identifier),
|
|
"application_no": document.application_no,
|
|
"application_date": (
|
|
document.application_date.isoformat() if document.application_date else None
|
|
),
|
|
"agency": document.agency,
|
|
"operator_id": operator_id,
|
|
"operator_decision": document.operator_decision,
|
|
"confirmed_at": datetime.now(UTC).replace(tzinfo=None).isoformat(timespec="seconds"),
|
|
"anomalies": anomalies,
|
|
}
|
|
|
|
@staticmethod
|
|
def _mask_account(value: str | None) -> str | None:
|
|
if value is None or len(value) <= 4:
|
|
return value
|
|
return f"{value[:2]}***{value[-2:]}"
|