218 lines
8.3 KiB
Python
218 lines
8.3 KiB
Python
"""Assess structured product data without creating conversations or re-parsing fields with an LLM."""
|
||
|
||
import asyncio
|
||
import logging
|
||
from datetime import date
|
||
|
||
from sqlalchemy import or_, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.models.entities import AnomalyCase, StandardRule
|
||
from app.schemas.integration_risk import IntegrationRiskRequest
|
||
from app.schemas.risk_prediction import (
|
||
CategoryWarning,
|
||
RiskAssessmentResult,
|
||
RiskDimensionScore,
|
||
RiskFactorResult,
|
||
StandardJudgment,
|
||
SupplierFrequency,
|
||
)
|
||
from app.services.database_risk_scoring import (
|
||
DatabaseRiskScoringService,
|
||
normalize_category,
|
||
normalize_material,
|
||
)
|
||
from app.services.llm_client import LLMClient
|
||
from app.services.purchase_orders import calendar_year_window, supplier_order_counts
|
||
from app.services.rag import AnomalyCaseRAG
|
||
from app.services.supplier_identity import resolve_supplier
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class ExternalRiskReportService:
|
||
def __init__(self, session: AsyncSession):
|
||
self.session = session
|
||
|
||
async def assess(self, payload: IntegrationRiskRequest, as_of: date) -> RiskAssessmentResult:
|
||
product = payload.assessment
|
||
start, _ = calendar_year_window(as_of, 3)
|
||
# Bound both scoring and case retrieval by the supplied cutoff date.
|
||
fields = (
|
||
AnomalyCase.id,
|
||
AnomalyCase.date,
|
||
AnomalyCase.product_name,
|
||
AnomalyCase.product_category,
|
||
AnomalyCase.material_type,
|
||
AnomalyCase.supplier,
|
||
AnomalyCase.factory,
|
||
AnomalyCase.stage_new,
|
||
AnomalyCase.defect_level,
|
||
AnomalyCase.anomaly_type,
|
||
AnomalyCase.description,
|
||
AnomalyCase.root_cause,
|
||
AnomalyCase.solution,
|
||
AnomalyCase.result,
|
||
)
|
||
rows = [
|
||
dict(row)
|
||
for row in (
|
||
await self.session.execute(
|
||
select(*fields).where(
|
||
AnomalyCase.date >= start,
|
||
AnomalyCase.date <= as_of,
|
||
)
|
||
)
|
||
)
|
||
.mappings()
|
||
.all()
|
||
]
|
||
identity = await resolve_supplier(self.session, product.supplier)
|
||
supplier = identity.canonical_name if identity else product.supplier
|
||
cooperation = await supplier_order_counts(self.session, supplier, start, as_of) if supplier else 0
|
||
score = await DatabaseRiskScoringService(self.session, as_of=as_of).score(
|
||
category=product.product_category,
|
||
material=product.material_type,
|
||
supplier=supplier,
|
||
factory=product.factory,
|
||
stage=product.stage_new,
|
||
defect_level=product.defect_level,
|
||
rows=rows,
|
||
cooperation_count=cooperation or 0,
|
||
)
|
||
query = "，".join(
|
||
value
|
||
for value in (
|
||
product.product_name,
|
||
product.product_category,
|
||
product.material_type,
|
||
supplier,
|
||
product.factory,
|
||
product.stage_new,
|
||
product.description,
|
||
)
|
||
if value
|
||
)
|
||
retrieved = await AnomalyCaseRAG(self.session).search(
|
||
query=query,
|
||
category=product.product_category,
|
||
material=product.material_type,
|
||
supplier=supplier,
|
||
factory=product.factory,
|
||
stage=product.stage_new,
|
||
rows=rows,
|
||
)
|
||
category = normalize_category(product.product_category)
|
||
material = normalize_material(product.material_type)
|
||
terms = list(
|
||
dict.fromkeys(
|
||
value
|
||
for value in (
|
||
product.product_category,
|
||
*(category or "").split("/"),
|
||
material,
|
||
*(term for term in ("PET", "PP", "软管", "泵头", "瓶", "纸盒", "密封", "相容性") if term in query),
|
||
)
|
||
if value
|
||
)
|
||
)
|
||
clauses = [
|
||
column.ilike(f"%{term}%")
|
||
for term in terms
|
||
for column in (
|
||
StandardRule.category_lv1,
|
||
StandardRule.category_lv2,
|
||
StandardRule.category_lv3,
|
||
StandardRule.risk_point,
|
||
)
|
||
]
|
||
standard_rows = list(
|
||
(
|
||
await self.session.scalars(
|
||
select(StandardRule)
|
||
.where(
|
||
StandardRule.status == "启用",
|
||
or_(*clauses),
|
||
)
|
||
.order_by(StandardRule.id)
|
||
.limit(6)
|
||
)
|
||
).all()
|
||
)
|
||
standards = [
|
||
StandardJudgment(
|
||
category=row.category_lv1,
|
||
risk_point=row.risk_point,
|
||
rule=row.judge_standard,
|
||
source=row.regulation or "标准文件.xlsx",
|
||
status="待检测",
|
||
)
|
||
for row in standard_rows
|
||
]
|
||
dimensions = [
|
||
RiskDimensionScore(
|
||
dimension=item.name,
|
||
score=item.score,
|
||
weight=item.weight,
|
||
status=item.status,
|
||
reason=item.reason,
|
||
sample_size=item.sample_size,
|
||
raw_value=item.raw_value,
|
||
)
|
||
for item in score.dimensions
|
||
]
|
||
category_dimension = next(item for item in score.dimensions if item.name == "产品品类")
|
||
supplier_dimension = next(item for item in score.dimensions if item.name == "供应商表现")
|
||
result = RiskAssessmentResult(
|
||
subject_id=product.subject_id,
|
||
score=score.score,
|
||
level=score.level,
|
||
model_version="database-seven-dimension-v1",
|
||
factors=[
|
||
RiskFactorResult(
|
||
name=item.name,
|
||
value=round(100 - item.score, 2),
|
||
weight=item.weight,
|
||
contribution=round(item.score * item.weight / 100, 2),
|
||
)
|
||
for item in score.dimensions
|
||
],
|
||
dimension_scores=dimensions,
|
||
category_warning=CategoryWarning(
|
||
category=category or "无法归类",
|
||
anomaly_count=category_dimension.sample_size or 0,
|
||
risk_level="案例较少" if (category_dimension.sample_size or 0) < 10 else score.level.value,
|
||
suggestion=category_dimension.reason,
|
||
),
|
||
supplier_freq=SupplierFrequency(
|
||
supplier_name=supplier,
|
||
cooperation_count=cooperation if cooperation else None,
|
||
anomaly_count=supplier_dimension.sample_size,
|
||
alpha=supplier_dimension.frequency_alpha,
|
||
frequency_basis=f"{start} 至 {as_of} 采购订单去重单据数",
|
||
note=supplier_dimension.reason,
|
||
data_available=bool(cooperation),
|
||
),
|
||
standard_judgment=standards,
|
||
similar_cases=[item.case for item in retrieved],
|
||
)
|
||
fallback = [
|
||
"打样阶段完成内容物与包装材料相容性、密封性及加速老化验证。",
|
||
"结合历史异常和命中标准制定来料检验、过程巡检和首批试产验证清单。",
|
||
"补充供应商、加工厂及实际检测数据，由品控复核默认分维度后再作放行决策。",
|
||
"对验证项明确责任人、完成节点及验收依据，并将结果回填形成闭环。",
|
||
]
|
||
try:
|
||
async with asyncio.timeout(20):
|
||
result.ai_suggestions = await LLMClient().generate_prevention_suggestions(
|
||
{
|
||
"input": product.model_dump(mode="json", by_alias=True),
|
||
"assessment": result.model_dump(mode="json", by_alias=True),
|
||
}
|
||
)
|
||
except Exception as exc: # noqa: BLE001 - rules and actual scores remain usable without the LLM
|
||
logger.warning("External report suggestions unavailable: %s", exc.__class__.__name__)
|
||
result.ai_suggestions = fallback
|
||
result.alert_text = "AI 建议服务不可用，本报告使用固定验证建议；评分仍来自数据库及规则。"
|
||
return result
|