dieya/dieya_risk_prediction_backend
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: dieya:680b0c20cfe35f8a295d72ee29c050ebafc64596
Branches Tags
dieya:main
dieya:dev
..
compare: dieya:b0b6ef014f3b06621c05f1f45cf46b1999c0ff38
Branches Tags
dieya:dev
dieya:main

No commits in common. "680b0c20cfe35f8a295d72ee29c050ebafc64596" and "b0b6ef014f3b06621c05f1f45cf46b1999c0ff38" have entirely different histories.
680b0c20cf ... b0b6ef014f

44 changed files with 128 additions and 2923 deletions
Show Stats Expand all files Collapse all files

14
.env.example
View File

@ -26,20 +26,6 @@ RISK_DINGTALK_APP_SECRET=
RISK_DINGTALK_CORP_ID=
RISK_DINGTALK_AGENT_ID=
# ==================== 百度 OCR ====================
# 图片调用 WebImage，PDF 调用 PaddleOCR-VL 文档解析；上传原件仅保存在本机目录。
BAIDU_OCR_API_KEY=
BAIDU_OCR_SECRET_KEY=
RISK_BAIDU_OCR_IMAGE_URL=https://aip.baidubce.com/rest/2.0/ocr/v1/webimage
RISK_BAIDU_OCR_DOCUMENT_URL=https://aip.baidubce.com/rest/2.0/brain/online/v2/paddle-vl-parser/task
RISK_BAIDU_OCR_TIMEOUT_SECONDS=75
RISK_BAIDU_OCR_POLL_INTERVAL_SECONDS=2
RISK_OCR_UPLOAD_DIR=.runtime-uploads/ocr
RISK_OCR_MAX_FILE_MB=10
RISK_OCR_MAX_BATCH_FILES=5
RISK_OCR_REVIEW_CONFIDENCE=0.8
RISK_OCR_JUDGMENT_CONFIDENCE=0.85
# ==================== 主 LLM 模型配置 ====================
# 主模型 API 密钥、OpenAI 兼容接口地址、模型名、请求超时时间和生成温度。
RISK_LLM_API_KEY=

1
.gitignore vendored
View File

@ -11,4 +11,3 @@ logs/
dist/
build/
data/
.runtime-uploads/

4
README.md
View File

@ -22,10 +22,6 @@ python scripts/create_user.py --emp-no admin001 --name 管理员 --role admin --
审核接口：`POST /api/v1/risk-items`、`GET /api/v1/risk-items/pending`、`POST /api/v1/risk-items/{id}/review`。
OCR 接口：`POST /api/v1/ocr/recognize` 接收最多 5 个 JPG、PNG 或 PDF 文件，图片调用百度 WebImage，PDF 调用 PaddleOCR-VL 文档解析；返回可编辑字段、检测数值、置信度及标准库判定。`POST /api/v1/ocr/judge` 用于人工修正检测值后重新判定，`GET /api/v1/ocr/files/{file_id}` 通过登录鉴权读取原始附件。启用前配置 `BAIDU_OCR_API_KEY` 和 `BAIDU_OCR_SECRET_KEY`，并执行数据库迁移。
角色采用管理员 `admin`、产品经理 `product_manager`、审核员 `reviewer`、普通用户 `普通用户` 四类。模型权重由管理员通过 `GET/PATCH /api/v1/model-config` 维护，七维权重合计必须为 100%，修改仅对新评估生效。
健康检查：`GET http://127.0.0.1:18474/api/v1/health`
数据库表结构和字段释义通过 Alembic 管理。执行 `python -m alembic upgrade head` 会创建或更新表、索引、外键及 PostgreSQL 表/字段注释。

8
app/api/router.py
View File

@ -6,23 +6,15 @@ 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.model_config import router as model_config_router
from app.api.v1.ocr import router as ocr_router
from app.api.v1.reports import router as reports_router
from app.api.v1.risk_items import router as risk_items_router
from app.api.v1.risk_predictions import router as risk_predictions_router
from app.api.v1.standards import router as standards_router
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(model_config_router)
api_router.include_router(ocr_router)
api_router.include_router(auth_router)
api_router.include_router(conversations_router)
api_router.include_router(dashboard_router)
api_router.include_router(risk_items_router)
api_router.include_router(reports_router)
api_router.include_router(risk_predictions_router)
api_router.include_router(standards_router)
api_router.include_router(suppliers_router)

30
app/api/v1/model_config.py
View File

@ -1,30 +0,0 @@
"""Administrator-only risk-model configuration endpoints."""
from typing import Annotated
from fastapi import APIRouter, Depends
from app.api.dependencies import DatabaseSession, require_permission
from app.models.entities import UserAccount
from app.schemas.common import ApiResponse
from app.schemas.model_config import ModelWeightConfigData, ModelWeightConfigUpdate
from app.services.model_config import ModelConfigService
router = APIRouter(prefix="/model-config", tags=["model-config"])
@router.get("", response_model=ApiResponse[ModelWeightConfigData])
async def get_model_config(
session: DatabaseSession,
_current_user: Annotated[UserAccount, Depends(require_permission("model_weight.manage"))],
) -> ApiResponse[ModelWeightConfigData]:
return ApiResponse(data=await ModelConfigService(session).get())
@router.patch("", response_model=ApiResponse[ModelWeightConfigData])
async def update_model_config(
payload: ModelWeightConfigUpdate,
session: DatabaseSession,
current_user: Annotated[UserAccount, Depends(require_permission("model_weight.manage", operate=True))],
) -> ApiResponse[ModelWeightConfigData]:
return ApiResponse(data=await ModelConfigService(session).update(payload, current_user))

50
app/api/v1/ocr.py
View File

@ -1,50 +0,0 @@
"""Authenticated OCR recognition and source-file endpoints."""
from typing import Annotated
from fastapi import APIRouter, Depends, File, UploadFile
from fastapi.responses import FileResponse
from app.api.dependencies import DatabaseSession, require_permission
from app.core.exceptions import AppException
from app.models.entities import UserAccount
from app.schemas.common import ApiResponse
from app.schemas.ocr import OcrBatchResult, OcrJudgmentRequest, OcrStandardJudgment
from app.services.ocr import OcrService
router = APIRouter(prefix="/ocr", tags=["ocr"])
@router.post("/recognize", response_model=ApiResponse[OcrBatchResult])
async def recognize_files(
files: Annotated[list[UploadFile], File(description="1-5 JPG, PNG, or PDF files")],
session: DatabaseSession,
_current_user: Annotated[UserAccount, Depends(require_permission("risk_item.submit", operate=True))],
) -> ApiResponse[OcrBatchResult]:
return ApiResponse(data=await OcrService(session).recognize(files))
@router.post("/judge", response_model=ApiResponse[list[OcrStandardJudgment]])
async def judge_measurements(
payload: OcrJudgmentRequest,
session: DatabaseSession,
_current_user: Annotated[UserAccount, Depends(require_permission("risk_item.submit", operate=True))],
) -> ApiResponse[list[OcrStandardJudgment]]:
return ApiResponse(data=await OcrService(session).judge(payload.measurements))
@router.get("/files/{file_id}", response_class=FileResponse)
async def get_source_file(
file_id: str,
session: DatabaseSession,
_current_user: Annotated[UserAccount, Depends(require_permission("risk_item.submit"))],
) -> FileResponse:
path = OcrService(session).upload_path(file_id)
if not path.is_file():
raise AppException("附件不存在", code=404, status_code=404)
media_type = {
".jpg": "image/jpeg",
".png": "image/png",
".pdf": "application/pdf",
}[path.suffix.lower()]
return FileResponse(path, media_type=media_type, filename=file_id)

38
app/api/v1/reports.py
View File

@ -1,38 +0,0 @@
"""Downloadable risk report endpoints."""
from typing import Annotated
from fastapi import APIRouter, Depends
from fastapi.responses import Response
from app.api.dependencies import require_permission
from app.models.entities import UserAccount
from app.schemas.risk_prediction import RiskAssessmentResult
from app.services.reports import build_risk_report_docx, build_risk_report_pdf
router = APIRouter(prefix="/reports", tags=["reports"])
@router.post("/pdf")
async def download_report_pdf(
payload: RiskAssessmentResult,
_current_user: Annotated[UserAccount, Depends(require_permission("risk.assess", operate=True))],
) -> Response:
pdf = build_risk_report_pdf(payload)
return Response(
content=pdf,
media_type="application/pdf",
headers={"Content-Disposition": 'attachment; filename="dieya-risk-report.pdf"'},
)
@router.post("/word")
async def download_report_word(
payload: RiskAssessmentResult,
_current_user: Annotated[UserAccount, Depends(require_permission("risk.assess", operate=True))],
) -> Response:
return Response(
content=build_risk_report_docx(payload),
media_type="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
headers={"Content-Disposition": 'attachment; filename="dieya-risk-report.docx"'},
)

32
app/api/v1/suppliers.py
View File

@ -1,32 +0,0 @@
"""Supplier profile endpoints."""
from typing import Annotated
from fastapi import APIRouter, Depends, Query
from app.api.dependencies import DatabaseSession, require_permission
from app.models.entities import UserAccount
from app.schemas.common import ApiResponse
from app.schemas.suppliers import SupplierListData, SupplierProfileData
from app.services.suppliers import SupplierProfileService
router = APIRouter(prefix="/suppliers", tags=["suppliers"])
@router.get("", response_model=ApiResponse[SupplierListData])
async def list_suppliers(
session: DatabaseSession,
_current_user: Annotated[UserAccount, Depends(require_permission("supplier.read"))],
keyword: str | None = Query(default=None, max_length=100),
limit: int = Query(default=100, ge=1, le=100),
) -> ApiResponse[SupplierListData]:
return ApiResponse(data=await SupplierProfileService(session).list(keyword=keyword, limit=limit))
@router.get("/{supplier_name}", response_model=ApiResponse[SupplierProfileData])
async def get_supplier_profile(
supplier_name: str,
session: DatabaseSession,
_current_user: Annotated[UserAccount, Depends(require_permission("supplier.read"))],
) -> ApiResponse[SupplierProfileData]:
return ApiResponse(data=await SupplierProfileService(session).detail(supplier_name))

20
app/core/config.py
View File

@ -1,7 +1,6 @@
"""Environment-backed application settings."""
from functools import lru_cache
from pathlib import Path
from typing import Annotated, Literal
from pydantic import AliasChoices, Field, SecretStr, field_validator
@ -25,25 +24,6 @@ class Settings(BaseSettings):
dingtalk_app_secret: SecretStr | None = Field(default=None, repr=False)
dingtalk_corp_id: str = ""
dingtalk_agent_id: str = ""
baidu_ocr_api_key: SecretStr | None = Field(
default=None,
repr=False,
validation_alias=AliasChoices("BAIDU_OCR_API_KEY", "RISK_BAIDU_OCR_API_KEY"),
)
baidu_ocr_secret_key: SecretStr | None = Field(
default=None,
repr=False,
validation_alias=AliasChoices("BAIDU_OCR_SECRET_KEY", "RISK_BAIDU_OCR_SECRET_KEY"),
)
baidu_ocr_image_url: str = "https://aip.baidubce.com/rest/2.0/ocr/v1/webimage"
baidu_ocr_document_url: str = "https://aip.baidubce.com/rest/2.0/brain/online/v2/paddle-vl-parser/task"
baidu_ocr_timeout_seconds: int = Field(default=75, ge=10, le=300)
baidu_ocr_poll_interval_seconds: float = Field(default=2.0, ge=0.5, le=10)
ocr_upload_dir: Path = Path(".runtime-uploads/ocr")
ocr_max_file_mb: int = Field(default=10, ge=1, le=20)
ocr_max_batch_files: int = Field(default=5, ge=1, le=10)
ocr_review_confidence: float = Field(default=0.8, ge=0, le=1)
ocr_judgment_confidence: float = Field(default=0.85, ge=0, le=1)
llm_api_key: SecretStr | None = Field(
default=None,
repr=False,

24
app/domain/model_config.py
View File

@ -1,24 +0,0 @@
"""Canonical risk-model configuration shared by scoring and administration."""
DEFAULT_WEIGHTS = {
"供应商表现": 25.0,
"物料品类": 20.0,
"产品品类": 15.0,
"加工厂质量": 15.0,
"标准符合性": 10.0,
"节点阶段": 10.0,
"缺陷等级": 5.0,
}
WEIGHT_DESCRIPTIONS = {
"供应商表现": "含频次风险系数 α 修正",
"物料品类": "按物料历史异常率",
"产品品类": "按具体产品品类历史异常密度",
"加工厂质量": "巡检 / 批次合格率",
"标准符合性": "按判定标准库量化达标率计分",
"节点阶段": "开发阶段越靠后风险越高",
"缺陷等级": "A 级缺陷加权",
}
DEFAULT_FREQUENCY_COEFFICIENT = 0.3
DEFAULT_FREQUENCY_CAP = 1.5

4
app/models/__init__.py
View File

@ -5,11 +5,9 @@ from app.models.entities import (
Conversation,
ConversationMessage,
OperationAudit,
PurchaseOrderLine,
ReviewRecord,
RiskAssessment,
RiskItem,
RiskModelConfig,
RolePermission,
StandardRule,
SupplierProfile,
@ -24,8 +22,6 @@ __all__ = [
"ReviewRecord",
"RiskAssessment",
"RiskItem",
"RiskModelConfig",
"PurchaseOrderLine",
"RolePermission",
"StandardRule",
"SupplierProfile",

59
app/models/entities.py
View File

@ -94,36 +94,6 @@ class SupplierProfile(Base):
)
class PurchaseOrderLine(Base):
"""A source row from the procurement order ledger."""
__tablename__ = "purchase_order_lines"
__table_args__ = (
Index("ix_purchase_order_lines_supplier_date", "supplier", "purchase_date"),
Index("ix_purchase_order_lines_document", "document_number"),
Index("uq_purchase_order_lines_source", "import_fingerprint", unique=True),
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
document_number: Mapped[str] = mapped_column(String(64), nullable=False)
purchase_date: Mapped[date | None] = mapped_column(Date)
supplier: Mapped[str] = mapped_column(String(300), nullable=False)
document_status: Mapped[str | None] = mapped_column(String(30))
close_status: Mapped[str | None] = mapped_column(String(30))
material_code: Mapped[str | None] = mapped_column(String(200))
material_name: Mapped[str | None] = mapped_column(Text)
purchase_unit: Mapped[str | None] = mapped_column(String(30))
purchase_quantity: Mapped[Decimal | None] = mapped_column(Numeric(18, 4))
delivery_date: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
tax_inclusive_unit_price: Mapped[Decimal | None] = mapped_column(Numeric(18, 6))
is_gift: Mapped[str | None] = mapped_column(String(10))
business_close: Mapped[str | None] = mapped_column(String(30))
remark: Mapped[str | None] = mapped_column(Text)
source_file: Mapped[str] = mapped_column(String(300), nullable=False)
import_fingerprint: Mapped[str] = mapped_column(String(64), nullable=False, unique=True)
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), nullable=False)
class RiskAssessment(Base):
"""A user-triggered risk assessment and its structured result."""
@ -149,7 +119,9 @@ class Conversation(Base):
"""A persistent assistant conversation owned by one authenticated user."""
__tablename__ = "conversations"
__table_args__ = (Index("ix_conversations_user_updated", "user_id", "updated_at"),)
__table_args__ = (
Index("ix_conversations_user_updated", "user_id", "updated_at"),
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
user_id: Mapped[int] = mapped_column(
@ -177,7 +149,9 @@ class ConversationMessage(Base):
"""One user or assistant message, including optional structured results."""
__tablename__ = "conversation_messages"
__table_args__ = (Index("ix_conversation_messages_conversation_created", "conversation_id", "created_at"),)
__table_args__ = (
Index("ix_conversation_messages_conversation_created", "conversation_id", "created_at"),
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
conversation_id: Mapped[int] = mapped_column(
@ -202,17 +176,13 @@ class RiskItem(Base):
Index("ix_risk_items_status", "status"),
Index("ix_risk_items_submitted_by", "submitted_by"),
Index("ix_risk_items_submitted_at", "submitted_at"),
Index("ix_risk_items_occurred_on", "occurred_on"),
Index("ix_risk_items_material_type", "material_type"),
)
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True)
source_type: Mapped[str] = mapped_column(String(20), nullable=False)
source_file_id: Mapped[str | None] = mapped_column(String(50))
occurred_on: Mapped[date | None] = mapped_column(Date)
product_name: Mapped[str | None] = mapped_column(String(200))
product_category: Mapped[str | None] = mapped_column(String(50))
material_type: Mapped[str | None] = mapped_column(String(200))
supplier: Mapped[str | None] = mapped_column(String(100))
factory: Mapped[str | None] = mapped_column(String(100))
stage_new: Mapped[str | None] = mapped_column(String(50))
@ -220,8 +190,6 @@ class RiskItem(Base):
defect_level: Mapped[str | None] = mapped_column(String(5))
solution: Mapped[str | None] = mapped_column(Text)
photos: Mapped[list[str]] = mapped_column(JSONB, default=list, server_default="[]", nullable=False)
recognition_confidence: Mapped[Decimal | None] = mapped_column(Numeric(5, 4))
ocr_payload: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict, server_default="{}", nullable=False)
submitted_by: Mapped[str | None] = mapped_column(String(50))
submitted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
status: Mapped[str] = mapped_column(String(20), default="待预审", server_default="待预审", nullable=False)
@ -355,21 +323,6 @@ class RolePermission(Base):
)
class RiskModelConfig(Base):
"""Administrator-managed seven-dimension weights and frequency parameters."""
__tablename__ = "risk_model_configs"
id: Mapped[int] = mapped_column(BigInteger, primary_key=True, default=1)
weights: Mapped[dict[str, float]] = mapped_column(JSONB, nullable=False)
frequency_coefficient: Mapped[Decimal] = mapped_column(Numeric(5, 4), nullable=False)
frequency_cap: Mapped[Decimal] = mapped_column(Numeric(4, 2), nullable=False)
updated_by: Mapped[str | None] = mapped_column(String(20))
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
)
class OperationAudit(Base):
"""Append-only audit record for authentication, permission, and review actions."""

