diff --git a/.env.example b/.env.example index ba55b2c..360e066 100644 --- a/.env.example +++ b/.env.example @@ -57,13 +57,16 @@ RISK_SCAN_POLL_SECONDS=30 RISK_EVIDENCE_DIR=storage/risk_evidence RISK_EVIDENCE_MAX_FILE_SIZE_MB=10 -RISK_DAILY_REPORT_MAIL_ENABLED=false +RISK_ALERT_MAIL_ENABLED=true +RISK_ALERT_MAIL_DRY_RUN=false +RISK_ALERT_MAIL_RECIPIENTS=你的邮箱账号 +RISK_DAILY_REPORT_MAIL_ENABLED=true RISK_DAILY_REPORT_MAIL_DRY_RUN=true -RISK_SMTP_HOST= +RISK_SMTP_HOST=smtp.163.com RISK_SMTP_PORT=465 -RISK_SMTP_USERNAME= -RISK_SMTP_PASSWORD= -RISK_SMTP_SENDER= +RISK_SMTP_USERNAME=你的邮箱账号 +RISK_SMTP_PASSWORD=你的SMTP授权码 +RISK_SMTP_SENDER=你的邮箱账号 RISK_SMTP_USE_SSL=true RISK_SMTP_TIMEOUT_SECONDS=30 diff --git a/app/api/controllers/risk.py b/app/api/controllers/risk.py index d9f380e..b9637f0 100644 --- a/app/api/controllers/risk.py +++ b/app/api/controllers/risk.py @@ -98,7 +98,7 @@ async def scan_risk_alerts( async with mysql_scan_lock() as acquired: if not acquired: raise RiskScanBusyError("规则扫描正在执行，请稍后重试") - return await RiskScanService(inner).scan(context) + return await RiskScanService.from_settings(inner).scan(context) data = await _idempotent_write( session, context, key, "POST /api/v1/risk/alerts/scan", {}, run diff --git a/app/core/config.py b/app/core/config.py index e0b3e06..f80514f 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -117,6 +117,9 @@ class Settings(BaseSettings): risk_scan_run_immediately: bool = False risk_scan_retry_limit: int = Field(default=2, ge=0, le=5) risk_scan_poll_seconds: float = Field(default=30, gt=0) + risk_alert_mail_enabled: bool = False + risk_alert_mail_dry_run: bool = True + risk_alert_mail_recipients: str = "" # 投顾灰度默认关闭，只有显式开启并配置客户白名单后才限制客户流量。 # 管理员角色始终可进入，便于审核、发布和故障处置。 advisor_rollout_enabled: bool = False diff --git a/app/service/risk_daily_report_mail_service.py b/app/service/risk_daily_report_mail_service.py index 3bff8a0..024316d 100644 --- a/app/service/risk_daily_report_mail_service.py +++ b/app/service/risk_daily_report_mail_service.py @@ -3,18 +3,27 @@ from __future__ import annotations import os -import smtplib from collections.abc import Mapping -from email.message import EmailMessage -from email.utils import formatdate, make_msgid from app.core.contracts import RequestContext from app.service.authorization_service import AuthorizationService +from app.service.risk_smtp_mail_service import ( + RiskSmtpConfigurationError, + RiskSmtpMailService, +) class RiskDailyReportMailService: - def __init__(self, *, environment: Mapping[str, str] | None = None) -> None: - self.environment = environment or os.environ + def __init__( + self, + *, + environment: Mapping[str, str] | None = None, + smtp_service: RiskSmtpMailService | None = None, + ) -> None: + self.environment = environment if environment is not None else os.environ + self.smtp_service = smtp_service or RiskSmtpMailService( + environment=environment + ) async def send( self, @@ -36,28 +45,10 @@ class RiskDailyReportMailService: return {"status": "disabled", "recipient_count": len(recipients)} if _enabled(self.environment, "RISK_DAILY_REPORT_MAIL_DRY_RUN", default=True): return {"status": "dry_run", "recipient_count": len(recipients)} - host = self.environment.get("RISK_SMTP_HOST", "").strip() - sender = self.environment.get("RISK_SMTP_SENDER", "").strip() - if not host or not sender: + try: + return await self.smtp_service.send(recipients, subject, content) + except RiskSmtpConfigurationError: return {"status": "configuration_error", "recipient_count": len(recipients)} - message = EmailMessage() - message["From"] = sender - message["To"] = ", ".join(recipients) - message["Subject"] = subject - message["Date"] = formatdate(localtime=True) - message["Message-ID"] = make_msgid(domain="risk-report.local") - message.set_content(content) - port = int(self.environment.get("RISK_SMTP_PORT", "465")) - use_ssl = _enabled(self.environment, "RISK_SMTP_USE_SSL", default=True) - timeout = float(self.environment.get("RISK_SMTP_TIMEOUT_SECONDS", "30")) - smtp_class = smtplib.SMTP_SSL if use_ssl else smtplib.SMTP - with smtp_class(host, port, timeout=timeout) as connection: - username = self.environment.get("RISK_SMTP_USERNAME", "").strip() - password = self.environment.get("RISK_SMTP_PASSWORD", "") - if username: - connection.login(username, password) - connection.send_message(message) - return {"status": "sent", "recipient_count": len(recipients)} def _enabled(environment: Mapping[str, str], name: str, *, default: bool = False) -> bool: diff --git a/app/service/risk_notification_service.py b/app/service/risk_notification_service.py index 9cbddf3..c3eba08 100644 --- a/app/service/risk_notification_service.py +++ b/app/service/risk_notification_service.py @@ -93,6 +93,22 @@ class RiskNotificationService: self.session.add(notification) return notification + def mark_mail_sent(self, notification: FundRiskNotification) -> None: + """邮件发送成功后回写通知状态。""" + notification.send_status = "已发送" + notification.sent_at = _now() + notification.fail_reason = None + + def mark_mail_failed( + self, + notification: FundRiskNotification, + reason: str, + ) -> None: + """邮件发送失败后保留通知记录并记录安全失败原因。""" + notification.send_status = "发送失败" + notification.sent_at = None + notification.fail_reason = reason.strip()[:500] or "邮件发送失败" + def create_high_risk_records( self, alerts: list[FundRiskAlert], diff --git a/app/service/risk_scan_service.py b/app/service/risk_scan_service.py index 761bef5..1a469d0 100644 --- a/app/service/risk_scan_service.py +++ b/app/service/risk_scan_service.py @@ -17,6 +17,7 @@ from uuid import uuid4 from sqlalchemy import Select, func, select from sqlalchemy.ext.asyncio import AsyncSession +from app.core.config import get_settings from app.core.contracts import RequestContext from app.core.errors import AgentError, ConflictAgentError from app.core.timeutil import local_date, local_hour @@ -27,11 +28,17 @@ from app.model.fund import ( FundHolding, FundProduct, FundRiskAlert, + FundRiskNotification, FundTransaction, ) from app.model.risk import RiskLoginRecord, RiskUser, RiskWorkOrder from app.service.authorization_service import AuthorizationService from app.service.risk_notification_service import RiskNotificationService +from app.service.risk_smtp_mail_service import ( + RiskSmtpConfigurationError, + RiskSmtpMailService, + RiskSmtpSendError, +) HIGH_RISK = "高" MEDIUM_RISK = "中" @@ -513,6 +520,8 @@ class RiskScanService: notification_enabled: bool = True, notification_email: str | None = None, mail_enabled: bool = False, + mail_dry_run: bool = False, + smtp_mail_service: RiskSmtpMailService | None = None, ) -> None: self.session = session self.rule_engine = rule_engine or RiskRuleEngine(session) @@ -520,6 +529,31 @@ class RiskScanService: self.notification_enabled = notification_enabled self.notification_email = notification_email self.mail_enabled = mail_enabled + self.mail_dry_run = mail_dry_run + self.smtp_mail_service = smtp_mail_service or RiskSmtpMailService() + + @classmethod + def from_settings( + cls, + session: AsyncSession, + *, + rule_engine: RiskRuleEngine | None = None, + notification_service: RiskNotificationService | None = None, + ) -> RiskScanService: + """从环境配置创建扫描服务，供手工扫描和定时扫描共用。""" + settings = get_settings() + recipients = _parse_recipients(settings.risk_alert_mail_recipients) + notification_email = recipients[0] if recipients else None + if settings.risk_alert_mail_enabled and not notification_email: + logger.warning("高风险邮件通知已启用，但未配置收件人") + return cls( + session, + rule_engine=rule_engine, + notification_service=notification_service, + notification_email=notification_email, + mail_enabled=settings.risk_alert_mail_enabled and bool(notification_email), + mail_dry_run=settings.risk_alert_mail_dry_run, + ) async def scan(self, context: RequestContext) -> dict[str, int | str]: await AuthorizationService.require(context, "risk:alert:scan") @@ -575,7 +609,7 @@ class RiskScanService: ) count += 1 if self.notification_email: - self.notification_service.create_mail_record( + notification = self.notification_service.create_mail_record( alert, receiver_email=self.notification_email, title=title, @@ -583,11 +617,48 @@ class RiskScanService: mail_enabled=self.mail_enabled, ) count += 1 + if self.mail_enabled: + await self._deliver_mail( + alert, + notification, + title, + ) return count, "" except Exception as error: logger.exception("高风险通知记录创建失败，预警扫描继续提交") return 0, f"{type(error).__name__}: {error}"[:200] + async def _deliver_mail( + self, + alert: FundRiskAlert, + notification: FundRiskNotification, + title: str, + ) -> None: + """发送高风险通知邮件，失败只回写通知记录，不影响预警落库。""" + if self.mail_dry_run: + self.notification_service.mark_mail_failed( + notification, + "邮件发送处于 dry-run 模式", + ) + return + try: + result = await self.smtp_mail_service.send( + [self.notification_email or ""], + title, + f"预警编号：{alert.alert_no}；{alert.evidence_summary}", + ) + except (RiskSmtpConfigurationError, RiskSmtpSendError) as error: + logger.warning("高风险预警邮件发送失败：alert_no=%s error=%s", alert.alert_no, error) + self.notification_service.mark_mail_failed(notification, str(error)) + except Exception: + logger.exception("高风险预警邮件发送发生未预期异常：alert_no=%s", alert.alert_no) + self.notification_service.mark_mail_failed(notification, "邮件发送发生未预期异常") + else: + if result.get("status") == "sent": + self.notification_service.mark_mail_sent(notification) + else: + self.notification_service.mark_mail_failed(notification, "邮件发送未完成") + def _new_alert_id() -> int: """生成非零 63 位正整数；预警表主键不自增。""" @@ -617,3 +688,14 @@ def _level_value(level: str, prefix: str) -> int | None: """ digits = level.strip().upper().removeprefix(prefix.upper()) return int(digits) if digits.isdigit() else None + + +def _parse_recipients(value: str | None) -> list[str]: + """解析逗号或分号分隔的邮件收件人配置。""" + if not value: + return [] + return [ + recipient.strip() + for recipient in value.replace(";", ",").split(",") + if recipient.strip() + ] diff --git a/app/service/risk_smtp_mail_service.py b/app/service/risk_smtp_mail_service.py new file mode 100644 index 0000000..3dbb13a --- /dev/null +++ b/app/service/risk_smtp_mail_service.py @@ -0,0 +1,117 @@ +"""风控模块共用的 SMTP 邮件发送服务。""" + +from __future__ import annotations + +import asyncio +import os +import smtplib +from collections.abc import Mapping +from email.message import EmailMessage +from email.utils import formatdate, make_msgid, parseaddr + + +class RiskSmtpConfigurationError(ValueError): + """SMTP 配置缺失或格式无效。""" + + +class RiskSmtpSendError(RuntimeError): + """SMTP 邮件发送失败。""" + + +class RiskSmtpMailService: + """使用 ``RISK_SMTP_*`` 配置发送风控邮件。""" + + def __init__(self, *, environment: Mapping[str, str] | None = None) -> None: + self.environment = environment if environment is not None else os.environ + + async def send( + self, + recipients: list[str], + subject: str, + content: str, + ) -> dict[str, object]: + return await asyncio.to_thread( + self._send_sync, + recipients, + subject, + content, + ) + + def _send_sync( + self, + recipients: list[str], + subject: str, + content: str, + ) -> dict[str, object]: + normalized = self._validated_recipients(recipients) + host = self.environment.get("RISK_SMTP_HOST", "").strip() + sender = self.environment.get("RISK_SMTP_SENDER", "").strip() + if not host or not sender: + raise RiskSmtpConfigurationError("SMTP 发件配置缺失") + if not _is_email(sender): + raise RiskSmtpConfigurationError("SMTP 发件地址格式无效") + if not subject.strip() or not content.strip(): + raise RiskSmtpConfigurationError("邮件主题和正文不能为空") + if "\r" in subject or "\n" in subject: + raise RiskSmtpConfigurationError("邮件主题包含非法换行") + + message = EmailMessage() + message["From"] = sender + message["To"] = ", ".join(normalized) + message["Subject"] = subject + message["Date"] = formatdate(localtime=True) + message["Message-ID"] = make_msgid(domain="risk-alert.local") + message.set_content(content) + + port = int(self.environment.get("RISK_SMTP_PORT", "465")) + use_ssl = _enabled(self.environment, "RISK_SMTP_USE_SSL", default=True) + timeout = float(self.environment.get("RISK_SMTP_TIMEOUT_SECONDS", "30")) + smtp_class = smtplib.SMTP_SSL if use_ssl else smtplib.SMTP + try: + with smtp_class(host, port, timeout=timeout) as connection: + username = self.environment.get("RISK_SMTP_USERNAME", "").strip() + password = self.environment.get("RISK_SMTP_PASSWORD", "") + if username: + connection.login(username, password) + connection.send_message(message) + except smtplib.SMTPAuthenticationError as error: + raise RiskSmtpSendError("SMTP 身份验证失败") from error + except smtplib.SMTPRecipientsRefused as error: + raise RiskSmtpSendError("收件人被邮件服务器拒绝") from error + except smtplib.SMTPSenderRefused as error: + raise RiskSmtpSendError("发件人被邮件服务器拒绝") from error + except TimeoutError as error: + raise RiskSmtpSendError("SMTP 连接超时") from error + except OSError as error: + raise RiskSmtpSendError(f"SMTP 连接失败：{error}") from error + except smtplib.SMTPException as error: + raise RiskSmtpSendError("SMTP 邮件发送失败") from error + return {"status": "sent", "recipient_count": len(normalized)} + + @staticmethod + def _validated_recipients(recipients: list[str]) -> list[str]: + normalized: list[str] = [] + seen: set[str] = set() + for value in recipients: + recipient = value.strip() + if not _is_email(recipient): + raise RiskSmtpConfigurationError("收件人地址格式无效") + lowered = recipient.lower() + if lowered not in seen: + normalized.append(recipient) + seen.add(lowered) + if not normalized: + raise RiskSmtpConfigurationError("收件人不能为空") + return normalized + + +def _enabled(environment: Mapping[str, str], name: str, *, default: bool = False) -> bool: + value = environment.get(name) + if value is None: + return default + return value.strip().lower() in {"1", "true", "yes", "on"} + + +def _is_email(value: str) -> bool: + _, parsed = parseaddr(value) + return bool(value and parsed == value and "@" in value and len(value) <= 254) diff --git a/app/worker/risk_scan_scheduler.py b/app/worker/risk_scan_scheduler.py index 9c5a4f1..4cd9e07 100644 --- a/app/worker/risk_scan_scheduler.py +++ b/app/worker/risk_scan_scheduler.py @@ -46,7 +46,7 @@ async def default_scan_executor() -> dict[str, int | str]: portal="worker", ) async with SessionFactory() as session: - return await RiskScanService(session).scan(context) + return await RiskScanService.from_settings(session).scan(context) async def default_audit_writer(status: str, detail: dict[str, Any]) -> None: diff --git a/docs/21-风控业务第二版迁移清单.md b/docs/21-风控业务第二版迁移清单.md index bd6803b..7619b12 100644 --- a/docs/21-风控业务第二版迁移清单.md +++ b/docs/21-风控业务第二版迁移清单.md @@ -539,12 +539,14 @@ tests/unit/api/test_risk_controller.py 已实现： - 站内通知记录创建。 -- 邮件通知记录创建，本阶段不外发 SMTP。 +- 邮件通知记录创建；迁移阶段当时尚未接入 SMTP。 - 通知主键和通知编号显式生成。 - 通知内容自动包含“预警编号：xxx”。 - 高风险批量通知记录。 - 通知分页复用 `RiskQueryService` 和 `RiskRepository`。 +> 2026-09-12 更新：本节“本阶段不外发 SMTP”只描述 R6.1 当时状态。当前手动扫描和定时扫描命中高风险后已恢复真实邮件发送，通知状态会回写 `已发送` 或 `发送失败`。当前高风险预警邮件只支持一个收件人。 + 验证结果：当前全部风控专项测试共 `58 passed`。 下一步：R6.2 将高风险通知创建接入扫描事务，并验证通知失败不破坏预警主事务。等待确认后执行。 @@ -556,7 +558,8 @@ tests/unit/api/test_risk_controller.py 已实现： - 扫描生成高风险预警后自动创建站内通知记录。 -- 可选邮件通知记录，默认不发送 SMTP。 +- 高风险邮件通知记录和真实 SMTP 外发，发送失败不回滚预警。 +- 高风险预警邮件当前只支持一个收件人，多个邮箱值只取第一个。 - 通知创建使用嵌套保存点隔离。 - 通知创建失败只记录日志，不回滚预警扫描主事务。 - 扫描结果增加 `notification_count`。 diff --git a/docs/风控业务演示文档/06-模块接口与字段映射.md b/docs/风控业务演示文档/06-模块接口与字段映射.md index 53d22a2..d63a104 100644 --- a/docs/风控业务演示文档/06-模块接口与字段映射.md +++ b/docs/风控业务演示文档/06-模块接口与字段映射.md @@ -79,6 +79,16 @@ 同一用户、同一路径、同一键的重复请求直接回放首次响应；同一键换了请求正文返回 `409 IDEMPOTENCY_CONFLICT`。证据上传靠"同一预警只能归档一次"的冲突保护去重。 +## 扫描通知与高风险邮件 + +- `POST /alerts/scan` 同时用于手动扫描，定时扫描复用同一 `RiskScanService`。 +- 高风险预警会创建站内通知和邮件通知。 +- 邮件发送成功时通知状态为 `已发送`；发送失败时为 `发送失败` 并返回失败原因。 +- 邮件发送失败不会改变扫描接口的成功状态，也不会回滚预警和站内通知。 +- `notification_count` 统计站内通知和邮件通知记录总数。 +- `notification_failure` 只表示通知记录创建失败，不表示 SMTP 邮件发送失败。 +- `RISK_ALERT_MAIL_RECIPIENTS` 当前只支持一个收件人，多个邮箱只取第一个。 + ## 日报和邮件 | 方法 | 路径 | 功能 | 权限 | diff --git a/docs/风控业务演示文档/12-日报通知与邮件规则.md b/docs/风控业务演示文档/12-日报通知与邮件规则.md index 8189d5f..116a3d9 100644 --- a/docs/风控业务演示文档/12-日报通知与邮件规则.md +++ b/docs/风控业务演示文档/12-日报通知与邮件规则.md @@ -48,7 +48,33 @@ - 数据库保留 `read_at`、`acknowledged_at` 基线字段，但当前未实现写入逻辑。 - 通知接口和前端不返回或展示阅读时间、确认时间。 -## 邮件开关 +### 高风险预警邮件 + +- 手动扫描和定时扫描共用同一套高风险预警邮件发送逻辑。 +- 高风险预警生成后，后端创建站内通知和邮件通知，并尝试调用真实 SMTP。 +- 邮件发送成功时，通知状态更新为 `已发送` 并写入发送时间。 +- 邮件发送失败时，通知状态更新为 `发送失败` 并写入安全失败原因。 +- 邮件发送失败不回滚预警，也不回滚站内通知。 +- 当前 `RISK_ALERT_MAIL_RECIPIENTS` 只支持一个收件人。即使配置多个邮箱，也只有第一个邮箱生效。 +- 高风险预警邮件收件人由后端配置，前端不提供收件人输入框。 + +高风险预警邮件配置： + +- `RISK_ALERT_MAIL_ENABLED` +- `RISK_ALERT_MAIL_DRY_RUN` +- `RISK_ALERT_MAIL_RECIPIENTS` + +高风险预警邮件复用以下 SMTP 配置： + +- `RISK_SMTP_HOST` +- `RISK_SMTP_PORT` +- `RISK_SMTP_USERNAME` +- `RISK_SMTP_PASSWORD` +- `RISK_SMTP_SENDER` +- `RISK_SMTP_USE_SSL` +- `RISK_SMTP_TIMEOUT_SECONDS` + +## 日报邮件开关 发送日报邮件需要 `risk:report:mail` 权限。 @@ -72,6 +98,7 @@ ## 收件人 +- 本节多个收件人规则只适用于日报邮件；高风险预警邮件当前只支持一个收件人。 - 支持多个收件人。 - 收件人去重并校验格式。 - 单次最多 10 个收件人。 diff --git a/docs/风控业务演示文档/15-模块验收与演示清单.md b/docs/风控业务演示文档/15-模块验收与演示清单.md index 641e0d4..93af2d0 100644 --- a/docs/风控业务演示文档/15-模块验收与演示清单.md +++ b/docs/风控业务演示文档/15-模块验收与演示清单.md @@ -40,6 +40,8 @@ - 定时扫描默认关闭，多个 Worker 同时运行时不重复执行。 - 定时扫描成功和失败写入系统审计。 - 扫描返回 `notification_failure` 时，页面必须区分成功与通知失败。 +- 手动扫描和定时扫描命中高风险后，高风险预警邮件按后端配置自动发送；当前只支持一个收件人。 +- 高风险邮件发送成功后通知状态为 `已发送`，失败时为 `发送失败`，且不影响预警落库。 - 重复扫描不重复生成同一交易和规则的预警。 - 多规则命中时正确合并。 @@ -69,6 +71,7 @@ - `data_truncated=true` 时不得把计数当作全量。 - 误报、处置结果和规则效果正确统计。 - 邮件接口要求 `risk:report:mail`，开关关闭时不发送真实邮件。 +- 日报邮件支持多个收件人；高风险预警邮件只支持一个收件人，两类邮件互不替代。 ## Agent 验收 @@ -88,7 +91,7 @@ 5. 展示行为分变化。 6. 使用奶龙风控智能助手查询和研判。 7. 生成并查看日报。 -8. 查看通知或邮件 dry-run 结果。 +8. 查看高风险预警邮件状态，并单独演示日报邮件 dry-run 或真实发送。 ## 通过标准 diff --git a/docs/风控业务演示文档/16-已知限制与待办.md b/docs/风控业务演示文档/16-已知限制与待办.md index e7f03c1..b3c2225 100644 --- a/docs/风控业务演示文档/16-已知限制与待办.md +++ b/docs/风控业务演示文档/16-已知限制与待办.md @@ -27,7 +27,14 @@ - Redis 不可用时，缓存和限流可能降级，不影响结构化业务主流程。 - Milvus 不可用时，语义记忆关闭，结构化查询继续运行。 - 模型服务不可用时，使用模板或规则化降级。 -- SMTP 未开启时，邮件接口返回禁用或 dry-run。 +- 日报邮件在 SMTP 未开启或 dry-run 时不会发送真实邮件。 +- 高风险预警邮件在 SMTP 发送失败时只更新通知状态，不回滚预警。 + +## 邮件通知限制 + +- `RISK_ALERT_MAIL_RECIPIENTS` 当前只支持一个收件人，多个邮箱配置只取第一个。 +- 高风险预警邮件和日报邮件是两条独立链路，收件人配置和开关互不替代。 +- 正式使用不能要求业务人员手工启动 `python -m app.worker.risk_scan_scheduler`。 ## 前端范围 @@ -52,6 +59,7 @@ - 主项目合并完成后对齐统一 RBAC 和客户数据范围。 - 修复公共 `data_scope` 最高权限跨资源扩散问题。 +- 将风控定时扫描接入主项目统一 Worker 或统一部署编排，并补充一个收件人以外的多收件人能力评估。 - 确认幂等回执与业务 Action 内部提交的事务边界。 - 私有前端最后再适配 `data/meta` 列表信封和写接口 `Idempotency-Key`。 - 根据合规要求确定会话保留期、脱敏和归档策略。 diff --git a/docs/风控业务演示文档/17-前端合并提示词与验收约束.md b/docs/风控业务演示文档/17-前端合并提示词与验收约束.md index 744bede..ca20e72 100644 --- a/docs/风控业务演示文档/17-前端合并提示词与验收约束.md +++ b/docs/风控业务演示文档/17-前端合并提示词与验收约束.md @@ -21,11 +21,35 @@ 后续模型开始前端合并前，必须先阅读本目录全部文档，并扫描上述代码和路由。 +## 前端合并前必须对齐的补充口径（2026-09-12） + +本节优先于后文中的概括性描述。前端合并时以本节为准。 + +### 风险概览口径 + +- 风险概览接口的 `total` 表示当前未闭环预警，状态范围为 `待处理` 和 `调查中`。 +- `total` 不是当天新增预警，不受 `created_at` 当天范围限制。 +- 前端指标名称应使用“未闭环预警”，不得继续显示为“今日预警”。 + +### 高风险预警邮件 + +- 手动扫描和定时扫描命中高风险预警后，后端都会创建站内通知并尝试发送邮件。 +- 高风险预警邮件由后端配置决定收件人，前端不提供收件人输入框。 +- 高风险预警邮件当前只支持一个收件人，前端不能按“多收件人”设计配置界面。 +- 邮件状态取值包括 `待发送`、`已发送`、`发送失败`、`未启用`。 +- 前端通知记录列表必须展示发送状态和失败原因，不能把“已创建通知记录”显示为“邮件已发送”。 +- 邮件发送失败不回滚预警，也不等于整次扫描失败。扫描成功后仅刷新通知记录即可看到真实邮件状态。 + +### 两类邮件必须区分 + +- 高风险预警邮件：手动扫描或定时扫描自动触发，收件人来自风控邮件通知配置。 +- 日报邮件：用户在高风险日报弹窗中填写收件人后手动发送，与预警扫描邮件互不替代。 + ## 二、前端功能范围 | 页面或区域 | 必需功能 | |---|---| -| 风险概览 | 总量、风险等级、待处理、超时、重点预警 | +| 风险概览 | 当前未闭环总量、风险等级、待处理、超时、重点预警；不得把总量显示为“今日预警” | | 预警队列 | 风险等级排序、筛选、每页 5 条、分页、弹窗详情 | | 预警详情 | 预警编号、状态、规则、证据、回执、人工处置 | | 证据区域 | 客户、产品、交易、资金、持仓、登录、预警、通知八类证据 | @@ -34,7 +58,7 @@ | 证据归档 | 图片和文档上传、归档状态、失败提示 | | 奶龙风控智能助手 | 对话、SSE 输出、工具调用展示、能力边界和免责声明 | | 日报 | 弹窗展示、流式生成、内容编辑、多邮箱发送 | -| 通知 | 通知记录、预警编号、发送状态 | +| 通知 | 通知记录、预警编号、站内或邮件渠道、发送状态、失败原因 | | 系统提示 | 政策解读、日报入口和预留模块 | ## 三、排版和交互约束 @@ -112,7 +136,9 @@ | 上传证据 | 证据上传成功 | | 手动扫描 | 预警扫描完成 | | 生成日报 | 日报生成完成 | -| 邮件发送 | 日报发送成功 | +| 日报邮件发送 | 日报邮件发送成功 | + +手动扫描成功只表示预警扫描完成。高风险预警邮件发送失败不会让扫描接口返回失败，前端应刷新通知记录，并通过通知的发送状态和失败原因展示真实结果。 ### 失败提示 diff --git a/docs/风控业务演示文档/18-当前项目完成进度.md b/docs/风控业务演示文档/18-当前项目完成进度.md index 4f8d256..9d16b80 100644 --- a/docs/风控业务演示文档/18-当前项目完成进度.md +++ b/docs/风控业务演示文档/18-当前项目完成进度.md @@ -8,10 +8,10 @@ | 项目 | 当前状态 | |---|---| -| 统计日期 | 2026-09-11 | +| 统计日期 | 2026-09-12 | | 当前分支 | `RM2_develop` | | 当前合并基线 | `origin/qyqy_develop` 主项目风控修复批次 | -| 代码状态 | 已完成主项目风控修复合并，全量测试通过 | +| 代码状态 | 已完成主项目风控修复合并，专项回归 54 项通过 | | 已推送分支 | `origin/RM2_develop` | | 已合并分支 | `origin/qyqy_develop` | | 私有前端 | `private_frontend/`，未提交、未推送 | @@ -63,11 +63,12 @@ ### 证据、通知和日报 - 图片和文档证据归档。 -- 高风险通知记录。 +- 高风险站内通知和邮件通知，邮件成功或失败状态可回查。 - 九段式日报。 - 历史未闭环完整统计。 - 日报流式生成。 -- 多邮箱校验和 dry-run 邮件发送。 +- 日报多邮箱校验和 dry-run 邮件发送。 +- 高风险预警邮件真实 SMTP 外发，当前只支持一个收件人。 ### 奶龙风控智能助手 @@ -124,14 +125,15 @@ ### 定时规则扫描 -状态：已完成。 +状态：功能已完成，主项目 Worker 接入待办。 - 新增独立 `RiskScanSchedulerWorker`，不在 Web 进程启动后台线程。 -- 扫描开关、周期、是否立即执行和重试次数由 `.env` 环境变量控制。 +- 扫描开关、间隔、轮询频率和重试次数由 `.env` 环境变量控制。 - 使用 MySQL 咨询锁防止多个 Worker 重复执行。 - 手动扫描和定时扫描共用同一个 `RiskScanService`。 - 扫描成功和失败写入系统审计。 - 默认关闭，配置开启后才执行。 +- 当前仍需独立启动 `python -m app.worker.risk_scan_scheduler`；主项目需要改为统一 Worker 注册或统一部署编排。 ### 主项目正式前端 @@ -170,11 +172,12 @@ ## 下一阶段建议 1. 将 `RM2_develop` 与最新 `qyqy_develop` 保持同步。 -2. 将定时规则扫描的环境变量配置纳入主项目部署配置。 -3. 按 `17-前端合并提示词与验收约束.md` 合并正式前端。 -4. 使用主项目真实登录、账号、角色和客户归属完成联调。 -5. 执行桌面端、移动端、权限、降级和完整业务链路验收。 -6. 主项目稳定后再实施对话历史、Redis 缓存和长期留存。 +2. 将定时规则扫描和高风险预警邮件配置纳入主项目部署配置。 +3. 将定时规则扫描接入主项目统一 Worker 或部署编排，取消业务人员手工启动独立进程。 +4. 按 `17-前端合并提示词与验收约束.md` 合并正式前端。 +5. 使用主项目真实登录、账号、角色和客户归属完成联调。 +6. 执行桌面端、移动端、权限、降级和完整业务链路验收。 +7. 主项目稳定后再实施对话历史、Redis 缓存和长期留存。 ## 完成判定 diff --git a/docs/风控业务演示文档/19-风控模块配置项清单.md b/docs/风控业务演示文档/19-风控模块配置项清单.md index 4a07738..cc0e374 100644 --- a/docs/风控业务演示文档/19-风控模块配置项清单.md +++ b/docs/风控业务演示文档/19-风控模块配置项清单.md @@ -20,9 +20,9 @@ |---|---:|---| | `RISK_SCAN_SCHEDULE_ENABLED` | `false` | 是否开启定时扫描 | | `RISK_SCAN_INTERVAL_MINUTES` | `5` | 扫描间隔，单位分钟 | -| `RISK_SCAN_RUN_IMMEDIATELY` | `false` | Worker 启动后是否立即执行 | +| `RISK_SCAN_RUN_IMMEDIATELY` | `false` | 仅保留兼容字段，当前不控制启动后的首次执行 | | `RISK_SCAN_RETRY_LIMIT` | `2` | 单轮失败后的重试次数 | -| `RISK_SCAN_POLL_SECONDS` | `30` | Worker 轮询配置和到期时间的间隔 | +| `RISK_SCAN_POLL_SECONDS` | `30` | Worker 检查扫描是否到期的频率，不是实际扫描间隔 | ### 证据归档 @@ -31,6 +31,14 @@ | `RISK_EVIDENCE_DIR` | `storage/risk_evidence` | 证据文件归档根目录 | | `RISK_EVIDENCE_MAX_FILE_SIZE_MB` | `10` | 单个证据文件最大大小，单位 MB | +### 高风险预警邮件 + +| 配置项 | 默认值 | 说明 | +|---|---:|---| +| `RISK_ALERT_MAIL_ENABLED` | `false` | 是否允许手动扫描和定时扫描发送高风险预警邮件 | +| `RISK_ALERT_MAIL_DRY_RUN` | `true` | 为真时只记录 dry-run 结果，不连接 SMTP | +| `RISK_ALERT_MAIL_RECIPIENTS` | 空 | 高风险预警邮件收件人；当前只支持一个，多个值只取第一个 | + ### 日报邮件 | 配置项 | 默认值 | 说明 | @@ -87,7 +95,9 @@ - 联调初期保持 `RISK_SCAN_SCHEDULE_ENABLED=false`。 - 确认扫描规则和演示数据后，再按需改为 `true`。 -- 邮件先保持 `RISK_DAILY_REPORT_MAIL_DRY_RUN=true`。 +- 高风险预警邮件先保持 `RISK_ALERT_MAIL_DRY_RUN=true`，确认预警落库和通知记录无误后再关闭 dry-run。 +- 当前高风险预警邮件只支持一个收件人，多邮箱配置只会有第一个生效。 +- 日报邮件先保持 `RISK_DAILY_REPORT_MAIL_DRY_RUN=true`。 - 正式发送前配置 SMTP，并将 `RISK_DAILY_REPORT_MAIL_ENABLED=true`。 - 证据目录应挂载到持久化存储，不能只保存在临时容器目录。 diff --git a/docs/风控业务演示文档/21-主项目合并后后端必改清单.md b/docs/风控业务演示文档/21-主项目合并后后端必改清单.md index 2b21b50..5b180d4 100644 --- a/docs/风控业务演示文档/21-主项目合并后后端必改清单.md +++ b/docs/风控业务演示文档/21-主项目合并后后端必改清单.md @@ -7,7 +7,7 @@ 公共底座、公共事务能力、公共数据范围、公共鉴权基座和公共 SSE 协商等问题不纳入本文， 由主项目统一修复和发布。 -## 本轮已完成 +## 已处理事项 ### 1. 证据查询时间口径统一 @@ -53,6 +53,23 @@ Agent 不能把截断结果表述成覆盖全部数据，也不能据此给出 `tools/grant_risk_permissions.py` 的文档字符串已改为实际文件名， 避免执行人员按错误路径操作。 +## 主项目仍需完成 + +### 6. 高风险预警邮件通知接入 + +- 主项目部署配置需要接入 `RISK_ALERT_MAIL_ENABLED`、`RISK_ALERT_MAIL_DRY_RUN` 和 `RISK_ALERT_MAIL_RECIPIENTS`。 +- 手动扫描和定时扫描必须继续共用同一套高风险邮件发送逻辑。 +- 高风险预警邮件当前只支持一个收件人，多个邮箱配置只有第一个生效。 +- 邮件发送成功写 `已发送`，失败写 `发送失败`；失败不得回滚预警和站内通知。 +- 日报邮件仍由用户在日报弹窗中填写多个收件人，和预警邮件配置相互独立。 + +### 7. 定时扫描接入主 Worker + +- 当前 `python -m app.worker` 不包含风控定时扫描，独立调度进程只适合本地联调。 +- 主项目需要提供通用后台任务注册入口，或在部署编排中统一拉起风控调度进程。 +- 正式使用不能要求业务人员额外手工执行 `python -m app.worker.risk_scan_scheduler`。 +- 无论采用哪种方式，都必须保留扫描互斥锁和幂等去重。 + ## 联调前需要执行的动作 以下内容属于环境准备或数据初始化，不是代码缺陷： @@ -68,7 +85,7 @@ Agent 不能把截断结果表述成覆盖全部数据，也不能据此给出 3. 执行 `python tools/publish_risk_agent_config.py`，确认奶龙风控智能助手的工具白名单和意图配置已发布。 4. 按主项目发布的迁移流程执行 Alembic 升级，确认 `trigger_rule_codes` 多值索引已生效。 5. 演示前准备足够的客户、交易、资金、持仓、登录和预警数据。 -6. 使用前确认日报邮件开关、SMTP 配置和收件人范围符合演示要求。 +6. 使用前确认高风险预警邮件开关、日报邮件开关、SMTP 配置和收件人范围符合演示要求。 ## 当前保留限制 diff --git a/tests/unit/api/test_risk_controller.py b/tests/unit/api/test_risk_controller.py index 995789e..840294c 100644 --- a/tests/unit/api/test_risk_controller.py +++ b/tests/unit/api/test_risk_controller.py @@ -56,6 +56,10 @@ class StubRiskScanService: def __init__(self, _session: Any) -> None: pass + @classmethod + def from_settings(cls, session: Any) -> "StubRiskScanService": + return cls(session) + async def scan(self, _context: RequestContext) -> dict[str, Any]: return {"message": "规则扫描完成", "created_count": 2, "high_risk_count": 1} diff --git a/tests/unit/service/test_risk_scan_alert_mail.py b/tests/unit/service/test_risk_scan_alert_mail.py new file mode 100644 index 0000000..15d7db1 --- /dev/null +++ b/tests/unit/service/test_risk_scan_alert_mail.py @@ -0,0 +1,125 @@ +"""验证高风险预警扫描会真实调用 SMTP，并正确回写通知状态。""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any, cast + +import pytest + +from app.service.risk_scan_service import HIGH_RISK, RiskScanService +from app.service.risk_smtp_mail_service import RiskSmtpSendError + + +class FakeNested: + async def __aenter__(self) -> FakeNested: + return self + + async def __aexit__(self, *_args: object) -> bool: + return False + + +class FakeSession: + def begin_nested(self) -> FakeNested: + return FakeNested() + + +class FakeNotificationService: + def __init__(self) -> None: + self.mail_records: list[Any] = [] + self.mail_marks: list[tuple[str, str | None]] = [] + + def create_in_app(self, _alert: Any, **_kwargs: Any) -> None: + return None + + def create_mail_record(self, alert: Any, **_kwargs: Any) -> Any: + notification = SimpleNamespace( + alert=alert, + send_status="待发送", + sent_at=None, + fail_reason=None, + ) + self.mail_records.append(notification) + return notification + + def mark_mail_sent(self, notification: Any) -> None: + self.mail_marks.append(("sent", None)) + notification.send_status = "已发送" + + def mark_mail_failed(self, notification: Any, reason: str) -> None: + self.mail_marks.append(("failed", reason)) + notification.send_status = "发送失败" + + +def alert() -> Any: + return SimpleNamespace( + alert_level=HIGH_RISK, + alert_no="ALTEST", + alert_type="大额快进快出", + evidence_summary="3 日内入金后大额赎回", + handler_id=9002, + ) + + +def service() -> RiskScanService: + instance = object.__new__(RiskScanService) + instance.session = cast(Any, FakeSession()) + instance.notification_enabled = True + instance.notification_email = "risk@example.com" + instance.mail_enabled = True + instance.mail_dry_run = False + instance.notification_service = cast(Any, FakeNotificationService()) + return instance + + +@pytest.mark.asyncio +async def test_high_risk_mail_is_sent_and_marked_successful() -> None: + class SmtpStub: + def __init__(self) -> None: + self.calls: list[tuple[list[str], str, str]] = [] + + async def send( + self, + recipients: list[str], + subject: str, + content: str, + ) -> dict[str, object]: + self.calls.append((recipients, subject, content)) + return {"status": "sent", "recipient_count": len(recipients)} + + instance = service() + notifier = cast(FakeNotificationService, instance.notification_service) + smtp = SmtpStub() + instance.smtp_mail_service = cast(Any, smtp) + + count, failure = await instance._create_notifications([alert()]) + + assert count == 2 + assert failure == "" + assert smtp.calls[0][0] == ["risk@example.com"] + assert "预警编号：ALTEST" in smtp.calls[0][2] + assert notifier.mail_marks == [("sent", None)] + assert notifier.mail_records[0].send_status == "已发送" + + +@pytest.mark.asyncio +async def test_high_risk_mail_failure_does_not_abort_scan() -> None: + class FailingSmtp: + async def send( + self, + _recipients: list[str], + _subject: str, + _content: str, + ) -> dict[str, object]: + raise RiskSmtpSendError("SMTP 身份验证失败") + + instance = service() + notifier = cast(FakeNotificationService, instance.notification_service) + instance.smtp_mail_service = cast(Any, FailingSmtp()) + + count, failure = await instance._create_notifications([alert()]) + + assert count == 2 + assert failure == "" + assert notifier.mail_marks == [("failed", "SMTP 身份验证失败")] + assert notifier.mail_records[0].send_status == "发送失败" diff --git a/tests/unit/service/test_risk_scan_notification.py b/tests/unit/service/test_risk_scan_notification.py index ec3a1fe..3bcb32b 100644 --- a/tests/unit/service/test_risk_scan_notification.py +++ b/tests/unit/service/test_risk_scan_notification.py @@ -38,6 +38,8 @@ class _FakeNotificationService: def __init__(self, *, fail: bool = False) -> None: self.fail = fail self.in_app: list[str] = [] + self.mail_records: list[Any] = [] + self.mail_marks: list[tuple[str, str | None]] = [] def create_in_app(self, alert: Any, **kwargs: Any) -> None: if self.fail: @@ -47,6 +49,22 @@ class _FakeNotificationService: def create_mail_record(self, alert: Any, **kwargs: Any) -> None: if self.fail: raise RuntimeError("邮件记录写库失败") + notification = SimpleNamespace( + alert=alert, + send_status="待发送", + sent_at=None, + fail_reason=None, + ) + self.mail_records.append(notification) + return notification + + def mark_mail_sent(self, notification: Any) -> None: + self.mail_marks.append(("sent", None)) + notification.send_status = "已发送" + + def mark_mail_failed(self, notification: Any, reason: str) -> None: + self.mail_marks.append(("failed", reason)) + notification.send_status = "发送失败" def _alert(level: str, alert_no: str = "ALTEST") -> Any: diff --git a/tests/unit/service/test_risk_smtp_mail_service.py b/tests/unit/service/test_risk_smtp_mail_service.py new file mode 100644 index 0000000..e2d0145 --- /dev/null +++ b/tests/unit/service/test_risk_smtp_mail_service.py @@ -0,0 +1,71 @@ +"""验证风控共用 SMTP 服务的基础发送行为。""" + +from __future__ import annotations + +from typing import Any + +import pytest + +from app.service import risk_smtp_mail_service +from app.service.risk_smtp_mail_service import ( + RiskSmtpConfigurationError, + RiskSmtpMailService, +) + + +class SmtpStub: + instances: list[SmtpStub] = [] + + def __init__(self, host: str, port: int, timeout: float) -> None: + self.host = host + self.port = port + self.timeout = timeout + self.login_args: tuple[str, str] | None = None + self.message: Any = None + self.__class__.instances.append(self) + + def __enter__(self) -> SmtpStub: + return self + + def __exit__(self, *_args: object) -> None: + return None + + def login(self, username: str, password: str) -> None: + self.login_args = (username, password) + + def send_message(self, message: Any) -> None: + self.message = message + + +@pytest.mark.asyncio +async def test_smtp_service_sends_message(monkeypatch: pytest.MonkeyPatch) -> None: + SmtpStub.instances = [] + monkeypatch.setattr(risk_smtp_mail_service.smtplib, "SMTP_SSL", SmtpStub) + + result = await RiskSmtpMailService( + environment={ + "RISK_SMTP_HOST": "smtp.example.com", + "RISK_SMTP_PORT": "465", + "RISK_SMTP_USERNAME": "sender@example.com", + "RISK_SMTP_PASSWORD": "secret", + "RISK_SMTP_SENDER": "sender@example.com", + "RISK_SMTP_USE_SSL": "true", + } + ).send(["risk@example.com"], "高风险预警", "请复核") + + assert result == {"status": "sent", "recipient_count": 1} + client = SmtpStub.instances[0] + assert (client.host, client.port) == ("smtp.example.com", 465) + assert client.login_args == ("sender@example.com", "secret") + assert client.message["To"] == "risk@example.com" + assert client.message["Subject"] == "高风险预警" + + +@pytest.mark.asyncio +async def test_smtp_service_rejects_missing_configuration() -> None: + with pytest.raises(RiskSmtpConfigurationError): + await RiskSmtpMailService(environment={}).send( + ["risk@example.com"], + "高风险预警", + "请复核", + )