From 6995fe1f973d6dfc6ac2b4ab3f25f61caeb03e48 Mon Sep 17 00:00:00 2001 From: kx_guo Date: Tue, 27 Oct 2026 16:08:41 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=B0=E5=A2=9E=E7=AC=AC=E4=B8=89=E6=96=B9?= =?UTF-8?q?=E9=A3=8E=E9=99=A9=E8=AF=84=E4=BC=B0=E6=8A=A5=E5=91=8A=E6=8E=A5?= =?UTF-8?q?=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 11 + README.md | 5 + app/api/router.py | 2 + app/api/v1/integration_risk.py | 218 +++++++ app/core/config.py | 4 + app/models/__init__.py | 6 +- app/models/entities.py | 64 +- app/repositories/integration_risk.py | 120 ++++ app/schemas/integration_risk.py | 61 ++ app/services/external_risk_report.py | 217 +++++++ app/services/integration_auth.py | 168 ++++++ app/services/markdown_reports.py | 140 +++++ app/services/rag.py | 3 +- docs/PLM接口测试脚本使用说明.md | 53 ++ docs/examples/plm-risk-request-20261009.json | 21 + docs/外部系统风险评估报告接口.md | 550 ++++++++++++++++++ .../20261008_0022_integration_reports.py | 49 ++ scripts/call_external_risk_report.py | 68 +++ scripts/setup_external_risk.py | 177 ++++++ scripts/test_plm_risk_report.py | 292 ++++++++++ tests/test_external_setup.py | 43 ++ tests/test_integration_risk.py | 494 ++++++++++++++++ tests/test_models.py | 2 + tests/test_plm_report_client.py | 77 +++ 24 files changed, 2841 insertions(+), 4 deletions(-) create mode 100644 app/api/v1/integration_risk.py create mode 100644 app/repositories/integration_risk.py create mode 100644 app/schemas/integration_risk.py create mode 100644 app/services/external_risk_report.py create mode 100644 app/services/integration_auth.py create mode 100644 app/services/markdown_reports.py create mode 100644 docs/PLM接口测试脚本使用说明.md create mode 100644 docs/examples/plm-risk-request-20261009.json create mode 100644 docs/外部系统风险评估报告接口.md create mode 100644 migrations/versions/20261008_0022_integration_reports.py create mode 100644 scripts/call_external_risk_report.py create mode 100644 scripts/setup_external_risk.py create mode 100644 scripts/test_plm_risk_report.py create mode 100644 tests/test_external_setup.py create mode 100644 tests/test_integration_risk.py create mode 100644 tests/test_plm_report_client.py 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 "