4
app/schemas/dashboard.py
View File

@ -32,9 +32,9 @@ class DashboardMatrixRow(ApiSchema):
name: str
party_type: str
categories: dict[str, int]
cooperation_total: int | None
cooperation_total: int
anomaly_count: int
anomaly_rate: float | None
anomaly_rate: float
tier: str

32
app/schemas/model_config.py
View File

@ -1,32 +0,0 @@
"""Risk-model configuration API contracts."""
from datetime import datetime
from pydantic import Field, model_validator
from app.domain.model_config import DEFAULT_WEIGHTS
from app.schemas.common import ApiSchema
class ModelWeightConfigData(ApiSchema):
weights: dict[str, float]
frequency_coefficient: float
frequency_cap: float
updated_by: str | None = None
updated_at: datetime | None = None
class ModelWeightConfigUpdate(ApiSchema):
weights: dict[str, float]
frequency_coefficient: float = Field(ge=0, le=1)
frequency_cap: float = Field(ge=1, le=3)
@model_validator(mode="after")
def validate_weights(self) -> "ModelWeightConfigUpdate":
if set(self.weights) != set(DEFAULT_WEIGHTS):
raise ValueError("权重必须完整包含七个固定维度")
if any(value <= 0 or value > 100 for value in self.weights.values()):
raise ValueError("每个维度权重必须大于0且不超过100")
if abs(sum(self.weights.values()) - 100) > 0.01:
raise ValueError("七个维度权重合计必须为100%")
return self

73
app/schemas/ocr.py
View File

@ -1,73 +0,0 @@
"""OCR recognition and standard-judgment contracts."""
from datetime import date
from typing import Literal
from pydantic import Field
from app.schemas.common import ApiSchema
class OcrField(ApiSchema):
value: str | None = None
confidence: float | None = Field(default=None, ge=0, le=1)
needs_review: bool = True
class OcrExtractedFields(ApiSchema):
product_name: OcrField = Field(default_factory=OcrField)
product_category: OcrField = Field(default_factory=OcrField)
supplier: OcrField = Field(default_factory=OcrField)
factory: OcrField = Field(default_factory=OcrField)
material_type: OcrField = Field(default_factory=OcrField)
stage_new: OcrField = Field(default_factory=OcrField)
description: OcrField = Field(default_factory=OcrField)
defect_level: OcrField = Field(default_factory=OcrField)
occurred_on: date | None = None
class OcrMeasurement(ApiSchema):
item: str
raw_value: str
value: float | None = None
unit: str | None = None
confidence: float | None = Field(default=None, ge=0, le=1)
needs_review: bool = True
class OcrStandardJudgment(ApiSchema):
measurement_item: str
measured_value: str
standard_rule_id: int | None = None
risk_point: str | None = None
standard_value: str | None = None
result: Literal["合格", "不合格", "待人工确认", "未匹配标准"]
basis: str | None = None
reason: str | None = None
class OcrFileResult(ApiSchema):
file_id: str
file_name: str
content_type: str
engine: Literal["webimage", "paddle-vl-parser"]
raw_text: str
confidence: float | None = Field(default=None, ge=0, le=1)
page_count: int = Field(default=1, ge=1)
fields: OcrExtractedFields
measurements: list[OcrMeasurement] = Field(default_factory=list)
judgments: list[OcrStandardJudgment] = Field(default_factory=list)
warnings: list[str] = Field(default_factory=list)
requires_review: bool = True
class OcrBatchResult(ApiSchema):
files: list[OcrFileResult]
fields: OcrExtractedFields
measurements: list[OcrMeasurement] = Field(default_factory=list)
judgments: list[OcrStandardJudgment] = Field(default_factory=list)
requires_review: bool = True
class OcrJudgmentRequest(ApiSchema):
measurements: list[OcrMeasurement] = Field(min_length=1, max_length=50)

15
app/schemas/risk_item.py
View File

@ -1,7 +1,6 @@
"""Risk item submission and review contracts."""
from datetime import date, datetime
from typing import Any
from datetime import datetime
from pydantic import Field
@ -11,10 +10,8 @@ from app.schemas.common import ApiSchema
class RiskItemCreate(ApiSchema):
source_type: str = Field(default="表单", min_length=1, max_length=20)
source_file_id: str | None = Field(default=None, max_length=50)
occurred_on: date | None = None
product_name: str = Field(min_length=1, max_length=200)
product_category: str | None = Field(default=None, max_length=100)
material_type: str | None = Field(default=None, max_length=200)
supplier: str | None = Field(default=None, max_length=300)
factory: str | None = Field(default=None, max_length=300)
stage_new: str | None = Field(default=None, max_length=100)
@ -22,15 +19,11 @@ class RiskItemCreate(ApiSchema):
defect_level: str | None = Field(default=None, max_length=10)
solution: str | None = None
photos: list[str] = Field(default_factory=list, max_length=10)
recognition_confidence: float | None = Field(default=None, ge=0, le=1)
ocr_payload: dict[str, Any] = Field(default_factory=dict)
class RiskItemUpdate(ApiSchema):
product_name: str | None = Field(default=None, min_length=1, max_length=200)
product_category: str | None = Field(default=None, max_length=100)
material_type: str | None = Field(default=None, max_length=200)
occurred_on: date | None = None
supplier: str | None = Field(default=None, max_length=300)
factory: str | None = Field(default=None, max_length=300)
stage_new: str | None = Field(default=None, max_length=100)
@ -38,18 +31,14 @@ class RiskItemUpdate(ApiSchema):
defect_level: str | None = Field(default=None, max_length=10)
solution: str | None = None
photos: list[str] | None = Field(default=None, max_length=10)
recognition_confidence: float | None = Field(default=None, ge=0, le=1)
ocr_payload: dict[str, Any] | None = None
class RiskItemData(ApiSchema):
id: int
source_type: str
source_file_id: str | None
occurred_on: date | None
product_name: str | None
product_category: str | None
material_type: str | None
supplier: str | None
factory: str | None
stage_new: str | None
@ -57,8 +46,6 @@ class RiskItemData(ApiSchema):
defect_level: str | None
solution: str | None
photos: list[str]
recognition_confidence: float | None
ocr_payload: dict[str, Any]
submitted_by: str | None
submitted_at: datetime | None
status: str

5
app/schemas/risk_prediction.py
View File

@ -43,12 +43,9 @@ class SupplierFrequency(ApiSchema):
supplier_name: str | None = None
cooperation_count: int | None = None
anomaly_count: int | None = None
alpha: float | None = None
alpha: float = 1.0
tier: str | None = None
suggestion: str | None = None
frequency_basis: str = "暂无匹配采购订单"
note: str | None = None
data_available: bool = False
class StandardJudgment(ApiSchema):

64
app/schemas/suppliers.py
View File

@ -1,64 +0,0 @@
"""Supplier profile API contracts."""
from pydantic import Field
from app.schemas.common import ApiSchema
class SupplierOption(ApiSchema):
name: str
anomaly_count: int
tier: str
class SupplierListData(ApiSchema):
items: list[SupplierOption] = Field(default_factory=list)
default_supplier: str | None = None
class SupplierMetric(ApiSchema):
cooperation_total: int | None = None
anomaly_total: int
anomaly_rate: float | None = None
frequency_alpha: float
risk_score: float | None = None
main_category_count: int
cooperation_count_basis: str = "按近3年采购订单去重单据编号统计"
anomaly_rate_basis: str = "暂无采购订单/生产批次总量，无法计算真实异常率"
class SupplierCategoryBreakdown(ApiSchema):
category: str
anomaly_count: int
anomaly_rate: float | None = None
top_material: str | None = None
top_problem: str | None = None
anomaly_rate_basis: str = "暂无供应商品类合作总量，无法计算真实异常率"
class SupplierRecentRecord(ApiSchema):
case_id: int
month: str | None = None
product: str | None = None
anomaly: str | None = None
defect_level: str | None = None
class FactoryRecommendation(ApiSchema):
category: str
preferred_factory: str | None = None
alternative_factory: str | None = None
reason: str
recommendation_basis: str = "暂无加工厂生产总量，无法计算真实异常率"
class SupplierProfileData(ApiSchema):
supplier_name: str
tier: str
tier_condition: str
metric: SupplierMetric
category_breakdown: list[SupplierCategoryBreakdown] = Field(default_factory=list)
monthly_trend: list[dict[str, int | str]] = Field(default_factory=list)
recent_records: list[SupplierRecentRecord] = Field(default_factory=list)
factory_recommendations: list[FactoryRecommendation] = Field(default_factory=list)
recommendation_note: str | None = None

179
app/services/baidu_ocr.py
View File

@ -1,179 +0,0 @@
"""Small async client for the Baidu OCR endpoints used by the application."""
import asyncio
import base64
import time
from typing import Any
from urllib.parse import urlparse
import httpx
from app.core.config import get_settings
from app.core.exceptions import AppException
class BaiduOcrClient:
token_url = "https://aip.baidubce.com/oauth/2.0/token"
def __init__(self, transport: httpx.AsyncBaseTransport | None = None) -> None:
settings = get_settings()
self.api_key = settings.baidu_ocr_api_key.get_secret_value() if settings.baidu_ocr_api_key else ""
self.secret_key = settings.baidu_ocr_secret_key.get_secret_value() if settings.baidu_ocr_secret_key else ""
self.image_url = settings.baidu_ocr_image_url.strip()
self.document_url = settings.baidu_ocr_document_url.strip().rstrip("/")
self.timeout_seconds = settings.baidu_ocr_timeout_seconds
self.poll_interval = settings.baidu_ocr_poll_interval_seconds
self.transport = transport
self._access_token = ""
self._token_expires_at = 0.0
@property
def configured(self) -> bool:
return bool(self.api_key and self.secret_key)
async def recognize_image(self, content: bytes) -> tuple[str, float | None, int]:
token = await self._get_access_token()
payload = {
"image": base64.b64encode(content).decode("ascii"),
"language_type": "CHN_ENG",
"detect_direction": "true",
"probability": "true",
}
body = await self._post_form(self.image_url, token, payload)
words = body.get("words_result") or []
lines = [str(item.get("words") or "").strip() for item in words if isinstance(item, dict)]
confidences = [
float(item["probability"]["average"])
for item in words
if isinstance(item, dict)
and isinstance(item.get("probability"), dict)
and isinstance(item["probability"].get("average"), (int, float))
]
text = "\n".join(line for line in lines if line)
if not text:
raise AppException("图片中未识别到文字，请重新拍摄或手动填写", code=422, status_code=422)
confidence = sum(confidences) / len(confidences) if confidences else None
return text, confidence, 1
async def recognize_document(self, content: bytes, file_name: str) -> tuple[str, float | None, int]:
token = await self._get_access_token()
body = await self._post_form(
self.document_url,
token,
{
"file_data": base64.b64encode(content).decode("ascii"),
"file_name": file_name,
"merge_tables": "true",
"return_span_boxes": "true",
"angle_adjust": "true",
},
)
task_id = str((body.get("result") or {}).get("task_id") or "")
if not task_id:
raise AppException("百度文档识别未返回任务编号", code=502, status_code=502)
deadline = time.monotonic() + self.timeout_seconds
query_url = f"{self.document_url}/query"
result: dict[str, Any] = {}
while time.monotonic() < deadline:
await asyncio.sleep(self.poll_interval)
query_body = await self._post_form(query_url, token, {"task_id": task_id})
result = query_body.get("result") or {}
status = result.get("status")
if status == "success":
break
if status == "failed":
detail = str(result.get("task_error") or "文档解析失败")
raise AppException(f"百度文档识别失败：{detail}", code=502, status_code=502)
else:
raise AppException("文档识别超时，请稍后重试", code=504, status_code=504)
parse_url = str(result.get("parse_result_url") or "")
markdown_url = str(result.get("markdown_url") or "")
if parse_url:
parsed = await self._download_result(parse_url, expect_json=True)
text, pages = self._text_from_parse_result(parsed)
elif markdown_url:
text = str(await self._download_result(markdown_url, expect_json=False))
pages = max(1, text.count("\n---\n") + 1)
else:
raise AppException("百度文档识别未返回解析结果", code=502, status_code=502)
if not text.strip():
raise AppException("文档中未识别到文字，请重新上传或手动填写", code=422, status_code=422)
return text[:100_000], None, pages
async def _get_access_token(self) -> str:
if not self.configured:
raise AppException("未配置百度 OCR API Key 或 Secret Key", code=503, status_code=503)
if self._access_token and time.monotonic() < self._token_expires_at:
return self._access_token
timeout = httpx.Timeout(15, connect=10)
async with httpx.AsyncClient(timeout=timeout, transport=self.transport) as client:
try:
response = await client.post(
self.token_url,
params={
"grant_type": "client_credentials",
"client_id": self.api_key,
"client_secret": self.secret_key,
},
)
response.raise_for_status()
body = response.json()
except (httpx.HTTPError, ValueError) as exc:
raise AppException("无法连接百度 OCR 鉴权服务", code=502, status_code=502) from exc
token = str(body.get("access_token") or "")
if not token:
raise AppException("百度 OCR 密钥无效或服务尚未开通", code=502, status_code=502)
expires_in = int(body.get("expires_in") or 2_592_000)
self._access_token = token
self._token_expires_at = time.monotonic() + max(60, expires_in - 300)
return token
async def _post_form(self, url: str, token: str, data: dict[str, str]) -> dict[str, Any]:
timeout = httpx.Timeout(self.timeout_seconds, connect=15)
async with httpx.AsyncClient(timeout=timeout, transport=self.transport) as client:
try:
response = await client.post(url, params={"access_token": token}, data=data)
response.raise_for_status()
body = response.json()
except httpx.TimeoutException as exc:
raise AppException("百度 OCR 请求超时，请重试", code=504, status_code=504) from exc
except (httpx.HTTPError, ValueError) as exc:
raise AppException("百度 OCR 服务暂时不可用", code=502, status_code=502) from exc
if body.get("error_code") not in (None, 0, "0"):
message = str(body.get("error_msg") or "识别失败")
raise AppException(f"百度 OCR 返回错误：{message}", code=502, status_code=502)
return body
async def _download_result(self, url: str, *, expect_json: bool) -> dict[str, Any] | str:
parsed = urlparse(url)
allowed = parsed.scheme == "https" and (
parsed.hostname == "baidubce.com"
or bool(parsed.hostname and parsed.hostname.endswith((".baidubce.com", ".bcebos.com")))
)
if not allowed:
raise AppException("百度 OCR 返回了不受信任的结果地址", code=502, status_code=502)
timeout = httpx.Timeout(self.timeout_seconds, connect=15)
async with httpx.AsyncClient(timeout=timeout, transport=self.transport, follow_redirects=True) as client:
try:
response = await client.get(url)
response.raise_for_status()
return response.json() if expect_json else response.text
except (httpx.HTTPError, ValueError) as exc:
raise AppException("无法下载百度 OCR 解析结果", code=502, status_code=502) from exc
@staticmethod
def _text_from_parse_result(body: dict[str, Any]) -> tuple[str, int]:
pages = body.get("pages") or []
chunks: list[str] = []
for page in pages:
if not isinstance(page, dict):
continue
page_text = str(page.get("text") or "").strip()
if page_text:
chunks.append(page_text)
for table in page.get("tables") or []:
if isinstance(table, dict) and table.get("markdown"):
chunks.append(str(table["markdown"]).strip())
return "\n\n".join(chunks), max(1, len(pages))

