430 lines
18 KiB
Python
430 lines
18 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 String, func, or_, select
|
||
from sqlalchemy.exc import ProgrammingError
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.exceptions import AppException
|
||
from app.models.entities import AnomalyCase, AnomalyCaseSupplier, SupplierMaster
|
||
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:
|
||
canonical = await self._load_master_summary(keyword=keyword, limit=limit)
|
||
if canonical:
|
||
return SupplierListData(items=canonical, default_supplier=canonical[0].name)
|
||
grouped = await self._load_supplier_summary(keyword=keyword)
|
||
# Some imported legacy rows have no anomaly date. They are excluded
|
||
# from the normal three-year risk window, but must still be discoverable
|
||
# when a user searches by a supplier/city name.
|
||
if keyword and not grouped:
|
||
grouped = await self._load_supplier_summary(
|
||
keyword=keyword,
|
||
include_undated=True,
|
||
)
|
||
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:
|
||
linked_rows = await self._load_linked_rows(supplier_name)
|
||
all_rows = linked_rows or await self._load_rows()
|
||
rows = linked_rows or [row for row in all_rows if self.matches(row.get("supplier"), supplier_name)]
|
||
used_legacy_fallback = False
|
||
if not rows:
|
||
all_rows = await self._load_rows(include_undated=True)
|
||
rows = [row for row in all_rows if self.matches(row.get("supplier"), supplier_name)]
|
||
used_legacy_fallback = bool(rows)
|
||
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=(
|
||
"该供应商仅命中历史导入记录，部分异常缺少发生日期；本页展示可识别的历史记录，"
|
||
"不将日期缺失数据混入近3年趋势。"
|
||
if used_legacy_fallback
|
||
else "供应商画像按标准供应商关联统计；合作次数按采购订单去重单据编号统计。"
|
||
),
|
||
)
|
||
|
||
async def _load_master_summary(self, *, keyword: str | None, limit: int) -> list[SupplierOption]:
|
||
filters = [SupplierMaster.status == "启用"]
|
||
if keyword:
|
||
pattern = f"%{keyword.strip()}%"
|
||
filters.append(or_(
|
||
SupplierMaster.canonical_name.ilike(pattern),
|
||
SupplierMaster.short_name.ilike(pattern),
|
||
SupplierMaster.city.ilike(pattern),
|
||
SupplierMaster.province.ilike(pattern),
|
||
SupplierMaster.aliases.cast(String).ilike(pattern),
|
||
))
|
||
statement = (
|
||
select(SupplierMaster, func.count(AnomalyCaseSupplier.id).label("anomaly_count"))
|
||
.outerjoin(AnomalyCaseSupplier, AnomalyCaseSupplier.supplier_id == SupplierMaster.id)
|
||
.where(*filters)
|
||
.group_by(SupplierMaster.id)
|
||
.order_by(func.count(AnomalyCaseSupplier.id).desc(), SupplierMaster.canonical_name)
|
||
.limit(limit)
|
||
)
|
||
try:
|
||
async with self.session.begin_nested():
|
||
rows = (await self.session.execute(statement)).all()
|
||
except ProgrammingError:
|
||
# Allows a rolling deployment: the legacy raw supplier search stays
|
||
# available until the privileged migration has created master tables.
|
||
return []
|
||
return [SupplierOption(
|
||
name=master.canonical_name,
|
||
anomaly_count=int(count or 0),
|
||
tier="待确认" if master.review_status != "已确认" else self.tier_for(int(count or 0), 0, 0),
|
||
city=master.city,
|
||
province=master.province,
|
||
short_name=master.short_name,
|
||
review_status=master.review_status,
|
||
) for master, count in rows]
|
||
|
||
async def _load_linked_rows(self, supplier_name: str) -> list[dict]:
|
||
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,
|
||
)
|
||
.join(AnomalyCaseSupplier, AnomalyCaseSupplier.case_id == AnomalyCase.id)
|
||
.join(SupplierMaster, SupplierMaster.id == AnomalyCaseSupplier.supplier_id)
|
||
.where(or_(
|
||
SupplierMaster.canonical_name == supplier_name,
|
||
SupplierMaster.short_name == supplier_name,
|
||
SupplierMaster.aliases.cast(String).ilike(f"%{supplier_name}%"),
|
||
))
|
||
)
|
||
try:
|
||
async with self.session.begin_nested():
|
||
result = await self.session.execute(statement)
|
||
return [dict(row) for row in result.mappings().all()]
|
||
except ProgrammingError:
|
||
return []
|
||
|
||
async def _load_supplier_summary(
|
||
self,
|
||
*,
|
||
keyword: str | None = None,
|
||
include_undated: bool = False,
|
||
) -> 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))
|
||
.group_by(AnomalyCase.supplier, AnomalyCase.defect_level)
|
||
)
|
||
if not include_undated:
|
||
statement = statement.where(
|
||
AnomalyCase.date.is_not(None),
|
||
AnomalyCase.date >= start,
|
||
AnomalyCase.date <= end,
|
||
)
|
||
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
|
||
if keyword and not self.matches(name, keyword):
|
||
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,
|
||
include_undated: bool = False,
|
||
) -> 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()
|
||
if not include_undated:
|
||
statement = statement.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:
|
||
if not value or not target:
|
||
return False
|
||
# Supplier cells in the source workbooks often contain several names
|
||
# such as ``皓盛/佛山执得``. Match each segment separately and make
|
||
# common city/company suffixes optional so ``佛山市`` can find
|
||
# ``佛山执得`` while preserving the canonical raw value for display.
|
||
queries = SupplierProfileService._search_terms(target)
|
||
segments = SupplierProfileService._supplier_segments(value)
|
||
return any(
|
||
query and (query in segment or segment in query)
|
||
for query in queries
|
||
for segment in segments
|
||
)
|
||
|
||
@staticmethod
|
||
def _supplier_segments(value: str) -> list[str]:
|
||
segments = re.split(r"[/／、,，;；|\\\\]+", value)
|
||
return [SupplierProfileService._compact_text(item) for item in segments if SupplierProfileService._compact_text(item)]
|
||
|
||
@staticmethod
|
||
def _compact_text(value: str) -> str:
|
||
return re.sub(r"[\s()（）【】\[\]{}]+", "", value.casefold())
|
||
|
||
@classmethod
|
||
def _search_terms(cls, value: str) -> set[str]:
|
||
compact = cls._compact_text(value)
|
||
terms = {compact}
|
||
# A user commonly enters a city with the administrative suffix while
|
||
# the workbook keeps the company/trading name (e.g. 佛山市 vs 佛山执得).
|
||
if compact.endswith("市") and len(compact) > 2:
|
||
terms.add(compact[:-1])
|
||
for suffix in ("有限公司", "有限责任公司", "公司"):
|
||
if compact.endswith(suffix) and len(compact) > len(suffix) + 1:
|
||
terms.add(compact[: -len(suffix)])
|
||
return {term for term in terms if term}
|
||
|
||
@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]
|
||
]
|