300 lines
12 KiB
Python

"""Live supplier profile aggregation from the approved anomaly case data."""
from __future__ import annotations
import re
from collections import Counter, defaultdict
from datetime import UTC, date, datetime
from sqlalchemy import func, 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 calendar_year_window, supplier_order_counts
class SupplierProfileService:
history_years = 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:
grouped = await self._load_supplier_summary(keyword=keyword)
items = [
SupplierOption(
name=name,
anomaly_count=summary["total"],
tier=self.tier_for(
summary["total"],
summary["a"],
summary["b"],
),
)
for name, summary in sorted(
grouped.items(), key=lambda item: (-item[1]["total"], item[0])
)[:limit]
]
return SupplierListData(
items=items,
default_supplier=items[0].name if items else None,
)
async def detail(self, supplier_name: str) -> SupplierProfileData:
all_rows = await self._load_rows()
rows = [row for row in all_rows if self.matches(row.get("supplier"), supplier_name)]
if not rows:
raise AppException("供应商不存在或暂无近3个自然年度异常记录", code=404, status_code=404)
anomaly_total = len(rows)
history_start, history_end = calendar_year_window(self.as_of, self.history_years)
order_count = await supplier_order_counts(
self.session,
supplier_name,
history_start,
history_end,
)
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,
rows=all_rows,
cooperation_count=order_count,
)
supplier_dimension = next(
(item for item in score.dimensions if item.name == "供应商表现"),
None,
)
cooperation_display = order_count if order_count > 0 else None
display_tier = {
"S级": "S级 · 战略",
"A级": "A级 · 优质",
"B级": "B级 · 关注",
"C级": "C级 · 高风险",
}.get(tier, tier)
return SupplierProfileData(
supplier_name=supplier_name,
tier=display_tier,
tier_condition=self.tier_condition(tier),
metric=SupplierMetric(
cooperation_total=cooperation_display,
anomaly_total=anomaly_total,
anomaly_rate=None,
frequency_alpha=(
supplier_dimension.frequency_alpha
if supplier_dimension and supplier_dimension.frequency_alpha is not None
else self.alpha(
cooperation_display or anomaly_total,
coefficient=scoring.frequency_coefficient,
cap=scoring.frequency_cap,
)
),
supplier_base_score=(
supplier_dimension.base_score if supplier_dimension else None
),
supplier_score=supplier_dimension.score if supplier_dimension else None,
risk_score=score.score,
main_category_count=len(categories),
),
category_breakdown=self._display_categories(categories),
monthly_trend=self._monthly_trend(rows),
recent_records=self._recent_records(rows),
factory_recommendations=[],
recommendation_note="合作次数按自然年度边界采购订单去重单据编号统计;异常数和供应商层级仍按异常案例库统计。",
)
async def _load_supplier_summary(
self, *, keyword: str | None = None
) -> dict[str, dict[str, int]]:
"""Load only supplier aggregates for the selector.
The selector used to fetch every anomaly row and group it in Python.
This grouped query returns a handful of rows even when the anomaly
table grows, which materially reduces first-navigation latency over
the remote PostgreSQL connection.
"""
start, end = calendar_year_window(self.as_of, self.history_years)
statement = (
select(AnomalyCase.supplier, AnomalyCase.defect_level, func.count())
.where(
AnomalyCase.supplier.is_not(None),
AnomalyCase.date.is_not(None),
AnomalyCase.date >= start,
AnomalyCase.date <= end,
)
.group_by(AnomalyCase.supplier, AnomalyCase.defect_level)
)
if keyword and keyword.strip():
statement = statement.where(AnomalyCase.supplier.ilike(f"%{keyword.strip()}%"))
grouped: dict[str, dict[str, int]] = defaultdict(
lambda: {"total": 0, "a": 0, "b": 0}
)
for supplier, defect_level, count in (await self.session.execute(statement)).all():
name = (supplier or "").strip()
if not name:
continue
value = grouped[name]
amount = int(count or 0)
value["total"] += amount
code = self.defect_code(defect_level)
if code == "A":
value["a"] += amount
elif code == "B":
value["b"] += amount
return grouped
async def _load_rows(self, *, supplier_name: str | None = None) -> list[dict]:
start, end = calendar_year_window(self.as_of, self.history_years)
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 <= end,
)
if supplier_name:
statement = statement.where(AnomalyCase.supplier.ilike(f"%{supplier_name}%"))
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:
match = re.search(r"[ABCD]", (value or "").upper())
return match.group(0) if match 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():
is_unclassified = category == "待补标品类"
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,
is_unclassified=is_unclassified,
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.is_unclassified, -item.anomaly_count, item.category))
@classmethod
def _display_categories(
cls, categories: list[SupplierCategoryBreakdown]
) -> list[SupplierCategoryBreakdown]:
"""Keep up to five real categories and always retain the unclassified row."""
real = [item for item in categories if not item.is_unclassified][: cls.max_categories]
unclassified = next((item for item in categories if item.is_unclassified), None)
return real + ([unclassified] if unclassified else [])
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]
]