272
app/services/conversations.py
View File

@ -4,7 +4,7 @@ import json
import logging
import re
from collections.abc import AsyncGenerator, Sequence
from datetime import UTC, date, datetime, timedelta
from datetime import UTC, datetime
from typing import Any, Literal
from sqlalchemy import func, or_, select
@ -23,20 +23,17 @@ from app.schemas.risk_prediction import (
RiskAssessmentResult,
RiskDimensionScore,
RiskFactorResult,
SimilarCase,
StandardJudgment,
SupplierFrequency,
)
from app.services.database_risk_scoring import DatabaseRiskScoringService
from app.services.llm_client import LLMClient
from app.services.purchase_orders import supplier_order_counts
from app.services.rag import AnomalyCaseRAG
logger = logging.getLogger("dieya_risk_prediction.conversations")
class ConversationService:
_supplier_history_days = 365 * 3
_engine = BaselineRiskEngine()
_risk_intents = {
"assessment",
@ -235,53 +232,37 @@ class ConversationService:
yield {"type": "error", "message": "LLM 未返回有效内容"}
return
# Persist the complete assistant text before optional card enrichment.
# A slow or failed RAG/score lookup must not lose the conversation turn.
show_risk = self._is_risk_request(text, semantics)
response_type = self._response_type(text, semantics)
response_type = await self._use_solution_card_without_history(response_type, text, semantics)
risk_result = (
await self._enrich_risk_result(text, semantics, self._predict(text))
if show_risk and response_type == "risk_assessment"
else None
)
response_card = (
await self._build_response_card(response_type, text, semantics)
if response_type != "general"
else None
)
assistant_message = ConversationMessage(
conversation_id=conversation.id,
role="assistant",
message_type="text",
content=reply,
structured_data={"semantic": semantics},
structured_data={
"semantic": semantics,
**(
{"riskResult": risk_result.model_dump(mode="json", by_alias=False)}
if risk_result
else {}
),
**({"responseCard": response_card} if response_card else {}),
},
)
self.session.add(assistant_message)
await self.session.flush()
await self.session.commit()
show_risk = self._is_risk_request(text, semantics)
response_type = self._response_type(text, semantics)
risk_result = None
response_card = None
try:
response_type = await self._use_solution_card_without_history(response_type, text, semantics)
risk_result = (
await self._enrich_risk_result(text, semantics, self._predict(text))
if show_risk and response_type == "risk_assessment"
else None
)
response_card = (
await self._build_response_card(response_type, text, semantics)
if response_type != "general"
else None
)
except Exception as exc: # noqa: BLE001 - text reply must remain persisted
logger.exception(
"Conversation enrichment failed after assistant text was persisted error=%s",
exc.__class__.__name__,
)
await self.session.rollback()
assistant_message.structured_data = {
"semantic": semantics,
**(
{"riskResult": risk_result.model_dump(mode="json", by_alias=False)}
if risk_result
else {}
),
**({"responseCard": response_card} if response_card else {}),
}
await self.session.flush()
await self.session.commit()
persisted = await self.repository.get_for_user(conversation.id, user.id)
if persisted is None: # pragma: no cover - the row already existed
yield {"type": "error", "message": "对话读取失败"}
@ -556,46 +537,25 @@ class ConversationService:
name = match.group(1).strip() if match else "未指定"
is_factory = any(word in text for word in ("加工厂", "工厂", "生产厂")) and "供应商" not in text
column = AnomalyCase.factory if is_factory else AnomalyCase.supplier
anomaly_count, a_count, b_count = await self._supplier_history_stats(column, name)
cooperation_count = (
await supplier_order_counts(
self.session,
name,
self._supplier_history_start(),
datetime.now(UTC).date(),
)
if hasattr(self, "session")
else None
)
has_order_data = bool(cooperation_count)
cooperation_display = cooperation_count if has_order_data else None
problems = await self._top_anomaly_types(
column,
name,
since=self._supplier_history_start(),
)
anomaly_count = await self._count_anomalies(column, name)
problems = await self._top_anomaly_types(column, name)
party_type = "加工厂" if is_factory else "供应商"
tier = self._supplier_tier(anomaly_count, a_count=a_count, b_count=b_count)
return {
"type": "supplier",
"title": "供应商深度画像",
"name": name,
"partyType": party_type,
"anomalyCount": anomaly_count,
"cooperationCount": cooperation_display,
"tier": tier,
"frequencyCoefficient": round(min(1.5, 1 + cooperation_display / 100 * 0.3), 2)
if cooperation_display is not None
else 1.0,
"cooperationCount": None,
"tier": self._supplier_tier(anomaly_count),
"frequencyCoefficient": round(min(1.5, 1 + anomaly_count / 100 * 0.3), 2) if anomaly_count else 1.0,
"topProblems": problems,
"suggestions": self._supplier_management_suggestions(tier),
"frequencyBasis": "近3年采购订单去重单据数" if has_order_data else "暂无匹配采购订单",
"note": (
"合作次数按近3年采购订单去重单据编号统计。"
if has_order_data
else "未匹配到该供应商采购订单，不展示合作次数。"
"suggestions": (
["加强巡检并准备备选供应商", "重点复核高频异常对应的来料和过程检验项目"]
if anomaly_count
else ["暂无历史合作数据，建议首单全检并安排驻厂巡检"]
),
"dataAvailable": has_order_data,
"note": "合作次数需要采购合作台账接入后统计，当前仅展示异常案例库中的真实记录。",
}
async def _material_card(self, text: str, semantics: dict[str, Any]) -> dict[str, Any]:
@ -841,27 +801,12 @@ class ConversationService:
for row in rows
]
async def _top_anomaly_types(
self,
column, # noqa: ANN001
value: str | None,
*,
since: date | None = None,
) -> Sequence[dict[str, Any]]:
async def _top_anomaly_types(self, column, value: str | None) -> Sequence[dict[str, Any]]: # noqa: ANN001
if not value:
return []
filters = [column.ilike(f"%{value}%"), AnomalyCase.anomaly_type.is_not(None)]
if since:
filters.extend(
[
AnomalyCase.date.is_not(None),
AnomalyCase.date >= since,
AnomalyCase.date <= datetime.now(UTC).date(),
]
)
rows = await self.session.execute(
select(AnomalyCase.anomaly_type, func.count())
.where(*filters)
.where(column.ilike(f"%{value}%"), AnomalyCase.anomaly_type.is_not(None))
.group_by(AnomalyCase.anomaly_type)
.order_by(func.count().desc())
.limit(5)
@ -972,10 +917,7 @@ class ConversationService:
defect_level=self._first_match(text, ("A级", "B级", "C级")),
)
category_count = await self._count_category_anomalies(category)
supplier_count, supplier_a_count, supplier_b_count = await self._supplier_history_stats(
AnomalyCase.supplier,
supplier,
)
supplier_count = await self._count_anomalies(AnomalyCase.supplier, supplier)
factory_count = await self._count_anomalies(AnomalyCase.factory, factory)
anomaly_count = supplier_count if supplier else factory_count
standard_rows = await self._standards_for(category, text)
@ -992,27 +934,9 @@ class ConversationService:
anomaly_type=self._semantic_text(semantics, "anomaly_type"),
)
cooperation_count = (
await supplier_order_counts(
self.session,
supplier,
self._supplier_history_start(),
datetime.now(UTC).date(),
)
if supplier
else None
)
alpha_count = cooperation_count or 0
alpha = round(min(1.5, 1 + (alpha_count / 100) * 0.3), 2) if cooperation_count else 1.0
tier = (
self._supplier_tier(
alpha_count,
a_count=supplier_a_count,
b_count=supplier_b_count,
)
if supplier
else None
)
alpha_count = supplier_count if supplier else 0
alpha = round(min(1.5, 1 + (alpha_count / 100) * 0.3), 2) if alpha_count else 1.0
tier = self._supplier_tier(alpha_count) if supplier else None
level_text = self._risk_level_text(database_score.level)
alert_text = self._alert_text(
category,
@ -1074,34 +998,21 @@ class ConversationService:
else "建议结合当前品类高频异常制定验证计划"
),
),
"supplier_freq": (
SupplierFrequency(
supplier_name=supplier,
cooperation_count=cooperation_count,
anomaly_count=supplier_count,
alpha=alpha,
tier=tier,
suggestion="；".join(self._supplier_management_suggestions(tier)),
frequency_basis="近3年采购订单去重单据数" if cooperation_count else "暂无匹配采购订单",
note=(
"合作次数按采购订单去重单据编号统计；异常数仍按近3年异常案例统计。"
if cooperation_count
else "未匹配到该供应商采购订单，不展示合作次数。"
),
data_available=bool(cooperation_count),
)
if supplier
else SupplierFrequency(
note="未指定供应商，暂无供应商频次分析。",
frequency_basis="未提供供应商",
)
"supplier_freq": SupplierFrequency(
supplier_name=supplier_or_factory,
cooperation_count=None,
anomaly_count=anomaly_count or None,
alpha=alpha,
tier=tier,
suggestion=(
"无历史数据，建议首单全检并安排驻厂巡检"
if not anomaly_count
else "建议加强巡检并准备备选供应商"
),
),
"standard_judgment": standard_rows,
"dimension_scores": dimensions,
"similar_cases": [
SimilarCase.model_validate(item.case.model_dump(mode="json", by_alias=False))
for item in similar_cases
],
"similar_cases": [item.case.model_dump(mode="json", by_alias=False) for item in similar_cases],
"ai_suggestions": suggestions,
}
)
@ -1110,47 +1021,10 @@ class ConversationService:
if not value or value == "未归类":
return 0
count = await self.session.scalar(
select(func.count())
.select_from(AnomalyCase)
.where(
column.ilike(f"%{value}%"),
AnomalyCase.date.is_not(None),
AnomalyCase.date >= self._supplier_history_start(),
AnomalyCase.date <= datetime.now(UTC).date(),
)
select(func.count()).select_from(AnomalyCase).where(column.ilike(f"%{value}%"))
)
return int(count or 0)
@classmethod
def _supplier_history_start(cls) -> date:
return datetime.now(UTC).date() - timedelta(days=cls._supplier_history_days)
async def _supplier_history_stats(
self,
column, # noqa: ANN001
value: str | None,
) -> tuple[int, int, int]:
"""Return the PRD's three-year N, A-count, and B-count inputs."""
if not value or value == "未归类":
return 0, 0, 0
rows = (
await self.session.execute(
select(AnomalyCase.defect_level, func.count())
.where(
column.ilike(f"%{value}%"),
AnomalyCase.date.is_not(None),
AnomalyCase.date >= self._supplier_history_start(),
AnomalyCase.date <= datetime.now(UTC).date(),
)
.group_by(AnomalyCase.defect_level)
)
).all()
total = sum(int(count or 0) for _, count in rows)
a_count = sum(int(count or 0) for level, count in rows if self._defect_code(level) == "A")
b_count = sum(int(count or 0) for level, count in rows if self._defect_code(level) == "B")
return total, a_count, b_count
async def _generate_ai_suggestions(
self,
*,
@ -1258,10 +1132,7 @@ class ConversationService:
or_(
AnomalyCase.product_category.ilike(f"%{category}%"),
AnomalyCase.product_name.ilike(f"%{category}%"),
),
AnomalyCase.date.is_not(None),
AnomalyCase.date >= self._supplier_history_start(),
AnomalyCase.date <= datetime.now(UTC).date(),
)
)
)
return int(count or 0)
@ -1342,40 +1213,17 @@ class ConversationService:
}[level]
@staticmethod
def _defect_code(value: str | None) -> str | None:
match = re.search(r"[ABCD]", (value or "").upper())
return match.group(0) if match else None
@staticmethod
def _supplier_tier(
anomaly_count: int,
*,
a_count: int = 0,
b_count: int = 0,
) -> str:
if anomaly_count == 0:
def _supplier_tier(anomaly_count: int) -> str | None:
if not anomaly_count:
return "新供应商"
ab_ratio = (a_count + b_count) / anomaly_count
if anomaly_count > 120 or a_count > 0 or ab_ratio > 0.5:
return "C级 · 高风险"
if anomaly_count > 60 or ab_ratio >= 0.3:
return "B级 · 关注"
if anomaly_count >= 20:
return "A级 · 优质"
if a_count == 0:
if anomaly_count < 20:
return "S级 · 战略"
if anomaly_count <= 60:
return "A级 · 优质"
if anomaly_count <= 120:
return "B级 · 关注"
return "C级 · 高风险"
@staticmethod
def _supplier_management_suggestions(tier: str | None) -> Sequence[str]:
return {
"S级 · 战略": ["正常合作", "季度回顾"],
"A级 · 优质": ["常规管理", "年度评审"],
"B级 · 关注": ["加强巡检", "新品打样加严"],
"C级 · 高风险": ["安排驻厂巡检", "储备备选供应商", "限制新品使用"],
"新供应商": ["无历史数据", "首单全检", "安排驻厂巡检"],
}.get(tier, [])
@staticmethod
def _alert_text(
category: str,

10
app/services/dashboard.py
View File

@ -286,8 +286,8 @@ class DashboardService:
result = []
for (name, party_type), item in sorted(actors.items(), key=lambda entry: -entry[1]["count"]):
anomaly_count = item["count"]
# No order or production denominator exists in anomaly_cases.
cooperation_total = None
# Keep the existing cooperation-total calculation unchanged.
cooperation_total = item["count"]
values = {category: 0 for category in categories}
for row in item["items"]:
category = cls._category(row["product_name"], row["material_type"])
@ -300,7 +300,11 @@ class DashboardService:
categories=values,
cooperation_total=cooperation_total,
anomaly_count=anomaly_count,
anomaly_rate=None,
anomaly_rate=(
round(anomaly_count / cooperation_total * 100, 1)
if cooperation_total
else 0
),
tier=item["tier"],
)
)

92
app/services/database_risk_scoring.py
View File

