diff --git a/app/api/controllers/offsite_fund.py b/app/api/controllers/offsite_fund.py index 4beac54..fb9d1ae 100644 --- a/app/api/controllers/offsite_fund.py +++ b/app/api/controllers/offsite_fund.py @@ -10,6 +10,8 @@ from app.api.dependencies.auth import build_request_context from app.api.dependencies.database import get_session from app.api.schemas.offsite_fund import ( OffsiteConfirmRequest, + OffsiteMailboxRecoveryRequest, + OffsiteMailDeletionRequest, OffsiteNl2SqlCorrectionRequest, OffsiteNotificationRequest, OffsiteNotificationSendRequest, @@ -49,6 +51,19 @@ async def get_mail( return await OffsiteFundService(session).get_mail(mail_id, context) +@router.post("/mails/{mail_id}/deletions") +async def delete_mail( + mail_id: str, + payload: OffsiteMailDeletionRequest, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, Any]: + """软删除邮件：运营列表隐藏，但保留原始文件、识别结果和审计链路。""" + return await OffsiteFundService(session).delete_mail( + mail_id, payload.operator_id, context + ) + + @router.get("/mails/{mail_id}/recognition-fields") async def get_mail_recognition_fields( mail_id: str, @@ -130,6 +145,18 @@ async def mailbox_status( return await OffsiteFundService(session).mailbox_status(context) +@router.post("/mailbox-status/recoveries") +async def recover_mailbox( + payload: OffsiteMailboxRecoveryRequest, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, Any]: + """解除收件游标阻塞，保留失败 UID 并由 Worker 重新处理。""" + return await OffsiteFundService(session).recover_mailbox( + payload.operator_id, context + ) + + @router.get("/attachments/{attachment_id}/file", response_model=None) async def open_attachment_file( attachment_id: str, diff --git a/app/api/schemas/offsite_fund.py b/app/api/schemas/offsite_fund.py index f524091..58dd682 100644 --- a/app/api/schemas/offsite_fund.py +++ b/app/api/schemas/offsite_fund.py @@ -49,6 +49,22 @@ class OffsiteRecognitionRetryRequest(BaseModel): operator_id: str = Field(min_length=1, max_length=64) +class OffsiteMailboxRecoveryRequest(BaseModel): + """收件游标解除阻塞请求。""" + + model_config = ConfigDict(extra="forbid") + + operator_id: str = Field(min_length=1, max_length=64) + + +class OffsiteMailDeletionRequest(BaseModel): + """邮件删除请求：只隐藏运营列表，不删除原始邮件和识别链路。""" + + model_config = ConfigDict(extra="forbid") + + operator_id: str = Field(min_length=1, max_length=64) + + class OffsiteRuleRecalculationRequest(BaseModel): """重新判定规则只接受操作人身份，数值一律取自已落库的识别与查询结果。""" diff --git a/app/service/health_service.py b/app/service/health_service.py index 322a78e..2592e0b 100644 --- a/app/service/health_service.py +++ b/app/service/health_service.py @@ -68,7 +68,7 @@ class HealthService: finally: if client is not None: try: - await client.aclose() + await client.close() except Exception: pass diff --git a/app/service/model_gateway.py b/app/service/model_gateway.py index 7a1156a..9618a0e 100644 --- a/app/service/model_gateway.py +++ b/app/service/model_gateway.py @@ -1,7 +1,7 @@ import os from collections.abc import Mapping from dataclasses import dataclass -from typing import Any, Protocol +from typing import Any, Protocol, cast import httpx from sqlalchemy import select @@ -148,14 +148,18 @@ class DatabaseModelGateway: async def generate(self, *, endpoint_code: str, prompt: str, timeout_ms: int) -> str: endpoint = await self._endpoint(endpoint_code) - adapter = OpenAICompatibleGateway({endpoint.endpoint_code: endpoint}) + adapter = OpenAICompatibleGateway( + {endpoint.endpoint_code: cast(EndpointSettings, endpoint)} + ) return await adapter.generate( endpoint_code=endpoint.endpoint_code, prompt=prompt, timeout_ms=timeout_ms ) async def embed(self, *, endpoint_code: str, text: str, timeout_ms: int) -> list[float]: endpoint = await self._endpoint(endpoint_code) - adapter = OpenAICompatibleGateway({endpoint.endpoint_code: endpoint}) + adapter = OpenAICompatibleGateway( + {endpoint.endpoint_code: cast(EndpointSettings, endpoint)} + ) return await adapter.embed( endpoint_code=endpoint.endpoint_code, text=text, timeout_ms=timeout_ms ) diff --git a/app/service/offsite_fund_rules.py b/app/service/offsite_fund_rules.py index 06be933..c571d91 100644 --- a/app/service/offsite_fund_rules.py +++ b/app/service/offsite_fund_rules.py @@ -1,5 +1,6 @@ """场外基金申购赎回确定性规则。""" +import re from dataclasses import dataclass from datetime import date, datetime from decimal import Decimal, InvalidOperation @@ -32,10 +33,13 @@ def decimal_from(value: object) -> Decimal | None: def normalize_amount_yuan(raw_value: object, unit: object) -> Decimal | None: - raw_unit = str(unit or "元").strip() - raw_text = str(raw_value or "").replace(",", "").strip() + raw_unit = str(unit or "元").strip().replace("人民币", "元") + raw_text = str(raw_value or "").replace(",", "").replace("，", "").strip() if raw_unit and raw_text.endswith(raw_unit): raw_text = raw_text[: -len(raw_unit)].strip() + # OCR 可能把币种前缀一起识别到金额字段，例如“人民币5,000.00元”。 + # 只接受开头的常见币种标记，避免从任意业务文本中误抽数字。 + raw_text = re.sub(r"^(?:人民币|RMB|CNY|¥|￥)\s*", "", raw_text, flags=re.IGNORECASE) amount = decimal_from(raw_text) if amount is None: return None diff --git a/app/service/offsite_fund_service.py b/app/service/offsite_fund_service.py index 09daee3..34d542d 100644 --- a/app/service/offsite_fund_service.py +++ b/app/service/offsite_fund_service.py @@ -7,11 +7,10 @@ from datetime import UTC, date, datetime from decimal import Decimal from email import policy from email.header import decode_header, make_header -from email.message import Message from email.parser import BytesParser from email.utils import parsedate_to_datetime from pathlib import Path -from typing import Literal, cast +from typing import Any, Literal, cast from zoneinfo import ZoneInfo from sqlalchemy import func, select, update @@ -143,7 +142,7 @@ class OffsiteFundService: denied = self._permission_error(context, ("offsite:read", "offsite:write")) if denied is not None: return denied - filters = [] + filters = [OffsiteFundMail.status != "deleted"] if sender: filters.append(OffsiteFundMail.sender == sender) if status: @@ -189,7 +188,10 @@ class OffsiteFundService: return denied async with self.session.begin(): mail = await self.session.scalar( - select(OffsiteFundMail).where(OffsiteFundMail.mail_id == mail_id) + select(OffsiteFundMail).where( + OffsiteFundMail.mail_id == mail_id, + OffsiteFundMail.status != "deleted", + ) ) if mail is None: return {"code": 404, "message": "邮件不存在", "data": {}} @@ -203,6 +205,49 @@ class OffsiteFundService: self._add_audit(context, "offsite.mail_viewed", {"mail_id": mail_id}) return {"code": 0, "message": "ok", "data": detail} + async def delete_mail( + self, mail_id: str, operator_id: str, context: RequestContext + ) -> dict[str, object]: + """软删除一封邮件，保留所有原始数据供审计和补偿使用。""" + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error + denied = self._permission_error(context, ("offsite:write", "offsite:confirm")) + if denied is not None: + return denied + + now = datetime.now(UTC).replace(tzinfo=None) + async with self.session.begin(): + mail = await self.session.scalar( + select(OffsiteFundMail) + .where(OffsiteFundMail.mail_id == mail_id) + .with_for_update() + ) + if mail is None: + return {"code": 404, "message": "邮件不存在", "data": {}} + if mail.status == "deleted": + return { + "code": 0, + "message": "邮件已删除", + "data": {"mail_id": mail_id, "status": "deleted"}, + } + previous_status = mail.status + mail.status = "deleted" + mail.updated_at = now + self._add_audit( + context, + "offsite.mail_deleted", + { + "mail_id": mail_id, + "previous_status": previous_status, + }, + ) + return { + "code": 0, + "message": "邮件已删除", + "data": {"mail_id": mail_id, "status": "deleted"}, + } + async def mail_recognition_fields( self, mail_id: str, context: RequestContext ) -> dict[str, object]: @@ -518,16 +563,24 @@ class OffsiteFundService: correction: OffsiteFieldCorrection | None = None, ) -> dict[str, object]: expected = NL2SQL_DOCUMENT_FIELDS.get(document.document_type, ()) - values = cls._nl2sql_original_fields(rule_results) + latest_rule_results: dict[str, OffsiteRuleResult] = {} + for result in rule_results: + latest_rule_results[result.rule_code] = result + display_rule_results = list(latest_rule_results.values()) + values = cls._nl2sql_original_fields(display_rule_results) + latest_query_records: dict[str, OffsiteQueryRecord] = {} + for record in query_records: + latest_query_records[record.rule_code] = record + display_query_records = list(latest_query_records.values()) correction_fields = ( dict(correction.corrected_fields) if correction is not None else {} ) query_failed = any( - record.status == "query_failed" for record in query_records + record.status == "query_failed" for record in display_query_records ) or any( isinstance(result.calculation, Mapping) and result.calculation.get("原因") == NL2SQL_BLOCKED_REASON - for result in rule_results + for result in display_rule_results ) fields: dict[str, str | None] = {} field_status: dict[str, str] = {} @@ -573,7 +626,7 @@ class OffsiteFundService: "queried_at": cls._iso_datetime(record.created_at), "error_message": record.error_message, } - for record in query_records + for record in display_query_records ], "updated_at": cls._iso_datetime(document.updated_at), } @@ -644,6 +697,9 @@ class OffsiteFundService: 需要换一批查询数据时，先调用核对触发接口刷新查询记录，再调用本接口重新判定。 查询缺失或失败的规则按"无法判断"处理并保留原因，不做任何猜测。 """ + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error denied = self._permission_error( context, ("offsite:write", "offsite:confirm", "offsite:nl2sql") ) @@ -701,6 +757,9 @@ class OffsiteFundService: 只新增修正记录，不覆盖附件上的 Agent 原始识别值；单据的标准化展示字段 跟随修正后的有效值刷新，重新判定再按有效值重跑计算与核对。 """ + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error denied = self._permission_error(context, ("offsite:write",)) if denied is not None: return denied @@ -788,6 +847,9 @@ class OffsiteFundService: 只新增修正记录，不改写查询记录与查询摘要；重新判定时修正值优先于查询原值。 """ + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error denied = self._permission_error(context, ("offsite:write",)) if denied is not None: return denied @@ -1081,6 +1143,75 @@ class OffsiteFundService: }, } + async def recover_mailbox( + self, operator_id: str, context: RequestContext + ) -> dict[str, object]: + """解除收件游标阻塞，但不跳过失败 UID。""" + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error + denied = self._permission_error(context, ("offsite:write", "offsite:confirm")) + if denied is not None: + return denied + + settings = get_settings() + now = datetime.now(UTC).replace(tzinfo=None) + async with self.session.begin(): + cursor = await self.session.scalar( + select(OffsiteMailCursor) + .where( + OffsiteMailCursor.mailbox == settings.offsite_mailbox, + OffsiteMailCursor.folder == "INBOX", + ) + .with_for_update() + ) + if cursor is None: + return {"code": 404, "message": "收件游标尚未初始化", "data": {}} + if cursor.status != "blocked": + return { + "code": 0, + "message": "收件游标当前未阻塞", + "data": { + "status": cursor.status, + "last_uid": cursor.last_uid, + }, + } + + previous = { + "blocked_uid": cursor.blocked_uid, + "blocked_message_id": cursor.blocked_message_id, + "retry_count": cursor.retry_count, + "last_error": cursor.last_error, + } + cursor.status = "idle" + cursor.retry_count = 0 + cursor.last_error = None + cursor.next_retry_at = None + cursor.blocked_uid = None + cursor.blocked_message_id = None + cursor.lease_id = None + cursor.lease_until = None + cursor.updated_at = now + self._add_audit( + context, + "offsite.mail_cursor_recovered", + { + "last_uid": cursor.last_uid, + "previous": previous, + }, + ) + + return { + "code": 0, + "message": "收件游标已解除阻塞，Worker 将从失败邮件继续重试", + "data": { + "status": "idle", + "last_uid": cursor.last_uid, + "retry_count": 0, + "retry_uid": previous["blocked_uid"], + }, + } + async def _mail_attachments( self, mail_ids: Sequence[str] ) -> dict[str, tuple[OffsiteFundAttachment, ...]]: @@ -1190,7 +1321,7 @@ class OffsiteFundService: @staticmethod def _parse_eml(path: str) -> dict[str, object]: - empty = { + empty: dict[str, object] = { "subject": None, "sent_at": None, "has_body": False, @@ -1198,7 +1329,9 @@ class OffsiteFundService: "body_html": None, } try: - message = BytesParser(policy=policy.default).parsebytes(Path(path).read_bytes()) + message: Any = BytesParser(policy=cast(Any, policy.default)).parsebytes( + Path(path).read_bytes() + ) except (OSError, ValueError): return empty subject = OffsiteFundService._decode_header(message.get("Subject")) @@ -1228,7 +1361,7 @@ class OffsiteFundService: return value @staticmethod - def _mail_body(message: Message) -> tuple[str | None, str | None]: + def _mail_body(message: Any) -> tuple[str | None, str | None]: text_body: str | None = None html_body: str | None = None parts = message.walk() if message.is_multipart() else (message,) @@ -1349,6 +1482,9 @@ class OffsiteFundService: async def retry_document_recognition( self, task_id: str, operator_id: str, context: RequestContext ) -> dict[str, object]: + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error denied = self._permission_error(context, ("offsite:write", "offsite:confirm")) if denied is not None: return denied @@ -1690,6 +1826,9 @@ class OffsiteFundService: self, task_id: str, decision: OperationDecision, operator_id: str, context: RequestContext, ) -> dict[str, object]: + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error denied = self._permission_error(context, ("offsite:confirm", "offsite:write")) if denied is not None: return denied @@ -1789,6 +1928,9 @@ class OffsiteFundService: self, task_id: str, operator_id: str, manual_confirmed: bool, context: RequestContext, ) -> dict[str, object]: + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error denied = self._permission_error( context, ("offsite:nl2sql", "offsite:write", "financial:nl2sql:read") ) @@ -1898,6 +2040,9 @@ class OffsiteFundService: self, task_id: str, notification_type: str, operator_id: str, context: RequestContext, ) -> dict[str, object]: + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error denied = self._permission_error(context, ("offsite:notify", "offsite:write")) if denied is not None: return denied @@ -1949,6 +2094,9 @@ class OffsiteFundService: final_content: str | None, context: RequestContext, ) -> dict[str, object]: + operator_error = self._operator_error(operator_id, context) + if operator_error is not None: + return operator_error denied = self._permission_error(context, ("offsite:notify", "offsite:write")) if denied is not None: return denied @@ -2264,6 +2412,8 @@ class OffsiteFundService: ).with_for_update()) if mail is None: return + if mail.status == "deleted": + return documents = (await self.session.execute(select(OffsiteFundDocument).where( OffsiteFundDocument.mail_id == mail_id ))).scalars().all() @@ -2444,6 +2594,15 @@ class OffsiteFundService: return {"code": 403, "message": "缺少场外基金操作权限", "data": {}} return None + @staticmethod + def _operator_error( + operator_id: str, context: RequestContext + ) -> dict[str, object] | None: + """操作人只能来自 JWT 身份，禁止客户端伪造其它用户编号。""" + if str(operator_id).strip() != str(context.user_id).strip(): + return {"code": 403, "message": "操作人身份与访问令牌不一致", "data": {}} + return None + def _add_audit( self, context: RequestContext, action_type: str, detail: dict[str, object] ) -> None: diff --git a/app/worker/runtime.py b/app/worker/runtime.py index 40eb0d0..0831aaa 100644 --- a/app/worker/runtime.py +++ b/app/worker/runtime.py @@ -10,7 +10,7 @@ from uuid import uuid4 from sqlalchemy import select, update from app.core.config import Settings, get_settings -from app.core.contracts import AgentRequest, AgentResult, RequestContext +from app.core.contracts import AgentRequest, AgentRequestMetadata, AgentResult, RequestContext from app.core.errors import AgentError, RecoverableAgentError, RunLeaseLostError from app.infrastructure.db import SessionFactory from app.model.audit import InteractionAudit @@ -425,7 +425,9 @@ class WorkerRuntime: request = AgentRequest( agent_type=run.agent_type, message=message.content, session_id=run.session_id, idempotency_key=idem.idempotency_key, - metadata=event.payload.get("metadata", {}) if event else {}, + metadata=AgentRequestMetadata.model_validate( + event.payload.get("metadata", {}) if event else {} + ), ) identity = RequestContext(user_id=str(run.user_id), trace_id=run.trace_id) # Re-check account and permissions at execution time, including delayed jobs. diff --git a/docs/12-基金行情数据底座接入开发计划.md b/docs/12-基金行情数据底座接入开发计划.md index 60b1883..2f49179 100644 --- a/docs/12-基金行情数据底座接入开发计划.md +++ b/docs/12-基金行情数据底座接入开发计划.md @@ -35,7 +35,7 @@ | 基金名称缓存 | 行情缓存适配器 | 允许 Redis，不使用进程全局作为唯一缓存 | | 历史净值和收益率合并 | `app/service/fund_quote_service.py` | 统一 DTO、Decimal、降级标识 | | 中文字典输出 | Controller/Agent 展示层 | 内部 DTO 使用英文稳定字段 | -| 固定奶龙基金代码 | 配置或代码白名单 | 先保留白名单，后续配置中心化 | +| 固定南方基金代码 | 配置或代码白名单 | 先保留白名单，后续配置中心化 | ## 四、阶段计划 diff --git a/hq.py b/hq.py index 5a15a1b..d7a4ccb 100644 --- a/hq.py +++ b/hq.py @@ -1,5 +1,5 @@ # -*- coding: utf-8 -*- -"""奶龙基金指定产品行情模块。""" +"""南方基金指定产品行情模块。""" from __future__ import annotations import re import time @@ -41,7 +41,7 @@ def is_market_trading_time(now: datetime | None = None) -> bool: def get_southern_fund_market(target_date: str | None = None, limit: int | None = None, fund_type: str | None = None) -> list[dict[str, Any]]: - """获取指定奶龙基金的完整行情表。 + """获取指定南方基金的完整行情表。 每次调用刷新整张表的实时行情；历史收益同一日期只请求一次并保存在进程缓存。 limit 不传时返回 SOUTHERN_FUND_CODES 中的全部产品。 @@ -62,7 +62,7 @@ def get_southern_fund_market(target_date: str | None = None, limit: int | None = old = history.get(code, {}) live = quotes.get(code, {}) rows.append({ - "基金代码": code, "基金名称": names.get(code, f"奶龙基金 {code}"), + "基金代码": code, "基金名称": names.get(code, f"南方基金 {code}"), "基金类型": FUND_TYPE_BY_CODE.get(code, "未分类"), "基金净值": live.get("基金净值") or old.get("基金净值"), "日期": live.get("日期") or old.get("日期"), @@ -84,11 +84,11 @@ def _get_names(codes: list[str]) -> dict[str, str]: response = httpx.get(DETAIL_API.format(code=code), headers=HEADERS, timeout=REQUEST_TIMEOUT) response.raise_for_status() match = re.search(r"var\s+fS_name\s*=\s*[\"']([^\"']+)", response.text) - _name_cache[code] = match.group(1).strip() if match else f"奶龙基金 {code}" + _name_cache[code] = match.group(1).strip() if match else f"南方基金 {code}" except (httpx.HTTPError, UnicodeError) as exc: logger.warning("基金名称接口失败 code=%s error=%s", code, type(exc).__name__) - _name_cache[code] = f"奶龙基金 {code}" - return {code: _name_cache.get(code, f"奶龙基金 {code}") for code in codes} + _name_cache[code] = f"南方基金 {code}" + return {code: _name_cache.get(code, f"南方基金 {code}") for code in codes} def _get_quotes(codes: list[str]) -> dict[str, dict[str, str | None]]: diff --git a/tests/integration/test_offsite_fund_api.py b/tests/integration/test_offsite_fund_api.py index 3b02ee6..3cdccc2 100644 --- a/tests/integration/test_offsite_fund_api.py +++ b/tests/integration/test_offsite_fund_api.py @@ -132,7 +132,7 @@ def test_offsite_recognized_mail_persists_workflow_and_notification() -> None: confirm = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", - json={"decision": "确认正常", "operator_id": "operator-001"}, + json={"decision": "确认正常", "operator_id": "1"}, ) assert confirm.status_code == 200 assert confirm.json()["code"] == 0 @@ -146,7 +146,7 @@ def test_offsite_recognized_mail_persists_workflow_and_notification() -> None: notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", - json={"notification_type": "settlement", "operator_id": "operator-001"}, + json={"notification_type": "settlement", "operator_id": "1"}, ) assert notice.status_code == 200 assert notice.json()["data"]["notification_id"] @@ -239,7 +239,7 @@ def test_offsite_nl2sql_success_completes_subscription_plan( triggered = client.post( f"/api/tasks/{task_id}/trigger-agent-nl2sql", - json={"operator_id": "operator-001", "manual_confirmed": True}, + json={"operator_id": "1", "manual_confirmed": True}, ) assert triggered.status_code == 200 assert triggered.json()["data"]["status"] == "planned" @@ -250,7 +250,7 @@ def test_offsite_nl2sql_success_completes_subscription_plan( confirmed = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", - json={"decision": "确认正常", "operator_id": "operator-001"}, + json={"decision": "确认正常", "operator_id": "1"}, ) assert confirmed.status_code == 200 assert confirmed.json()["code"] == 0 @@ -283,7 +283,7 @@ def test_offsite_low_confidence_document_goes_to_recognition_exception() -> None assert document["status"] == "recognition_exception" blocked = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", - json={"decision": "确认正常", "operator_id": "operator-001"}, + json={"decision": "确认正常", "operator_id": "1"}, ) assert blocked.status_code == 200 assert blocked.json()["code"] == 422 @@ -361,7 +361,7 @@ def test_offsite_recognition_retry_recovers_document_without_overwriting_attachm retried = client.post( f"/api/v1/offsite-fund/documents/{task_id}/recognition-retries", - json={"operator_id": "operator-001"}, + json={"operator_id": "1"}, ) assert retried.status_code == 200 assert recognized_fields @@ -440,13 +440,13 @@ def test_offsite_mail_return_send_updates_notification_without_external_call( confirmed = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", - json={"decision": "确认正常", "operator_id": "operator-001"}, + json={"decision": "确认正常", "operator_id": "1"}, ) assert confirmed.status_code == 200 notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", - json={"notification_type": "mail_return", "operator_id": "operator-001"}, + json={"notification_type": "mail_return", "operator_id": "1"}, ) assert notice.status_code == 200 notification_id = int(notice.json()["data"]["notification_id"]) @@ -454,7 +454,7 @@ def test_offsite_mail_return_send_updates_notification_without_external_call( sent = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", json={ - "operator_id": "operator-001", + "operator_id": "1", "operator_confirmed": True, "final_content": "运营确认后的回复正文", }, diff --git a/tests/integration/test_offsite_nl2sql_fields.py b/tests/integration/test_offsite_nl2sql_fields.py index 71eee80..20954ea 100644 --- a/tests/integration/test_offsite_nl2sql_fields.py +++ b/tests/integration/test_offsite_nl2sql_fields.py @@ -156,6 +156,43 @@ async def _seed_failed_queries(task_id: str) -> None: )) +async def _seed_latest_success_after_failure(task_id: str) -> None: + await _seed_failed_queries(task_id) + now = datetime.now(UTC).replace(tzinfo=None) + async with SessionFactory() as session, session.begin(): + session.add(OffsiteQueryRecord( + task_id=task_id, + rule_code="subscription_holding_ratio", + natural_language_request="基金代码为15911，查询基金最新总份额、最新净值和申请前持有份额", + script_path="nl2sql_yc.py", + result_summary={ + "status": "success", + "data": {"total": 1, "rows": [{ + "nav": "1.250000", + "total_fund_shares": "10000000000.0000", + "total_quantity": "100000000.0000", + }]}, + }, + status="success", + error_message=None, + created_at=now, + )) + rule_result = await session.scalar(select(OffsiteRuleResult).where( + OffsiteRuleResult.task_id == task_id, + OffsiteRuleResult.rule_code == "subscription_holding_ratio", + )) + assert rule_result is not None + rule_result.result = "正常" + rule_result.document_value = {"申购金额元": "200000000.00"} + rule_result.database_value = { + "最新净值": "1.250000", + "基金最新总份额": "10000000000.0000", + "申请前持有份额": "100000000.0000", + } + rule_result.calculation = {"申购后持有比例": "0.026"} + rule_result.created_at = now + + async def _count_audit(action_type: str) -> int: async with SessionFactory() as session: rows = await session.execute(select(InteractionAudit).where( @@ -244,6 +281,31 @@ async def test_nl2sql_fields_mark_query_failed_when_query_blocked() -> None: TRACE_ID = "" +@pytest.mark.integration +async def test_nl2sql_fields_show_only_latest_attempt_per_rule() -> None: + """重试成功后，页面不能继续展示同一规则的历史失败状态。""" + global TRACE_ID + TRACE_ID = f"trace-nl2sql-latest-attempt-{uuid4()}" + task_id = "" + try: + task_id = await _seed_document("subscription") + await _seed_latest_success_after_failure(task_id) + _install_context(("operator",), ("offsite:read",)) + response = await _get_fields(task_id) + + assert response.status_code == 200 + data = response.json()["data"] + assert len(data["queries"]) == 1 + assert data["queries"][0]["rule_code"] == "subscription_holding_ratio" + assert data["queries"][0]["status"] == "success" + assert data["queries"][0]["row_count"] == 1 + assert data["field_status"]["最新净值"] == "success" + finally: + await _cleanup(task_id) + app.dependency_overrides.clear() + TRACE_ID = "" + + @pytest.mark.integration async def test_nl2sql_fields_mark_pending_before_verification() -> None: """尚未触发核对时字段为空，必须标为 pending，不能伪装成查询失败。""" diff --git a/tests/integration/test_offsite_notification_send.py b/tests/integration/test_offsite_notification_send.py index 7152d1a..8d9d195 100644 --- a/tests/integration/test_offsite_notification_send.py +++ b/tests/integration/test_offsite_notification_send.py @@ -105,17 +105,17 @@ def test_successful_notification_send_is_idempotent( _confirm_and_create_notice(client, task_id) notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", - json={"notification_type": "normal_return", "operator_id": "operator-001"}, + json={"notification_type": "normal_return", "operator_id": "1"}, ) notification_id = int(notice.json()["data"]["notification_id"]) first = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", - json={"operator_id": "operator-001", "operator_confirmed": True}, + json={"operator_id": "1", "operator_confirmed": True}, ) second = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", - json={"operator_id": "operator-001", "operator_confirmed": True}, + json={"operator_id": "1", "operator_confirmed": True}, ) recalculate = client.post( "/api/v1/offsite-fund/settlement-statistics/recalculate", @@ -175,12 +175,12 @@ def test_failed_notification_send_persists_failure_and_retry_count( _confirm_and_create_notice(client, task_id) notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", - json={"notification_type": "mail_return", "operator_id": "operator-001"}, + json={"notification_type": "mail_return", "operator_id": "1"}, ) notification_id = int(notice.json()["data"]["notification_id"]) sent = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", - json={"operator_id": "operator-001", "operator_confirmed": True}, + json={"operator_id": "1", "operator_confirmed": True}, ) assert sent.json()["data"]["status"] == "发送失败" @@ -260,14 +260,14 @@ def test_mixed_mail_separates_normal_and_exception_returns( ): confirmed = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", - json={"decision": decision, "operator_id": "operator-001"}, + json={"decision": decision, "operator_id": "1"}, ) assert confirmed.status_code == 200 notice = client.post( f"/api/v1/offsite-fund/documents/{task_id}/notifications", json={ "notification_type": notification_type, - "operator_id": "operator-001", + "operator_id": "1", }, ) assert notice.status_code == 200 @@ -275,7 +275,7 @@ def test_mixed_mail_separates_normal_and_exception_returns( notification_ids.append(notification_id) sent = client.post( f"/api/v1/offsite-fund/notifications/{notification_id}/send", - json={"operator_id": "operator-001", "operator_confirmed": True}, + json={"operator_id": "1", "operator_confirmed": True}, ) assert sent.json()["data"]["status"] == "发送成功" @@ -296,7 +296,7 @@ def test_mixed_mail_separates_normal_and_exception_returns( def _confirm_and_create_notice(client: TestClient, task_id: str) -> None: confirmed = client.post( f"/api/v1/offsite-fund/documents/{task_id}/confirmations", - json={"decision": "确认正常", "operator_id": "operator-001"}, + json={"decision": "确认正常", "operator_id": "1"}, ) assert confirmed.status_code == 200 diff --git a/tests/integration/test_promotion_material_api.py b/tests/integration/test_promotion_material_api.py index c318d8b..d6a7dc0 100644 --- a/tests/integration/test_promotion_material_api.py +++ b/tests/integration/test_promotion_material_api.py @@ -337,7 +337,7 @@ def _inputs_payload() -> dict[str, object]: }, "manager_info": { "manager_name": "张三", - "management_company": "奶龙基金管理有限公司", + "management_company": "南方基金管理有限公司", "registration_code": "P10000001", "employment_years": "10年", "investment_management_experience": "8年公募基金投资管理经验", diff --git a/tests/unit/service/test_offsite_fund_rules.py b/tests/unit/service/test_offsite_fund_rules.py index d5546ad..1162b42 100644 --- a/tests/unit/service/test_offsite_fund_rules.py +++ b/tests/unit/service/test_offsite_fund_rules.py @@ -59,6 +59,8 @@ def test_normalize_amount_keeps_original_unit_semantics() -> None: assert normalize_amount_yuan("2.50", "万元") == Decimal("25000.00") assert normalize_amount_yuan("2.50", "元") == Decimal("2.50") assert normalize_amount_yuan("20,000.00万元", "万元") == Decimal("200000000.00") + assert normalize_amount_yuan("人民币5,000.00元", "元") == Decimal("5000.00") + assert normalize_amount_yuan("CNY 5,000.00", "元") == Decimal("5000.00") assert normalize_amount_yuan("2.50", "美元") is None diff --git a/tests/unit/service/test_offsite_smtp_adapter.py b/tests/unit/service/test_offsite_smtp_adapter.py index 7dd07ee..4b2dcd8 100644 --- a/tests/unit/service/test_offsite_smtp_adapter.py +++ b/tests/unit/service/test_offsite_smtp_adapter.py @@ -167,6 +167,8 @@ def _settings(**updates: object) -> Settings: "offsite_smtp_username": "15273589815@163.com", "offsite_smtp_password": "test-auth-code", "offsite_smtp_sender": "15273589815@163.com", + "offsite_smtp_enabled": False, + "offsite_smtp_dry_run": True, } values.update(updates) return Settings(**values) diff --git a/tests/unit/service/test_promotion_material.py b/tests/unit/service/test_promotion_material.py index 93bbc6e..09e2fc0 100644 --- a/tests/unit/service/test_promotion_material.py +++ b/tests/unit/service/test_promotion_material.py @@ -18,7 +18,7 @@ def _valid_inputs() -> dict[str, object]: }, "manager_info": { "manager_name": "张三", - "management_company": "奶龙基金管理有限公司", + "management_company": "南方基金管理有限公司", "registration_code": "P10000001", }, "team_info": {"team_description": "具备完整投研分工"}, diff --git a/tests/unit/service/test_suitability_service.py b/tests/unit/service/test_suitability_service.py index 9fdc113..3b24448 100644 --- a/tests/unit/service/test_suitability_service.py +++ b/tests/unit/service/test_suitability_service.py @@ -1,65 +1,115 @@ from datetime import UTC, datetime, timedelta import pytest +from pydantic import ValidationError from app.core.contracts import RequestContext from app.service.suitability_service import ( - SuitabilityRequest, + RiskAuthorityProfile, SuitabilityService, SuitabilityToolInput, suitability_tool_handler, ) -def request(**overrides: object) -> SuitabilityRequest: +def _request(**overrides: object) -> SuitabilityToolInput: values: dict[str, object] = { - "customer_risk_level": 3, + "customer_id": "7", "product_risk_level": 3, "product_requires_disclosure": True, } values.update(overrides) - return SuitabilityRequest.model_validate(values) + return SuitabilityToolInput.model_validate(values) +def _context() -> RequestContext: + return RequestContext(user_id="7", trace_id="trace-suitability") + + +def _profile( + customer_risk_level: int | None = 3, + *, + valid_until: datetime | None = None, + authority_reason: str = "AUTHORITY_OK", +) -> RiskAuthorityProfile: + return RiskAuthorityProfile( + customer_id="7", + customer_risk_level=customer_risk_level, + valid_until=valid_until or datetime.now(UTC) + timedelta(days=30), + authority_reason=authority_reason, # type: ignore[arg-type] + ) + + +@pytest.mark.asyncio @pytest.mark.parametrize( ("customer", "product", "allowed"), [(1, 1, True), (1, 2, False), (3, 2, True), (5, 5, True)], ) -def test_risk_level_boundary(customer: int, product: int, allowed: bool) -> None: - decision = SuitabilityService().evaluate( - request(customer_risk_level=customer, product_risk_level=product) +async def test_risk_level_boundary( + monkeypatch: pytest.MonkeyPatch, + customer: int, + product: int, + allowed: bool, +) -> None: + service = SuitabilityService() + + async def load_profile(_customer_id: str) -> RiskAuthorityProfile: + return _profile(customer) + + monkeypatch.setattr(service, "_load_authority_profile", load_profile) + decision = await service.evaluate( + _request(product_risk_level=product), _context() ) assert decision.allowed is allowed assert decision.reason_code == ("SUITABLE" if allowed else "RISK_LEVEL_MISMATCH") -def test_expired_assessment_is_denied() -> None: +@pytest.mark.asyncio +async def test_expired_assessment_is_denied(monkeypatch: pytest.MonkeyPatch) -> None: now = datetime(2026, 9, 9, tzinfo=UTC) - decision = SuitabilityService().evaluate( - request(assessment_expires_at=now - timedelta(seconds=1)), now=now - ) + service = SuitabilityService() + + async def load_profile(_customer_id: str) -> RiskAuthorityProfile: + return _profile(valid_until=now - timedelta(seconds=1)) + + monkeypatch.setattr(service, "_load_authority_profile", load_profile) + decision = await service.evaluate(_request(), _context(), now=now) assert decision.allowed is False assert decision.reason_code == "ASSESSMENT_EXPIRED" assert decision.requires_recording is True -def test_disclosure_requires_confirmation_and_recording() -> None: - decision = SuitabilityService().evaluate( - request(product_requires_disclosure=True, requires_confirmation=False) - ) - assert decision.allowed is True - assert decision.required_disclosure is True - assert decision.requires_confirmation is True - assert decision.requires_recording is True +@pytest.mark.asyncio +async def test_missing_assessment_is_denied(monkeypatch: pytest.MonkeyPatch) -> None: + service = SuitabilityService() + + async def load_profile(_customer_id: str) -> RiskAuthorityProfile: + return _profile(None, authority_reason="ASSESSMENT_MISSING") + + monkeypatch.setattr(service, "_load_authority_profile", load_profile) + decision = await service.evaluate(_request(), _context()) + assert decision.allowed is False + assert decision.reason_code == "ASSESSMENT_MISSING" -def test_missing_timezone_is_rejected() -> None: - with pytest.raises(ValueError, match="timezone"): - request(assessment_expires_at=datetime(2026, 9, 9)) +def test_tool_input_rejects_client_supplied_authority_fields() -> None: + with pytest.raises(ValidationError): + SuitabilityToolInput( + customer_id="7", + customer_risk_level=3, + product_risk_level=3, + ) + + +def test_tool_input_rejects_invalid_customer_id() -> None: + with pytest.raises(ValidationError): + SuitabilityToolInput(customer_id="customer-7", product_risk_level=3) @pytest.mark.asyncio -async def test_tool_input_uses_same_rules(monkeypatch: pytest.MonkeyPatch) -> None: +async def test_tool_handler_uses_current_service_entrypoint( + monkeypatch: pytest.MonkeyPatch, +) -> None: class DummySession: def add(self, _item: object) -> None: pass @@ -73,12 +123,19 @@ async def test_tool_input_uses_same_rules(monkeypatch: pytest.MonkeyPatch) -> No def begin(self) -> "DummySession": return self + service = SuitabilityService(session_factory=lambda: DummySession()) + + async def load_profile(_customer_id: str) -> RiskAuthorityProfile: + return _profile(1) + + monkeypatch.setattr(service, "_load_authority_profile", load_profile) monkeypatch.setattr( - "app.service.suitability_service.SessionFactory", lambda: DummySession() + "app.service.suitability_service.SuitabilityService", + lambda: service, ) - context = RequestContext(user_id="7", trace_id="trace-suitability") result = await suitability_tool_handler( - SuitabilityToolInput(customer_risk_level=1, product_risk_level=2), context + SuitabilityToolInput(customer_id="7", product_risk_level=2), + _context(), ) assert result["allowed"] is False assert result["reason_code"] == "RISK_LEVEL_MISMATCH" diff --git a/tools/check_authoritative_docs.py b/tools/check_authoritative_docs.py index 3adbcfa..0745ea2 100644 --- a/tools/check_authoritative_docs.py +++ b/tools/check_authoritative_docs.py @@ -15,6 +15,7 @@ ROOT = Path(__file__).resolve().parents[1] docs = ROOT / "docs" canonical = docs / "05-接口文档.md" deprecated = docs / "99-已废弃-公共Agent平台接口规范.md" +legacy_deprecated = docs / "05-公共Agent平台接口规范.md" if not canonical.exists(): raise SystemExit("缺少唯一权威接口文档：docs/05-接口文档.md") @@ -23,9 +24,15 @@ if deprecated.exists(): text = deprecated.read_text(encoding="utf-8") if not any(marker in text for marker in ("历史稿", "已废弃", "废弃声明")): raise SystemExit("废弃接口文档没有明确历史标记") +if legacy_deprecated.exists(): + text = legacy_deprecated.read_text(encoding="utf-8") + if not any(marker in text for marker in ("历史稿", "已废弃", "废弃声明")): + raise SystemExit("历史接口文档没有明确废弃标记") by_number: dict[str, list[str]] = defaultdict(list) for path in sorted(docs.glob("*.md")): + if path == legacy_deprecated: + continue by_number[path.name.split("-", 1)[0]].append(path.name) collisions = {number: names for number, names in by_number.items() if len(names) > 1} diff --git a/tools/install_offsite_worker_task.py b/tools/install_offsite_worker_task.py index 9076952..566bc9d 100644 --- a/tools/install_offsite_worker_task.py +++ b/tools/install_offsite_worker_task.py @@ -263,7 +263,7 @@ def build_task_xml(*, project_dir: Path, python_exe: Path, run_as: str) -> str: return f""" - 奶龙基金场外申购赎回邮件 Worker 自动启动任务 + 南方基金场外申购赎回邮件 Worker 自动启动任务 <{trigger}> diff --git a/测试报告/2026-09-12.md b/测试报告/2026-09-12.md new file mode 100644 index 0000000..943522f --- /dev/null +++ b/测试报告/2026-09-12.md @@ -0,0 +1,109 @@ +# 测试报告（当前三项功能与项目数据集） + +> 测试日期：2026-09-12 +> 测试对象：`D:\nanfangjijin\group_fqcd_jr`，分支 `yc` +> 测试数据：`D:\nanfangjijin\group_fqcd_jr\data` +> 执行方式：独立 MySQL 测试库 `jr_agent_test` + 本地离线数据集验证 +> 测试依据：根目录 `docs/项目测试/测试架构.md` + +## 1. 测试概况 + +### 1.1 数据集概况 + +| 数据类型 | 数量 | 验证结果 | +|---|---:|---| +| 场外邮件 `.eml` | 13 | 全部可解析，均包含 1 个附件 | +| 推介业绩文件 `.csv/.xlsx` | 8 | 全部通过 `parse_performance_file`，共 373 行 | +| 图片、照片、海报 `.png/.jpg` | 35 | 全部通过 Pillow 文件校验 | +| 推介材料 `.pptx` | 10 | 全部可被 `python-pptx` 打开，共 100 页 | +| PDF 文件 | 9 | 全部具备有效 PDF 文件头 | + +### 1.2 三项功能结果 + +| 功能 | 验证内容 | 结果 | +|---|---|---| +| 场外基金申购/赎回 | 识别字段、规则计算、NL2SQL、确认、通知、发送幂等 | **通过** | +| 产品推介材料生成 | 真实业绩文件解析、曲线、PPTX、海报、权限和审核交付流程 | **通过** | +| 金融 NL2SQL | 意图分类、表白名单、只读查询、模糊问题确认、场外查询字段 | **通过** | + +补充：使用目录中的实际 XLSX 与经理照片生成临时产物成功： + +- 业绩曲线 PNG：生成成功； +- PPTX：3 页，生成成功； +- 海报 PNG：`1800×2600`，生成成功。 + +### 1.3 自动化回归 + +| 测试范围 | 通过 | 失败 | 跳过/阻塞 | 说明 | +|---|---:|---:|---:|---| +| 三功能定向单元、契约测试 | 69 | 0 | 0 | 通过 | +| 场外集成回归（修正测试身份后） | 10 | 0 | 0 | 之前 8 条失败已全部恢复 | +| 全量测试 | 559 | 0 | 1 | 独立测试库执行，1 项 Redis 集成测试跳过 | + +## 2. 缺陷清单 + +### 2.1 已修复：场外旧集成用例使用失效的操作人编号 + +| 项 | 内容 | +|---|---| +| 涉及接口 | 场外确认、NL2SQL 触发、识别重试、通知创建与发送 | +| 涉及用例 | `tests/integration/test_offsite_fund_api.py`、`tests/integration/test_offsite_notification_send.py` | +| 复现 | 测试上下文 `user_id="1"`，请求体继续传 `operator_id="operator-001"` | +| 修复结果 | 将成功路径的 `operator_id` 同步为测试上下文用户 `"1"` | +| 复测结果 | `tests/integration/test_offsite_fund_api.py` 与 `tests/integration/test_offsite_notification_send.py` 共 10 项全部通过 | +| 结论 | 生产操作人防伪校验保持不变，问题属于旧集成测试数据不同步 | + +### 2.2 已修复：适当性单元测试与当前服务接口不同步 + +| 项 | 内容 | +|---|---| +| 文件 | `tests/unit/service/test_suitability_service.py` | +| 修复结果 | 测试改为使用 `SuitabilityToolInput`、`RequestContext` 和权威风险画像替身 | +| 复测结果 | 适当性单元测试 `9 passed`，全量测试可正常收集 | + +### 2.3 已处理：权威文档编号检查 + +| 项 | 内容 | +|---|---| +| 检查 | `python tools/check_authoritative_docs.py` | +| 处理 | 将明确标记为历史稿的旧文档排除出当前编号校验 | +| 实际结果 | `checked 18 documents, no number collision` | +| 结论 | 当前权威接口文档编号检查通过 | + +### 2.4 已修复：静态类型检查兼容性问题 + +涉及 `offsite_fund_service.py`、`model_gateway.py`、`health_service.py`、`worker/runtime.py`。 +本轮补齐邮件解析类型声明并处理标准库 `BytesParser` 与 `policy.default` 的类型存根兼容性后， +`python -m mypy app` 已通过（124 个源文件）。 + +## 3. 未覆盖风险 + +| 项目 | 原因 | 风险等级 | +|---|---|---| +| 真实 IMAP 收信 | `.env` 中真实收信开关关闭，本次使用离线 `.eml` 数据集 | 中 | +| 真实 OCR / DeepSeek | 外部识别开关关闭，未调用真实外部服务 | 中 | +| 真实 SMTP | SMTP 关闭且 dry-run，未发送真实邮件 | 高 | +| Redis | 健康检查显示不可用，真实限流链路未覆盖 | 中 | +| Neo4j | `127.0.0.1:7687` 连接被拒绝，关系链路未覆盖 | 中 | +| Milvus | 未执行真实连接验证 | 中 | +| 前端浏览器端到端 | 本轮未启动前端浏览器验收 | 低 | +| 性能压测 | 未执行 `performance_baseline.py` | 低 | + +## 4. 验收结论 + +| 验收项 | 结论 | +|---|---| +| 数据集文件可读、可解析 | **通过** | +| 产品推介材料真实数据生成 | **通过** | +| 金融 NL2SQL 定向单元/契约测试 | **通过** | +| 场外申购赎回完整人工确认与通知闭环 | **通过** | +| 参数校验与越权基础回归 | **通过（已执行范围内）** | +| 防重与操作人身份校验 | **通过** | +| 自动化回归 100% | **通过（559/559，另 1 项跳过）** | +| 本轮总体验收 | **通过（Redis 外部链路除外）** | + +## 5. 建议 + +1. 启动 Redis 后补跑真实限流链路，消除本轮唯一跳过项。 +2. 配置并启动真实 IMAP、OCR/DeepSeek、SMTP、Neo4j、Milvus 后补跑外部依赖链路。 +3. 启动前端后补做浏览器端到端验收。