diff --git a/.env.example b/.env.example
index ba07f09..264d9c7 100644
--- a/.env.example
+++ b/.env.example
@@ -26,6 +26,17 @@ RISK_DINGTALK_APP_SECRET=
RISK_DINGTALK_CORP_ID=
RISK_DINGTALK_AGENT_ID=
+# ==================== 外部系统风险评估接口 ====================
+# JSON 对象：调用系统编号 -> 独立 HMAC 密钥。默认 {} 禁止所有外部调用。
+# 测试/正式环境使用不同密钥，密钥至少 32 个字符，仅保存在双方后端。
+RISK_INTEGRATION_CLIENTS={}
+# 已批准的非钉钉用户映射：调用系统 -> 对方账号标识 -> 本地 emp_no；默认不允许。
+RISK_INTEGRATION_OPERATOR_MAPPINGS={}
+# 示例：{"dieya-plm":{"superadmin":"plm_superadmin"}}
+# 配置示例（请替换密钥）：{"dieya-plm":"replace-with-at-least-32-random-characters"}
+RISK_INTEGRATION_REQUESTS_PER_MINUTE=60
+RISK_INTEGRATION_REPORT_TIMEOUT_SECONDS=110
+
# ==================== 百度 OCR ====================
# 图片调用 WebImage，PDF 调用 PaddleOCR-VL 文档解析；上传原件仅保存在本机目录。
BAIDU_OCR_API_KEY=
diff --git a/README.md b/README.md
index f475e47..5844525 100644
--- a/README.md
+++ b/README.md
@@ -54,6 +54,11 @@ python scripts/clean_suppliers.py
风险评估预览：`POST http://127.0.0.1:18474/api/v1/risk-predictions/preview`。指标值表示风险程度（0-100），接口返回的综合分按 PRD 转换为安全得分，分数越高风险越低；等级阈值为 80/60/40/20。
+外部系统风险报告：`POST /api/v1/integrations/risk-assessments/report` 使用 HMAC 签名鉴权，返回 Markdown 报告及 `X-Risk-Score`、`X-Risk-Level` 等响应头。`GET` 同路径按 `requestId` 查询已保存报告。启用前执行 `python -m alembic upgrade head`，配置 `RISK_INTEGRATION_CLIENTS` JSON 密钥表，并确保操作人已在本地绑定钉钉 UnionId 且拥有 `risk.assess` 操作权限；默认不自动创建用户。完整参数、GET/POST 签名、处理中状态与部署说明见 [外部接口文档](docs/外部系统风险评估报告接口.md)。
+
+非钉钉操作人可用 `operator.externalUserId`，通过 `RISK_INTEGRATION_OPERATOR_MAPPINGS` 按调用系统显式映射到本地账号；仍校验本地状态及权限。开通工具：`python scripts/setup_external_risk.py allow-user --client-id dieya-plm --external-user-id superadmin --employee-no plm_superadmin --name "周子墨"`。该模式与钉钉身份二选一，配置后需重启后端。
+
+
对话页真实 AI：配置 `RISK_LLM_API_KEY`、`RISK_LLM_BASE_URL` 和 `RISK_LLM_MODEL` 后，调用 `POST /api/v1/conversations/{id}/stream`。接口使用 OpenAI 兼容 Chat Completions 协议，先提取 `intent/category/specification/packaging/supplier/factory/stage/confidence` 语义 JSON，再通过 SSE 返回增量文本；最近 20 条消息作为上下文，完整助手消息、语义结果和风险评分会在流结束后持久化。也兼容参考项目的 `OPENAI_API_KEY`、`LLM_BASE_URL`、`LLM_MODEL` 配置名。未配置密钥时返回明确的 503，不生成模拟回复。
架构说明见 [docs/architecture.md](docs/architecture.md)。
diff --git a/app/api/router.py b/app/api/router.py
index 05b141a..43cbef7 100644
--- a/app/api/router.py
+++ b/app/api/router.py
@@ -6,6 +6,7 @@ from app.api.v1.auth import router as auth_router
from app.api.v1.conversations import router as conversations_router
from app.api.v1.dashboard import router as dashboard_router
from app.api.v1.health import router as health_router
+from app.api.v1.integration_risk import router as integration_risk_router
from app.api.v1.model_config import router as model_config_router
from app.api.v1.monitoring import router as monitoring_router
from app.api.v1.ocr import router as ocr_router
@@ -17,6 +18,7 @@ from app.api.v1.suppliers import router as suppliers_router
api_router = APIRouter(prefix="/api/v1")
api_router.include_router(health_router)
+api_router.include_router(integration_risk_router)
api_router.include_router(model_config_router)
api_router.include_router(monitoring_router)
api_router.include_router(ocr_router)
diff --git a/app/api/v1/integration_risk.py b/app/api/v1/integration_risk.py
new file mode 100644
index 0000000..b2b690f
--- /dev/null
+++ b/app/api/v1/integration_risk.py
@@ -0,0 +1,218 @@
+"""Signed external assessments and report lookup; no browser JWT or conversation required."""
+
+import asyncio
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse, Response
+from fastapi.routing import APIRoute
+from pydantic import TypeAdapter, ValidationError
+from sqlalchemy.exc import SQLAlchemyError
+
+from app.api.dependencies import DatabaseSession
+from app.core.config import get_settings
+from app.core.exceptions import AppException
+from app.models.entities import IntegrationAssessment, OperationAudit
+from app.repositories.integration_risk import IntegrationRiskRepository
+from app.schemas.integration_risk import IntegrationOperator, IntegrationRiskRequest, RequestId
+from app.services.external_risk_report import ExternalRiskReportService
+from app.services.integration_auth import authenticate_integration, authorize_operator
+from app.services.markdown_reports import render_markdown_report
+
+logger = logging.getLogger(__name__)
+
+
+class IntegrationRoute(APIRoute):
+ def get_route_handler(self):
+ handler = super().get_route_handler()
+
+ async def wrapped(request: Request):
+ try:
+ return await handler(request)
+ except AppException as exc:
+ headers = {"Retry-After": "60"} if exc.status_code == 429 else {}
+ return JSONResponse(
+ status_code=exc.status_code,
+ headers=headers,
+ content={
+ "code": exc.code,
+ "message": exc.message,
+ "data": exc.details,
+ },
+ )
+ except SQLAlchemyError:
+ logger.error("External report database operation failed")
+ return JSONResponse(
+ status_code=503,
+ content={
+ "code": 503,
+ "message": "风险数据库暂不可用，请使用原请求编号重试",
+ "data": None,
+ },
+ )
+ except RequestValidationError as exc:
+ invalid_json = any(error["type"] == "json_invalid" for error in exc.errors())
+ status = 400 if invalid_json else 422
+ locations = [".".join(map(str, error["loc"])) for error in exc.errors()]
+ return JSONResponse(
+ status_code=status,
+ content={
+ "code": status,
+ "message": "请求体必须是合法 JSON"
+ if invalid_json
+ else "请求参数不符合约束：" + "、".join(locations),
+ "data": None,
+ },
+ )
+ except Exception as exc: # noqa: BLE001 - external callers receive no internal traceback
+ logger.error("External interface failed: %s", exc.__class__.__name__)
+ return JSONResponse(
+ status_code=500,
+ content={
+ "code": 500,
+ "message": "平台内部处理异常，请使用原请求编号重试",
+ "data": None,
+ },
+ )
+
+ return wrapped
+
+
+router = APIRouter(prefix="/integrations/risk-assessments", tags=["integrations"], route_class=IntegrationRoute)
+
+
+async def integration_client(request: Request, session: DatabaseSession) -> str:
+ if (
+ request.method == "POST"
+ and request.headers.get("Content-Type", "").split(";", 1)[0].strip().lower() != "application/json"
+ ):
+ raise AppException("Content-Type 必须为 application/json", code=400, status_code=400)
+ if len(await request.body()) > 65536:
+ raise AppException("请求体超过 64KB", code=422, status_code=422)
+ return await authenticate_integration(request, session)
+
+
+ClientId = Annotated[str, Depends(integration_client)]
+
+
+def report_response(record: IntegrationAssessment) -> Response:
+ if record.status == "processing":
+ return JSONResponse(
+ status_code=202,
+ headers={"Retry-After": "5"},
+ content={
+ "code": 202,
+ "message": "报告正在生成，请稍后查询",
+ "data": {"requestId": record.request_id},
+ },
+ )
+ if record.status == "failed":
+ raise AppException("报告生成失败，请使用原请求编号和内容重新提交", code=503, status_code=503)
+ result = record.result_data
+ return Response(
+ content=record.report_markdown,
+ media_type="text/markdown; charset=utf-8",
+ headers={
+ "X-Request-Id": record.request_id,
+ "X-Assessment-Id": str(record.id),
+ "X-Model-Version": result["modelVersion"],
+ "X-Risk-Score": f"{result['score']:.2f}",
+ "X-Risk-Level": result["level"],
+ "Cache-Control": "no-store",
+ },
+ )
+
+
+@router.post(
+ "/report",
+ response_class=Response,
+ responses={
+ 200: {"description": "Markdown 报告", "content": {"text/markdown": {"schema": {"type": "string"}}}},
+ 202: {"description": "同一请求正在生成"},
+ },
+)
+async def create_external_report(
+ request: Request,
+ payload: IntegrationRiskRequest,
+ session: DatabaseSession,
+ client_id: ClientId,
+) -> Response:
+ if payload.source_system != client_id:
+ raise AppException("sourceSystem 必须与 X-Client-Id 一致", code=403, status_code=403)
+ if request.headers.get("Idempotency-Key") != payload.request_id:
+ raise AppException("Idempotency-Key 必须与 requestId 一致", code=422, status_code=422)
+ today = datetime.now(timezone(timedelta(hours=8))).date()
+ if payload.assessment.as_of_date and payload.assessment.as_of_date > today:
+ raise AppException("asOfDate 不允许使用未来日期", code=422, status_code=422)
+ user = await authorize_operator(session, payload.operator, client_id)
+ user_id, operator_name, emp_no = user.id, user.name, user.emp_no
+ repository = IntegrationRiskRepository(session)
+ record, owned = await repository.claim(payload, user_id, today)
+ if not owned:
+ return report_response(record)
+ record_id, token, as_of = record.id, record.owner_token, record.as_of_date
+ try:
+ async with asyncio.timeout(get_settings().integration_report_timeout_seconds):
+ result = await ExternalRiskReportService(session).assess(payload, as_of)
+ markdown = render_markdown_report(payload, result, as_of, operator_name)
+ session.add(
+ OperationAudit(
+ actor_user_id=user_id,
+ actor_emp_no=emp_no,
+ action="integration.risk_assess",
+ resource_type="integration_assessment",
+ resource_id=str(record_id),
+ audit_metadata={
+ "clientId": client_id,
+ "requestId": payload.request_id,
+ "modelVersion": result.model_version,
+ },
+ )
+ )
+ if not await repository.finish(
+ record_id, token, result=result.model_dump(mode="json", by_alias=True), markdown=markdown
+ ):
+ raise AppException("评估处理权已变更，请查询原请求", code=503, status_code=503)
+ except asyncio.CancelledError:
+ await session.rollback()
+ await repository.finish(record_id, token, error="请求处理被中断")
+ raise
+ except Exception as exc: # noqa: BLE001 - preserve retryable failure without exposing database/provider errors
+ logger.warning("External assessment failed request=%s error=%s", payload.request_id, exc.__class__.__name__)
+ await session.rollback()
+ await repository.finish(record_id, token, error="报告生成失败")
+ raise AppException("报告生成失败，请使用原请求编号和内容重试", code=503, status_code=503) from exc
+ record = await repository.get(client_id, payload.request_id)
+ return report_response(record)
+
+
+@router.get(
+ "/report",
+ response_class=Response,
+ openapi_extra={
+ "parameters": [
+ {"name": "requestId", "in": "query", "required": True, "schema": {"type": "string", "maxLength": 128}}
+ ],
+ "responses": {
+ "200": {"description": "原始 Markdown 报告", "content": {"text/markdown": {"schema": {"type": "string"}}}},
+ "202": {"description": "报告正在生成"},
+ },
+ },
+)
+async def get_external_report(request: Request, session: DatabaseSession, client_id: ClientId) -> Response:
+ if list(request.query_params.keys()) != ["requestId"] or len(request.query_params.getlist("requestId")) != 1:
+ raise AppException("查询参数必须且只能包含一个 requestId", code=422, status_code=422)
+ if await request.body():
+ raise AppException("报告查询不能携带请求体", code=400, status_code=400)
+ try:
+ request_id = TypeAdapter(RequestId).validate_python(request.query_params["requestId"])
+ except ValidationError as exc:
+ raise AppException("requestId 格式不符合约束", code=422, status_code=422) from exc
+ record = await IntegrationRiskRepository(session).get(client_id, request_id)
+ if record is None:
+ raise AppException("当前调用系统未找到该请求", code=404, status_code=404)
+ await authorize_operator(session, IntegrationOperator.model_validate(record.request_data["operator"]), client_id)
+ return report_response(record)
diff --git a/app/core/config.py b/app/core/config.py
index c603c9f..55b81a4 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -25,6 +25,10 @@ class Settings(BaseSettings):
dingtalk_app_secret: SecretStr | None = Field(default=None, repr=False)
dingtalk_corp_id: str = ""
dingtalk_agent_id: str = ""
+ integration_clients: SecretStr = Field(default=SecretStr("{}"), repr=False)
+ integration_operator_mappings: SecretStr = Field(default=SecretStr("{}"), repr=False)
+ integration_requests_per_minute: int = Field(default=60, ge=1, le=10000)
+ integration_report_timeout_seconds: int = Field(default=110, ge=10, le=110)
baidu_ocr_api_key: SecretStr | None = Field(
default=None,
repr=False,
diff --git a/app/models/__init__.py b/app/models/__init__.py
index c07312e..f44706b 100644
--- a/app/models/__init__.py
+++ b/app/models/__init__.py
@@ -5,6 +5,8 @@ from app.models.entities import (
AnomalyCaseSupplier,
Conversation,
ConversationMessage,
+ IntegrationAssessment,
+ IntegrationNonce,
OperationAudit,
PurchaseOrderLine,
ReviewRecord,
@@ -13,8 +15,8 @@ from app.models.entities import (
RiskModelConfig,
RolePermission,
StandardRule,
- SupplierProfile,
SupplierMaster,
+ SupplierProfile,
UserAccount,
)
@@ -23,6 +25,8 @@ __all__ = [
"AnomalyCaseSupplier",
"Conversation",
"ConversationMessage",
+ "IntegrationAssessment",
+ "IntegrationNonce",
"OperationAudit",
"ReviewRecord",
"RiskAssessment",
diff --git a/app/models/entities.py b/app/models/entities.py
index ea87d86..354fc77 100644
--- a/app/models/entities.py
+++ b/app/models/entities.py
@@ -4,7 +4,20 @@ from datetime import date, datetime
from decimal import Decimal
from typing import Any
-from sqlalchemy import BigInteger, Boolean, Date, DateTime, ForeignKey, Index, Integer, Numeric, String, Text, UniqueConstraint, func
+from sqlalchemy import (
+ BigInteger,
+ Boolean,
+ Date,
+ DateTime,
+ ForeignKey,
+ Index,
+ Integer,
+ Numeric,
+ String,
+ Text,
+ UniqueConstraint,
+ func,
+)
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship
@@ -106,7 +119,9 @@ class AnomalyCaseSupplier(Base):
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
case_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("anomaly_cases.id", ondelete="CASCADE"), nullable=False)
- supplier_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("supplier_master.id", ondelete="CASCADE"), nullable=False)
+ supplier_id: Mapped[int] = mapped_column(
+ BigInteger, ForeignKey("supplier_master.id", ondelete="CASCADE"), nullable=False
+ )
supplier_role: Mapped[str | None] = mapped_column(String(50))
material_type: Mapped[str | None] = mapped_column(String(200))
source_text: Mapped[str | None] = mapped_column(Text)
@@ -252,6 +267,51 @@ class ConversationMessage(Base):
conversation: Mapped[Conversation] = relationship(back_populates="messages")
+class IntegrationNonce(Base):
+ """Durable replay protection shared by all API workers."""
+
+ __tablename__ = "integration_nonces"
+ __table_args__ = (
+ UniqueConstraint("client_id", "nonce", name="uq_integration_nonce"),
+ Index("ix_integration_nonce_expires", "expires_at"),
+ Index("ix_integration_nonce_client_created", "client_id", "created_at"),
+ )
+
+ id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+ client_id: Mapped[str] = mapped_column(String(64), nullable=False)
+ nonce: Mapped[str] = mapped_column(String(64), nullable=False)
+ created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+ expires_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+
+class IntegrationAssessment(Base):
+ """Immutable successful reports and recoverable requests from external systems."""
+
+ __tablename__ = "integration_assessments"
+ __table_args__ = (
+ UniqueConstraint("client_id", "request_id", name="uq_integration_assessment_request"),
+ Index("ix_integration_assessment_user", "user_id"),
+ )
+
+ id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
+ client_id: Mapped[str] = mapped_column(String(64), nullable=False)
+ request_id: Mapped[str] = mapped_column(String(128), nullable=False)
+ # Logical reference, like OperationAudit: migration does not require
+ # REFERENCES on the pre-existing user table, and reports survive account removal.
+ user_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+ request_hash: Mapped[str] = mapped_column(String(64), nullable=False)
+ request_data: Mapped[dict[str, Any]] = mapped_column(JSONB, nullable=False)
+ as_of_date: Mapped[date] = mapped_column(Date, nullable=False)
+ status: Mapped[str] = mapped_column(String(20), nullable=False)
+ owner_token: Mapped[str] = mapped_column(String(32), nullable=False)
+ lease_until: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+ result_data: Mapped[dict[str, Any] | None] = mapped_column(JSONB)
+ report_markdown: Mapped[str | None] = mapped_column(Text)
+ error_message: Mapped[str | None] = mapped_column(String(200))
+ created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+ updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+
+
class RiskItem(Base):
"""Risk item submitted through form, voice, OCR, or import before approval."""
diff --git a/app/repositories/integration_risk.py b/app/repositories/integration_risk.py
new file mode 100644
index 0000000..f492f09
--- /dev/null
+++ b/app/repositories/integration_risk.py
@@ -0,0 +1,120 @@
+"""Atomic idempotency claims with a lease and fenced completion."""
+
+import hashlib
+import json
+import uuid
+from datetime import UTC, date, datetime, timedelta
+
+from sqlalchemy import or_, select, update
+from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.exceptions import AppException
+from app.models.entities import IntegrationAssessment
+from app.schemas.integration_risk import IntegrationRiskRequest
+
+
+class IntegrationRiskRepository:
+ def __init__(self, session: AsyncSession):
+ self.session = session
+
+ async def get(self, client_id: str, request_id: str) -> IntegrationAssessment | None:
+ return await self.session.scalar(
+ select(IntegrationAssessment)
+ .where(
+ IntegrationAssessment.client_id == client_id,
+ IntegrationAssessment.request_id == request_id,
+ )
+ .execution_options(populate_existing=True)
+ )
+
+ async def claim(
+ self,
+ payload: IntegrationRiskRequest,
+ user_id: int,
+ default_date: date,
+ ) -> tuple[IntegrationAssessment, bool]:
+ data = payload.model_dump(mode="json", by_alias=True)
+ # Preserve the digest of DingTalk requests created before externalUserId was introduced.
+ if data["operator"]["externalUserId"] is None:
+ data["operator"].pop("externalUserId")
+ digest = hashlib.sha256(
+ json.dumps(data, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()
+ ).hexdigest()
+ now = datetime.now(UTC)
+ token = uuid.uuid4().hex
+ lease_until = now + timedelta(seconds=150)
+ inserted = await self.session.scalar(
+ insert(IntegrationAssessment)
+ .values(
+ client_id=payload.source_system,
+ request_id=payload.request_id,
+ user_id=user_id,
+ request_hash=digest,
+ request_data=data,
+ as_of_date=payload.assessment.as_of_date or default_date,
+ status="processing",
+ owner_token=token,
+ lease_until=lease_until,
+ created_at=now,
+ updated_at=now,
+ )
+ .on_conflict_do_nothing(index_elements=["client_id", "request_id"])
+ .returning(IntegrationAssessment.id)
+ )
+ record = await self.get(payload.source_system, payload.request_id)
+ if record is None:
+ raise AppException("评估请求记录失败", code=503, status_code=503)
+ if record.request_hash != digest:
+ raise AppException("requestId 已被不同请求内容使用", code=409, status_code=409)
+ if inserted is not None:
+ await self.session.commit()
+ return record, True
+ if record.status == "completed":
+ return record, False
+ reclaimed = await self.session.scalar(
+ update(IntegrationAssessment)
+ .where(
+ IntegrationAssessment.id == record.id,
+ or_(IntegrationAssessment.status == "failed", IntegrationAssessment.lease_until <= now),
+ IntegrationAssessment.status != "completed",
+ )
+ .values(status="processing", owner_token=token, lease_until=lease_until, error_message=None, updated_at=now)
+ .execution_options(synchronize_session=False)
+ .returning(IntegrationAssessment.id)
+ )
+ await self.session.commit()
+ record = await self.get(payload.source_system, payload.request_id)
+ return record, reclaimed is not None
+
+ async def finish(
+ self,
+ record_id: int,
+ owner_token: str,
+ *,
+ result: dict | None = None,
+ markdown: str | None = None,
+ error: str | None = None,
+ ) -> bool:
+ updated = await self.session.scalar(
+ update(IntegrationAssessment)
+ .where(
+ IntegrationAssessment.id == record_id,
+ IntegrationAssessment.owner_token == owner_token,
+ IntegrationAssessment.status == "processing",
+ )
+ .values(
+ status="failed" if error else "completed",
+ result_data=result,
+ report_markdown=markdown,
+ error_message=error,
+ updated_at=datetime.now(UTC),
+ )
+ .execution_options(synchronize_session=False)
+ .returning(IntegrationAssessment.id)
+ )
+ if updated is None:
+ await self.session.rollback()
+ return False
+ await self.session.commit()
+ return True
diff --git a/app/schemas/integration_risk.py b/app/schemas/integration_risk.py
new file mode 100644
index 0000000..dd27256
--- /dev/null
+++ b/app/schemas/integration_risk.py
@@ -0,0 +1,61 @@
+"""Validated external assessment inputs; no caller-supplied scores or roles."""
+
+from datetime import date
+from typing import Annotated, Literal
+
+from pydantic import ConfigDict, Field, StringConstraints, field_validator, model_validator
+
+from app.schemas.common import ApiSchema
+
+
+class IntegrationSchema(ApiSchema):
+ model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)
+
+
+class IntegrationOperator(IntegrationSchema):
+ dingtalk_union_id: str | None = Field(default=None, min_length=1, max_length=64)
+ external_user_id: str | None = Field(default=None, min_length=1, max_length=128)
+ dingtalk_user_id: str | None = Field(default=None, min_length=1, max_length=50)
+ employee_no: str | None = Field(default=None, max_length=64)
+ name: str | None = Field(default=None, max_length=50)
+
+ @model_validator(mode="after")
+ def identity_source(self):
+ if bool(self.dingtalk_union_id) == bool(self.external_user_id):
+ raise ValueError("dingtalkUnionId 和 externalUserId 必须且只能提供一个")
+ if self.external_user_id and self.dingtalk_user_id:
+ raise ValueError("外部账号模式不能同时提供 dingtalkUserId")
+ return self
+
+
+class IntegrationProduct(IntegrationSchema):
+ subject_id: str = Field(min_length=1, max_length=128)
+ product_name: str | None = Field(default=None, max_length=200)
+ product_category: str = Field(min_length=1, max_length=100)
+ specification: str | None = Field(default=None, max_length=100)
+ material_type: str = Field(min_length=1, max_length=200)
+ supplier: str | None = Field(default=None, max_length=300)
+ factory: str | None = Field(default=None, max_length=300)
+ stage_new: Literal["需求发现", "产品立项", "产品开发", "量产上市", "产品生命管理"] | None = None
+ defect_level: Literal["A", "B", "C"] | None = None
+ description: str | None = Field(default=None, max_length=2000)
+ as_of_date: date | None = None
+
+ @field_validator("as_of_date", mode="before")
+ @classmethod
+ def iso_date(cls, value): # noqa: ANN001
+ if value is not None and (not isinstance(value, str) or len(value) != 10):
+ raise ValueError("asOfDate 必须为 YYYY-MM-DD")
+ if value is not None and date.fromisoformat(value).year < 1900:
+ raise ValueError("asOfDate 不早于 1900-01-01")
+ return value
+
+
+RequestId = Annotated[str, StringConstraints(min_length=1, max_length=128, pattern=r"^[A-Za-z0-9_.:-]+$")]
+
+
+class IntegrationRiskRequest(IntegrationSchema):
+ request_id: RequestId
+ source_system: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$")
+ operator: IntegrationOperator
+ assessment: IntegrationProduct
diff --git a/app/services/external_risk_report.py b/app/services/external_risk_report.py
new file mode 100644
index 0000000..d4d8945
--- /dev/null
+++ b/app/services/external_risk_report.py
@@ -0,0 +1,217 @@
+"""Assess structured product data without creating conversations or re-parsing fields with an LLM."""
+
+import asyncio
+import logging
+from datetime import date
+
+from sqlalchemy import or_, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.entities import AnomalyCase, StandardRule
+from app.schemas.integration_risk import IntegrationRiskRequest
+from app.schemas.risk_prediction import (
+ CategoryWarning,
+ RiskAssessmentResult,
+ RiskDimensionScore,
+ RiskFactorResult,
+ StandardJudgment,
+ SupplierFrequency,
+)
+from app.services.database_risk_scoring import (
+ DatabaseRiskScoringService,
+ normalize_category,
+ normalize_material,
+)
+from app.services.llm_client import LLMClient
+from app.services.purchase_orders import calendar_year_window, supplier_order_counts
+from app.services.rag import AnomalyCaseRAG
+from app.services.supplier_identity import resolve_supplier
+
+logger = logging.getLogger(__name__)
+
+
+class ExternalRiskReportService:
+ def __init__(self, session: AsyncSession):
+ self.session = session
+
+ async def assess(self, payload: IntegrationRiskRequest, as_of: date) -> RiskAssessmentResult:
+ product = payload.assessment
+ start, _ = calendar_year_window(as_of, 3)
+ # Bound both scoring and case retrieval by the supplied cutoff date.
+ fields = (
+ AnomalyCase.id,
+ AnomalyCase.date,
+ AnomalyCase.product_name,
+ AnomalyCase.product_category,
+ AnomalyCase.material_type,
+ AnomalyCase.supplier,
+ AnomalyCase.factory,
+ AnomalyCase.stage_new,
+ AnomalyCase.defect_level,
+ AnomalyCase.anomaly_type,
+ AnomalyCase.description,
+ AnomalyCase.root_cause,
+ AnomalyCase.solution,
+ AnomalyCase.result,
+ )
+ rows = [
+ dict(row)
+ for row in (
+ await self.session.execute(
+ select(*fields).where(
+ AnomalyCase.date >= start,
+ AnomalyCase.date <= as_of,
+ )
+ )
+ )
+ .mappings()
+ .all()
+ ]
+ identity = await resolve_supplier(self.session, product.supplier)
+ supplier = identity.canonical_name if identity else product.supplier
+ cooperation = await supplier_order_counts(self.session, supplier, start, as_of) if supplier else 0
+ score = await DatabaseRiskScoringService(self.session, as_of=as_of).score(
+ category=product.product_category,
+ material=product.material_type,
+ supplier=supplier,
+ factory=product.factory,
+ stage=product.stage_new,
+ defect_level=product.defect_level,
+ rows=rows,
+ cooperation_count=cooperation or 0,
+ )
+ query = "，".join(
+ value
+ for value in (
+ product.product_name,
+ product.product_category,
+ product.material_type,
+ supplier,
+ product.factory,
+ product.stage_new,
+ product.description,
+ )
+ if value
+ )
+ retrieved = await AnomalyCaseRAG(self.session).search(
+ query=query,
+ category=product.product_category,
+ material=product.material_type,
+ supplier=supplier,
+ factory=product.factory,
+ stage=product.stage_new,
+ rows=rows,
+ )
+ category = normalize_category(product.product_category)
+ material = normalize_material(product.material_type)
+ terms = list(
+ dict.fromkeys(
+ value
+ for value in (
+ product.product_category,
+ *(category or "").split("/"),
+ material,
+ *(term for term in ("PET", "PP", "软管", "泵头", "瓶", "纸盒", "密封", "相容性") if term in query),
+ )
+ if value
+ )
+ )
+ clauses = [
+ column.ilike(f"%{term}%")
+ for term in terms
+ for column in (
+ StandardRule.category_lv1,
+ StandardRule.category_lv2,
+ StandardRule.category_lv3,
+ StandardRule.risk_point,
+ )
+ ]
+ standard_rows = list(
+ (
+ await self.session.scalars(
+ select(StandardRule)
+ .where(
+ StandardRule.status == "启用",
+ or_(*clauses),
+ )
+ .order_by(StandardRule.id)
+ .limit(6)
+ )
+ ).all()
+ )
+ standards = [
+ StandardJudgment(
+ category=row.category_lv1,
+ risk_point=row.risk_point,
+ rule=row.judge_standard,
+ source=row.regulation or "标准文件.xlsx",
+ status="待检测",
+ )
+ for row in standard_rows
+ ]
+ dimensions = [
+ RiskDimensionScore(
+ dimension=item.name,
+ score=item.score,
+ weight=item.weight,
+ status=item.status,
+ reason=item.reason,
+ sample_size=item.sample_size,
+ raw_value=item.raw_value,
+ )
+ for item in score.dimensions
+ ]
+ category_dimension = next(item for item in score.dimensions if item.name == "产品品类")
+ supplier_dimension = next(item for item in score.dimensions if item.name == "供应商表现")
+ result = RiskAssessmentResult(
+ subject_id=product.subject_id,
+ score=score.score,
+ level=score.level,
+ model_version="database-seven-dimension-v1",
+ factors=[
+ RiskFactorResult(
+ name=item.name,
+ value=round(100 - item.score, 2),
+ weight=item.weight,
+ contribution=round(item.score * item.weight / 100, 2),
+ )
+ for item in score.dimensions
+ ],
+ dimension_scores=dimensions,
+ category_warning=CategoryWarning(
+ category=category or "无法归类",
+ anomaly_count=category_dimension.sample_size or 0,
+ risk_level="案例较少" if (category_dimension.sample_size or 0) < 10 else score.level.value,
+ suggestion=category_dimension.reason,
+ ),
+ supplier_freq=SupplierFrequency(
+ supplier_name=supplier,
+ cooperation_count=cooperation if cooperation else None,
+ anomaly_count=supplier_dimension.sample_size,
+ alpha=supplier_dimension.frequency_alpha,
+ frequency_basis=f"{start} 至 {as_of} 采购订单去重单据数",
+ note=supplier_dimension.reason,
+ data_available=bool(cooperation),
+ ),
+ standard_judgment=standards,
+ similar_cases=[item.case for item in retrieved],
+ )
+ fallback = [
+ "打样阶段完成内容物与包装材料相容性、密封性及加速老化验证。",
+ "结合历史异常和命中标准制定来料检验、过程巡检和首批试产验证清单。",
+ "补充供应商、加工厂及实际检测数据，由品控复核默认分维度后再作放行决策。",
+ "对验证项明确责任人、完成节点及验收依据，并将结果回填形成闭环。",
+ ]
+ try:
+ async with asyncio.timeout(20):
+ result.ai_suggestions = await LLMClient().generate_prevention_suggestions(
+ {
+ "input": product.model_dump(mode="json", by_alias=True),
+ "assessment": result.model_dump(mode="json", by_alias=True),
+ }
+ )
+ except Exception as exc: # noqa: BLE001 - rules and actual scores remain usable without the LLM
+ logger.warning("External report suggestions unavailable: %s", exc.__class__.__name__)
+ result.ai_suggestions = fallback
+ result.alert_text = "AI 建议服务不可用，本报告使用固定验证建议；评分仍来自数据库及规则。"
+ return result
diff --git a/app/services/integration_auth.py b/app/services/integration_auth.py
new file mode 100644
index 0000000..a8b73d2
--- /dev/null
+++ b/app/services/integration_auth.py
@@ -0,0 +1,168 @@
+"""System HMAC authentication, durable replay protection and DingTalk authorization."""
+
+import base64
+import hashlib
+import hmac
+import json
+import re
+from datetime import UTC, datetime, timedelta
+
+from fastapi import Request
+from sqlalchemy import delete, func, select, text
+from sqlalchemy.dialects.postgresql import insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.config import get_settings
+from app.core.exceptions import AppException
+from app.models.entities import IntegrationNonce, UserAccount
+from app.repositories.users import UserRepository
+from app.schemas.integration_risk import IntegrationOperator
+
+
+def signature_message(request: Request, timestamp: str, nonce: str, body: bytes) -> bytes:
+ prefix = timestamp.encode("ascii") + b"\n" + nonce.encode("ascii") + b"\n"
+ if request.method == "GET":
+ # GET binds both the endpoint and query so a signature cannot be moved to another report.
+ target = request.scope.get("raw_path", request.url.path.encode("ascii"))
+ query = request.scope.get("query_string", b"")
+ return b"GET\n" + target + b"\n" + query + b"\n" + prefix
+ return prefix + body
+
+
+async def authenticate_integration(request: Request, session: AsyncSession) -> str:
+ settings = get_settings()
+ client_id = request.headers.get("X-Client-Id", "")
+ timestamp = request.headers.get("X-Timestamp", "")
+ nonce = request.headers.get("X-Nonce", "")
+ signature = request.headers.get("X-Signature", "")
+ if not all((client_id, timestamp, nonce, signature)):
+ raise AppException("缺少系统签名认证信息", code=401, status_code=401)
+ try:
+ clients = json.loads(settings.integration_clients.get_secret_value())
+ if not isinstance(clients, dict) or any(
+ not isinstance(key, str)
+ or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", key)
+ or not isinstance(value, str)
+ or len(value) < 32
+ for key, value in clients.items()
+ ):
+ raise ValueError
+ except (ValueError, TypeError) as exc:
+ raise AppException("外部系统鉴权配置无效", code=503, status_code=503) from exc
+ secret = clients.get(client_id)
+ if secret is None:
+ raise AppException("调用系统未开通接口", code=403, status_code=403)
+ if not re.fullmatch(r"[0-9]{1,12}", timestamp) or not re.fullmatch(r"[A-Za-z0-9_-]{8,64}", nonce):
+ raise AppException("时间戳或随机串格式错误", code=401, status_code=401)
+ now = datetime.now(UTC)
+ if abs(now.timestamp() - int(timestamp)) > 300:
+ raise AppException("系统签名时间戳已过期", code=401, status_code=401)
+ expected = base64.b64encode(
+ hmac.new(
+ secret.encode("utf-8"), signature_message(request, timestamp, nonce, await request.body()), hashlib.sha256
+ ).digest()
+ ).decode("ascii")
+ if not re.fullmatch(r"[A-Za-z0-9+/]{43}=", signature) or not hmac.compare_digest(signature, expected):
+ raise AppException("系统签名校验失败", code=401, status_code=401)
+
+ # Serialize only the short authentication transaction for this client across workers.
+ lock_key = int.from_bytes(hashlib.sha256(f"integration-auth:{client_id}".encode()).digest()[:8], signed=True)
+ await session.execute(text("SELECT pg_advisory_xact_lock(:key)"), {"key": lock_key})
+ await session.execute(delete(IntegrationNonce).where(IntegrationNonce.expires_at <= now))
+ replay = await session.scalar(
+ select(IntegrationNonce.id).where(
+ IntegrationNonce.client_id == client_id,
+ IntegrationNonce.nonce == nonce,
+ )
+ )
+ if replay is not None:
+ raise AppException("随机串已使用，请重新签名", code=401, status_code=401)
+ count = await session.scalar(
+ select(func.count())
+ .select_from(IntegrationNonce)
+ .where(
+ IntegrationNonce.client_id == client_id,
+ IntegrationNonce.created_at > now - timedelta(seconds=60),
+ )
+ )
+ if count >= settings.integration_requests_per_minute:
+ raise AppException("超过系统调用频率限制", code=429, status_code=429)
+ await session.execute(
+ insert(IntegrationNonce).values(
+ client_id=client_id,
+ nonce=nonce,
+ created_at=now,
+ expires_at=now + timedelta(minutes=10),
+ )
+ )
+ # Persist nonce even if body validation or assessment later fails.
+ await session.commit()
+ return client_id
+
+
+async def authorize_operator(
+ session: AsyncSession,
+ operator: IntegrationOperator,
+ client_id: str | None = None,
+) -> UserAccount:
+ if operator.external_user_id:
+ if not client_id:
+ raise AppException("外部账号必须关联已认证的调用系统", code=403, status_code=403)
+ try:
+ mappings = json.loads(get_settings().integration_operator_mappings.get_secret_value())
+ if not isinstance(mappings, dict) or any(
+ not isinstance(client, str)
+ or not isinstance(accounts, dict)
+ or any(
+ not isinstance(external, str)
+ or not external
+ or not isinstance(local, str)
+ or not local
+ or len(local) > 20
+ for external, local in accounts.items()
+ )
+ for client, accounts in mappings.items()
+ ):
+ raise ValueError
+ except (ValueError, TypeError) as exc:
+ raise AppException("外部账号映射配置无效", code=503, status_code=503) from exc
+ local_employee_no = mappings.get(client_id, {}).get(operator.external_user_id)
+ if not local_employee_no:
+ raise AppException("外部用户未在当前调用系统开通", code=403, status_code=403)
+ users = list(
+ (
+ await session.scalars(
+ select(UserAccount)
+ .where(
+ UserAccount.emp_no == local_employee_no,
+ )
+ .limit(2)
+ )
+ ).all()
+ )
+ else:
+ users = list(
+ (
+ await session.scalars(
+ select(UserAccount)
+ .where(
+ UserAccount.dingtalk_union_id == operator.dingtalk_union_id,
+ )
+ .limit(2)
+ )
+ ).all()
+ )
+ if len(users) != 1:
+ kind = "外部" if operator.external_user_id else "钉钉"
+ message = f"{kind}用户未在风险平台开通" if not users else f"{kind}用户映射不唯一，请联系管理员"
+ raise AppException(message, code=403, status_code=403)
+ user = users[0]
+ if user.status != "启用":
+ raise AppException("用户已停用", code=403, status_code=403)
+ if operator.dingtalk_user_id and operator.dingtalk_user_id != user.dingtalk_id:
+ raise AppException("钉钉 UnionId 与 UserId 不一致", code=403, status_code=403)
+ if user.role != "admin":
+ permissions = await UserRepository(session).permissions_for_role(user.role, operate=True)
+ if not any(item.func_key == "risk.assess" for item in permissions):
+ raise AppException("用户无风险评估权限", code=403, status_code=403)
+ return user
diff --git a/app/services/markdown_reports.py b/app/services/markdown_reports.py
new file mode 100644
index 0000000..e1e6b5c
--- /dev/null
+++ b/app/services/markdown_reports.py
@@ -0,0 +1,140 @@
+"""Deterministic Markdown reports with escaped untrusted text and explicit data boundaries."""
+
+import html
+from datetime import date
+
+from app.schemas.integration_risk import IntegrationRiskRequest
+from app.schemas.risk_prediction import RiskAssessmentResult
+from app.services.database_risk_scoring import normalize_category, normalize_material
+from app.services.purchase_orders import calendar_year_window
+from app.services.reports import LEVEL_LABELS
+
+
+def md_text(value: object) -> str:
+ if value is None or value == "":
+ return "未提供"
+ value = html.escape(str(value), quote=False).replace("\r", " ").replace("\n", " ")
+ for char in ("\\", "`", "*", "_", "[", "]", "|", "#", "!"):
+ value = value.replace(char, "\\" + char)
+ return value
+
+
+def render_markdown_report(
+ payload: IntegrationRiskRequest,
+ result: RiskAssessmentResult,
+ as_of: date,
+ operator_name: str,
+) -> str:
+ product = payload.assessment
+ start, _ = calendar_year_window(as_of, 3)
+ info = (
+ ("评估对象", product.product_name or product.subject_id),
+ ("对象编号", product.subject_id),
+ ("请求编号", payload.request_id),
+ ("提交品类", product.product_category),
+ ("评分品类", normalize_category(product.product_category) or "无法归类，品类维度使用默认 50 分"),
+ ("规格", product.specification),
+ ("包装原文", product.material_type),
+ ("评分物料标签", normalize_material(product.material_type)),
+ ("供应商", product.supplier),
+ ("加工厂", product.factory),
+ ("开发阶段", product.stage_new),
+ ("已知缺陷等级", product.defect_level),
+ ("评估人", operator_name),
+ ("数据截止日期", as_of),
+ ("评估背景", product.description),
+ )
+ lines = ["# 新品风险评估报告", "", "## 一、评估信息", "", "| 项目 | 内容 |", "|---|---|"]
+ lines.extend(f"| {label} | {md_text(value)} |" for label, value in info)
+ lines.extend(
+ [
+ "",
+ "## 二、综合风险结论",
+ "",
+ f"- 综合安全得分：{result.score:.2f}（越高风险越低）",
+ f"- 风险等级：{LEVEL_LABELS[result.level.value]}",
+ f"- 模型版本：{md_text(result.model_version)}",
+ ]
+ )
+ if result.alert_text:
+ lines.append(f"- 数据与建议提示：{md_text(result.alert_text)}")
+ lines.extend(
+ [
+ "",
+ "## 三、七维风险评分",
+ "",
+ "| 风险维度 | 得分 | 权重 | 数据状态 | 说明 |",
+ "|---|---:|---:|---|---|",
+ ]
+ )
+ lines.extend(
+ f"| {md_text(item.dimension)} | {item.score:.2f} | {item.weight:g}% | "
+ f"{md_text(item.status)} | {md_text(item.reason)} |"
+ for item in result.dimension_scores
+ )
+ lines.extend(["", "## 四、历史统计与相似异常案例", ""])
+ if result.category_warning:
+ lines.append(f"- 评分品类历史异常：{result.category_warning.anomaly_count} 条。")
+ if result.supplier_freq:
+ freq = result.supplier_freq
+ lines.extend(
+ [
+ f"- 供应商：{md_text(freq.supplier_name)}；合作次数：{md_text(freq.cooperation_count)}；"
+ f"历史异常：{md_text(freq.anomaly_count)}。",
+ f"- 供应商统计说明：{md_text(freq.note)}",
+ ]
+ )
+ lines.extend(["", "| 日期 | 产品 | 异常现象 | 原因 | 历史处理方案 | 相似度 |", "|---|---|---|---|---|---:|"])
+ for case in result.similar_cases:
+ similarity = f"{case.similarity:.1%}" if case.similarity is not None else "未提供"
+ lines.append(
+ "| "
+ + " | ".join(
+ md_text(value)
+ for value in (
+ case.date,
+ case.product,
+ case.problem or case.description,
+ case.root_cause,
+ case.solution,
+ similarity,
+ )
+ )
+ + " |"
+ )
+ if not result.similar_cases:
+ lines.extend(["", "暂无匹配的历史案例，不能据此认定产品无风险。"])
+ lines.extend(["", "## 五、标准规则", "", "| 风险点 | 判定规则 | 来源 | 状态 |", "|---|---|---|---|"])
+ lines.extend(
+ "| "
+ + " | ".join(
+ md_text(value)
+ for value in (
+ rule.risk_point,
+ rule.rule,
+ rule.source,
+ rule.status,
+ )
+ )
+ + " |"
+ for rule in result.standard_judgment
+ )
+ if not result.standard_judgment:
+ lines.extend(["", "暂无匹配标准，需由品控补充核验；不代表已通过合规或检测判定。"])
+ lines.extend(["", "## 六、风险预防建议", ""])
+ lines.extend(f"{index}. {md_text(value)}" for index, value in enumerate(result.ai_suggestions, 1))
+ lines.extend(
+ [
+ "",
+ "## 七、数据口径说明",
+ "",
+ f"- 数据来自风险平台本地异常案例、采购订单及启用的标准规则；统计区间为 {start} 至 {as_of}。",
+ "- 沿用现有评分口径，以截止年份往前 3 年的 1 月 1 日为区间起点。",
+ "- 包装组合沿用现有关键词归一化规则，只使用一个匹配物料标签进行评分。",
+ "- 缺少巡检、检测结果或可用历史记录的维度按现有默认分处理，并在明细中标注。",
+ "- 缺陷维度基于历史 A/B/C 记录计算，不预测新品实际缺陷等级。",
+ "- 标准匹配和预防建议不替代实际检测与品控审核；最终放行由业务和品控人员确认。",
+ "",
+ ]
+ )
+ return "\n".join(lines)
diff --git a/app/services/rag.py b/app/services/rag.py
index ade10fc..5c28fb1 100644
--- a/app/services/rag.py
+++ b/app/services/rag.py
@@ -53,6 +53,7 @@ class AnomalyCaseRAG:
factory: str | None = None,
stage: str | None = None,
anomaly_type: str | None = None,
+ rows: list[dict[str, Any]] | None = None,
) -> list[RetrievedCase]:
identity = await resolve_supplier(self.session, supplier)
supplier_match = (
@@ -61,7 +62,7 @@ class AnomalyCaseRAG:
else supplier
)
terms = self._terms(query, category, material, supplier, factory, stage, anomaly_type)
- rows = await self._load_rows(
+ rows = rows if rows is not None else await self._load_rows(
query,
category=category,
material=material,
diff --git a/docs/PLM接口测试脚本使用说明.md b/docs/PLM接口测试脚本使用说明.md
new file mode 100644
index 0000000..05fd2a0
--- /dev/null
+++ b/docs/PLM接口测试脚本使用说明.md
@@ -0,0 +1,53 @@
+# PLM 接口测试脚本使用说明
+
+脚本 `scripts/test_plm_risk_report.py` 使用对方 2026-10-09 调用说明中第 3 节的完整参数。
+样例保存在 `docs/examples/plm-risk-request-20261009.json`，品类保持“护发素”，不改为洗发水。
+
+## 原报告与幂等验证
+
+在后端项目根目录执行：
+
+```powershell
+.\.venv\Scripts\python.exe scripts/test_plm_risk_report.py --local-config
+```
+
+默认访问 `http://192.168.9.28:18474/api/v1/integrations/risk-assessments/report`。
+`--local-config` 表示从本项目 `.env` 读取 `dieya-plm` 对应密钥，密钥和签名不写入输出。
+在外部环境可以改为设置 `RISK_CLIENT_SECRET` 或 `RISK_INTEGRATION_CLIENT_SECRET`，不传 `--local-config`。
+
+脚本首先查询原请求。不存在时才提交原参数生成，失败记录允许使用原快照重新提交。
+取得报告后，重复 POST 同一业务参数验证幂等，再 GET 查询验证正文及元数据一致。
+HTTP 202、429 会遵守 `Retry-After` 等待；完整测试总期限 180 秒，单次连接最多 10 秒、请求最多 130 秒。
+POST 网络异常后先 GET 查询，避免直接换编号重复生成。其他错误会保留诊断与快照并以退出码 1 结束。
+
+## 新生成一份报告
+
+```powershell
+.\.venv\Scripts\python.exe scripts/test_plm_risk_report.py --local-config --new-request
+```
+
+这会生成唯一的 `PLM-TEST-...` 请求编号。其他业务字段（包括截止日期）保持原样，便于与原报告比较。
+如需要不同产品、日期或操作人，准备新的 JSON 文件，再通过 `--body-file` 指定。
+
+可指定访问地址和输出目录：
+
+```powershell
+.\.venv\Scripts\python.exe scripts/test_plm_risk_report.py --local-config --url http://127.0.0.1:18474/api/v1/integrations/risk-assessments/report --output-dir .runtime-uploads/integration-tests/my-run
+```
+
+## 输出文件
+
+默认保存在 `.runtime-uploads/integration-tests/<北京时间及随机后缀>/`，该目录已被 Git 忽略。
+
+| 文件 | 内容 |
+|---|---|
+| `request.json` | 实际发送的紧凑 UTF-8 请求快照，用于恢复 |
+| `report.md` | 接口实际返回的完整 Markdown，不进行重写 |
+| `response.json` | 状态、耗时、安全响应头、报告摘要、测试检查项或失败诊断 |
+| `summary.md` | 成功时的分数、等级、维度数量及报告链接 |
+
+响应头校验包括请求编号、评估 ID、模型版本、0～100 安全分、合法风险等级；还会核对七个维度行。
+HTTP 客户端只解压一次，报告以 UTF-8 校验，解压后的上限为 1MiB。
+
+当前“护发素”不匹配八类品类评分规则，报告会显示无法归类和品类维度默认 50 分。
+HTTP 200 表示调用成功，不代表所有维度具备真实数据。新旧报告的评分可比较，AI 建议文本可能不同。
diff --git a/docs/examples/plm-risk-request-20261009.json b/docs/examples/plm-risk-request-20261009.json
new file mode 100644
index 0000000..6e9eced
--- /dev/null
+++ b/docs/examples/plm-risk-request-20261009.json
@@ -0,0 +1,21 @@
+{
+ "requestId": "PLM-REQ202610090001-002",
+ "sourceSystem": "dieya-plm",
+ "operator": {
+ "externalUserId": "superadmin",
+ "name": "周子墨"
+ },
+ "assessment": {
+ "subjectId": "REQ202610090001",
+ "productName": "泊紫汀兰PDRN鱼子酱角蛋白护发素",
+ "productCategory": "护发素",
+ "specification": "750ml",
+ "materialType": "PET瓶身",
+ "supplier": null,
+ "factory": null,
+ "stageNew": "需求发现",
+ "defectLevel": null,
+ "description": "PLM需求 REQ202610090001 新品上市前风险评估",
+ "asOfDate": "2026-10-09"
+ }
+}
diff --git a/docs/外部系统风险评估报告接口.md b/docs/外部系统风险评估报告接口.md
new file mode 100644
index 0000000..746fc76
--- /dev/null
+++ b/docs/外部系统风险评估报告接口.md
@@ -0,0 +1,550 @@
+# 外部系统风险评估报告接口
+
+## 1. 文档说明
+
+本文档定义外部业务系统调用蝶雅风险预警平台生成新品风险评估 Markdown 报告的接口协议。
+
+> 外部接口代码已实现。启用前需执行数据库迁移、配置系统调用密钥并开通本地钉钉用户；测试和正式地址由部署方提供。本接口与内部 `POST /api/v1/risk-predictions/preview` 分离，不依赖内部用户 JWT 或聊天会话。
+
+### 1.1 接口用途
+
+外部系统提交产品、包装、供应商和开发阶段等风险评估参数，蝶雅平台完成以下处理：
+
+1. 校验调用方系统身份和请求签名。
+2. 根据钉钉用户标识，或管理员预先批准的外部账号映射，匹配蝶雅平台用户。
+3. 使用蝶雅平台风险数据库、标准库和评分模型完成风险评估。
+4. 生成 Markdown 格式的风险评估报告并直接返回。
+
+### 1.2 当前数据边界
+
+外部系统与蝶雅平台使用不同数据库时，蝶雅平台默认只使用自己的历史异常案例、采购订单、供应商主数据和标准规则库。外部系统数据库中的历史数据不会自动参与评分；如需使用，双方另行约定数据同步或 `externalEvidence` 扩展协议。
+
+## 2. 接口信息
+
+| 项目 | 内容 |
+|---|---|
+| 请求方式 | `POST` |
+| 请求路径 | `/api/v1/integrations/risk-assessments/report` |
+| 请求类型 | `application/json` |
+| 成功响应类型 | `text/markdown; charset=utf-8` |
+| 字符编码 | UTF-8 |
+| 接口超时时间建议 | 120 秒 |
+| 幂等依据 | `sourceSystem + requestId`；`Idempotency-Key` 必须与 `requestId` 一致 |
+
+生产环境完整地址由蝶雅平台提供，例如：
+
+```text
+https://risk.example.com/api/v1/integrations/risk-assessments/report
+```
+
+## 3. 调用鉴权
+
+接口使用“调用方系统身份 + HMAC-SHA256 签名”鉴权。外部系统不使用蝶雅平台用户 JWT，也不直接传入蝶雅平台本地 `userId`。
+
+### 3.1 请求头
+
+| 请求头 | 必填 | 说明 |
+|---|---:|---|
+| `Content-Type` | 是 | 固定为 `application/json` |
+| `X-Client-Id` | 是 | 蝶雅平台分配的调用方系统编号，例如 `dieya-plm` |
+| `X-Timestamp` | 是 | Unix 时间戳，单位为秒 |
+| `X-Nonce` | 是 | 8～64 位随机字符串，仅允许字母、数字、下划线、短横线，单次请求唯一 |
+| `X-Signature` | 是 | HMAC-SHA256 签名，固定使用标准 Base64 编码（带填充 `=`） |
+| `Idempotency-Key` | 是 | 与请求体中的 `requestId` 保持一致 |
+
+### 3.2 签名计算
+
+蝶雅平台为每个 `X-Client-Id` 分配独立密钥 `clientSecret`。POST 调用使用以下原文计算签名：
+
+```text
+X-Timestamp + "\n" + X-Nonce + "\n" + 原始请求Body
+```
+
+伪代码：
+
+```python
+message = f"{timestamp}\n{nonce}\n{raw_body}"
+signature = base64.b64encode(
+ hmac.new(client_secret.encode("utf-8"), message.encode("utf-8"), hashlib.sha256).digest()
+).decode("ascii")
+```
+
+注意事项：
+
+- 签名必须基于发送时的原始 JSON 字节内容计算，不能先格式化后再计算。
+- 服务端默认只接受时间误差不超过 5 分钟的请求。
+- 已认证请求的 `X-Nonce` 在数据库中保留 10 分钟，跨进程和重启后仍拒绝重复使用；正常 JSON 的字段校验失败也会消耗 Nonce。
+- `clientSecret` 只通过双方约定的安全渠道保存，不放在前端代码、日志或请求体中。
+- 生产环境建议叠加 IP 白名单；高安全场景可使用 mTLS。
+
+### 3.3 用户身份映射
+
+请求中的用户身份使用钉钉标识：
+
+- `dingtalkUnionId`：首选跨系统关联标识。
+- `dingtalkUserId`：钉钉用户 ID，作为辅助校验和追踪字段。
+- `employeeNo`、`name`：展示和审计信息，不能单独作为身份依据。
+
+蝶雅平台按以下顺序处理：
+
+1. 按 `dingtalkUnionId` 查找本地用户。
+2. 必要时结合 `dingtalkUserId` 校验用户身份。
+3. 用户不存在、停用或 UnionId 匹配到多个本地用户时返回 `403`，不自动创建账号。
+4. 评分、审计和报告记录使用蝶雅平台本地用户 ID；该 ID 不返回给外部系统作为跨库主键。
+
+用户需要拥有本地 `risk.assess` 操作权限（管理员直接允许）。如传入 `dingtalkUserId`，必须与本地用户一致。报告中的姓名使用本地用户资料，不使用调用方提交的姓名作为认证或显示依据。查询报告时再次校验原操作人的账号状态及权限。
+
+### 3.4 非钉钉操作人
+
+没有钉钉身份的用户可使用 `operator.externalUserId`，前提是风险平台管理员已为当前调用系统批准映射。两种身份模式必须且只能选一种；外部账号模式不传 `dingtalkUnionId`、`dingtalkUserId`，可以省略或传 `null`。不允许按姓名、员工编号自动匹配用户。
+
+例如 PLM 账号 `superadmin`（周子墨）的操作人参数为：
+
+```json
+{
+ "externalUserId": "superadmin",
+ "name": "周子墨"
+}
+```
+
+这个对象替换请求中的整个 `operator`，其他字段及系统签名要求不变。PLM 后端必须从当前已登录用户取出该账号标识，不能让前端任意指定别人的 `externalUserId`。平台通过 `X-Client-Id + externalUserId` 查映射并检查本地账号及权限，未批准映射返回 `403`。外部系统的管理员角色不授予风险平台管理员权限。
+
+GET 查询时也会重新校验记录中的外部账号映射；撤销映射或停用本地账号后禁止查询。
+
+如果两个钉钉应用返回的 UnionId 不一致，双方需要先确定统一的钉钉身份源或增加统一身份映射表。
+
+## 4. 请求参数
+
+### 4.1 请求体
+
+```json
+{
+ "requestId": "PLM-D-FT-003-0001",
+ "sourceSystem": "dieya-plm",
+ "operator": {
+ "dingtalkUnionId": "ding_union_xxx",
+ "dingtalkUserId": "ding_user_xxx",
+ "employeeNo": "A10001",
+ "name": "张三"
+ },
+ "assessment": {
+ "subjectId": "SKU-20261007-001",
+ "productName": "新品洗发水",
+ "productCategory": "洗发水/洗护类",
+ "specification": "500ml",
+ "materialType": "PET瓶/泵头",
+ "supplier": "某某供应商",
+ "factory": "某某加工厂",
+ "stageNew": "产品开发",
+ "defectLevel": null,
+ "description": "新品上市前风险评估",
+ "asOfDate": "2026-10-07"
+ }
+}
+```
+
+### 4.2 顶层字段
+
+| 字段 | 类型 | 必填 | 说明 | 校验建议 |
+|---|---|---:|---|---|
+| `requestId` | string | 是 | 调用方请求唯一编号，用于幂等 | 1～128 个字符，仅允许字母、数字、`_`、`.`、`:`、`-` |
+| `sourceSystem` | string | 是 | 来源系统编号 | 必须是已登记的 `X-Client-Id` |
+| `operator` | object | 是 | 发起评估的钉钉用户 | 见下表 |
+| `assessment` | object | 是 | 风险评估数据 | 见下表 |
+
+### 4.3 `operator` 字段
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---:|---|
+| `dingtalkUnionId` | string/null | 条件必填 | 钉钉用户模式必填，与 `externalUserId` 二选一 |
+| `externalUserId` | string/null | 条件必填 | 非钉钉用户的调用方稳定账号标识，1～128 字符，必须在当前调用系统下预先批准映射 |
+| `dingtalkUserId` | string | 建议 | 钉钉用户 ID |
+| `employeeNo` | string | 否 | 员工编号，用于审计展示 |
+| `name` | string | 否 | 用户姓名，用于报告展示 |
+
+### 4.4 `assessment` 字段
+
+| 字段 | 类型 | 必填 | 说明 |
+|---|---|---:|---|
+| `subjectId` | string | 是 | 产品、SKU 或评估对象编号 |
+| `productName` | string | 否 | 产品名称 |
+| `productCategory` | string | 是 | 按当前风险评分的八类传值，完整清单及兼容规则见 4.6 |
+| `specification` | string | 否 | 规格或容量，例如 `500ml`、`100g` |
+| `materialType` | string | 是 | 包装材质或物料组合，例如 `PET瓶/泵头` |
+| `supplier` | string | 否 | 供应商名称 |
+| `factory` | string | 否 | 加工厂名称 |
+| `stageNew` | string | 否 | 开发阶段 |
+| `defectLevel` | string/null | 否 | 缺陷等级，可取 `A`、`B`、`C` |
+| `description` | string | 否 | 评估背景或补充说明 |
+| `asOfDate` | string/null | 否 | 评估数据截止日期，格式 `YYYY-MM-DD`，不早于 1900-01-01、不允许未来日期；省略时使用首次受理的上海日期 |
+
+`stageNew` 建议使用以下枚举值：
+
+```text
+需求发现
+产品立项
+产品开发
+量产上市
+产品生命管理
+```
+
+### 4.5 字段约束
+
+- `productCategory`、`materialType`、`subjectId` 缺失时，接口返回 `422`。
+- `description` 仅作为评估背景，不直接作为已确认的检测结果或历史事实。
+- 调用方不要提交最终风险分数、风险等级或模型版本；这些字段由蝶雅平台统一计算。
+- 当前版本不接收外部数据库连接信息、SQL、文件路径或可执行脚本。
+- 可选字段允许省略或传 `null`。`stageNew`、`defectLevel` 非空时必须符合枚举；未知字段（包括调用方提交的分数、角色）返回 `422`。
+- 参数长度：`subjectId` 128、`productName` 200、`productCategory` 100、`specification` 100、`materialType` 200、`supplier/factory` 300、`description` 2000；`dingtalkUnionId` 64、`dingtalkUserId` 50、`employeeNo` 64、`name` 50。请求体最大 64KB。
+
+### 4.6 产品品类字典（当前评分口径）
+
+本次 PLM 对接采用现有七维风险评分服务 `normalize_category()` 的八类口径。以下为当前评分分类清单，并非数据库字典接口的返回结果。
+
+PLM 优先将产品品类映射成下表的标准值后提交 `assessment.productCategory`，并保存映射关系。无法确定具体类别时，不应仅凭 PLM 的宽泛业务大类强行映射。
+
+| 标准值（建议提交值） | 当前兼容识别关键词 |
+|---|---|
+| `洗发水/洗护类` | 洗发、洗护、洗发水、洗发露、洗发乳 |
+| `沐浴露类` | 沐浴、沐浴露 |
+| `面霜/护肤类` | 面霜、乳液、精华、面膜、护肤 |
+| `牙膏/口腔护理类` | 牙膏、口腔 |
+| `清洁剂类` | 清洁剂、清洁、洗洁 |
+| `脱毛类` | 脱毛 |
+| `彩妆/唇部类` | 彩妆、口红、粉底、眼影 |
+| `香氛/止汗类` | 香氛、香水、止汗 |
+
+匹配规则及评分边界：
+
+- 当前按上表顺序进行关键词子串匹配，多个类别同时命中时取第一类。
+- 空值仍按本接口必填规则返回 `422`。非空值无法识别时，按现有评分逻辑将“产品品类”维度设为默认 50 分，并在报告中注明无法归类；这不代表产品风险已经由历史数据验证。
+- 已识别品类的历史异常案例少于 10 条时，该维度按 45 分处理；其余情况按现有异常数量归一化规则计算。
+- 当前 `防晒`、`洁面`、`护发素` 等名称本身没有专门的评分匹配规则；包含其他兼容关键词的完整名称仍按现有规则归类，例如“防晒乳液”命中“乳液”。
+- `个护清洁` 会命中“清洁”并归为“清洁剂类”，不能将该 PLM 大类整体作为防晒等产品的评分分类。
+- 报告应同时展示调用方提交的品类和实际评分归类；无法识别时注明默认分及原因。
+
+历史案例的品类维度统计优先归一化案例的 `product_category`，无法识别时再尝试案例的 `product_name`。本对接口径采用评分规则；看板中的“身体护理类”“洁面/卸妆类”“头发护理类”“包材/物料专项”“待补标品类”不属于当前八类评分字典。
+
+## 5. 成功响应
+
+### 5.1 响应头
+
+```http
+HTTP/1.1 200 OK
+Content-Type: text/markdown; charset=utf-8
+X-Request-Id: PLM-D-FT-003-0001
+X-Assessment-Id: 10086
+X-Model-Version: database-seven-dimension-v1
+X-Risk-Score: 72.50
+X-Risk-Level: moderately_low
+```
+
+### 5.2 响应体
+
+响应体直接为 Markdown 文本，不再包裹 `data`、`code` 等 JSON 字段：
+
+```markdown
+# 新品风险评估报告
+
+## 一、评估信息
+
+| 项目 | 内容 |
+|---|---|
+| 评估对象 | 新品洗发水 |
+| 对象编号 | SKU-20261007-001 |
+| 产品品类 | 洗发水/洗护类 |
+| 规格 | 500ml |
+| 包装材质 | PET瓶/泵头 |
+| 供应商 | 某某供应商 |
+| 加工厂 | 某某加工厂 |
+| 开发阶段 | 产品开发 |
+| 评估人 | 张三 |
+| 评估时间 | 2026-10-07 |
+
+## 二、综合风险结论
+
+- 综合得分：72.50
+- 风险等级：较低风险
+- 模型版本：database-seven-dimension-v1
+
+## 三、七维风险评分
+
+| 风险维度 | 得分 | 权重 | 数据状态 | 说明 |
+|---|---:|---:|---|---|
+| 供应商表现 | 80.00 | 25% | database | 近三年采购订单和异常案例统计 |
+| 物料品类 | 75.00 | 20% | database | 包装材质历史异常统计 |
+| 品类历史 | 70.00 | 15% | database | 产品品类历史案例统计 |
+
+## 四、相似异常案例
+
+暂无直接匹配案例，建议结合相似品类进行验证。
+
+## 五、标准规则
+
+| 风险点 | 判定规则 | 来源 | 状态 |
+|---|---|---|---|
+| 密封性 | 按标准要求完成密封性验证 | 标准文件.xlsx | 待检测 |
+
+## 六、风险预防建议
+
+1. 打样阶段完成内容物与包装材料相容性测试。
+2. 增加密封性和加速老化测试。
+3. 首单加强来料检验和过程巡检。
+
+## 七、数据口径说明
+
+本报告基于蝶雅风险平台本地数据库中的历史案例、采购订单和标准库生成。
+部分维度无实际数据时，按系统配置的中性值计算。
+```
+
+报告中的分数是安全得分，分数越高代表风险越低。风险等级阈值以蝶雅平台当前模型配置为准，调用方不要自行重新计算等级。
+
+`X-Risk-Score` 为 0～100 的安全分，保留两位小数。`X-Risk-Level` 为 `low`（低风险）、`moderately_low`（较低风险）、`medium`（中等风险）、`high`（高风险）、`very_high`（极高风险）。`X-Assessment-Id` 按字符串存储。GET 和幂等重试返回原报告及相同结论响应头。
+
+## 6. 幂等规则
+
+`requestId` 是调用方生成的业务请求唯一编号，同一个 `sourceSystem + requestId` 只能对应一次评估请求。
+
+### 首次请求
+
+返回 `200` 和 Markdown 报告。
+
+### 相同请求重复提交
+
+如果请求体和原请求一致，服务端返回同一份已生成报告，仍返回 `200`，并在响应头中带回原始 `X-Assessment-Id`。
+
+### 同一编号提交不同内容
+
+如果 `requestId` 已经使用，但请求内容摘要不同，返回 `409`，避免覆盖历史报告。
+
+### 处理中、失败及重试
+
+- 幂等比较基于校验后的 JSON 对象摘要，忽略 JSON 键顺序和排版，省略的可选字段与 `null` 等价；变更任意有效字段视为不同请求。
+- 重试保留原请求体（包括操作人和 `asOfDate`），重新生成时间戳、Nonce 和签名。主动重新评估必须使用新 `requestId`。
+- 同一请求仍在生成时，POST 或 GET 返回 `202` JSON 和 `Retry-After: 5`，不会重复计算。
+- 每次生成最多 110 秒（可配置为 10～110 秒），超时或失败记录为 failed，返回 `503`；相同请求内容可以重新 POST 重试。生成权租约为 150 秒，进程异常退出后租约到期可重新 POST 接管。
+- 调用方网络超时并不代表服务端已经停止。应先 GET 查询，再决定是否使用原请求重试。
+- 省略 `asOfDate` 时首次受理日期单独持久化，跨天重试仍沿用首次日期。
+
+处理中响应示例：
+
+```json
+{"code":202,"message":"报告正在生成，请稍后查询","data":{"requestId":"PLM-D-FT-003-0001"}}
+```
+
+### 报告查询
+
+```http
+GET /api/v1/integrations/risk-assessments/report?requestId=PLM-D-FT-003-0001
+```
+
+查询参数必须且只能包含一个 `requestId`；GET 不携带 Body，也不要求 `Idempotency-Key`。仅能查询与 `X-Client-Id` 相同来源的记录。成功返回 `200` Markdown，处理中返回 `202`，失败返回 `503`，不存在返回 `404`。
+
+GET 签名原文为：
+
+```text
+"GET" + "\n" + 请求路径 + "\n" + 原始查询字符串 + "\n" + X-Timestamp + "\n" + X-Nonce + "\n"
+```
+
+请求路径包含 `/api/v1`，不含域名、查询字符串；原始查询字符串不含 `?`，包括实际发送的 URL 编码，签名后不得重新编码。网关不能重写对外接口路径。以 UTF-8 字节计算 HMAC-SHA256，再以标准 Base64 编码。报告持久化保存，当前未启用自动清理策略。
+
+## 7. 错误响应
+
+错误响应统一使用 JSON：
+
+```json
+{
+ "code": 422,
+ "message": "产品品类不能为空",
+ "data": null
+}
+```
+
+| HTTP 状态码 | `code` | 场景 | 调用方处理建议 |
+|---:|---:|---|---|
+| `400` | `400` | 请求格式错误或签名原文不合法 | 修正请求格式后重试 |
+| `401` | `401` | 缺少签名、签名错误、时间戳过期 | 检查系统密钥和签名算法 |
+| `403` | `403` | 调用方未开通接口或用户无风险评估权限 | 联系平台管理员开通权限 |
+| `404` | `404` | 当前调用系统未找到该请求 | 核对 requestId 和调用系统 |
+| `409` | `409` | `requestId` 重复但内容不一致 | 更换请求编号或查询原请求 |
+| `422` | `422` | 参数缺失或字段值不符合约束 | 修正参数后重试 |
+| `429` | `429` | 超过调用频率限制 | 按 `Retry-After` 延迟重试 |
+| `500` | `500` | 平台内部处理异常 | 使用同一 `requestId` 重试并联系平台方 |
+| `503` | `503` | 风险数据库、模型或依赖服务暂不可用 | 稍后重试，不要生成新的业务编号 |
+
+## 8. 调用示例
+
+### 8.1 cURL
+
+```bash
+curl -X POST 'https://risk.example.com/api/v1/integrations/risk-assessments/report' \
+ -H 'Content-Type: application/json' \
+ -H 'X-Client-Id: dieya-plm' \
+ -H 'X-Timestamp: 1791331200' \
+ -H 'X-Nonce: 8f3d4c1a9e' \
+ -H 'X-Signature: Base64HmacSha256Value' \
+ -H 'Idempotency-Key: PLM-D-FT-003-0001' \
+ --data-raw '{
+ "requestId": "PLM-D-FT-003-0001",
+ "sourceSystem": "dieya-plm",
+ "operator": {
+ "dingtalkUnionId": "ding_union_xxx",
+ "dingtalkUserId": "ding_user_xxx",
+ "employeeNo": "A10001",
+ "name": "张三"
+ },
+ "assessment": {
+ "subjectId": "SKU-20261007-001",
+ "productName": "新品洗发水",
+ "productCategory": "洗发水/洗护类",
+ "specification": "500ml",
+ "materialType": "PET瓶/泵头",
+ "supplier": "某某供应商",
+ "factory": "某某加工厂",
+ "stageNew": "产品开发",
+ "description": "新品上市前风险评估",
+ "asOfDate": "2026-10-07"
+ }
+ }' \
+ --output risk-report.md
+```
+
+### 8.2 Python 签名示例
+
+```python
+import base64
+import hashlib
+import hmac
+import json
+import secrets
+import time
+
+body = {
+ "requestId": "PLM-D-FT-003-0001",
+ "sourceSystem": "dieya-plm",
+ "operator": {"dingtalkUnionId": "ding_union_xxx"},
+ "assessment": {
+ "subjectId": "SKU-20261007-001",
+ "productCategory": "洗发水/洗护类",
+ "materialType": "PET瓶/泵头",
+ },
+}
+raw_body = json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+timestamp = str(int(time.time()))
+nonce = secrets.token_urlsafe(16)
+message = timestamp.encode() + b"\n" + nonce.encode() + b"\n" + raw_body
+signature = base64.b64encode(
+ hmac.new(client_secret.encode(), message, hashlib.sha256).digest()
+).decode()
+```
+
+调用方必须使用计算签名时的同一份 `raw_body` 发送请求。
+
+## 9. 评分和报告处理说明
+
+平台内部处理逻辑建议统一复用现有七维风险评分服务：
+
+- 供应商表现
+- 物料品类
+- 品类历史
+- 加工厂质量
+- 标准符合性
+- 节点阶段
+- 缺陷严重度
+
+平台会在报告中标注数据状态。没有真实数据库数据的维度不得伪造历史结论，按模型配置的中性值处理并在报告中说明。
+
+外部接口使用结构化参数直接评分，不再通过聊天语义提取参数。沿用现有评分区间起点（截止年份往前 3 年的 1 月 1 日），区间终点为 `asOfDate`，不包含截止日期之后的案例及采购订单。相似案例使用现有检索服务的文本和业务字段排序，输入同一统计区间内的案例；本接口不调用向量嵌入服务。
+
+包装组合沿用现有单标签评分口径，例如 `PET 瓶身 + PP 泵头` 命中“泵头”，不代表分别评估了所有组件。报告展示原文及评分标签。缺陷维度来自历史 A/B/C 记录，不预测新品实际缺陷等级。
+
+AI 只用于生成预防建议，等待最多 20 秒。AI 不可用时仍返回数据库评分及固定验证建议，并在报告中标注；数据库评分失败则返回 `503`，不生成模拟分数。
+
+报告建议包含：
+
+1. 评估信息。
+2. 综合风险得分和风险等级。
+3. 七维评分明细。
+4. 品类、供应商和物料历史统计。
+5. 相似异常案例。
+6. 标准规则匹配结果。
+7. 风险预防建议。
+8. 数据口径和模型版本。
+
+## 10. 联调前需要双方确认的事项
+
+1. 正式接口域名、环境地址和接口版本。
+2. `sourceSystem`、`X-Client-Id` 和密钥分配方式。
+3. 钉钉 `unionId` 是否可以作为双方统一用户标识；如果不能，需要提供统一身份映射方案。
+4. 操作人提前开通、UnionId 唯一映射及 `risk.assess` 权限；本接口已采用不存在返回 `403` 的策略。
+5. 是否启用 IP 白名单、mTLS 和调用频率限制。
+6. `stageNew`、`defectLevel` 枚举，以及 PLM 到第 4.6 节八类评分品类的映射关系；确认无法归类时的默认分提示。
+7. 外部系统历史数据是否需要同步到蝶雅平台。
+8. Markdown 报告的保存、下载和展示方式。
+9. 超时、重试和 `requestId` 查询机制。
+
+## 11. 服务端启用配置
+
+先执行迁移：
+
+```powershell
+python -m alembic upgrade head
+```
+
+新增表为 `integration_nonces` 和 `integration_assessments`，分别保存已认证随机串及请求快照、处理状态、评分结果和 Markdown 报告；成功评估同时写入操作审计。现有用户表不自动添加外部用户。报告记录通过本地用户编号逻辑关联，不对已有用户表创建外键，迁移账号不需要该表的 `REFERENCES` 权限；用户状态和权限仍由接口校验。
+
+在服务端环境配置中设置：
+
+```dotenv
+RISK_INTEGRATION_CLIENTS={"dieya-plm":"replace-with-at-least-32-random-characters"}
+RISK_INTEGRATION_REQUESTS_PER_MINUTE=60
+RISK_INTEGRATION_REPORT_TIMEOUT_SECONDS=110
+```
+
+上面的密钥仅为占位，须替换为至少 32 个字符的随机密钥；测试、正式各使用一套。调用系统编号仅允许字母、数字、下划线和短横线，最长 64 位。默认 `{}` 不允许任何系统调用。配置后重启服务。
+
+也可以在后端项目根目录使用启用工具生成密钥，工具会保留 `.env` 中的其他设置及已有系统密钥，不打印密钥、不自动轮换已有密钥：
+
+```powershell
+.\.venv\Scripts\python.exe scripts/setup_external_risk.py configure --client-id dieya-plm
+```
+
+让操作人在钉钉工作台中登录风险平台，正常免登会同步用户资料和 UnionId。然后使用 PLM 保存的真实 UnionId 检查映射与权限：
+
+```powershell
+.\.venv\Scripts\python.exe scripts/setup_external_risk.py check-user --union-id "PLM操作人的真实UnionId" --dingtalk-user-id "同一人的真实UserId"
+```
+
+工具只检查、不修改账号或权限。不存在时先通过钉钉免登开通；已停用或无权限时由账号管理员确认处理；映射不唯一或身份不一致时核对两边钉钉身份源，不能按姓名猜测绑定。当前启用的 admin、product_manager、reviewer、user 角色均有风险评估操作权限，以本地 `role_permissions` 表实际记录为准。
+
+限流按调用系统统计已通过签名的请求，POST、GET、幂等重试共用滚动 60 秒窗口；达到阈值返回 `429` 和 `Retry-After: 60`。IP 白名单和 mTLS 可在网关配置，当前应用层不提供这两项配置。
+
+## 12. 联调调用工具
+
+非钉钉用户需由风险平台管理员在本地显式开通，例如：
+
+```powershell
+.\.venv\Scripts\python.exe scripts/setup_external_risk.py allow-user --client-id dieya-plm --external-user-id superadmin --employee-no plm_superadmin --name "周子墨"
+```
+
+工具创建无密码、无钉钉绑定的普通本地账号，并写入 `.env`：
+
+```dotenv
+RISK_INTEGRATION_OPERATOR_MAPPINGS='{"dieya-plm":{"superadmin":"plm_superadmin"}}'
+```
+
+映射目标是本地 `emp_no`。默认 `{}` 禁止所有非钉钉账号；已有映射不会自动指向其他账号，已有账号不会自动更改角色、姓名或启用状态。配置后重启后端，无需数据库迁移。
+
+项目提供 `scripts/call_external_risk_report.py`，自动生成 POST/GET 签名并保存 Markdown。调用方将本系统分配的密钥写入环境变量 `RISK_INTEGRATION_CLIENT_SECRET`（客户端单个密钥，与服务端 JSON 密钥表是不同配置）。`assessment.json` 使用本文第 4.1 节请求结构，填入已开通的真实 UnionId。
+
+```powershell
+python scripts/call_external_risk_report.py --url https://risk.example.com/api/v1/integrations/risk-assessments/report --client-id dieya-plm --body-file assessment.json --output risk-report.md
+```
+
+查询原请求：
+
+```powershell
+python scripts/call_external_risk_report.py --url https://risk.example.com/api/v1/integrations/risk-assessments/report --client-id dieya-plm --request-id PLM-D-FT-003-0001 --output risk-report.md
+```
+
+只有 `200` 会保存报告；`202`/错误响应会展示 JSON 和重试等待时间，不会覆盖已有 Markdown。
diff --git a/migrations/versions/20261008_0022_integration_reports.py b/migrations/versions/20261008_0022_integration_reports.py
new file mode 100644
index 0000000..07716b2
--- /dev/null
+++ b/migrations/versions/20261008_0022_integration_reports.py
@@ -0,0 +1,49 @@
+"""External reports, idempotency leases and durable HMAC replay protection."""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision = "20261008_0022"
+down_revision = "20260911_0021"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+ op.create_table(
+ "integration_nonces",
+ sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
+ sa.Column("client_id", sa.String(64), nullable=False),
+ sa.Column("nonce", sa.String(64), nullable=False),
+ sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+ sa.UniqueConstraint("client_id", "nonce", name="uq_integration_nonce"),
+ )
+ op.create_index("ix_integration_nonce_expires", "integration_nonces", ["expires_at"])
+ op.create_index("ix_integration_nonce_client_created", "integration_nonces", ["client_id", "created_at"])
+ op.create_table(
+ "integration_assessments",
+ sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
+ sa.Column("client_id", sa.String(64), nullable=False),
+ sa.Column("request_id", sa.String(128), nullable=False),
+ sa.Column("user_id", sa.BigInteger(), nullable=False),
+ sa.Column("request_hash", sa.String(64), nullable=False),
+ sa.Column("request_data", postgresql.JSONB(), nullable=False),
+ sa.Column("as_of_date", sa.Date(), nullable=False),
+ sa.Column("status", sa.String(20), nullable=False),
+ sa.Column("owner_token", sa.String(32), nullable=False),
+ sa.Column("lease_until", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("result_data", postgresql.JSONB()),
+ sa.Column("report_markdown", sa.Text()),
+ sa.Column("error_message", sa.String(200)),
+ sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+ sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+ sa.UniqueConstraint("client_id", "request_id", name="uq_integration_assessment_request"),
+ )
+ op.create_index("ix_integration_assessment_user", "integration_assessments", ["user_id"])
+
+
+def downgrade() -> None:
+ op.drop_table("integration_assessments")
+ op.drop_table("integration_nonces")
diff --git a/scripts/call_external_risk_report.py b/scripts/call_external_risk_report.py
new file mode 100644
index 0000000..e4209a3
--- /dev/null
+++ b/scripts/call_external_risk_report.py
@@ -0,0 +1,68 @@
+"""POST an assessment or GET a persisted report using the documented HMAC protocol."""
+
+import argparse
+import base64
+import hashlib
+import hmac
+import os
+import secrets
+import time
+from pathlib import Path
+from urllib.parse import urlencode, urlsplit
+
+import httpx
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--url", required=True, help="Full /api/v1/integrations/risk-assessments/report URL")
+ parser.add_argument("--client-id", default="dieya-plm")
+ inputs = parser.add_mutually_exclusive_group(required=True)
+ inputs.add_argument("--body-file", type=Path, help="UTF-8 JSON assessment to POST")
+ inputs.add_argument("--request-id", help="Existing requestId to GET")
+ parser.add_argument("--output", type=Path, default=Path("risk-report.md"))
+ args = parser.parse_args()
+ secret = os.environ.get("RISK_INTEGRATION_CLIENT_SECRET", "")
+ if len(secret) < 32:
+ parser.error("Set RISK_INTEGRATION_CLIENT_SECRET to the system's secret (at least 32 characters)")
+ url = urlsplit(args.url)
+ if url.scheme not in {"https", "http"} or url.query or url.fragment:
+ parser.error("--url must be an HTTP(S) endpoint without query or fragment")
+ timestamp, nonce = str(int(time.time())), secrets.token_urlsafe(16)
+ message = timestamp.encode() + b"\n" + nonce.encode() + b"\n"
+ headers = {"X-Client-Id": args.client_id, "X-Timestamp": timestamp, "X-Nonce": nonce}
+ if args.body_file:
+ import json
+
+ body = args.body_file.read_bytes()
+ data = json.loads(body)
+ headers.update({"Content-Type": "application/json", "Idempotency-Key": data["requestId"]})
+ message += body
+ method, target = "POST", args.url
+ else:
+ body = b""
+ query = urlencode({"requestId": args.request_id})
+ message = b"GET\n" + url.path.encode() + b"\n" + query.encode() + b"\n" + message
+ method, target = "GET", args.url + "?" + query
+ headers["X-Signature"] = base64.b64encode(hmac.new(secret.encode(), message, hashlib.sha256).digest()).decode()
+ try:
+ response = httpx.request(method, target, content=body, headers=headers, timeout=120)
+ except httpx.HTTPError as exc:
+ print(f"Request did not complete ({exc.__class__.__name__}); query the same requestId before retrying.")
+ return 1
+ if response.status_code != 200:
+ print(f"HTTP {response.status_code}: {response.text}")
+ if response.headers.get("Retry-After"):
+ print(f"Retry-After: {response.headers['Retry-After']} seconds")
+ return 1
+ args.output.write_text(response.text, encoding="utf-8")
+ print(f"Saved {args.output.resolve()}")
+ print(
+ f"Assessment: {response.headers.get('X-Assessment-Id')}, "
+ f"score: {response.headers.get('X-Risk-Score')}, level: {response.headers.get('X-Risk-Level')}"
+ )
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/setup_external_risk.py b/scripts/setup_external_risk.py
new file mode 100644
index 0000000..89f126d
--- /dev/null
+++ b/scripts/setup_external_risk.py
@@ -0,0 +1,177 @@
+"""Configure a backend integration secret or verify an existing DingTalk operator."""
+
+import argparse
+import asyncio
+import json
+import re
+import secrets
+from pathlib import Path
+
+from app.core.exceptions import AppException
+from app.database.session import SessionFactory, engine
+from app.models.entities import UserAccount
+from app.repositories.users import UserRepository
+from app.schemas.integration_risk import IntegrationOperator
+from app.services.integration_auth import authorize_operator
+from dotenv import dotenv_values
+from sqlalchemy import select
+
+
+def configure_client(env_file: Path, client_id: str) -> bool:
+ """Keep existing secrets and unrelated settings; never print secret material."""
+ if not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", client_id):
+ raise ValueError("client-id 仅允许字母、数字、下划线和短横线，最长 64 位")
+ text = env_file.read_text(encoding="utf-8-sig") if env_file.exists() else ""
+ raw = dotenv_values(env_file).get("RISK_INTEGRATION_CLIENTS") if env_file.exists() else None
+ clients = json.loads(raw or "{}")
+ if not isinstance(clients, dict) or any(
+ not isinstance(key, str)
+ or not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", key)
+ or not isinstance(value, str)
+ or len(value) < 32
+ for key, value in clients.items()
+ ):
+ raise ValueError("已有 RISK_INTEGRATION_CLIENTS 配置无效，请检查 JSON 和密钥长度")
+ if client_id in clients:
+ return False
+ clients[client_id] = secrets.token_urlsafe(32)
+ # Prefer readable single-quoted JSON; double-escape if an existing secret has an apostrophe.
+ encoded = json.dumps(clients, ensure_ascii=True, separators=(",", ":"))
+ value = "'" + encoded + "'" if "'" not in encoded else json.dumps(encoded, ensure_ascii=True)
+ line = f"RISK_INTEGRATION_CLIENTS={value}"
+ lines = text.splitlines()
+ indexes = [
+ i for i, current in enumerate(lines) if re.match(r"^\s*(?:export\s+)?RISK_INTEGRATION_CLIENTS\s*=", current)
+ ]
+ if indexes:
+ lines[indexes[0]] = line
+ lines = [current for i, current in enumerate(lines) if i not in indexes[1:]]
+ else:
+ lines.extend(["", "# 外部系统调用密钥（仅保存在双方后端）", line])
+ env_file.write_text("\n".join(lines) + "\n", encoding="utf-8")
+ return True
+
+
+async def check_operator(union_id: str, user_id: str | None) -> None:
+ try:
+ async with SessionFactory() as session:
+ user = await authorize_operator(
+ session,
+ IntegrationOperator(
+ dingtalk_union_id=union_id,
+ dingtalk_user_id=user_id,
+ ),
+ )
+ print(f"检查通过：本地用户 {user.id}，姓名 {user.name}，角色 {user.role}，账号启用，有风险评估操作权限。")
+ finally:
+ await engine.dispose()
+
+
+def read_operator_mappings(env_file: Path) -> dict:
+ raw = dotenv_values(env_file).get("RISK_INTEGRATION_OPERATOR_MAPPINGS") if env_file.exists() else None
+ mappings = json.loads(raw or "{}")
+ if not isinstance(mappings, dict) or any(
+ not isinstance(client, str)
+ or not isinstance(accounts, dict)
+ or any(
+ not isinstance(external, str) or not external or not isinstance(local, str) or not local or len(local) > 20
+ for external, local in accounts.items()
+ )
+ for client, accounts in mappings.items()
+ ):
+ raise ValueError("外部账号映射配置无效")
+ return mappings
+
+
+def write_operator_mapping(env_file: Path, client_id: str, external_id: str, employee_no: str) -> None:
+ mappings = read_operator_mappings(env_file)
+ accounts = mappings.setdefault(client_id, {})
+ if external_id in accounts and accounts[external_id] != employee_no:
+ raise ValueError("外部账号已映射到其他本地账号，不能自动覆盖")
+ accounts[external_id] = employee_no
+ encoded = json.dumps(mappings, ensure_ascii=True, separators=(",", ":"))
+ value = "'" + encoded + "'" if "'" not in encoded else json.dumps(encoded)
+ key = "RISK_INTEGRATION_OPERATOR_MAPPINGS"
+ lines = env_file.read_text(encoding="utf-8-sig").splitlines() if env_file.exists() else []
+ indexes = [i for i, line in enumerate(lines) if re.match(r"^\s*(?:export\s+)?" + key + r"\s*=", line)]
+ if indexes:
+ lines[indexes[0]] = f"{key}={value}"
+ lines = [line for i, line in enumerate(lines) if i not in indexes[1:]]
+ else:
+ lines.extend(["", "# 已批准的外部操作人映射", f"{key}={value}"])
+ env_file.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+async def allow_external_user(env_file: Path, client_id: str, external_id: str, employee_no: str, name: str) -> None:
+ if not re.fullmatch(r"[A-Za-z0-9_-]{1,64}", client_id):
+ raise ValueError("调用系统格式错误")
+ if not external_id.strip() or len(external_id) > 128 or not name.strip() or len(name) > 50:
+ raise ValueError("账号标识或姓名格式错误")
+ if not re.fullmatch(r"[A-Za-z0-9_-]{1,20}", employee_no):
+ raise ValueError("本地账号编号格式错误")
+ mappings = read_operator_mappings(env_file)
+ existing = mappings.get(client_id, {}).get(external_id)
+ if existing and existing != employee_no:
+ raise ValueError("已有映射不能自动覆盖")
+ clients = json.loads(dotenv_values(env_file).get("RISK_INTEGRATION_CLIENTS") or "{}")
+ if client_id not in clients:
+ raise ValueError("请先配置调用系统密钥")
+ try:
+ async with SessionFactory() as session:
+ user = await session.scalar(select(UserAccount).where(UserAccount.emp_no == employee_no))
+ if user is None:
+ permissions = await UserRepository(session).permissions_for_role("user", operate=True)
+ if not any(item.func_key == "risk.assess" for item in permissions):
+ raise ValueError("普通角色无评估权限，请由管理员配置")
+ user = UserAccount(emp_no=employee_no, name=name, role="user", status="启用")
+ session.add(user)
+ await session.flush()
+ elif user.name != name or user.status != "启用":
+ raise ValueError("已有本地账号姓名不一致或已停用，不能自动修改")
+ user_id = user.id
+ await session.commit()
+ write_operator_mapping(env_file, client_id, external_id, employee_no)
+ print(f"已配置外部用户：{name}（{external_id}），调用系统 {client_id}，本地用户编号 {user_id}。")
+ print("保持本地账号权限；新建账号为普通用户，无密码及钉钉绑定。请重启后端。")
+ finally:
+ await engine.dispose()
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ actions = parser.add_subparsers(dest="action", required=True)
+ configure = actions.add_parser("configure", help="Generate a missing client secret without printing it")
+ configure.add_argument("--client-id", default="dieya-plm")
+ configure.add_argument("--env-file", type=Path, default=Path(".env"))
+ check = actions.add_parser("check-user", help="Check UnionId, optional UserId and local permission")
+ check.add_argument("--union-id", required=True)
+ check.add_argument("--dingtalk-user-id")
+ allow = actions.add_parser("allow-user", help="Explicitly register a non-DingTalk operator for one client")
+ allow.add_argument("--client-id", default="dieya-plm")
+ allow.add_argument("--external-user-id", required=True)
+ allow.add_argument("--employee-no", required=True)
+ allow.add_argument("--name", required=True)
+ allow.add_argument("--env-file", type=Path, default=Path(".env"))
+ args = parser.parse_args()
+ try:
+ if args.action == "configure":
+ changed = configure_client(args.env_file, args.client_id)
+ print(f"{args.client_id}：" + ("密钥已生成并写入配置。" if changed else "已配置，保留现有密钥。"))
+ print("密钥不会打印。请重启后端，并通过安全渠道向调用方提供该 clientId 对应的密钥。")
+ elif args.action == "allow-user":
+ asyncio.run(
+ allow_external_user(args.env_file, args.client_id, args.external_user_id, args.employee_no, args.name)
+ )
+ else:
+ asyncio.run(check_operator(args.union_id, args.dingtalk_user_id))
+ except AppException as exc:
+ print(f"检查不通过（{exc.status_code}）：{exc.message}")
+ return 1
+ except (ValueError, OSError):
+ print("配置或参数无效，请检查文件、JSON、标识格式及密钥长度；未显示敏感内容。")
+ return 1
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/test_plm_risk_report.py b/scripts/test_plm_risk_report.py
new file mode 100644
index 0000000..271c983
--- /dev/null
+++ b/scripts/test_plm_risk_report.py
@@ -0,0 +1,292 @@
+"""Exercise the PLM POST/GET protocol and save the real report without logging credentials."""
+
+import argparse
+import base64
+import hashlib
+import hmac
+import json
+import math
+import os
+import re
+import secrets
+import time
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from urllib.parse import urlencode, urlsplit
+
+import httpx
+
+ROOT = Path(__file__).resolve().parents[1]
+DEFAULT_PAYLOAD = ROOT / "docs" / "examples" / "plm-risk-request-20261009.json"
+DEFAULT_URL = "http://192.168.9.28:18474/api/v1/integrations/risk-assessments/report"
+LEVELS = {
+ "low": "低风险",
+ "moderately_low": "较低风险",
+ "medium": "中等风险",
+ "high": "高风险",
+ "very_high": "极高风险",
+}
+SAFE_HEADERS = (
+ "content-type",
+ "content-encoding",
+ "x-request-id",
+ "x-assessment-id",
+ "x-model-version",
+ "x-risk-score",
+ "x-risk-level",
+ "retry-after",
+)
+
+
+def signed_headers(method: str, url: str, body: bytes, secret: str, client_id: str, request_id: str) -> dict:
+ timestamp, nonce = str(int(time.time())), secrets.token_urlsafe(16)
+ message = timestamp.encode() + b"\n" + nonce.encode() + b"\n"
+ headers = {"X-Client-Id": client_id, "X-Timestamp": timestamp, "X-Nonce": nonce}
+ if method == "GET":
+ target = urlsplit(url)
+ message = b"GET\n" + target.path.encode() + b"\n" + target.query.encode() + b"\n" + message
+ else:
+ message += body
+ headers.update({"Content-Type": "application/json", "Idempotency-Key": request_id})
+ headers["X-Signature"] = base64.b64encode(hmac.new(secret.encode(), message, hashlib.sha256).digest()).decode()
+ return headers
+
+
+def report_metadata(response: httpx.Response, request_id: str) -> dict:
+ if response.status_code != 200:
+ raise ValueError(f"Expected HTTP 200, received {response.status_code}")
+ if response.headers.get("content-type", "").split(";", 1)[0].strip().lower() != "text/markdown":
+ raise ValueError("Successful response must be text/markdown")
+ if response.headers.get("x-request-id") != request_id:
+ raise ValueError("X-Request-Id does not match this request")
+ score = float(response.headers.get("x-risk-score", "nan"))
+ if not math.isfinite(score) or not 0 <= score <= 100:
+ raise ValueError("Invalid X-Risk-Score")
+ level = response.headers.get("x-risk-level")
+ if level not in LEVELS:
+ raise ValueError("Invalid X-Risk-Level")
+ assessment_id, model = response.headers.get("x-assessment-id"), response.headers.get("x-model-version")
+ if not assessment_id or not model:
+ raise ValueError("Missing assessment ID or model version")
+ # httpx already decompresses gzip/deflate. Decode the resulting bytes exactly once.
+ if len(response.content) > 1024 * 1024:
+ raise ValueError("Decompressed report exceeds 1MiB")
+ markdown = response.content.decode("utf-8")
+ if not markdown.strip():
+ raise ValueError("Empty report")
+ dimensions = sum(
+ bool(re.match(r"^\|\s*(供应商表现|物料品类|产品品类|加工厂质量|标准符合性|节点阶段|缺陷等级)\s*\|", line))
+ for line in markdown.splitlines()
+ )
+ if dimensions != 7:
+ raise ValueError(f"Expected 7 dimension rows, received {dimensions}")
+ return {
+ "requestId": request_id,
+ "assessmentId": assessment_id,
+ "score": score,
+ "level": level,
+ "levelLabel": LEVELS[level],
+ "modelVersion": model,
+ "dimensionCount": dimensions,
+ "reportSha256": hashlib.sha256(response.content).hexdigest(),
+ }
+
+
+class PLMReportTest:
+ def __init__(
+ self, client: httpx.Client, url: str, secret: str, payload: dict, output: Path, total_timeout: float = 180
+ ):
+ self.client, self.url, self.secret, self.payload, self.output = client, url, secret, payload, output
+ self.request_id = payload["requestId"]
+ self.body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+ self.query_url = url + "?" + urlencode({"requestId": self.request_id})
+ self.events = []
+ self.deadline = time.monotonic() + total_timeout
+
+ def send(self, method: str) -> httpx.Response:
+ remaining = self.deadline - time.monotonic()
+ if remaining <= 0:
+ raise TimeoutError("Overall test deadline exceeded; use the saved snapshot to resume")
+ url = self.query_url if method == "GET" else self.url
+ body = b"" if method == "GET" else self.body
+ headers = signed_headers(method, url, body, self.secret, self.payload["sourceSystem"], self.request_id)
+ started = time.monotonic()
+ try:
+ response = self.client.request(
+ method,
+ url,
+ content=body,
+ headers=headers,
+ timeout=httpx.Timeout(min(130, remaining), connect=min(10, remaining)),
+ )
+ except httpx.HTTPError as exc:
+ self.events.append(
+ {
+ "method": method,
+ "transportError": exc.__class__.__name__,
+ "seconds": round(time.monotonic() - started, 2),
+ }
+ )
+ raise
+ event = {
+ "method": method,
+ "status": response.status_code,
+ "seconds": round(time.monotonic() - started, 2),
+ "headers": {key: response.headers[key] for key in SAFE_HEADERS if key in response.headers},
+ }
+ if response.status_code != 200:
+ try:
+ event["error"] = response.json()
+ except ValueError:
+ event["error"] = {"message": "Non-JSON error response"}
+ self.events.append(event)
+ print(f"{method} HTTP {response.status_code} ({event['seconds']}s)", flush=True)
+ return response
+
+ def wait(self, response: httpx.Response):
+ try:
+ delay = max(1, min(60, int(response.headers.get("retry-after", "5"))))
+ except ValueError:
+ delay = 5
+ if time.monotonic() + delay >= self.deadline:
+ raise TimeoutError("Retry-After exceeds remaining test deadline")
+ # Short waits keep progress visible even if rate limiting requests a long delay.
+ print(f"Wait {delay}s before retry", flush=True)
+ while delay:
+ step = min(5, delay)
+ time.sleep(step)
+ delay -= step
+
+ def resolve(self, response: httpx.Response) -> httpx.Response:
+ for _ in range(40):
+ if response.status_code == 200:
+ return response
+ if response.status_code not in {202, 429}:
+ raise ValueError(f"HTTP {response.status_code}; inspect response.json for platform diagnostics")
+ if response.status_code == 202:
+ data = response.json()
+ if data.get("code") != 202 or (data.get("data") or {}).get("requestId") != self.request_id:
+ raise ValueError("Processing response does not match this request")
+ self.wait(response)
+ response = self.send("GET")
+ raise TimeoutError("Polling limit reached; use the saved snapshot to resume")
+
+ def run(self) -> dict:
+ self.output.mkdir(parents=True, exist_ok=True)
+ (self.output / "request.json").write_bytes(self.body)
+ checks = []
+ first = self.send("GET")
+ while first.status_code == 429:
+ self.wait(first)
+ first = self.send("GET")
+ if first.status_code in {404, 503}:
+ try:
+ first = self.send("POST")
+ except httpx.HTTPError:
+ # A timed-out POST may already have persisted the request.
+ first = self.send("GET")
+ first = self.resolve(first)
+ metadata = report_metadata(first, self.request_id)
+ (self.output / "report.md").write_bytes(first.content)
+ checks.append("报告为 UTF-8 Markdown，编号、分数、等级及七个维度校验通过")
+ repeated = self.send("POST")
+ if repeated.status_code == 429:
+ self.wait(repeated)
+ repeated = self.send("POST")
+ repeated = self.resolve(repeated)
+ if report_metadata(repeated, self.request_id) != metadata or repeated.content != first.content:
+ raise ValueError("Repeated POST differs from the original report")
+ checks.append("同编号同内容 POST 返回原报告，幂等校验通过")
+ queried = self.resolve(self.send("GET"))
+ if report_metadata(queried, self.request_id) != metadata or queried.content != first.content:
+ raise ValueError("GET report differs from the original report")
+ checks.append("GET 报告正文、评估 ID、分数、等级及模型版本一致")
+ return {
+ "success": True,
+ "endpoint": self.url,
+ **metadata,
+ "checks": checks,
+ "requestSha256": hashlib.sha256(self.body).hexdigest(),
+ "events": self.events,
+ }
+
+
+def load_secret(client_id: str, use_local_config: bool) -> str:
+ secret = os.environ.get("RISK_CLIENT_SECRET") or os.environ.get("RISK_INTEGRATION_CLIENT_SECRET")
+ if not secret and use_local_config:
+ # Explicit opt-in for running this tool in the risk backend workspace.
+ from dotenv import dotenv_values
+
+ raw = os.environ.get("RISK_INTEGRATION_CLIENTS") or dotenv_values(ROOT / ".env").get("RISK_INTEGRATION_CLIENTS")
+ secret = json.loads(raw or "{}").get(client_id)
+ if not isinstance(secret, str) or len(secret) < 32:
+ raise ValueError("Set RISK_CLIENT_SECRET or use --local-config to read this backend's .env")
+ return secret
+
+
+def main() -> int:
+ parser = argparse.ArgumentParser(description=__doc__)
+ parser.add_argument("--url", default=DEFAULT_URL)
+ parser.add_argument("--body-file", type=Path, default=DEFAULT_PAYLOAD)
+ parser.add_argument("--output-dir", type=Path)
+ parser.add_argument("--local-config", action="store_true", help="Read secret from backend .env without printing it")
+ parser.add_argument(
+ "--new-request", action="store_true", help="Use a unique requestId to generate a new assessment"
+ )
+ args = parser.parse_args()
+ stamp = datetime.now(timezone(timedelta(hours=8))).strftime("%Y%m%d-%H%M%S")
+ output = args.output_dir or ROOT / ".runtime-uploads" / "integration-tests" / (stamp + "-" + secrets.token_hex(3))
+ output.mkdir(parents=True, exist_ok=True)
+ runner = None
+ try:
+ target = urlsplit(args.url)
+ if target.scheme not in {"http", "https"} or target.query or target.fragment or target.username:
+ raise ValueError("Endpoint must be HTTP(S), without credentials, query or fragment")
+ payload = json.loads(args.body_file.read_text(encoding="utf-8-sig"))
+ if args.new_request:
+ payload["requestId"] = f"PLM-TEST-{payload['assessment']['subjectId']}-{stamp}-{secrets.token_hex(4)}"
+ secret = load_secret(payload["sourceSystem"], args.local_config)
+ with httpx.Client(follow_redirects=False, trust_env=False) as client:
+ runner = PLMReportTest(client, args.url, secret, payload, output)
+ result = runner.run()
+ except (ValueError, KeyError, OSError, TimeoutError, httpx.HTTPError) as exc:
+ result = {
+ "success": False,
+ "errorType": exc.__class__.__name__,
+ "message": str(exc)
+ if not isinstance(exc, (httpx.HTTPError, OSError))
+ else "Transport or file operation failed",
+ "events": runner.events if runner else [],
+ }
+ (output / "response.json").write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
+ if result["success"]:
+ summary = [
+ "# PLM 风险接口测试结果",
+ "",
+ f"- 请求编号：{result['requestId']}",
+ f"- 平台评估 ID：{result['assessmentId']}",
+ f"- 综合安全得分：{result['score']:.2f}",
+ f"- 风险等级：{result['levelLabel']}",
+ f"- 模型版本：{result['modelVersion']}",
+ f"- 维度数量：{result['dimensionCount']}",
+ "",
+ "报告：[report.md](report.md)",
+ "",
+ ]
+ summary.extend(f"- {check}" for check in result["checks"])
+ summary.extend(
+ [
+ "",
+ "本次测试沿用接口现有评分规则；HTTP 200 不代表全部维度拥有真实数据。",
+ "当品类为护发素时，当前规则可能无法归类，报告会注明品类维度默认分。",
+ "",
+ ]
+ )
+ (output / "summary.md").write_text("\n".join(summary), encoding="utf-8")
+ print(f"Score: {result['score']:.2f}; level: {result['level']}; assessment: {result['assessmentId']}")
+ print(f"Outputs: {output.resolve()}")
+ return 0 if result["success"] else 1
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/tests/test_external_setup.py b/tests/test_external_setup.py
new file mode 100644
index 0000000..65ca13e
--- /dev/null
+++ b/tests/test_external_setup.py
@@ -0,0 +1,43 @@
+"""Secret configuration must preserve unrelated settings and existing client credentials."""
+
+import json
+
+from dotenv import dotenv_values
+from scripts.setup_external_risk import configure_client, read_operator_mappings, write_operator_mapping
+
+
+def test_configure_keeps_settings_and_does_not_rotate_existing_secret(tmp_path):
+ env_file = tmp_path / ".env"
+ env_file.write_text("RISK_DATABASE_URL=keep-this-value\n# 保留注释\n", encoding="utf-8")
+ assert configure_client(env_file, "dieya-plm")
+ values = dotenv_values(env_file)
+ clients = json.loads(values["RISK_INTEGRATION_CLIENTS"])
+ assert len(clients["dieya-plm"]) >= 32
+ assert values["RISK_DATABASE_URL"] == "keep-this-value"
+ previous = env_file.read_bytes()
+ assert not configure_client(env_file, "dieya-plm")
+ assert env_file.read_bytes() == previous
+
+
+def test_external_mapping_preserves_secret_and_other_mappings(tmp_path):
+ env_file = tmp_path / ".env"
+ assert configure_client(env_file, "dieya-plm")
+ secret = dotenv_values(env_file)["RISK_INTEGRATION_CLIENTS"]
+ write_operator_mapping(env_file, "other-system", "user-1", "employee-1")
+ write_operator_mapping(env_file, "dieya-plm", "superadmin", "plm_superadmin")
+ assert read_operator_mappings(env_file) == {
+ "other-system": {"user-1": "employee-1"},
+ "dieya-plm": {"superadmin": "plm_superadmin"},
+ }
+ assert dotenv_values(env_file)["RISK_INTEGRATION_CLIENTS"] == secret
+
+
+def test_configure_preserves_other_secret_with_quotes_and_backslashes(tmp_path):
+ env_file = tmp_path / ".env"
+ existing = "a'b\"c\\d" + "-" * 40
+ encoded = json.dumps(json.dumps({"other-system": existing}))
+ env_file.write_text(f"RISK_INTEGRATION_CLIENTS={encoded}\n", encoding="utf-8")
+ assert configure_client(env_file, "dieya-plm")
+ clients = json.loads(dotenv_values(env_file)["RISK_INTEGRATION_CLIENTS"])
+ assert clients["other-system"] == existing
+ assert len(clients["dieya-plm"]) >= 32
diff --git a/tests/test_integration_risk.py b/tests/test_integration_risk.py
new file mode 100644
index 0000000..6a604ac
--- /dev/null
+++ b/tests/test_integration_risk.py
@@ -0,0 +1,494 @@
+"""HTTP integration checks against an isolated relational test database.
+
+SQLite runs the real persistence/ON CONFLICT paths. Its advisory-lock stub does
+not simulate PostgreSQL worker locking; deployment still needs PostgreSQL migration.
+"""
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import time
+import uuid
+from datetime import UTC, date, datetime, timedelta
+from types import SimpleNamespace
+
+import httpx
+import pytest
+from app.api.v1 import integration_risk
+from app.database.base import Base
+from app.database.session import get_db_session
+from app.main import create_app
+from app.models.entities import (
+ AnomalyCase,
+ IntegrationAssessment,
+ IntegrationNonce,
+ RolePermission,
+ UserAccount,
+)
+from app.repositories.integration_risk import IntegrationRiskRepository
+from app.services import integration_auth
+from app.services.llm_client import LLMClient
+from pydantic import SecretStr
+from sqlalchemy import JSON, BigInteger, Integer, MetaData, event, select, update
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+PATH = "/api/v1/integrations/risk-assessments/report"
+SECRET = "test-only-integration-secret-32-characters"
+
+
+def payload():
+ return {
+ "requestId": "PLM-D-FT-003-0001",
+ "sourceSystem": "dieya-plm",
+ "operator": {"dingtalkUnionId": "union-1", "dingtalkUserId": "ding-1", "name": "伪造姓名"},
+ "assessment": {
+ "subjectId": "D-FT-003",
+ "productName": "洗发水",
+ "productCategory": "洗发水/洗护类",
+ "materialType": "PET 瓶身 + PP 泵头",
+ "stageNew": "产品开发",
+ "asOfDate": "2026-01-15",
+ "supplier": None,
+ "factory": None,
+ },
+ }
+
+
+def headers(body=b"", *, method="POST", query="", client="dieya-plm", nonce=None, timestamp=None):
+ timestamp = timestamp or str(int(time.time()))
+ nonce = nonce or uuid.uuid4().hex
+ message = timestamp.encode() + b"\n" + nonce.encode() + b"\n"
+ if method == "GET":
+ message = b"GET\n" + PATH.encode() + b"\n" + query.encode() + b"\n" + message
+ else:
+ message += body
+ return {
+ "Content-Type": "application/json",
+ "X-Client-Id": client,
+ "X-Timestamp": timestamp,
+ "X-Nonce": nonce,
+ "X-Signature": base64.b64encode(hmac.new(SECRET.encode(), message, hashlib.sha256).digest()).decode(),
+ "Idempotency-Key": "PLM-D-FT-003-0001",
+ }
+
+
+async def post(http_client, data=None, **options):
+ body = json.dumps(data if data is not None else payload(), ensure_ascii=False).encode()
+ return await http_client.post(PATH, content=body, headers=headers(body, **options))
+
+
+@pytest.fixture
+async def integration_db(tmp_path, monkeypatch):
+ engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'integration.db'}")
+
+ @event.listens_for(engine.sync_engine, "connect")
+ def sqlite_advisory_stub(connection, _):
+ connection.create_function("pg_advisory_xact_lock", 1, lambda key: None)
+
+ schema = MetaData()
+ for table in Base.metadata.sorted_tables:
+ copied = table.to_metadata(schema)
+ for column in copied.columns:
+ if isinstance(column.type, JSONB):
+ column.type = JSON()
+ elif isinstance(column.type, BigInteger):
+ column.type = Integer()
+ async with engine.begin() as connection:
+ await connection.run_sync(schema.create_all)
+ sessions = async_sessionmaker(engine, expire_on_commit=False)
+ async with sessions() as session:
+ session.add(
+ UserAccount(
+ id=1,
+ emp_no="real-employee",
+ name="真实评估人",
+ role="user",
+ status="启用",
+ dingtalk_union_id="union-1",
+ dingtalk_id="ding-1",
+ )
+ )
+ session.add(RolePermission(role="user", func_key="risk.assess", can_operate=True, can_view=True))
+ session.add_all(
+ [
+ AnomalyCase(
+ id=1,
+ date=date(2026, 1, 10),
+ product_name="洗发水",
+ product_category="洗发露",
+ material_type="泵头",
+ stage_new="产品开发",
+ defect_level="C",
+ description="泵头漏液",
+ solution="密封验证",
+ import_fingerprint="case1",
+ ),
+ AnomalyCase(
+ id=2,
+ date=date(2026, 1, 20),
+ product_name="洗发水",
+ product_category="洗发露",
+ material_type="泵头",
+ stage_new="产品开发",
+ defect_level="A",
+ description="未来记录",
+ import_fingerprint="case2",
+ ),
+ ]
+ )
+ await session.commit()
+ settings = SimpleNamespace(
+ integration_clients=SecretStr(json.dumps({"dieya-plm": SECRET, "other-system": SECRET})),
+ integration_operator_mappings=SecretStr("{}"),
+ integration_requests_per_minute=60,
+ integration_report_timeout_seconds=110,
+ )
+ monkeypatch.setattr(integration_auth, "get_settings", lambda: settings)
+ monkeypatch.setattr(integration_risk, "get_settings", lambda: settings)
+
+ async def suggestions(self, context):
+ return ["完成密封性验证。"]
+
+ monkeypatch.setattr(LLMClient, "generate_prevention_suggestions", suggestions)
+ app = create_app()
+
+ async def database():
+ async with sessions() as session:
+ try:
+ yield session
+ await session.commit()
+ except Exception:
+ await session.rollback()
+ raise
+
+ app.dependency_overrides[get_db_session] = database
+ async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://test") as client:
+ yield client, sessions, settings
+ await engine.dispose()
+
+
+async def test_success_uses_cutoff_actual_user_and_seven_dimensions(integration_db):
+ client, sessions, _ = integration_db
+ response = await post(client)
+ assert response.status_code == 200, response.text
+ assert response.headers["content-type"] == "text/markdown; charset=utf-8"
+ assert response.headers["x-risk-level"] in {"low", "moderately_low", "medium", "high", "very_high"}
+ assert "真实评估人" in response.text and "伪造姓名" not in response.text
+ assert "泵头" in response.text and "未来记录" not in response.text
+ assert response.headers["x-model-version"] == "database-seven-dimension-v1"
+ async with sessions() as session:
+ record = await session.scalar(select(IntegrationAssessment))
+ assert record.status == "completed"
+ assert len(record.result_data["dimensionScores"]) == 7
+ assert record.result_data["categoryWarning"]["anomalyCount"] == 1
+ assert record.result_data["similarCases"][0]["caseId"] == 1
+ assert record.result_data["score"] == float(response.headers["x-risk-score"])
+
+
+async def test_replay_and_bad_signature_do_not_generate_reports(integration_db):
+ client, sessions, _ = integration_db
+ response = await post(client, nonce="same-nonce-123")
+ assert response.status_code == 200
+ assert (await post(client, nonce="same-nonce-123")).status_code == 401
+ body = json.dumps(payload()).encode()
+ signed = headers(body)
+ assert (await client.post(PATH, content=body + b" ", headers=signed)).status_code == 401
+ assert (await post(client, timestamp=str(int(time.time()) - 301))).status_code == 401
+ assert (await client.post(PATH, json=payload())).status_code == 401
+ async with sessions() as session:
+ assert len(list((await session.scalars(select(IntegrationAssessment))).all())) == 1
+
+
+async def test_idempotency_query_isolation_and_query_signature(integration_db):
+ client, _, _ = integration_db
+ original = await post(client)
+ body = json.dumps(payload(), sort_keys=True, separators=(",", ":")).encode()
+ repeated = await client.post(PATH, content=body, headers=headers(body))
+ assert repeated.status_code == 200
+ assert repeated.text == original.text
+ assert repeated.headers["x-assessment-id"] == original.headers["x-assessment-id"]
+ changed = payload()
+ changed["assessment"]["materialType"] = "软管"
+ assert (await post(client, changed)).status_code == 409
+ query = "requestId=PLM-D-FT-003-0001"
+ retrieved = await client.get(PATH + "?" + query, headers=headers(method="GET", query=query))
+ assert retrieved.status_code == 200 and retrieved.text == original.text
+ missing = await client.get(PATH + "?" + query, headers=headers(method="GET", query=query, client="other-system"))
+ assert missing.status_code == 404
+ tampered = await client.get(PATH + "?requestId=OTHER", headers=headers(method="GET", query=query))
+ assert tampered.status_code == 401
+
+
+@pytest.mark.parametrize("change", ["missing_category", "blank_material", "bad_stage", "bad_date", "extra_score"])
+async def test_invalid_fields_return_json_422(integration_db, change):
+ client, _, _ = integration_db
+ data = payload()
+ if change == "missing_category":
+ del data["assessment"]["productCategory"]
+ elif change == "blank_material":
+ data["assessment"]["materialType"] = " "
+ elif change == "bad_stage":
+ data["assessment"]["stageNew"] = "未知阶段"
+ elif change == "bad_date":
+ data["assessment"]["asOfDate"] = "2099-01-01"
+ else:
+ data["assessment"]["score"] = 99
+ response = await post(client, data)
+ assert response.status_code == 422
+ assert response.json()["code"] == 422 and response.json()["data"] is None
+
+
+@pytest.mark.parametrize("change", ["missing_user", "duplicate_union", "disabled_user", "no_permission", "mismatch"])
+async def test_user_mapping_and_permissions(integration_db, change):
+ client, sessions, _ = integration_db
+ data = payload()
+ async with sessions() as session:
+ if change == "missing_user":
+ data["operator"]["dingtalkUnionId"] = "not-opened"
+ elif change == "duplicate_union":
+ session.add(
+ UserAccount(
+ emp_no="second",
+ name="重复映射",
+ role="admin",
+ status="启用",
+ dingtalk_union_id="union-1",
+ dingtalk_id="ding-2",
+ )
+ )
+ elif change == "disabled_user":
+ await session.execute(update(UserAccount).values(status="停用"))
+ elif change == "no_permission":
+ await session.execute(update(RolePermission).values(can_operate=False))
+ else:
+ data["operator"]["dingtalkUserId"] = "wrong-user"
+ await session.commit()
+ assert (await post(client, data)).status_code == 403
+ async with sessions() as session:
+ assert await session.scalar(select(IntegrationAssessment)) is None
+
+
+async def test_rate_limit_has_retry_after(integration_db):
+ client, _, settings = integration_db
+ settings.integration_requests_per_minute = 1
+ assert (await post(client)).status_code == 200
+ response = await post(client)
+ assert response.status_code == 429 and response.headers["retry-after"] == "60"
+
+
+async def test_failed_and_expired_processing_requests_can_retry(integration_db, monkeypatch):
+ client, sessions, _ = integration_db
+ original = integration_risk.ExternalRiskReportService.assess
+
+ async def fail(self, payload, as_of):
+ raise RuntimeError("provider secret must not be returned")
+
+ monkeypatch.setattr(integration_risk.ExternalRiskReportService, "assess", fail)
+ failed = await post(client)
+ assert failed.status_code == 503 and "provider secret" not in failed.text
+ monkeypatch.setattr(integration_risk.ExternalRiskReportService, "assess", original)
+ assert (await post(client)).status_code == 200
+ async with sessions() as session:
+ await session.execute(
+ update(IntegrationAssessment).values(
+ status="processing",
+ lease_until=datetime.now(UTC) + timedelta(seconds=100),
+ )
+ )
+ await session.commit()
+ pending = await post(client)
+ assert pending.status_code == 202 and pending.headers["retry-after"] == "5"
+ async with sessions() as session:
+ await session.execute(
+ update(IntegrationAssessment).values(
+ lease_until=datetime.now(UTC) - timedelta(seconds=1),
+ )
+ )
+ await session.commit()
+ assert (await post(client)).status_code == 200
+
+
+async def test_unrecognized_category_and_ai_failure_are_explained(integration_db, monkeypatch):
+ client, _, _ = integration_db
+
+ async def unavailable(self, context):
+ raise RuntimeError("LLM down")
+
+ monkeypatch.setattr(LLMClient, "generate_prevention_suggestions", unavailable)
+ data = payload()
+ data["assessment"]["productCategory"] = "防晒"
+ data["assessment"]["productName"] = "|[链接](javascript:test)"
+ response = await post(client, data)
+ assert response.status_code == 200
+ assert "无法归类" in response.text and "固定验证建议" in response.text
+ assert "