@ -9,10 +9,18 @@ from typing import Any
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.domain.model_config import DEFAULT_FREQUENCY_CAP, DEFAULT_FREQUENCY_COEFFICIENT, DEFAULT_WEIGHTS
from app.domain.risk_engine import RiskLevel
from app.models.entities import AnomalyCase, RiskModelConfig
from app.services.purchase_orders import supplier_order_counts
from app.models.entities import AnomalyCase
DEFAULT_WEIGHTS = {
"供应商表现": 25.0,
"物料品类": 20.0,
"产品品类": 15.0,
"加工厂质量": 15.0,
"标准符合性": 10.0,
"节点阶段": 10.0,
"缺陷等级": 5.0,
}
STAGES = ("需求发现", "产品立项", "产品开发", "量产上市", "产品生命管理")
@ -46,9 +54,6 @@ class DatabaseRiskScoringService:
def __init__(self, session: AsyncSession, *, as_of: date | None = None) -> None:
self.session = session
self.as_of = as_of or date.today()
self.weights = DEFAULT_WEIGHTS.copy()
self.frequency_coefficient = DEFAULT_FREQUENCY_COEFFICIENT
self.frequency_cap = DEFAULT_FREQUENCY_CAP
async def score(
self,
@ -60,29 +65,27 @@ class DatabaseRiskScoringService:
stage: str | None = None,
defect_level: str | None = None,
) -> DatabaseRiskScore:
await self._load_model_config()
rows = await self._load_rows()
category_name = normalize_category(category)
material_name = normalize_material(material)
stage_name = normalize_stage(stage)
party_name = clean_value(supplier)
cooperation_count = await self._supplier_order_count(party_name)
dimensions = (
self._supplier_dimension(rows, party_name, cooperation_count),
self._supplier_dimension(rows, party_name),
self._material_dimension(rows, material_name),
self._category_dimension(rows, category_name),
DimensionScore(
name="加工厂质量",
score=50.0,
weight=self.weights["加工厂质量"],
weight=DEFAULT_WEIGHTS["加工厂质量"],
status="default",
reason="当前数据库没有巡检合格率/批次数据，按 PRD 中性值 50 分计入",
),
DimensionScore(
name="标准符合性",
score=50.0,
weight=self.weights["标准符合性"],
weight=DEFAULT_WEIGHTS["标准符合性"],
status="default",
reason="当前数据库没有检测结果与标准判定记录，按 PRD 中性值 50 分计入",
),
@ -92,32 +95,6 @@ class DatabaseRiskScoringService:
score = round(sum(item.score * item.weight / 100 for item in dimensions), 2)
return DatabaseRiskScore(score=score, level=classify_score(score), dimensions=dimensions)
async def _supplier_order_count(self, party_name: str | None) -> int | None:
if not party_name or not hasattr(self.session, "execute"):
return None
try:
return await supplier_order_counts(
self.session,
party_name,
self.as_of - timedelta(days=365 * 3),
self.as_of,
)
except Exception: # noqa: BLE001 - legacy/test databases may not have the order ledger yet
return None
async def _load_model_config(self) -> None:
get = getattr(self.session, "get", None)
if get is None:
return
config = await get(RiskModelConfig, 1)
if config is None:
return
weights = {key: float(config.weights.get(key, value)) for key, value in DEFAULT_WEIGHTS.items()}
if abs(sum(weights.values()) - 100) <= 0.01 and all(value > 0 for value in weights.values()):
self.weights = weights
self.frequency_coefficient = float(config.frequency_coefficient)
self.frequency_cap = float(config.frequency_cap)
async def _load_rows(self) -> list[dict[str, Any]]:
start = self.as_of - timedelta(days=365 * 3)
statement = select(
@ -136,13 +113,8 @@ class DatabaseRiskScoringService:
)
return [dict(row) for row in (await self.session.execute(statement)).mappings().all()]
def _supplier_dimension(
self,
rows: list[dict[str, Any]],
party_name: str | None,
cooperation_count: int | None = None,
) -> DimensionScore:
weight = self.weights["供应商表现"]
def _supplier_dimension(self, rows: list[dict[str, Any]], party_name: str | None) -> DimensionScore:
weight = DEFAULT_WEIGHTS["供应商表现"]
if not party_name:
return DimensionScore(
name="供应商表现",
@ -152,8 +124,7 @@ class DatabaseRiskScoringService:
reason="未提供供应商/加工厂，按 PRD 缺失维度默认 50 分",
)
matched = [row for row in rows if party_matches(row, party_name)]
anomaly_count = len(matched)
count = cooperation_count if cooperation_count is not None and cooperation_count > 0 else anomaly_count
count = len(matched)
if count == 0:
base = 70.0
elif count <= 20:
@ -184,24 +155,21 @@ class DatabaseRiskScoringService:
else 0
)
base = max(0.0, min(100.0, base + trend_adjustment))
alpha = min(self.frequency_cap, 1.0 + count / 100 * self.frequency_coefficient)
alpha = min(1.5, 1.0 + count / 100 * 0.3)
score = min(100.0, round(base * alpha, 2))
return DimensionScore(
name="供应商表现",
score=score,
weight=weight,
status="database",
reason=(
f"近 3 年采购订单 {count} 单，异常案例 {anomaly_count} 条，基础分 {base:g}，α={alpha:.2f}"
if cooperation_count is not None and cooperation_count > 0
else f"近 3 年异常 {count} 条，基础分 {base:g}，α={alpha:.2f}"
),
sample_size=anomaly_count,
reason=f"近 3 年异常 {count} 条，基础分 {base:g}，α={alpha:.2f}",
sample_size=count,
raw_value=count,
)
def _material_dimension(self, rows: list[dict[str, Any]], material: str | None) -> DimensionScore:
weight = self.weights["物料品类"]
@staticmethod
def _material_dimension(rows: list[dict[str, Any]], material: str | None) -> DimensionScore:
weight = DEFAULT_WEIGHTS["物料品类"]
if not material:
return DimensionScore(
name="物料品类",
@ -228,8 +196,9 @@ class DatabaseRiskScoringService:
raw_value=count,
)
def _category_dimension(self, rows: list[dict[str, Any]], category: str | None) -> DimensionScore:
weight = self.weights["产品品类"]
@staticmethod
def _category_dimension(rows: list[dict[str, Any]], category: str | None) -> DimensionScore:
weight = DEFAULT_WEIGHTS["产品品类"]
if not category:
return DimensionScore(
name="产品品类",
@ -263,8 +232,9 @@ class DatabaseRiskScoringService:
raw_value=count,
)
def _stage_dimension(self, rows: list[dict[str, Any]], stage: str | None) -> DimensionScore:
weight = self.weights["节点阶段"]
@staticmethod
def _stage_dimension(rows: list[dict[str, Any]], stage: str | None) -> DimensionScore:
weight = DEFAULT_WEIGHTS["节点阶段"]
if not stage:
return DimensionScore(
name="节点阶段",
@ -296,15 +266,15 @@ class DatabaseRiskScoringService:
raw_value=share * 100,
)
@staticmethod
def _defect_dimension(
self,
rows: list[dict[str, Any]],
category: str | None,
material: str | None,
party_name: str | None,
defect_level: str | None,
) -> DimensionScore:
weight = self.weights["缺陷等级"]
weight = DEFAULT_WEIGHTS["缺陷等级"]
if not category and not material and not party_name and not defect_level:
return DimensionScore(
name="缺陷等级",

16
app/services/dingtalk_auth.py
View File

@ -170,15 +170,21 @@ class DingTalkAuthService:
@staticmethod
def map_role(dept: str, title: str, is_admin: bool = False) -> str:
"""Map DingTalk organization snapshots to the four-role administration model."""
"""Map DingTalk organization snapshots to the PRD seven-role model."""
text = f"{dept} {title}".lower()
if is_admin or "it" in text or "信息" in text or "系统" in text:
return "admin"
if "审核" in text or "review" in text:
return "reviewer"
if "产品经理" in text or "product manager" in text or (
"产品" in dept and ("经理" in title or "主管" in title)
):
return "product_manager"
if "采购" in text or "procurement" in text:
return "procurement"
if "品控" in text or "质量" in text or "qc" in text:
return "qc"
if "管理" in text or "主管" in text or "manager" in text:
return "manager"
if "运营" in text or "ops" in text:
return "ops"
if "开发" in text or "产品" in text or "dev" in text:
return "dev"
return "普通用户"

39
app/services/llm_client.py
View File

@ -135,40 +135,6 @@ class LLMClient:
status_code=503,
) from fallback_exc
async def extract_ocr_fields(self, text: str) -> dict[str, Any]:
"""Turn OCR text into an editable risk-item draft without inventing absent values."""
messages = [
{
"role": "system",
"content": (
"你是蝶雅产品风险预估系统的OCR字段抽取模块，只输出合法JSON对象。"
"只能提取输入文字明确出现的信息，不得推测或补造。输出字段：product_name、product_category、"
"supplier、factory、material_type、stage_new、description、defect_level、occurred_on、confidence、"
"measurements。缺失标量用null。defect_level只能是A、B、C或null；occurred_on使用YYYY-MM-DD或null。"
"measurements为数组，每项包含item、raw_value、value、unit、confidence；value必须是数字或null。"
"measurements只允许质量检测指标，例如pH、色差、重金属含量、力矩、喷出量、附着力、"
"不合格率；数量、抽检数量、产品规格、容量、日期、单号和批次不得作为检测指标。"
"confidence为0到1。description应保留异常现象和影响比例，但不得加入原文没有的结论。"
),
},
{"role": "user", "content": text[:20_000]},
]
last_error: Exception | None = None
providers: list[ProviderName] = ["primary"]
if self.fallback_configured:
providers.append("fallback")
for provider in providers:
try:
body = await self._request(messages, provider=provider, stream=False, json_mode=True)
return self._parse_json(self._response_text(body, provider=provider))
except Exception as exc: # noqa: BLE001 - deterministic extraction handles provider failures
last_error = exc
logger.warning("LLM OCR extraction failed provider=%s error=%s", provider, exc.__class__.__name__)
if last_error:
raise last_error
return {}
async def stream_reply(self, messages: list[dict[str, str]]) -> AsyncGenerator[str, None]:
"""Yield reply deltas and fail over only before any primary text is emitted."""
@ -213,8 +179,9 @@ class LLMClient:
},
{
"role": "user",
"content": "以下是本次风险评估数据，请生成预防建议：\n"
+ json.dumps(context, ensure_ascii=False, separators=(",", ":")),
"content": "以下是本次风险评估数据，请生成预防建议：\n" + json.dumps(
context, ensure_ascii=False, separators=(",", ":")
),
},
]
try:

57
app/services/model_config.py
View File

@ -1,57 +0,0 @@
"""Read and update administrator-managed risk-model configuration."""
from decimal import Decimal
from sqlalchemy.ext.asyncio import AsyncSession
from app.domain.model_config import DEFAULT_FREQUENCY_CAP, DEFAULT_FREQUENCY_COEFFICIENT, DEFAULT_WEIGHTS
from app.models.entities import OperationAudit, RiskModelConfig, UserAccount
from app.schemas.model_config import ModelWeightConfigData, ModelWeightConfigUpdate
class ModelConfigService:
def __init__(self, session: AsyncSession) -> None:
self.session = session
async def get(self) -> ModelWeightConfigData:
row = await self.session.get(RiskModelConfig, 1)
if row is None:
return ModelWeightConfigData(
weights=DEFAULT_WEIGHTS,
frequency_coefficient=DEFAULT_FREQUENCY_COEFFICIENT,
frequency_cap=DEFAULT_FREQUENCY_CAP,
)
return self._to_data(row)
async def update(self, payload: ModelWeightConfigUpdate, user: UserAccount) -> ModelWeightConfigData:
row = await self.session.get(RiskModelConfig, 1)
if row is None:
row = RiskModelConfig(id=1)
self.session.add(row)
row.weights = payload.weights
row.frequency_coefficient = Decimal(str(payload.frequency_coefficient))
row.frequency_cap = Decimal(str(payload.frequency_cap))
row.updated_by = user.emp_no
self.session.add(
OperationAudit(
actor_user_id=user.id,
actor_emp_no=user.emp_no,
action="model_weight.update",
resource_type="risk_model_config",
resource_id="1",
audit_metadata=payload.model_dump(),
)
)
await self.session.flush()
await self.session.refresh(row)
return self._to_data(row)
@staticmethod
def _to_data(row: RiskModelConfig) -> ModelWeightConfigData:
return ModelWeightConfigData(
weights={key: float(value) for key, value in row.weights.items()},
frequency_coefficient=float(row.frequency_coefficient),
frequency_cap=float(row.frequency_cap),
updated_by=row.updated_by,
updated_at=row.updated_at,
)

507
app/services/ocr.py
View File

