"""Database-backed dashboard aggregation.""" from collections import Counter, defaultdict from datetime import date, timedelta from typing import Literal from sqlalchemy import func, select from app.models.entities import AnomalyCase, StandardRule from app.schemas.dashboard import ( DashboardCategory, DashboardData, DashboardDefect, DashboardMatrixRow, DashboardMetric, DashboardStage, DashboardSupplierRank, DashboardTrendPoint, ) class DashboardService: _stages = ( ("需求发现", "第一阶段"), ("产品立项", "第二阶段"), ("产品开发", "第三阶段"), ("量产上市", "第四阶段"), ("产品生命管理", "第五阶段"), ) _category_order = ( "面霜/护肤类", "洗发水/洗护类", "沐浴露类", "洁面/卸妆类", "身体护理类", "头发护理类", "脱毛类", "牙膏/口腔护理类", "彩妆/唇部类", "香氛/止汗类", "清洁剂类", "包材/物料专项", "待补标品类", ) def __init__(self, session) -> None: # noqa: ANN001 self.session = session async def overview(self, year: Literal["all", "2024", "2025", "2026"]) -> DashboardData: rows = list( ( await self.session.execute( select( AnomalyCase.date, AnomalyCase.product_name, AnomalyCase.stage_new, AnomalyCase.anomaly_type, AnomalyCase.material_type, AnomalyCase.supplier, AnomalyCase.factory, AnomalyCase.defect_level, ) ) ).mappings() ) filtered = [row for row in rows if self._in_year(row["date"], year)] previous = self._previous_period(rows, year, filtered) total = len(filtered) previous_total = len(previous) standards_count = int( await self.session.scalar( select(func.count()).select_from(StandardRule).where(StandardRule.status == "启用") ) or 0 ) defects = self._defects(filtered, total) actors = self._actors(filtered) high_risk_count = sum(1 for item in actors.values() if item["tier"] in {"B", "C"}) ab_count = sum(item.count for item in defects if item.level in {"A", "B"}) metrics = [ DashboardMetric( value=total, label="异常案例总数", change_percent=self._change(total, previous_total), ), DashboardMetric( value=ab_count, label="A+B级缺陷", change_percent=self._change( ab_count, sum(1 for row in previous if row["defect_level"] in {"A", "B"}), ), ), DashboardMetric( value=high_risk_count, label="高风险供应商/工厂", change_percent=self._change( high_risk_count, sum(1 for item in self._actors(previous).values() if item["tier"] in {"B", "C"}), ), ), DashboardMetric(value=None, label="预测准确率", change_percent=None), DashboardMetric(value=standards_count, label="标准库覆盖", change_percent=None), ] stages = self._stages_data(filtered, total) categories = self._categories_data(filtered, total) matrix_categories = [item.name for item in categories] matrix = self._matrix(filtered, matrix_categories) ranking = self._ranking(actors) trend = self._trend(filtered, year) insight = self._insight(trend, stages, categories) warnings = self._warnings(trend, categories, ranking) return DashboardData( year=year, total_cases=total, metrics=metrics, stages=stages, categories=categories, matrix_categories=matrix_categories, matrix=matrix, defects=defects, supplier_ranking=ranking, trend=trend, insight=insight, warnings=warnings, ) @staticmethod def _in_year(value: date | None, year: str) -> bool: return year == "all" or (value is not None and value.year == int(year)) @staticmethod def _change(current: int, previous: int) -> float | None: if not previous: return None return round((current - previous) / previous * 100, 1) @staticmethod def _previous_period(rows, year: str, filtered) -> list: # noqa: ANN001 if year != "all": return [ row for row in rows if row["date"] is not None and row["date"].year == int(year) - 1 ] dates = [row["date"] for row in filtered if row["date"] is not None] if not dates: return [] current_end = max(dates) current_start = current_end - timedelta(days=365) previous_start = current_start - timedelta(days=365) return [ row for row in rows if row["date"] is not None and previous_start <= row["date"] < current_start ] @classmethod def _category(cls, product_name: str | None, material_type: str | None = None) -> str: value = (product_name or "").lower() if "脱毛" in value: return "脱毛类" if any(word in value for word in ("牙膏", "牙粉", "漱口", "口腔")): return "牙膏/口腔护理类" if any(word in value for word in ("洁面", "洁颜", "洗面", "卸妆")): return "洁面/卸妆类" if any(word in value for word in ("护发", "发膜", "护发素", "染发")): return "头发护理类" if any(word in value for word in ("唇", "口红", "唇膏", "气垫", "粉底", "彩妆", "腮红")): return "彩妆/唇部类" if any(word in value for word in ("香氛", "香水", "香体", "走珠", "狐臭", "止汗")): return "香氛/止汗类" if any(word in value for word in ("身体乳", "护手霜", "磨砂", "搓泥", "防晒")): return "身体护理类" if any(word in value for word in ("面霜", "乳液", "精华", "面膜", "素颜霜", "爽肤水", "化妆水", "眼霜")): return "面霜/护肤类" if any(word in value for word in ("洗发", "洗发水", "洗发露")): return "洗发水/洗护类" if "沐浴" in value: return "沐浴露类" if "牙膏" in value: return "牙膏类" if any(word in value for word in ("清洁剂", "消毒", "洁厕", "洗衣")): return "清洁剂类" if material_type and material_type.strip() not in {"--", "-"}: return "包材/物料专项" return "待补标品类" @staticmethod def _risk_level_by_ratio(ratio: float) -> str: """Use the four visual bands defined by the dashboard prototype.""" if ratio >= 0.8: return "高风险" if ratio >= 0.6: return "中高" if ratio >= 0.35: return "中等" return "较低" @classmethod def _stages_data(cls, rows, total: int) -> list[DashboardStage]: # noqa: ANN001 counts = Counter() for row in rows: raw = row["stage_new"] or "" for label, marker in cls._stages: if label in raw or marker in raw: counts[label] += 1 break maximum = max(counts.values(), default=0) return [ DashboardStage( name=label, count=counts[label], share=round(counts[label] / total * 100, 1) if total else 0, risk_level=cls._risk_level_by_ratio(counts[label] / maximum) if maximum else "较低", ) for label, _ in cls._stages ] @classmethod def _categories_data(cls, rows, total: int) -> list[DashboardCategory]: # noqa: ANN001 grouped: dict[str, list] = defaultdict(list) for row in rows: grouped[cls._category(row["product_name"], row["material_type"])].append(row) output = [] for name in cls._category_order: group = grouped.get(name, []) if not group: continue materials = Counter(row["material_type"] for row in group if row["material_type"]) output.append( DashboardCategory( name=name, count=len(group), share=round(len(group) / total * 100, 1) if total else 0, risk_level="较低", top_material=materials.most_common(1)[0][0] if materials else None, ) ) output = [item for item in output if item.count] maximum = max((item.count for item in output), default=0) return [ item.model_copy( update={ "risk_level": cls._risk_level_by_ratio(item.count / maximum) if maximum else "较低" } ) for item in sorted(output, key=lambda item: (-item.count, cls._category_order.index(item.name))) ] @staticmethod def _tier(count: int, has_a: bool = False) -> str: if count > 120 or has_a: return "C" if count > 60: return "B" if count >= 20: return "A" return "S" @classmethod def _actors(cls, rows) -> dict[tuple[str, str], dict]: # noqa: ANN001 grouped: dict[tuple[str, str], list] = defaultdict(list) for row in rows: for field, party_type in (("supplier", "供应商"), ("factory", "加工厂")): name = (row[field] or "").strip() if name and name != "--": grouped[(name, party_type)].append(row) return { key: { "count": len(items), "has_a": any(row["defect_level"] == "A" for row in items), "items": items, "tier": cls._tier(len(items), any(row["defect_level"] == "A" for row in items)), } for key, items in grouped.items() } @classmethod def _matrix(cls, rows, categories: list[str]) -> list[DashboardMatrixRow]: # noqa: ANN001 actors = cls._actors(rows) 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 values = {category: 0 for category in categories} for row in item["items"]: category = cls._category(row["product_name"], row["material_type"]) if category in values: values[category] += 1 result.append( DashboardMatrixRow( name=name, party_type=party_type, categories=values, cooperation_total=cooperation_total, anomaly_count=anomaly_count, anomaly_rate=None, tier=item["tier"], ) ) return result @classmethod def _ranking(cls, actors: dict[tuple[str, str], dict]) -> list[DashboardSupplierRank]: ranking = [] for (name, party_type), item in sorted(actors.items(), key=lambda entry: -entry[1]["count"]): problems = Counter( row["anomaly_type"] for row in item["items"] if row["anomaly_type"] ) ranking.append( DashboardSupplierRank( name=name, party_type=party_type, count=item["count"], tier=item["tier"], top_problem=problems.most_common(1)[0][0] if problems else None, ) ) return ranking @staticmethod def _defects(rows, total: int) -> list[DashboardDefect]: # noqa: ANN001 labels = (("A", "A级缺陷"), ("B", "B级缺陷"), ("C", "C级缺陷"), ("D", "D级/未标注")) counts = Counter(row["defect_level"] if row["defect_level"] in {"A", "B", "C"} else "D" for row in rows) return [ DashboardDefect( level=level, label=label, count=counts[level], share=round(counts[level] / total * 100, 1) if total else 0, ) for level, label in labels ] @classmethod def _trend(cls, rows, year: str) -> list[DashboardTrendPoint]: # noqa: ANN001 dated = [row["date"] for row in rows if row["date"] is not None] if not dated: return [] end = max(dated) if year != "all": end = date(int(year), 12, 1) months = [] current_year, current_month = end.year, end.month for _ in range(12): key = f"{current_year:04d}-{current_month:02d}" count = sum(1 for value in dated if value.year == current_year and value.month == current_month) months.append(DashboardTrendPoint(month=key, label=f"{current_month}月", count=count)) current_month -= 1 if current_month == 0: current_year -= 1 current_month = 12 return list(reversed(months)) @staticmethod def _insight(trend, stages, categories) -> str: # noqa: ANN001 if len([item for item in trend if item.count]) < 3: return "数据量不足，暂无法形成稳定的趋势洞察。" recent = sum(item.count for item in trend[-3:]) earlier = sum(item.count for item in trend[:3]) direction = "上升" if recent > earlier * 1.1 else "下降" if recent < earlier * 0.9 else "基本平稳" stage = max(stages, key=lambda item: item.count, default=None) category = max(categories, key=lambda item: item.count, default=None) stage_name = stage.name if stage else "产品开发" category_name = category.name if category else "品类" return f"近期开案趋势{direction}；{stage_name}阶段异常最多，{category_name}是当前主要风险来源。" @staticmethod def _warnings(trend, categories, ranking) -> list[str]: # noqa: ANN001 warnings = [] if trend and trend[-1].count > 10: warnings.append(f"最近月份异常 {trend[-1].count} 条，已超过供应商月异常预警参考阈值。") for item in ranking: if item.count > 60: warnings.append(f"{item.name}{item.party_type}累计异常 {item.count} 条，层级为 {item.tier} 级。") for item in categories: if item.count > 50: warnings.append(f"{item.name}累计异常 {item.count} 条，建议重点关注{item.top_material or '高频物料'}。") return warnings[:5]