@ -1,507 +0,0 @@
"""OCR upload orchestration, field extraction, and standards evaluation."""
import re
from datetime import date
from pathlib import Path
from typing import Any
from uuid import uuid4
from fastapi import UploadFile
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.config import get_settings
from app.core.exceptions import AppException
from app.models.entities import StandardRule
from app.schemas.ocr import (
OcrBatchResult,
OcrExtractedFields,
OcrField,
OcrFileResult,
OcrMeasurement,
OcrStandardJudgment,
)
from app.services.baidu_ocr import BaiduOcrClient
from app.services.llm_client import LLMClient
_FIELD_ALIASES = {
"product_name": ("产品名称", "产品名", "品名"),
"product_category": ("产品品类", "品类"),
"supplier": ("供应商", "供方"),
"factory": ("加工厂", "生产工厂", "工厂"),
"material_type": ("物料名称", "物料类型", "物料"),
"stage_new": ("发生阶段", "发现阶段", "产品阶段", "阶段"),
"description": ("异常描述", "问题描述", "不良描述", "异常现象"),
"defect_level": ("缺陷等级", "异常等级", "等级"),
}
_MEASUREMENT_WORDS = (
"pH",
"PH",
"色差",
"铅",
"汞",
"砷",
"镉",
"力矩",
"喷出量",
"附着力",
"含量",
"粘度",
"黏度",
"透光率",
"菌落",
"温度",
"湿度",
"不合格率",
)
_NON_MEASUREMENT_NAMES = {
"数量",
"抽检数量",
"产品规格",
"规格",
"容量",
"日期",
"生产日期",
"发生日期",
"单号",
"订单号",
"批次",
"批号",
"页码",
}
_UNIT_PATTERN = r"(?:mg/kg|μg/g|ug/g|g/L|mg/L|N[·.]?m|mPa[·.]?s|Pa[·.]?s|mm|cm|nm|%|级|℃|°C)?"
class OcrService:
def __init__(
self,
session: AsyncSession,
*,
client: BaiduOcrClient | None = None,
llm: LLMClient | None = None,
) -> None:
self.session = session
self.client = client or BaiduOcrClient()
self.llm = llm or LLMClient()
self.settings = get_settings()
async def recognize(self, files: list[UploadFile]) -> OcrBatchResult:
if not files:
raise AppException("请至少上传一个图片或 PDF", code=422, status_code=422)
if len(files) > self.settings.ocr_max_batch_files:
raise AppException(
f"一次最多上传 {self.settings.ocr_max_batch_files} 个文件",
code=422,
status_code=422,
)
rules = list(
(
await self.session.scalars(
select(StandardRule).where(StandardRule.status == "启用").order_by(StandardRule.id)
)
).all()
)
results: list[OcrFileResult] = []
try:
for upload in files:
results.append(await self._recognize_one(upload, rules))
except Exception:
for result in results:
self.upload_path(result.file_id).unlink(missing_ok=True)
raise
fields = self._merge_fields([item.fields for item in results])
measurements = [measurement for item in results for measurement in item.measurements]
judgments = [judgment for item in results for judgment in item.judgments]
return OcrBatchResult(
files=results,
fields=fields,
measurements=measurements,
judgments=judgments,
requires_review=any(item.requires_review for item in results),
)
async def judge(self, measurements: list[OcrMeasurement]) -> list[OcrStandardJudgment]:
rules = list(
(
await self.session.scalars(
select(StandardRule).where(StandardRule.status == "启用").order_by(StandardRule.id)
)
).all()
)
return self._judge_measurements(measurements, rules)
async def _recognize_one(self, upload: UploadFile, rules: list[StandardRule]) -> OcrFileResult:
content = await self._read_limited(upload)
content_type, suffix = self._detect_type(content)
original_name = Path(upload.filename or f"upload{suffix}").name
file_id = f"{uuid4().hex}{suffix}"
target = self.upload_path(file_id)
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(content)
try:
if content_type == "application/pdf":
raw_text, provider_confidence, page_count = await self.client.recognize_document(content, original_name)
engine = "paddle-vl-parser"
else:
raw_text, provider_confidence, page_count = await self.client.recognize_image(content)
engine = "webimage"
extracted = await self._extract(raw_text, provider_confidence)
fields, measurements, extraction_confidence = extracted
judgments = self._judge_measurements(measurements, rules)
warnings = self._warnings(fields, measurements, judgments, provider_confidence)
requires_review = bool(warnings) or any(field.needs_review for field in self._field_values(fields))
return OcrFileResult(
file_id=file_id,
file_name=original_name,
content_type=content_type,
engine=engine,
raw_text=raw_text,
confidence=provider_confidence if provider_confidence is not None else extraction_confidence,
page_count=page_count,
fields=fields,
measurements=measurements,
judgments=judgments,
warnings=warnings,
requires_review=requires_review,
)
except Exception:
target.unlink(missing_ok=True)
raise
finally:
await upload.close()
async def _extract(
self,
text: str,
provider_confidence: float | None,
) -> tuple[OcrExtractedFields, list[OcrMeasurement], float | None]:
fallback_fields = self._extract_fields_by_labels(text, provider_confidence)
fallback_measurements = self._extract_measurements(text, provider_confidence)
if not self.llm.configured:
return fallback_fields, fallback_measurements, provider_confidence
try:
data = await self.llm.extract_ocr_fields(text)
except Exception: # noqa: BLE001 - OCR remains usable when the optional entity extractor fails
return fallback_fields, fallback_measurements, provider_confidence
confidence = self._as_confidence(data.get("confidence"))
fields = self._fields_from_llm(data, confidence, fallback_fields)
measurements = self._measurements_from_llm(data.get("measurements"), text, confidence)
return fields, measurements or fallback_measurements, confidence
def _fields_from_llm(
self,
data: dict[str, Any],
confidence: float | None,
fallback: OcrExtractedFields,
) -> OcrExtractedFields:
values: dict[str, Any] = {}
for key in _FIELD_ALIASES:
value = data.get(key)
if key == "defect_level" and value:
value = str(value).upper().replace("级", "")
if value not in {"A", "B", "C"}:
value = None
if value is None or not str(value).strip():
values[key] = getattr(fallback, key)
else:
values[key] = self._field(str(value).strip(), confidence)
occurred_on = self._as_date(data.get("occurred_on")) or fallback.occurred_on
return OcrExtractedFields(**values, occurred_on=occurred_on)
def _measurements_from_llm(
self,
raw: Any,
source_text: str,
default_confidence: float | None,
) -> list[OcrMeasurement]:
if not isinstance(raw, list):
return []
results: list[OcrMeasurement] = []
for item in raw[:50]:
if not isinstance(item, dict):
continue
name = str(item.get("item") or "").strip()
raw_value = str(item.get("raw_value") or item.get("value") or "").strip()
if (
not name
or self._normalize(name) in {self._normalize(value) for value in _NON_MEASUREMENT_NAMES}
or not raw_value
or not self._numeric_appears(raw_value, source_text)
):
continue
value = self._as_float(item.get("value"))
if value is None:
value = self._as_float(raw_value)
confidence = self._as_confidence(item.get("confidence")) or default_confidence
results.append(
OcrMeasurement(
item=name[:100],
raw_value=raw_value[:100],
value=value,
unit=str(item.get("unit") or "").strip()[:30] or self._unit_from(raw_value),
confidence=confidence,
needs_review=confidence is None or confidence < self.settings.ocr_judgment_confidence,
)
)
return results
def _extract_fields_by_labels(self, text: str, confidence: float | None) -> OcrExtractedFields:
values: dict[str, OcrField] = {}
for field_name, aliases in _FIELD_ALIASES.items():
found: str | None = None
for alias in aliases:
match = re.search(rf"{re.escape(alias)}\s*[:：]?\s*([^\n|｜]{{1,120}})", text, re.IGNORECASE)
if match:
found = match.group(1).strip(" ：:;；")
break
if field_name == "defect_level" and found:
level = re.search(r"\b([ABC])\s*级?\b", found, re.IGNORECASE)
found = level.group(1).upper() if level else None
values[field_name] = self._field(found, confidence)
occurred = None
date_match = re.search(r"(?:发生日期|日期)\s*[:：]?\s*(20\d{2})[-/.年](\d{1,2})[-/.月](\d{1,2})", text)
if date_match:
occurred = self._as_date("-".join(date_match.groups()))
return OcrExtractedFields(**values, occurred_on=occurred)
def _extract_measurements(self, text: str, confidence: float | None) -> list[OcrMeasurement]:
results: list[OcrMeasurement] = []
seen: set[tuple[str, str]] = set()
pattern = re.compile(
rf"(?P<item>[A-Za-z\u4e00-\u9fffΔδ（）()·/\-]{{1,32}}?)\s*[:：]?\s*"
rf"(?P<value>[+\-]?\d+(?:\.\d+)?)\s*(?P<unit>{_UNIT_PATTERN})",
re.IGNORECASE,
)
for line in text.splitlines():
if not any(word.lower() in line.lower() for word in _MEASUREMENT_WORDS):
continue
for match in pattern.finditer(line):
name = match.group("item").strip(" ：:;；-|｜")
raw_value = f"{match.group('value')}{match.group('unit') or ''}"
key = (name, raw_value)
if not name or key in seen:
continue
seen.add(key)
results.append(
OcrMeasurement(
item=name[:100],
raw_value=raw_value,
value=float(match.group("value")),
unit=match.group("unit") or None,
confidence=confidence,
needs_review=confidence is None or confidence < self.settings.ocr_judgment_confidence,
)
)
return results[:50]
def _judge_measurements(
self,
measurements: list[OcrMeasurement],
rules: list[StandardRule],
) -> list[OcrStandardJudgment]:
judgments: list[OcrStandardJudgment] = []
for measurement in measurements:
rule = self._best_rule(measurement.item, rules)
if rule is None:
judgments.append(
OcrStandardJudgment(
measurement_item=measurement.item,
measured_value=measurement.raw_value,
result="未匹配标准",
reason="标准库中没有找到对应检测项",
)
)
continue
result, reason = self.evaluate_standard(measurement, rule.judge_standard)
if measurement.needs_review and result in {"合格", "不合格"}:
result = "待人工确认"
reason = (
f"识别置信度低于 {self.settings.ocr_judgment_confidence:.2f}，"
"需确认检测值后再采用自动判定"
)
judgments.append(
OcrStandardJudgment(
measurement_item=measurement.item,
measured_value=measurement.raw_value,
standard_rule_id=rule.id,
risk_point=rule.risk_point,
standard_value=rule.judge_standard,
result=result,
basis=rule.regulation,
reason=reason,
)
)
return judgments
@staticmethod
def _best_rule(item: str, rules: list[StandardRule]) -> StandardRule | None:
needle = OcrService._normalize(item)
if not needle:
return None
scored: list[tuple[float, StandardRule]] = []
for rule in rules:
haystack = OcrService._normalize(f"{rule.risk_point} {rule.judge_standard}")
if needle in haystack or haystack in needle:
score = 2.0 + min(len(needle), len(haystack)) / 100
else:
left = set(OcrService._bigrams(needle))
right = set(OcrService._bigrams(haystack))
score = len(left & right) / max(1, len(left | right))
threshold = 0.45 if len(needle) <= 4 else 0.3
if score >= threshold:
scored.append((score, rule))
return max(scored, key=lambda pair: pair[0])[1] if scored else None
@staticmethod
def evaluate_standard(measurement: OcrMeasurement, standard: str) -> tuple[str, str | None]:
if measurement.value is None:
return "待人工确认", "检测值不是可比较的数字"
normalized = standard.replace("～", "~").replace("—", "-").replace("至", "~")
range_match = re.search(r"(-?\d+(?:\.\d+)?)\s*(?:~|-)\s*(-?\d+(?:\.\d+)?)", normalized)
if range_match:
mismatch = OcrService._unit_mismatch(measurement.unit, normalized[range_match.end() :])
if mismatch:
return "待人工确认", mismatch
lower, upper = map(float, range_match.groups())
passed = lower <= measurement.value <= upper
return ("合格" if passed else "不合格"), None
comparator = re.search(r"(<=|>=|≤|≥|<|>)\s*(?:±\s*)?(-?\d+(?:\.\d+)?)", normalized)
if comparator:
mismatch = OcrService._unit_mismatch(measurement.unit, normalized[comparator.end() :])
if mismatch:
return "待人工确认", mismatch
operator, raw_limit = comparator.groups()
limit = float(raw_limit)
value = abs(measurement.value) if "±" in comparator.group(0) else measurement.value
passed = {
"<=": value <= limit,
"≤": value <= limit,
">=": value >= limit,
"≥": value >= limit,
"<": value < limit,
">": value > limit,
}[operator]
return ("合格" if passed else "不合格"), None
return "待人工确认", "该标准不是系统可自动计算的数值范围"
def _warnings(
self,
fields: OcrExtractedFields,
measurements: list[OcrMeasurement],
judgments: list[OcrStandardJudgment],
confidence: float | None,
) -> list[str]:
warnings: list[str] = []
if confidence is None:
warnings.append("服务未返回文字级置信度，请人工核对识别结果")
elif confidence < self.settings.ocr_review_confidence:
warnings.append("OCR 置信度低于 0.8，请重点核对标黄字段")
if not fields.product_name.value:
warnings.append("未识别到产品名称")
if not fields.description.value:
warnings.append("未识别到异常描述")
if measurements and any(item.result == "不合格" for item in judgments):
warnings.append("检测数值存在不合格项，请确认后提交审核")
return warnings
async def _read_limited(self, upload: UploadFile) -> bytes:
limit = self.settings.ocr_max_file_mb * 1024 * 1024
content = await upload.read(limit + 1)
if len(content) > limit:
raise AppException(f"单个文件不能超过 {self.settings.ocr_max_file_mb}MB", code=413, status_code=413)
if not content:
raise AppException("不能上传空文件", code=422, status_code=422)
return content
@staticmethod
def _detect_type(content: bytes) -> tuple[str, str]:
if content.startswith(b"%PDF-"):
return "application/pdf", ".pdf"
if content.startswith(b"\x89PNG\r\n\x1a\n"):
return "image/png", ".png"
if content.startswith(b"\xff\xd8\xff"):
return "image/jpeg", ".jpg"
raise AppException("仅支持 JPG、PNG 或 PDF 文件", code=422, status_code=422)
def upload_path(self, file_id: str) -> Path:
if not re.fullmatch(r"[0-9a-f]{32}\.(?:jpg|png|pdf)", file_id):
raise AppException("附件不存在", code=404, status_code=404)
root = self.settings.ocr_upload_dir.resolve()
return root / file_id
@staticmethod
def _field(value: str | None, confidence: float | None) -> OcrField:
threshold = get_settings().ocr_review_confidence
return OcrField(
value=value,
confidence=confidence,
needs_review=not value or confidence is None or confidence < threshold,
)
@staticmethod
def _field_values(fields: OcrExtractedFields) -> list[OcrField]:
return [getattr(fields, key) for key in _FIELD_ALIASES]
@staticmethod
def _merge_fields(items: list[OcrExtractedFields]) -> OcrExtractedFields:
merged: dict[str, Any] = {}
for key in _FIELD_ALIASES:
candidates = [getattr(item, key) for item in items if getattr(item, key).value]
merged[key] = max(candidates, key=lambda field: field.confidence or -1) if candidates else OcrField()
merged["occurred_on"] = next((item.occurred_on for item in items if item.occurred_on), None)
return OcrExtractedFields(**merged)
@staticmethod
def _as_confidence(value: Any) -> float | None:
try:
number = float(value)
except (TypeError, ValueError):
return None
return min(1.0, max(0.0, number))
@staticmethod
def _as_float(value: Any) -> float | None:
match = re.search(r"[+\-]?\d+(?:\.\d+)?", str(value or ""))
return float(match.group()) if match else None
@staticmethod
def _as_date(value: Any) -> date | None:
if not value:
return None
try:
return date.fromisoformat(str(value).strip()[:10])
except ValueError:
return None
@staticmethod
def _unit_from(value: str) -> str | None:
match = re.search(_UNIT_PATTERN + r"$", value, re.IGNORECASE)
return match.group() if match and match.group() else None
@staticmethod
def _unit_mismatch(measured_unit: str | None, standard_tail: str) -> str | None:
if not measured_unit:
return None
standard_unit = re.match(rf"\s*({_UNIT_PATTERN})", standard_tail, re.IGNORECASE)
expected = standard_unit.group(1) if standard_unit and standard_unit.group(1) else None
if not expected:
return None
def normalize(unit: str) -> str:
return unit.lower().replace(".", "·").replace("°c", "℃")
if normalize(measured_unit) != normalize(expected):
return f"检测值单位 {measured_unit} 与标准单位 {expected} 不一致"
return None
@staticmethod
def _numeric_appears(value: str, text: str) -> bool:
number = re.search(r"[+\-]?\d+(?:\.\d+)?", value)
return bool(number and number.group() in text)
@staticmethod
def _normalize(value: str) -> str:
return re.sub(r"[^a-z0-9\u4e00-\u9fffΔδ]", "", value.lower())
@staticmethod
def _bigrams(value: str) -> list[str]:
return [value[index : index + 2] for index in range(max(1, len(value) - 1))]

130
app/services/purchase_orders.py
View File

@ -1,130 +0,0 @@
"""Procurement-order ledger import and supplier cooperation aggregation."""
from __future__ import annotations
import hashlib
import re
from datetime import date, datetime
from pathlib import Path
from typing import Any
from openpyxl import load_workbook
from sqlalchemy import func, select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.asyncio import AsyncSession
from app.models.entities import PurchaseOrderLine
def clean_text(value: Any) -> str | None:
if value is None:
return None
text = str(value).replace("\u200b", "").strip()
return text or None
def parse_date(value: Any) -> date | None:
if isinstance(value, datetime):
return value.date()
if isinstance(value, date):
return value
text = clean_text(value)
if not text:
return None
text = re.sub(r"\s+.*$", "", text).replace("/", "-").replace("年", "-").replace("月", "-").replace("日", "")
for fmt in ("%Y-%m-%d", "%Y.%m.%d"):
try:
return datetime.strptime(text, fmt).date()
except ValueError:
continue
return None
def parse_datetime(value: Any) -> datetime | None:
if isinstance(value, datetime):
return value
text = clean_text(value)
if not text:
return None
for fmt in ("%Y/%m/%d %H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y/%m/%d", "%Y-%m-%d"):
try:
return datetime.strptime(text, fmt)
except ValueError:
continue
return None
def fingerprint(file_name: str, row_number: int) -> str:
return hashlib.sha256(f"{file_name}\x1fSheet1\x1f{row_number}".encode()).hexdigest()
def read_purchase_order_rows(path: Path) -> list[dict[str, Any]]:
workbook = load_workbook(path, read_only=True, data_only=True)
sheet = workbook.active
headers = [clean_text(value) for value in next(sheet.iter_rows(values_only=True))]
required = ["单据编号", "采购日期", "供应商"]
missing = [name for name in required if name not in headers]
if missing:
raise ValueError(f"采购订单缺少必要字段：{', '.join(missing)}")
indexes = {name: headers.index(name) for name in headers if name}
def value(row: tuple[Any, ...], name: str) -> Any:
index = indexes.get(name)
return row[index] if index is not None and index < len(row) else None
rows: list[dict[str, Any]] = []
for row_number, raw in enumerate(sheet.iter_rows(min_row=2, values_only=True), start=2):
document = clean_text(value(raw, "单据编号"))
supplier = clean_text(value(raw, "供应商"))
if not document or not supplier:
continue
rows.append(
{
"document_number": document,
"purchase_date": parse_date(value(raw, "采购日期")),
"supplier": supplier,
"document_status": clean_text(value(raw, "单据状态")),
"close_status": clean_text(value(raw, "关闭状态")),
"material_code": clean_text(value(raw, "物料编码")),
"material_name": clean_text(value(raw, "物料名称")),
"purchase_unit": clean_text(value(raw, "采购单位")),
"purchase_quantity": value(raw, "采购数量"),
"delivery_date": parse_datetime(value(raw, "交货日期")),
"tax_inclusive_unit_price": value(raw, "含税单价"),
"is_gift": clean_text(value(raw, "是否赠品")),
"business_close": clean_text(value(raw, "业务关闭")),
"remark": clean_text(value(raw, "备注")),
"source_file": path.name,
"import_fingerprint": fingerprint(path.name, row_number),
}
)
return rows
async def import_purchase_orders(session: AsyncSession, path: Path) -> dict[str, int]:
rows = read_purchase_order_rows(path)
inserted = 0
for start in range(0, len(rows), 500):
batch = rows[start : start + 500]
result = await session.execute(
insert(PurchaseOrderLine).values(batch).on_conflict_do_nothing(index_elements=["import_fingerprint"])
)
inserted += result.rowcount or 0
await session.flush()
return {"source_rows": len(rows), "inserted_rows": inserted, "skipped_rows": len(rows) - inserted}
async def supplier_order_counts(
session: AsyncSession,
supplier: str,
start: date,
end: date,
) -> int:
value = await session.scalar(
select(func.count(func.distinct(PurchaseOrderLine.document_number))).where(
PurchaseOrderLine.supplier.ilike(f"%{supplier}%"),
PurchaseOrderLine.purchase_date >= start,
PurchaseOrderLine.purchase_date <= end,
)
)
return int(value or 0)

377
app/services/reports.py
View File

@ -1,377 +0,0 @@
"""PDF rendering for the prototype-aligned risk report."""
from io import BytesIO
from pathlib import Path
from docx import Document
from docx.enum.table import WD_CELL_VERTICAL_ALIGNMENT
from docx.enum.text import WD_ALIGN_PARAGRAPH
from docx.oxml import OxmlElement
from docx.shared import Pt, RGBColor
from reportlab.lib import colors
from reportlab.lib.enums import TA_CENTER
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
from reportlab.lib.units import mm
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
from reportlab.platypus import Paragraph, SimpleDocTemplate, Table, TableStyle
from app.schemas.risk_prediction import RiskAssessmentResult
LEVEL_LABELS = {
"low": "低风险",
"moderately_low": "较低风险",
"medium": "中等风险",
"high": "高风险",
"very_high": "极高风险",
}
def _font_name() -> str:
candidates = [
Path("C:/Windows/Fonts/msyh.ttc"),
Path("C:/Windows/Fonts/msyh.ttf"),
Path("C:/Windows/Fonts/simhei.ttf"),
]
for path in candidates:
if path.exists():
name = "DieyaCjk"
pdfmetrics.registerFont(TTFont(name, str(path), subfontIndex=0))
return name
return "Helvetica"
def build_risk_report_pdf(result: RiskAssessmentResult) -> bytes:
font = _font_name()
styles = getSampleStyleSheet()
title = ParagraphStyle(
"title", parent=styles["Title"], fontName=font, fontSize=20, alignment=TA_CENTER, spaceAfter=4
)
subtitle = ParagraphStyle(
"subtitle",
parent=styles["Normal"],
fontName=font,
fontSize=9,
textColor=colors.HexColor("#64748b"),
alignment=TA_CENTER,
spaceAfter=14,
)
section = ParagraphStyle(
"section",
parent=styles["Heading2"],
fontName=font,
fontSize=11,
textColor=colors.HexColor("#111827"),
leftIndent=5,
borderPadding=3,
spaceBefore=12,
spaceAfter=7,
)
body = ParagraphStyle("body", parent=styles["BodyText"], fontName=font, fontSize=9, leading=15)
small = ParagraphStyle("small", parent=body, fontSize=8, textColor=colors.HexColor("#64748b"))
supplier_name = result.supplier_freq.supplier_name if result.supplier_freq else None
report_number = f"RP-{result.subject_id[:8]}-{round(result.score)}"
generated_at = __import__("datetime").datetime.now().strftime("%Y-%m-%d %H:%M")
story = [
Paragraph("新品风险预估报告", title),
Paragraph(f"蝶雅产品部 · 风险预估系统 · 自动生成 · {generated_at}", subtitle),
Paragraph(
f"产品/评估对象：{result.subject_id}　　供应商：{supplier_name or '未指定'}　　报告编号：{report_number}",
body,
),
]
def table(rows: list[list[str]], widths: list[float] | None = None) -> Table:
converted = [[Paragraph(str(cell), body) for cell in row] for row in rows]
item = Table(converted, colWidths=widths, repeatRows=1)
item.setStyle(
TableStyle(
[
("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#111827")),
("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
("FONTNAME", (0, 0), (-1, -1), font),
("GRID", (0, 0), (-1, -1), 0.45, colors.HexColor("#dbe3ef")),
("VALIGN", (0, 0), (-1, -1), "TOP"),
("LEFTPADDING", (0, 0), (-1, -1), 7),
("RIGHTPADDING", (0, 0), (-1, -1), 7),
("TOPPADDING", (0, 0), (-1, -1), 6),
("BOTTOMPADDING", (0, 0), (-1, -1), 6),
]
)
)
return item
story += [Paragraph("一、风险结论", section)]
rows = [
["评估项", "结论"],
[
"综合评分",
f"{round(result.score)} / 100 · "
f"{LEVEL_LABELS.get(result.level.value, result.level.value)} · "
f"{result.risk_badge or '需关注'}",
],
["风险警示", result.alert_text or "当前暂无足够历史数据，建议按标准库完成基础验证。"],
]
dimensions = result.dimension_scores or [
{"dimension": factor.name, "score": 100 - factor.value, "weight": factor.weight, "reason": None}
for factor in result.factors
]
for item in dimensions:
name = item.dimension if hasattr(item, "dimension") else item["dimension"]
if name in {"节点阶段", "缺陷等级"}:
continue
score = item.score if hasattr(item, "score") else item["score"]
weight = item.weight if hasattr(item, "weight") else item["weight"]
reason = item.reason if hasattr(item, "reason") else item.get("reason")
supplier = result.supplier_freq
if name == "供应商表现" and supplier:
order_text = f"合作 {supplier.cooperation_count} 次、" if supplier.cooperation_count is not None else ""
conclusion = f"{round(score)} 分 · {order_text}异常 {supplier.anomaly_count or 0} 次"
elif name == "标准符合性" and result.standard_judgment:
conclusion = f"{round(score)} 分 · 命中 {len(result.standard_judgment)} 条判定标准，待检测项需人工确认"
else:
conclusion = f"{round(score)} 分" + (f" · {reason}" if reason else "")
rows.append([f"{name}（{weight:g}%）", conclusion])
story += [table(rows, [55 * mm, 105 * mm])]
if result.category_warning:
story += [
Paragraph("二、品类风险预警", section),
table(
[
["品类", "历史异常", "风险等级", "建议"],
[
result.category_warning.category,
f"{result.category_warning.anomaly_count} 条",
result.category_warning.risk_level,
result.category_warning.suggestion or "结合相似品类和标准库预判",
],
],
[35 * mm, 30 * mm, 30 * mm, 65 * mm],
),
]
supplier = result.supplier_freq
if supplier and supplier.supplier_name:
supplier_rows = [["维度", "数值", "说明"]]
if supplier.cooperation_count is not None:
supplier_rows.append(["合作次数", f"{supplier.cooperation_count} 次", "近三年采购订单去重单据数"])
supplier_rows.append(["异常次数", f"{supplier.anomaly_count or 0} 次", "异常案例库统计"])
if supplier.alpha is not None:
supplier_rows.append(
["频次风险系数", f"×{supplier.alpha:.2f}", "α = 1.0 +（采购订单数 / 100）× 0.3，最高1.5"]
)
if supplier.tier:
supplier_rows.append(["供应商层级", supplier.tier, supplier.suggestion or ""])
story += [Paragraph("三、供应商合作频次分析", section), table(supplier_rows, [38 * mm, 35 * mm, 87 * mm])]
story += [Paragraph(f"四、相似案例匹配（{len(result.similar_cases)} 条）", section)]
if result.similar_cases:
story.append(
table(
[["日期", "产品", "问题", "处理方案"]]
+ [
[case.date or "未知", case.product or "未命名", case.problem or "未填写", case.solution or "未填写"]
for case in result.similar_cases
],
[28 * mm, 42 * mm, 48 * mm, 42 * mm],
)
)
else:
story.append(Paragraph("暂无相关案例，RAG 相似案例检索未返回匹配结果。", small))
story += [Paragraph("五、预防建议", section)]
for index, suggestion in enumerate(
result.ai_suggestions or ["优先针对低分维度安排专项验证，并明确责任人与完成节点。"], 1
):
story.append(Paragraph(f"{index}. {suggestion}", body))
story += [
Paragraph("六、AI 推理方案（无历史案例场景）", section),
Paragraph(
"本次评估已采用历史案例与标准库双引擎结论；若后续未匹配到足够历史案例，将基于相似品类、相似物料生成类比方案，并由标准库进行规则兜底。",
body,
),
]
story += [Paragraph("七、标准判定结果", section)]
if result.standard_judgment:
story.append(
table(
[["判定维度", "判定标准", "标准来源", "当前状态"]]
+ [
[f"{item.category} · {item.risk_point}", item.rule, item.source or "标准文件.xlsx", item.status]
for item in result.standard_judgment
],
[38 * mm, 55 * mm, 40 * mm, 27 * mm],
)
)
else:
story.append(Paragraph("暂无相关标准。", small))
story.append(Paragraph("双引擎融合：历史异常经验引擎提供风险优先级，标准库提供可执行的检测阈值与验证任务。", small))
output = BytesIO()
doc = SimpleDocTemplate(
output,
pagesize=A4,
rightMargin=18 * mm,
leftMargin=18 * mm,
topMargin=16 * mm,
bottomMargin=16 * mm,
title="蝶雅新品风险预估报告",
)
doc.build(story)
return output.getvalue()
def build_risk_report_docx(result: RiskAssessmentResult) -> bytes:
document = Document()
section = document.sections[0]
section.page_width = 210 * 36000
section.page_height = 297 * 36000
section.top_margin = 16 * 36000
section.bottom_margin = 16 * 36000
section.left_margin = 18 * 36000
section.right_margin = 18 * 36000
normal = document.styles["Normal"]
normal.font.name = "微软雅黑"
normal.font.size = Pt(9)
title = document.add_paragraph()
title.alignment = WD_ALIGN_PARAGRAPH.CENTER
run = title.add_run("新品风险预估报告")
run.bold = True
run.font.size = Pt(20)
subtitle = document.add_paragraph()
subtitle.alignment = WD_ALIGN_PARAGRAPH.CENTER
subtitle.add_run(f"蝶雅产品部 · 风险预估系统 · 自动生成 · {result.subject_id}").font.size = Pt(9)
def heading(text: str) -> None:
p = document.add_paragraph()
p.paragraph_format.space_before = Pt(12)
p.paragraph_format.space_after = Pt(6)
p.paragraph_format.left_indent = Pt(5)
r = p.add_run(text)
r.bold = True
r.font.size = Pt(11)
def add_table(rows: list[list[str]]) -> None:
table = document.add_table(rows=len(rows), cols=len(rows[0]))
table.style = "Table Grid"
table.autofit = True
for row_index, row in enumerate(rows):
for col_index, value in enumerate(row):
cell = table.cell(row_index, col_index)
cell.vertical_alignment = WD_CELL_VERTICAL_ALIGNMENT.TOP
cell.text = str(value)
for paragraph in cell.paragraphs:
paragraph.paragraph_format.space_after = Pt(0)
for run in paragraph.runs:
run.font.name = "微软雅黑"
run.font.size = Pt(9)
if row_index == 0:
run.bold = True
run.font.color.rgb = RGBColor(255, 255, 255)
if row_index == 0:
shading = OxmlElement("w:shd")
shading.set("{http://schemas.openxmlformats.org/wordprocessingml/2006/main}fill", "111827")
cell._tc.get_or_add_tcPr().append(shading)
document.add_paragraph()
heading("一、风险结论")
rows = [
["评估项", "结论"],
[
"综合评分",
f"{round(result.score)} / 100 · "
f"{LEVEL_LABELS.get(result.level.value, result.level.value)} · "
f"{result.risk_badge or '需关注'}",
],
["风险警示", result.alert_text or "当前暂无足够历史数据，建议按标准库完成基础验证。"],
]
dimensions = result.dimension_scores or [
{
"dimension": factor.name,
"score": 100 - factor.value,
"weight": factor.weight,
"reason": None,
}
for factor in result.factors
]
for item in dimensions:
name = item.dimension if hasattr(item, "dimension") else item["dimension"]
if name in {"节点阶段", "缺陷等级"}:
continue
score = item.score if hasattr(item, "score") else item["score"]
weight = item.weight if hasattr(item, "weight") else item["weight"]
reason = item.reason if hasattr(item, "reason") else item.get("reason")
supplier = result.supplier_freq
if name == "供应商表现" and supplier:
order_text = f"合作 {supplier.cooperation_count} 次、" if supplier.cooperation_count is not None else ""
conclusion = f"{round(score)} 分 · {order_text}异常 {supplier.anomaly_count or 0} 次"
elif name == "标准符合性" and result.standard_judgment:
conclusion = f"{round(score)} 分 · 命中 {len(result.standard_judgment)} 条判定标准，待检测项需人工确认"
else:
conclusion = f"{round(score)} 分" + (f" · {reason}" if reason else "")
rows.append([f"{name}（{weight:g}%）", conclusion])
add_table(rows)
if result.category_warning:
heading("二、品类风险预警")
add_table(
[
["品类", "历史异常", "风险等级", "建议"],
[
result.category_warning.category,
f"{result.category_warning.anomaly_count} 条",
result.category_warning.risk_level,
result.category_warning.suggestion or "结合相似品类和标准库预判",
],
]
)
if result.supplier_freq and result.supplier_freq.supplier_name:
heading("三、供应商合作频次分析")
rows = [["维度", "数值", "说明"]]
if result.supplier_freq.cooperation_count is not None:
rows.append(["合作次数", f"{result.supplier_freq.cooperation_count} 次", "近三年采购订单去重单据数"])
rows.append(["异常次数", f"{result.supplier_freq.anomaly_count or 0} 次", "异常案例库统计"])
if result.supplier_freq.alpha is not None:
rows.append(
["频次风险系数", f"×{result.supplier_freq.alpha:.2f}", "α = 1.0 +（采购订单数 / 100）× 0.3，最高1.5"]
)
if result.supplier_freq.tier:
rows.append(["供应商层级", result.supplier_freq.tier, result.supplier_freq.suggestion or ""])
add_table(rows)
heading(f"四、相似案例匹配（{len(result.similar_cases)} 条）")
if result.similar_cases:
add_table(
[["日期", "产品", "问题", "处理方案"]]
+ [
[case.date or "未知", case.product or "未命名", case.problem or "未填写", case.solution or "未填写"]
for case in result.similar_cases
]
)
else:
document.add_paragraph("暂无相关案例，RAG 相似案例检索未返回匹配结果。")
heading("五、预防建议")
for index, suggestion in enumerate(
result.ai_suggestions or ["优先针对低分维度安排专项验证，并明确责任人与完成节点。"], 1
):
document.add_paragraph(f"{index}. {suggestion}")
heading("六、AI 推理方案（无历史案例场景）")
document.add_paragraph(
"本次评估已采用历史案例与标准库双引擎结论；若后续未匹配到足够历史案例，将基于相似品类、相似物料生成类比方案，并由标准库进行规则兜底。"
)
heading("七、标准判定结果")
if result.standard_judgment:
add_table(
[["判定维度", "判定标准", "标准来源", "当前状态"]]
+ [
[f"{item.category} · {item.risk_point}", item.rule, item.source or "标准文件.xlsx", item.status]
for item in result.standard_judgment
]
)
else:
document.add_paragraph("暂无相关标准。")
document.add_paragraph("双引擎融合：历史异常经验引擎提供风险优先级，标准库提供可执行的检测阈值与验证任务。")
output = BytesIO()
document.save(output)
return output.getvalue()

11
app/services/risk_items.py
View File

@ -19,10 +19,8 @@ class RiskItemService:
item = RiskItem(
source_type=payload.source_type,
source_file_id=payload.source_file_id,
occurred_on=payload.occurred_on,
product_name=payload.product_name,
product_category=payload.product_category,
material_type=payload.material_type,
supplier=payload.supplier,
factory=payload.factory,
stage_new=payload.stage_new,
@ -30,8 +28,6 @@ class RiskItemService:
defect_level=payload.defect_level,
solution=payload.solution,
photos=payload.photos,
recognition_confidence=payload.recognition_confidence,
ocr_payload=payload.ocr_payload,
submitted_by=user.emp_no,
submitted_at=datetime.now(UTC),
status="待审核" if pre_result == "通过" else "待预审",
@ -115,7 +111,9 @@ class RiskItemService:
raise AppException("审核人不能审核自己提交的风险项", code=403, status_code=403)
record = await self.session.scalar(
select(ReviewRecord).where(ReviewRecord.risk_item_id == item_id).order_by(ReviewRecord.id.desc())
select(ReviewRecord)
.where(ReviewRecord.risk_item_id == item_id)
.order_by(ReviewRecord.id.desc())
)
if record is None:
raise AppException("审核记录不存在", code=409, status_code=409)
@ -126,9 +124,7 @@ class RiskItemService:
if action == "通过":
case = AnomalyCase(
product_name=item.product_name,
date=item.occurred_on,
product_category=item.product_category,
material_type=item.material_type,
supplier=item.supplier,
factory=item.factory,
stage_new=item.stage_new,
@ -136,7 +132,6 @@ class RiskItemService:
defect_level=item.defect_level,
solution=item.solution,
source_type=item.source_type,
source_file=item.source_file_id,
import_fingerprint=f"risk-item-{item.id}",
)
self.session.add(case)

217
app/services/suppliers.py
View File

@ -1,217 +0,0 @@
"""Live supplier profile aggregation from the approved anomaly case data."""
from __future__ import annotations
from collections import Counter, defaultdict
from datetime import UTC, date, datetime, timedelta
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from app.core.exceptions import AppException
from app.models.entities import AnomalyCase
from app.schemas.suppliers import (
SupplierCategoryBreakdown,
SupplierListData,
SupplierMetric,
SupplierOption,
SupplierProfileData,
SupplierRecentRecord,
)
from app.services.database_risk_scoring import DatabaseRiskScoringService, normalize_category
from app.services.purchase_orders import supplier_order_counts
class SupplierProfileService:
history_days = 365 * 3
recent_months = 6
max_categories = 5
max_recent_records = 10
def __init__(self, session: AsyncSession, *, as_of: date | None = None) -> None:
self.session = session
self.as_of = as_of or datetime.now(UTC).date()
async def list(self, *, keyword: str | None = None, limit: int = 100) -> SupplierListData:
rows = await self._load_rows()
grouped = self._group_suppliers(rows, keyword)
items = [
SupplierOption(
name=name,
anomaly_count=len(items),
tier=self.tier_for(
len(items),
sum(self.defect_code(row.get("defect_level")) == "A" for row in items),
sum(self.defect_code(row.get("defect_level")) == "B" for row in items),
),
)
for name, items in sorted(grouped.items(), key=lambda item: (-len(item[1]), item[0]))[:limit]
]
return SupplierListData(
items=items,
default_supplier=items[0].name if items else None,
)
async def detail(self, supplier_name: str) -> SupplierProfileData:
rows = [row for row in await self._load_rows() if self.matches(row.get("supplier"), supplier_name)]
if not rows:
raise AppException("供应商不存在或暂无近3年异常记录", code=404, status_code=404)
anomaly_total = len(rows)
order_count = await supplier_order_counts(
self.session,
supplier_name,
self.as_of - timedelta(days=self.history_days),
self.as_of,
)
a_count = sum(self.defect_code(row.get("defect_level")) == "A" for row in rows)
b_count = sum(self.defect_code(row.get("defect_level")) == "B" for row in rows)
tier = self.tier_for(anomaly_total, a_count, b_count)
categories = self._categories(rows)
scoring = DatabaseRiskScoringService(self.session, as_of=self.as_of)
score = await scoring.score(supplier=supplier_name)
return SupplierProfileData(
supplier_name=supplier_name,
tier=tier,
tier_condition=self.tier_condition(tier),
metric=SupplierMetric(
cooperation_total=order_count,
anomaly_total=anomaly_total,
anomaly_rate=None,
frequency_alpha=self.alpha(
order_count,
coefficient=scoring.frequency_coefficient,
cap=scoring.frequency_cap,
),
risk_score=score.score,
main_category_count=len(categories),
),
category_breakdown=categories[: self.max_categories],
monthly_trend=self._monthly_trend(rows),
recent_records=self._recent_records(rows),
factory_recommendations=[],
recommendation_note="合作次数按近3年采购订单去重单据编号统计；异常数和供应商层级仍按异常案例库统计。",
)
async def _load_rows(self) -> list[dict]:
start = self.as_of - timedelta(days=self.history_days)
statement = select(
AnomalyCase.id,
AnomalyCase.date,
AnomalyCase.product_name,
AnomalyCase.product_category,
AnomalyCase.material_type,
AnomalyCase.supplier,
AnomalyCase.factory,
AnomalyCase.anomaly_type,
AnomalyCase.defect_level,
).where(
AnomalyCase.date.is_not(None),
AnomalyCase.date >= start,
AnomalyCase.date <= self.as_of,
)
return [dict(row) for row in (await self.session.execute(statement)).mappings().all()]
@staticmethod
def matches(value: str | None, target: str) -> bool:
return bool(value and target.strip().lower() in value.lower())
@classmethod
def _group_suppliers(cls, rows: list[dict], keyword: str | None) -> dict[str, list[dict]]:
grouped: dict[str, list[dict]] = defaultdict(list)
for row in rows:
name = (row.get("supplier") or "").strip()
if name and (not keyword or cls.matches(name, keyword)):
grouped[name].append(row)
return grouped
@staticmethod
def defect_code(value: str | None) -> str | None:
value = (value or "").strip().upper()
return value if value in {"A", "B", "C"} else None
@staticmethod
def alpha(anomaly_count: int, *, coefficient: float = 0.3, cap: float = 1.5) -> float:
return round(min(cap, 1 + anomaly_count / 100 * coefficient), 2)
@classmethod
def tier_for(cls, anomaly_count: int, a_count: int, b_count: int) -> str:
if anomaly_count == 0:
return "新供应商"
ab_ratio = (a_count + b_count) / anomaly_count
if anomaly_count > 120 or a_count > 0 or ab_ratio > 0.5:
return "C级"
if anomaly_count > 60 or ab_ratio >= 0.3:
return "B级"
if anomaly_count >= 20:
return "A级"
return "S级"
@staticmethod
def tier_condition(tier: str) -> str:
return {
"S级": "异常记录 < 20 次，且无 A 级缺陷",
"A级": "异常记录 20~60 次，且 A/B 级缺陷占比 < 30%",
"B级": "异常记录 60~120 次，或 A/B 级缺陷占比 30%~50%",
"C级": "异常记录 > 120 次，或存在 A 级缺陷，或 A/B 级缺陷占比 > 50%",
"新供应商": "近3年异常记录为 0",
}[tier]
@classmethod
def _categories(cls, rows: list[dict]) -> list[SupplierCategoryBreakdown]:
grouped: dict[str, list[dict]] = defaultdict(list)
for row in rows:
category = normalize_category(row.get("product_category")) or normalize_category(row.get("product_name"))
grouped[category or "待补标品类"].append(row)
result = []
for category, items in grouped.items():
materials = Counter(row.get("material_type") for row in items if row.get("material_type"))
problems = Counter(row.get("anomaly_type") for row in items if row.get("anomaly_type"))
result.append(
SupplierCategoryBreakdown(
category=category,
anomaly_count=len(items),
anomaly_rate=None,
top_material=materials.most_common(1)[0][0] if materials else None,
top_problem=problems.most_common(1)[0][0] if problems else None,
)
)
return sorted(result, key=lambda item: (-item.anomaly_count, item.category))
def _monthly_trend(self, rows: list[dict]) -> list[dict[str, int | str]]:
start = self._month_offset(self.as_of.replace(day=1), -5)
counts = Counter(
row["date"].strftime("%Y-%m")
for row in rows
if row.get("date") and row["date"] >= start
)
output = []
cursor = start
for _ in range(self.recent_months):
key = cursor.strftime("%Y-%m")
output.append({"month": key, "count": counts.get(key, 0)})
cursor = date(cursor.year + (cursor.month == 12), 1 if cursor.month == 12 else cursor.month + 1, 1)
return output
@staticmethod
def _month_offset(value: date, months: int) -> date:
index = value.year * 12 + value.month - 1 + months
return date(index // 12, index % 12 + 1, 1)
def _recent_records(self, rows: list[dict]) -> list[SupplierRecentRecord]:
start = self._month_offset(self.as_of.replace(day=1), -5)
recent = sorted(
(row for row in rows if row.get("date") and row["date"] >= start),
key=lambda row: row["date"],
reverse=True,
)
return [
SupplierRecentRecord(
case_id=int(row["id"]),
month=row["date"].strftime("%Y-%m") if row.get("date") else None,
product=row.get("product_name"),
anomaly=row.get("anomaly_type"),
defect_level=self.defect_code(row.get("defect_level")),
)
for row in recent[: self.max_recent_records]
]

39
migrations/versions/20260907_0015_add_supplier_read_permissions.py
View File

@ -1,39 +0,0 @@
"""Align full supplier-profile access with the PRD's L2 role matrix."""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "20260907_0015"
down_revision: str | None = "20260905_0014"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
PERMISSIONS = ("procurement", "reviewer", "manager")
def upgrade() -> None:
connection = op.get_bind()
statement = sa.text(
"""
INSERT INTO role_permissions (role, func_key, data_level, can_view, can_operate)
VALUES (:role, 'supplier.read', 'L2', true, false)
ON CONFLICT (role, func_key) DO UPDATE SET
data_level = EXCLUDED.data_level,
can_view = EXCLUDED.can_view,
can_operate = EXCLUDED.can_operate,
updated_at = now()
"""
)
for role in PERMISSIONS:
connection.execute(statement, {"role": role})
def downgrade() -> None:
connection = op.get_bind()
for role in PERMISSIONS:
connection.execute(
sa.text("DELETE FROM role_permissions WHERE role = :role AND func_key = 'supplier.read'"),
{"role": role},
)

37
migrations/versions/20260907_0016_add_ocr_fields.py
View File

@ -1,37 +0,0 @@
"""Add OCR extraction and judgment fields to risk items."""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects import postgresql
revision: str = "20260907_0016"
down_revision: str | None = "20260907_0015"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.add_column("risk_items", sa.Column("occurred_on", sa.Date(), nullable=True))
op.add_column("risk_items", sa.Column("material_type", sa.String(length=200), nullable=True))
op.add_column("risk_items", sa.Column("recognition_confidence", sa.Numeric(5, 4), nullable=True))
op.add_column(
"risk_items",
sa.Column("ocr_payload", postgresql.JSONB(astext_type=sa.Text()), server_default="{}", nullable=False),
)
op.create_index("ix_risk_items_occurred_on", "risk_items", ["occurred_on"])
op.create_index("ix_risk_items_material_type", "risk_items", ["material_type"])
op.execute("COMMENT ON COLUMN risk_items.occurred_on IS 'OCR或人工录入的异常发生日期。'")
op.execute("COMMENT ON COLUMN risk_items.material_type IS 'OCR或人工录入的异常物料类型。'")
op.execute("COMMENT ON COLUMN risk_items.recognition_confidence IS 'OCR综合识别置信度，范围0到1。'")
op.execute("COMMENT ON COLUMN risk_items.ocr_payload IS 'OCR检测数值、标准判定和人工确认前的结构化结果。'")
def downgrade() -> None:
op.drop_index("ix_risk_items_material_type", table_name="risk_items")
op.drop_index("ix_risk_items_occurred_on", table_name="risk_items")
op.drop_column("risk_items", "ocr_payload")
op.drop_column("risk_items", "recognition_confidence")
op.drop_column("risk_items", "material_type")
op.drop_column("risk_items", "occurred_on")

103
migrations/versions/20260907_0017_four_role_model_config.py
View File

@ -1,103 +0,0 @@
"""Apply the four-role matrix and add configurable model weights."""
import json
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
from app.domain.model_config import DEFAULT_WEIGHTS
from sqlalchemy.dialects import postgresql
revision: str = "20260907_0017"
down_revision: str | None = "20260907_0016"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
PERMISSIONS = (
("普通用户", "risk.assess", "L1", True, True),
("普通用户", "standard.read", "L1", True, False),
("普通用户", "supplier.read", "L1", True, False),
("product_manager", "risk.assess", "L1", True, True),
("product_manager", "risk_item.submit", "L1", True, True),
("product_manager", "risk_item.review", "L2", True, True),
("product_manager", "standard.read", "L1", True, False),
("product_manager", "supplier.read", "L2", True, False),
("reviewer", "risk.assess", "L1", True, True),
("reviewer", "risk_item.review", "L2", True, True),
("reviewer", "standard.read", "L1", True, False),
("reviewer", "supplier.read", "L2", True, False),
("admin", "risk.assess", "L3", True, True),
("admin", "risk_item.submit", "L3", True, True),
("admin", "risk_item.review", "L3", True, True),
("admin", "model_weight.manage", "L3", True, True),
("admin", "rbac.manage", "L3", True, True),
("admin", "standard.read", "L3", True, True),
("admin", "standard.manage", "L3", True, True),
("admin", "supplier.read", "L3", True, True),
)
def upgrade() -> None:
op.create_table(
"risk_model_configs",
sa.Column("id", sa.BigInteger(), nullable=False),
sa.Column("weights", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
sa.Column("frequency_coefficient", sa.Numeric(5, 4), nullable=False),
sa.Column("frequency_cap", sa.Numeric(4, 2), nullable=False),
sa.Column("updated_by", sa.String(length=20), nullable=True),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
connection = op.get_bind()
connection.execute(
sa.text(
"""
INSERT INTO risk_model_configs (id, weights, frequency_coefficient, frequency_cap)
VALUES (1, CAST(:weights AS jsonb), 0.3, 1.5)
"""
),
{"weights": json.dumps(DEFAULT_WEIGHTS, ensure_ascii=False)},
)
connection.execute(
sa.text(
"""
UPDATE user_accounts
SET role = CASE
WHEN role IN ('dev', 'manager') THEN 'product_manager'
WHEN role IN ('qc', 'reviewer') THEN 'reviewer'
WHEN role IN ('procurement', 'ops', '普通用户') THEN '普通用户'
ELSE role
END,
updated_at = now()
"""
)
)
connection.execute(sa.text("DELETE FROM role_permissions"))
statement = sa.text(
"""
INSERT INTO role_permissions (role, func_key, data_level, can_view, can_operate)
VALUES (:role, :func_key, :data_level, :can_view, :can_operate)
"""
)
for role, func_key, data_level, can_view, can_operate in PERMISSIONS:
connection.execute(
statement,
{
"role": role,
"func_key": func_key,
"data_level": data_level,
"can_view": can_view,
"can_operate": can_operate,
},
)
op.execute("COMMENT ON TABLE risk_model_configs IS '管理员维护的七维权重与频次系数配置。'")
op.execute("COMMENT ON COLUMN user_accounts.role IS '角色：admin、product_manager、reviewer 或 普通用户。'")
def downgrade() -> None:
connection = op.get_bind()
connection.execute(
sa.text("UPDATE user_accounts SET role = 'manager', updated_at = now() WHERE role = 'product_manager'")
)
connection.execute(sa.text("DELETE FROM role_permissions"))
op.drop_table("risk_model_configs")

48
migrations/versions/20260908_0018_purchase_order_ledger.py
View File

@ -1,48 +0,0 @@
"""Create the procurement-order ledger used for real supplier cooperation counts."""
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "20260908_0018"
down_revision: str | None = "20260907_0017"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.create_table(
"purchase_order_lines",
sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
sa.Column("document_number", sa.String(length=64), nullable=False),
sa.Column("purchase_date", sa.Date(), nullable=True),
sa.Column("supplier", sa.String(length=300), nullable=False),
sa.Column("document_status", sa.String(length=30), nullable=True),
sa.Column("close_status", sa.String(length=30), nullable=True),
sa.Column("material_code", sa.String(length=200), nullable=True),
sa.Column("material_name", sa.Text(), nullable=True),
sa.Column("purchase_unit", sa.String(length=30), nullable=True),
sa.Column("purchase_quantity", sa.Numeric(18, 4), nullable=True),
sa.Column("delivery_date", sa.DateTime(timezone=True), nullable=True),
sa.Column("tax_inclusive_unit_price", sa.Numeric(18, 6), nullable=True),
sa.Column("is_gift", sa.String(length=10), nullable=True),
sa.Column("business_close", sa.String(length=30), nullable=True),
sa.Column("remark", sa.Text(), nullable=True),
sa.Column("source_file", sa.String(length=300), nullable=False),
sa.Column("import_fingerprint", sa.String(length=64), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False),
sa.PrimaryKeyConstraint("id"),
sa.UniqueConstraint("import_fingerprint"),
)
op.create_index("ix_purchase_order_lines_supplier_date", "purchase_order_lines", ["supplier", "purchase_date"])
op.create_index("ix_purchase_order_lines_document", "purchase_order_lines", ["document_number"])
op.create_index("uq_purchase_order_lines_source", "purchase_order_lines", ["import_fingerprint"], unique=True)
op.execute("COMMENT ON TABLE purchase_order_lines IS '采购订单明细台账；供应商合作频次按单据编号去重统计。'")
def downgrade() -> None:
op.drop_index("uq_purchase_order_lines_source", table_name="purchase_order_lines")
op.drop_index("ix_purchase_order_lines_document", table_name="purchase_order_lines")
op.drop_index("ix_purchase_order_lines_supplier_date", table_name="purchase_order_lines")
op.drop_table("purchase_order_lines")

3
pyproject.toml
View File

@ -14,9 +14,6 @@ dependencies = [
"fastapi>=0.115,<1.0",
"httpx>=0.27,<1.0",
"openpyxl>=3.1,<4.0",
"python-multipart>=0.0.9,<1.0",
"reportlab>=4.0,<5.0",
"python-docx>=1.1,<2.0",
"PyJWT>=2.8,<3.0",
"pydantic>=2.8,<3.0",
"pydantic-settings>=2.4,<3.0",

3
requirements.txt
View File

@ -3,9 +3,6 @@ asyncpg>=0.29,<1.0
fastapi>=0.115,<1.0
httpx>=0.27,<1.0
openpyxl>=3.1,<4.0
python-multipart>=0.0.9,<1.0
reportlab>=4.0,<5.0
python-docx>=1.1,<2.0
PyJWT>=2.8,<3.0
pydantic>=2.8,<3.0
pydantic-settings>=2.4,<3.0

32
scripts/import_purchase_orders.py
View File

@ -1,32 +0,0 @@
"""Import a procurement-order workbook into the purchase order ledger."""
import argparse
import asyncio
import logging
from pathlib import Path
from app.core.config import get_settings
from app.database.session import SessionFactory
from app.services.purchase_orders import import_purchase_orders
async def run(path: Path) -> None:
async with SessionFactory() as session:
result = await import_purchase_orders(session, path)
await session.commit()
for key, value in result.items():
print(f"{key}={value}")
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--file", type=Path, required=True)
args = parser.parse_args()
if not args.file.is_file():
parser.error(f"file does not exist: {args.file}")
logging.basicConfig(level=get_settings().log_level)
asyncio.run(run(args.file))
if __name__ == "__main__":
main()

6
tests/test_auth.py
View File

@ -2,9 +2,9 @@ from app.services.dingtalk_auth import DingTalkAuthService
def test_dingtalk_role_mapping() -> None:
assert DingTalkAuthService.map_role("产品部", "产品经理") == "product_manager"
assert DingTalkAuthService.map_role("品控部", "质量主管") == "普通用户"
assert DingTalkAuthService.map_role("采购部", "采购专员") == "普通用户"
assert DingTalkAuthService.map_role("产品开发部", "工程师") == "dev"
assert DingTalkAuthService.map_role("品控部", "质量主管") == "qc"
assert DingTalkAuthService.map_role("采购部", "采购专员") == "procurement"
assert DingTalkAuthService.map_role("", "审核员") == "reviewer"
assert DingTalkAuthService.map_role("", "", is_admin=True) == "admin"
assert DingTalkAuthService.map_role("未知部门", "普通员工") == "普通用户"

132
tests/test_baidu_ocr.py
View File

@ -1,132 +0,0 @@
import json
import httpx
import pytest
from app.schemas.ocr import OcrMeasurement
from app.services.baidu_ocr import BaiduOcrClient
from app.services.ocr import OcrService
@pytest.mark.asyncio
async def test_webimage_recognition_returns_text_and_confidence() -> None:
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/token"):
return httpx.Response(200, json={"access_token": "token", "expires_in": 3600})
assert request.url.path.endswith("/webimage")
return httpx.Response(
200,
json={
"words_result": [
{"words": "产品名称：测试面霜", "probability": {"average": 0.92}},
{"words": "pH：5.6", "probability": {"average": 0.88}},
]
},
)
client = BaiduOcrClient(httpx.MockTransport(handler))
client.api_key = "api-key"
client.secret_key = "secret-key"
text, confidence, pages = await client.recognize_image(b"fake-image")
assert text == "产品名称：测试面霜\npH：5.6"
assert confidence == pytest.approx(0.9)
assert pages == 1
@pytest.mark.asyncio
async def test_paddle_document_recognition_polls_and_downloads_result() -> None:
calls = {"query": 0}
def handler(request: httpx.Request) -> httpx.Response:
if request.url.path.endswith("/token"):
return httpx.Response(200, json={"access_token": "token", "expires_in": 3600})
if request.url.path.endswith("/task/query"):
calls["query"] += 1
status = "processing" if calls["query"] == 1 else "success"
result = {"status": status}
if status == "success":
result["parse_result_url"] = "https://mock.bcebos.com/result.json"
return httpx.Response(200, json={"error_code": 0, "result": result})
if request.url.path.endswith("/task"):
return httpx.Response(200, json={"error_code": 0, "result": {"task_id": "task-1"}})
if request.url.path.endswith("/result.json"):
return httpx.Response(
200,
content=json.dumps(
{
"pages": [
{
"text": "产品名称：测试面霜",
"tables": [{"markdown": "|检测项|值|\n|pH|5.6|"}],
}
]
},
ensure_ascii=False,
).encode(),
)
raise AssertionError(f"unexpected URL: {request.url}")
client = BaiduOcrClient(httpx.MockTransport(handler))
client.api_key = "api-key"
client.secret_key = "secret-key"
client.poll_interval = 0.001
text, confidence, pages = await client.recognize_document(b"%PDF-test", "report.pdf")
assert "测试面霜" in text
assert "|pH|5.6|" in text
assert confidence is None
assert pages == 1
assert calls["query"] == 2
@pytest.mark.parametrize(
("raw_value", "value", "standard", "expected"),
[
("5.6", 5.6, "pH 4.5-6.5", "合格"),
("3.8", 3.8, "色差 ΔE≤3.0", "不合格"),
("+18%", 18, "喷出量偏差≤±15%", "不合格"),
("0.9 N·m", 0.9, "开启力矩 0.8~3.2N·m", "合格"),
],
)
def test_standard_numeric_evaluation(raw_value: str, value: float, standard: str, expected: str) -> None:
measurement = OcrMeasurement(
item="检测项",
raw_value=raw_value,
value=value,
confidence=0.95,
needs_review=False,
)
result, reason = OcrService.evaluate_standard(measurement, standard)
assert result == expected
assert reason is None
def test_upload_type_is_detected_from_content_not_filename() -> None:
assert OcrService._detect_type(b"%PDF-1.7") == ("application/pdf", ".pdf")
assert OcrService._detect_type(b"\x89PNG\r\n\x1a\nrest") == ("image/png", ".png")
assert OcrService._detect_type(b"\xff\xd8\xffrest") == ("image/jpeg", ".jpg")
def test_standard_judgment_rejects_unit_mismatch() -> None:
measurement = OcrMeasurement(
item="铅含量",
raw_value="8mg/L",
value=8,
unit="mg/L",
confidence=0.95,
needs_review=False,
)
result, reason = OcrService.evaluate_standard(measurement, "铅≤10mg/kg")
assert result == "待人工确认"
assert reason == "检测值单位 mg/L 与标准单位 mg/kg 不一致"
def test_generic_quantity_does_not_fuzzy_match_a_standard() -> None:
class Rule:
risk_point = "销售包装可视面及最小销售单元完整标注"
judge_standard = "中英文标签内容完整"
assert OcrService._best_rule("数量", [Rule()]) is None

136
tests/test_conversations.py
View File

@ -2,7 +2,6 @@ from datetime import UTC, datetime
import pytest
from app.models.entities import Conversation, ConversationMessage
from app.schemas.risk_prediction import SupplierFrequency
from app.services.conversations import ConversationService
@ -32,32 +31,6 @@ def test_conversation_message_data_reads_persisted_risk_result() -> None:
assert result.risk_result.level.value == "low"
def test_conversation_message_data_reads_persisted_semantic_and_response_card() -> None:
message = ConversationMessage(
id=11,
conversation_id=3,
role="assistant",
content="已完成分析。",
structured_data={
"semantic": {"intent": "solution", "category": "洗发水"},
"responseCard": {
"type": "solution",
"title": "AI 案例推理生成",
"similarCases": [{"case_id": 101, "similarity": 0.82}],
"suggestions": ["完成包材相容性测试"],
},
},
created_at=datetime.now(UTC),
)
result = ConversationService._message(message)
assert result.semantic == {"intent": "solution", "category": "洗发水"}
assert result.response_card is not None
assert result.response_card["similarCases"][0]["case_id"] == 101
assert result.response_card["suggestions"] == ["完成包材相容性测试"]
def test_conversation_data_includes_all_messages() -> None:
conversation = Conversation(
id=3,
@ -171,115 +144,6 @@ def test_unknown_standalone_name_is_treated_as_new_supplier() -> None:
assert ConversationService._response_type("新供应商A", {"intent": "supplier"}) == "general"
@pytest.mark.parametrize(
("count", "a_count", "b_count", "expected"),
[
(0, 0, 0, "新供应商"),
(12, 0, 1, "S级 · 战略"),
(40, 0, 4, "A级 · 优质"),
(70, 0, 10, "B级 · 关注"),
(40, 0, 16, "B级 · 关注"),
(10, 1, 0, "C级 · 高风险"),
(121, 0, 0, "C级 · 高风险"),
],
)
def test_supplier_tier_uses_prd_frequency_and_defect_rules(
count: int,
a_count: int,
b_count: int,
expected: str,
) -> None:
assert ConversationService._supplier_tier(
count,
a_count=a_count,
b_count=b_count,
) == expected
def test_supplier_management_suggestions_follow_prd_actions() -> None:
assert ConversationService._supplier_management_suggestions("B级 · 关注") == [
"加强巡检",
"新品打样加严",
]
assert ConversationService._supplier_management_suggestions("新供应商") == [
"无历史数据",
"首单全检",
"安排驻厂巡检",
]
@pytest.mark.asyncio
async def test_supplier_card_uses_prd_proxy_frequency_and_tier_advice(
monkeypatch: pytest.MonkeyPatch,
) -> None:
service = object.__new__(ConversationService)
async def history_stats(*_args, **_kwargs):
return 70, 0, 10
async def top_anomaly_types(*_args, **_kwargs):
return [{"name": "泵头异常", "count": 12}]
monkeypatch.setattr(service, "_supplier_history_stats", history_stats)
monkeypatch.setattr(service, "_top_anomaly_types", top_anomaly_types)
result = await service._supplier_card(
"祥印供应商风险怎么样",
{"supplier": "祥印"},
)
assert result["anomalyCount"] == 70
assert result["cooperationCount"] is None
assert result["frequencyCoefficient"] == 1.0
assert result["tier"] == "B级 · 关注"
assert result["suggestions"] == ["加强巡检", "新品打样加严"]
assert result["frequencyBasis"] == "暂无匹配采购订单"
assert result["dataAvailable"] is False
def test_new_supplier_frequency_serializes_zero_proxy_count() -> None:
result = SupplierFrequency(
supplier_name="新供应商",
cooperation_count=0,
anomaly_count=0,
tier="新供应商",
suggestion="无历史数据；首单全检；安排驻厂巡检",
note="合作次数为近3年异常记录数代理值，不代表真实采购订单合作次数。",
).model_dump(by_alias=True)
assert result["cooperationCount"] == 0
assert result["anomalyCount"] == 0
assert result["frequencyBasis"] == "暂无匹配采购订单"
def test_unspecified_supplier_returns_displayable_no_data_frequency_card() -> None:
result = SupplierFrequency(
note="未指定供应商，暂无供应商频次分析。",
frequency_basis="未提供供应商",
).model_dump(by_alias=True)
assert result["dataAvailable"] is False
assert result["alpha"] is None
assert result["note"] == "未指定供应商，暂无供应商频次分析。"
def test_similar_cases_are_typed_before_risk_result_serialization() -> None:
from app.schemas.risk_prediction import RiskAssessmentResult, SimilarCase
result = RiskAssessmentResult(
subject_id="test",
score=52,
level="medium",
model_version="test",
factors=[],
similar_cases=[SimilarCase(case_id=1, similarity=0.75)],
)
payload = result.model_dump(mode="json")
assert payload["similar_cases"][0]["case_id"] == 1
@pytest.mark.asyncio
async def test_any_product_without_history_uses_solution_card(monkeypatch: pytest.MonkeyPatch) -> None:
service = object.__new__(ConversationService)

53
tests/test_database_risk_scoring.py
View File

@ -1,11 +1,7 @@
from datetime import date
from decimal import Decimal
from types import SimpleNamespace
import pytest
from app.schemas.model_config import ModelWeightConfigUpdate
from app.services.database_risk_scoring import DatabaseRiskScoringService
from pydantic import ValidationError
def _row(**values):
@ -67,52 +63,3 @@ async def test_supplier_dimension_does_not_count_factory_records() -> None:
supplier_dimension = result.dimensions[0]
assert supplier_dimension.score == 70
assert supplier_dimension.raw_value == 0
@pytest.mark.asyncio
async def test_score_uses_persisted_weight_and_frequency_config() -> None:
configured_weights = {
"供应商表现": 30,
"物料品类": 15,
"产品品类": 15,
"加工厂质量": 15,
"标准符合性": 10,
"节点阶段": 10,
"缺陷等级": 5,
}
class Session:
async def get(self, _model, _identifier):
return SimpleNamespace(
weights=configured_weights,
frequency_coefficient=Decimal("0.5"),
frequency_cap=Decimal("1.8"),
)
service = DatabaseRiskScoringService(Session(), as_of=date(2026, 9, 4))
async def load_rows():
return [_row(), _row(product_name="洗发水 250ml", defect_level="B")]
service._load_rows = load_rows # type: ignore[method-assign]
result = await service.score(category="洗发水", material="瓶子", supplier="祥印", stage="产品开发")
assert [item.weight for item in result.dimensions] == [30, 15, 15, 15, 10, 10, 5]
assert result.dimensions[0].score == 95.95
def test_weight_config_requires_exactly_one_hundred_percent() -> None:
with pytest.raises(ValidationError, match="合计必须为100%"):
ModelWeightConfigUpdate(
weights={
"供应商表现": 20,
"物料品类": 20,
"产品品类": 15,
"加工厂质量": 15,
"标准符合性": 10,
"节点阶段": 10,
"缺陷等级": 5,
},
frequency_coefficient=0.3,
frequency_cap=1.5,
)

3
tests/test_models.py
View File

@ -7,7 +7,6 @@ from app.models import ( # noqa: F401
ReviewRecord,
RiskAssessment,
RiskItem,
RiskModelConfig,
RolePermission,
StandardRule,
SupplierProfile,
@ -25,8 +24,6 @@ def test_prd_core_tables_are_registered() -> None:
"review_records",
"risk_assessments",
"risk_items",
"risk_model_configs",
"purchase_order_lines",
"role_permissions",
"standard_rules",
"supplier_profiles",

6
tests/test_risk_prediction.py
View File

@ -18,12 +18,6 @@ def test_preview_requires_authentication() -> None:
assert response.status_code == 401
def test_supplier_profile_requires_authentication() -> None:
response = TestClient(app).get("/api/v1/suppliers")
assert response.status_code == 401
def test_baseline_engine_calculates_weighted_score() -> None:
score = BaselineRiskEngine().predict(
[

66
tests/test_suppliers.py
View File

@ -1,66 +0,0 @@
from datetime import date
from app.schemas.suppliers import SupplierListData, SupplierOption
from app.services.suppliers import SupplierProfileService
def test_supplier_tier_rules_follow_prd() -> None:
assert SupplierProfileService.tier_for(0, 0, 0) == "新供应商"
assert SupplierProfileService.tier_for(12, 0, 0) == "S级"
assert SupplierProfileService.tier_for(40, 0, 4) == "A级"
assert SupplierProfileService.tier_for(70, 0, 10) == "B级"
assert SupplierProfileService.tier_for(40, 0, 16) == "B级"
assert SupplierProfileService.tier_for(10, 1, 0) == "C级"
assert SupplierProfileService.tier_for(121, 0, 0) == "C级"
def test_supplier_alpha_is_capped_and_rounded() -> None:
assert SupplierProfileService.alpha(0) == 1.0
assert SupplierProfileService.alpha(42) == 1.13
assert SupplierProfileService.alpha(148) == 1.44
assert SupplierProfileService.alpha(200) == 1.5
def test_supplier_list_exposes_default_highest_anomaly_supplier() -> None:
result = SupplierListData(
items=[
SupplierOption(name="甲供应商", anomaly_count=2, tier="S级"),
SupplierOption(name="乙供应商", anomaly_count=1, tier="S级"),
],
default_supplier="甲供应商",
)
assert result.default_supplier == result.items[0].name
def test_supplier_category_breakdown_is_top_five_with_material_and_problem() -> None:
rows = [
{"product_category": "洗发水", "product_name": "洗发水", "material_type": "泵头", "anomaly_type": "漏液"},
{"product_category": "洗发水", "product_name": "洗发水", "material_type": "泵头", "anomaly_type": "漏液"},
{"product_category": "面霜", "product_name": "面霜", "material_type": "瓶子", "anomaly_type": "刮花"},
]
result = SupplierProfileService._categories(rows)
assert result[0].category == "洗发水/洗护类"
assert result[0].top_material == "泵头"
assert result[0].top_problem == "漏液"
def test_supplier_recent_records_are_limited_and_sorted() -> None:
service = SupplierProfileService(object(), as_of=date(2026, 9, 7))
rows = [
{
"id": index,
"date": date(2026, 4, index),
"product_name": "产品",
"anomaly_type": str(index),
"defect_level": "B",
}
for index in range(1, 13)
]
result = service._recent_records(rows)
assert len(result) == 10
assert result[0].case_id == 12