498 lines
20 KiB
Python
498 lines
20 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import re
|
||
|
|
from collections import defaultdict
|
||
|
|
from datetime import UTC, date, datetime, time, timedelta
|
||
|
|
from decimal import Decimal
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from sqlalchemy import select
|
||
|
|
from sqlalchemy.orm import Session
|
||
|
|
|
||
|
|
from app.db.base import Base
|
||
|
|
from app.models.budget import BudgetAllocation
|
||
|
|
from app.models.financial_record import ExpenseClaim
|
||
|
|
from app.models.risk_observation import RiskObservation
|
||
|
|
from app.schemas.finance_dashboard import FinanceDashboardRead
|
||
|
|
from app.services.budget_support import BudgetSupportMixin
|
||
|
|
from app.services.expense_claim_constants import EXPENSE_TYPE_LABELS
|
||
|
|
|
||
|
|
SLA_TARGET_HOURS = Decimal("8.0")
|
||
|
|
PENDING_STATUSES = {
|
||
|
|
"submitted",
|
||
|
|
"review",
|
||
|
|
"pending_review",
|
||
|
|
"manager_review",
|
||
|
|
"budget_review",
|
||
|
|
"finance_review",
|
||
|
|
"approving",
|
||
|
|
}
|
||
|
|
SUCCESS_STATUSES = {"approved", "pending_payment", "paid", "completed"}
|
||
|
|
EXCLUDED_SPEND_STATUSES = {"draft", "rejected", "returned", "supplement", "deleted"}
|
||
|
|
EMPTY_DONUT = [{"name": "暂无数据", "value": 0, "color": "#cbd5e1"}]
|
||
|
|
CHART_COLORS = [
|
||
|
|
"var(--theme-primary)",
|
||
|
|
"var(--chart-blue)",
|
||
|
|
"var(--chart-amber)",
|
||
|
|
"var(--chart-purple)",
|
||
|
|
"var(--success)",
|
||
|
|
"var(--danger)",
|
||
|
|
]
|
||
|
|
STAGE_LABELS = {
|
||
|
|
"manager": "直属经理",
|
||
|
|
"manager_review": "直属经理",
|
||
|
|
"budget": "预算复核",
|
||
|
|
"budget_review": "预算复核",
|
||
|
|
"finance": "财务审核",
|
||
|
|
"finance_review": "财务审核",
|
||
|
|
"payment": "付款确认",
|
||
|
|
"pending_payment": "付款确认",
|
||
|
|
}
|
||
|
|
RISK_SIGNAL_LABELS = {
|
||
|
|
"duplicate_invoice": "重复发票",
|
||
|
|
"split_billing": "拆分报销",
|
||
|
|
"frequent_small_claims": "高频小额",
|
||
|
|
"location_mismatch": "地点不一致",
|
||
|
|
"amount_outlier": "金额异常",
|
||
|
|
"preapproval_absent": "缺少事前申请",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
class FinanceDashboardService(BudgetSupportMixin):
|
||
|
|
def __init__(self, db: Session) -> None:
|
||
|
|
self.db = db
|
||
|
|
|
||
|
|
def build_dashboard(
|
||
|
|
self,
|
||
|
|
*,
|
||
|
|
range_key: str = "近10日",
|
||
|
|
start_date: date | None = None,
|
||
|
|
end_date: date | None = None,
|
||
|
|
trend_range: str = "近12天",
|
||
|
|
department_range: str = "本月",
|
||
|
|
) -> FinanceDashboardRead:
|
||
|
|
self._ensure_storage_ready()
|
||
|
|
now = datetime.now(UTC)
|
||
|
|
start, end, resolved_key = self._resolve_scope(
|
||
|
|
range_key=range_key,
|
||
|
|
start_date=start_date,
|
||
|
|
end_date=end_date,
|
||
|
|
now=now,
|
||
|
|
)
|
||
|
|
previous_start = start - (end - start)
|
||
|
|
trend_start, trend_end, trend_labels = self._resolve_trend_scope(trend_range, now)
|
||
|
|
department_start, department_end = self._resolve_department_scope(department_range, now)
|
||
|
|
|
||
|
|
claims = self._fetch_claims()
|
||
|
|
observations = self._fetch_risk_observations()
|
||
|
|
scope_claims = self._claims_between(claims, start, end)
|
||
|
|
previous_claims = self._claims_between(claims, previous_start, start)
|
||
|
|
trend_claims = self._claims_between(claims, trend_start, trend_end)
|
||
|
|
department_claims = self._claims_between(claims, department_start, department_end)
|
||
|
|
scope_observations = self._observations_between(observations, start, end)
|
||
|
|
|
||
|
|
totals = self._totals(scope_claims, scope_observations, now)
|
||
|
|
previous_totals = self._totals(previous_claims, [], now)
|
||
|
|
|
||
|
|
return FinanceDashboardRead(
|
||
|
|
range_key=resolved_key,
|
||
|
|
start_date=start.date().isoformat(),
|
||
|
|
end_date=(end - timedelta(days=1)).date().isoformat(),
|
||
|
|
generated_at=now.isoformat(),
|
||
|
|
has_real_data=bool(claims or observations or self._fetch_budget_allocations(now.year)),
|
||
|
|
totals=totals,
|
||
|
|
metric_meta=self._metric_meta(totals, previous_totals),
|
||
|
|
trend=self._trend(trend_labels, trend_claims, now),
|
||
|
|
spend_by_category=self._spend_by_category(scope_claims),
|
||
|
|
exception_mix=self._exception_mix(scope_claims, scope_observations),
|
||
|
|
department_ranking=self._department_ranking(department_claims),
|
||
|
|
bottlenecks=self._bottlenecks(scope_claims, now),
|
||
|
|
budget_summary=self._budget_summary(now.year),
|
||
|
|
)
|
||
|
|
|
||
|
|
def _ensure_storage_ready(self) -> None:
|
||
|
|
Base.metadata.create_all(bind=self.db.get_bind())
|
||
|
|
|
||
|
|
def _fetch_claims(self) -> list[ExpenseClaim]:
|
||
|
|
stmt = select(ExpenseClaim).order_by(ExpenseClaim.created_at.asc())
|
||
|
|
return list(self.db.scalars(stmt).all())
|
||
|
|
|
||
|
|
def _fetch_risk_observations(self) -> list[RiskObservation]:
|
||
|
|
stmt = select(RiskObservation).order_by(RiskObservation.created_at.asc())
|
||
|
|
return list(self.db.scalars(stmt).all())
|
||
|
|
|
||
|
|
def _fetch_budget_allocations(self, fiscal_year: int) -> list[BudgetAllocation]:
|
||
|
|
stmt = (
|
||
|
|
select(BudgetAllocation)
|
||
|
|
.where(BudgetAllocation.fiscal_year == fiscal_year)
|
||
|
|
.order_by(BudgetAllocation.period_key.asc())
|
||
|
|
)
|
||
|
|
return list(self.db.scalars(stmt).all())
|
||
|
|
|
||
|
|
def _resolve_scope(
|
||
|
|
self,
|
||
|
|
*,
|
||
|
|
range_key: str,
|
||
|
|
start_date: date | None,
|
||
|
|
end_date: date | None,
|
||
|
|
now: datetime,
|
||
|
|
) -> tuple[datetime, datetime, str]:
|
||
|
|
today = now.date()
|
||
|
|
normalized_key = str(range_key or "").strip() or "近10日"
|
||
|
|
|
||
|
|
if start_date and end_date:
|
||
|
|
start_day = min(start_date, end_date)
|
||
|
|
end_day = max(start_date, end_date)
|
||
|
|
return self._day_start(start_day), self._day_after(end_day), "自定义"
|
||
|
|
|
||
|
|
if normalized_key == "今日":
|
||
|
|
start_day = today
|
||
|
|
elif normalized_key == "本周":
|
||
|
|
start_day = today - timedelta(days=today.weekday())
|
||
|
|
elif normalized_key == "本月":
|
||
|
|
start_day = today.replace(day=1)
|
||
|
|
else:
|
||
|
|
days = self._days_from_label(normalized_key, default=10)
|
||
|
|
start_day = today - timedelta(days=days - 1)
|
||
|
|
|
||
|
|
return self._day_start(start_day), self._day_after(today), normalized_key
|
||
|
|
|
||
|
|
def _resolve_trend_scope(
|
||
|
|
self,
|
||
|
|
trend_range: str,
|
||
|
|
now: datetime,
|
||
|
|
) -> tuple[datetime, datetime, list[str]]:
|
||
|
|
days = self._days_from_label(trend_range, default=12)
|
||
|
|
end_day = now.date()
|
||
|
|
start_day = end_day - timedelta(days=days - 1)
|
||
|
|
labels = [self._date_label(start_day + timedelta(days=index)) for index in range(days)]
|
||
|
|
return self._day_start(start_day), self._day_after(end_day), labels
|
||
|
|
|
||
|
|
def _resolve_department_scope(
|
||
|
|
self,
|
||
|
|
department_range: str,
|
||
|
|
now: datetime,
|
||
|
|
) -> tuple[datetime, datetime]:
|
||
|
|
today = now.date()
|
||
|
|
key = str(department_range or "").strip()
|
||
|
|
if key == "本周":
|
||
|
|
start_day = today - timedelta(days=today.weekday())
|
||
|
|
elif key == "本季度":
|
||
|
|
quarter_month = ((today.month - 1) // 3) * 3 + 1
|
||
|
|
start_day = today.replace(month=quarter_month, day=1)
|
||
|
|
else:
|
||
|
|
start_day = today.replace(day=1)
|
||
|
|
return self._day_start(start_day), self._day_after(today)
|
||
|
|
|
||
|
|
def _claims_between(
|
||
|
|
self,
|
||
|
|
claims: list[ExpenseClaim],
|
||
|
|
start: datetime,
|
||
|
|
end: datetime,
|
||
|
|
) -> list[ExpenseClaim]:
|
||
|
|
return [claim for claim in claims if start <= self._claim_time(claim) < end]
|
||
|
|
|
||
|
|
def _observations_between(
|
||
|
|
self,
|
||
|
|
observations: list[RiskObservation],
|
||
|
|
start: datetime,
|
||
|
|
end: datetime,
|
||
|
|
) -> list[RiskObservation]:
|
||
|
|
return [item for item in observations if start <= self._as_utc(item.created_at) < end]
|
||
|
|
|
||
|
|
def _totals(
|
||
|
|
self,
|
||
|
|
claims: list[ExpenseClaim],
|
||
|
|
observations: list[RiskObservation],
|
||
|
|
now: datetime,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
active_claims = [claim for claim in claims if self._status(claim) not in {"draft", "deleted"}]
|
||
|
|
pending_claims = [claim for claim in active_claims if self._status(claim) in PENDING_STATUSES]
|
||
|
|
success_claims = [claim for claim in active_claims if self._status(claim) in SUCCESS_STATUSES]
|
||
|
|
risk_claim_keys = {self._claim_key(claim) for claim in active_claims if self._has_claim_risk(claim)}
|
||
|
|
observation_keys = {
|
||
|
|
str(item.claim_no or item.subject_key or item.id).strip()
|
||
|
|
for item in observations
|
||
|
|
if str(item.status or "").strip().lower() != "false_positive"
|
||
|
|
}
|
||
|
|
sla_hours = [self._claim_sla_hours(claim, now) for claim in active_claims if claim.submitted_at]
|
||
|
|
sla_met = sum(1 for hours in sla_hours if hours <= SLA_TARGET_HOURS)
|
||
|
|
clean_success = sum(1 for claim in success_claims if not self._has_claim_risk(claim))
|
||
|
|
|
||
|
|
return {
|
||
|
|
"pendingCount": len(pending_claims),
|
||
|
|
"pendingAmount": self._decimal_number(sum((self._claim_amount(claim) for claim in pending_claims), Decimal("0.00"))),
|
||
|
|
"avgSla": self._decimal_number(self._average(sla_hours)),
|
||
|
|
"autoPassRate": self._percent(clean_success, len(active_claims)),
|
||
|
|
"riskCount": len({key for key in risk_claim_keys | observation_keys if key}),
|
||
|
|
"slaRate": self._percent(sla_met, len(sla_hours)),
|
||
|
|
}
|
||
|
|
|
||
|
|
def _metric_meta(self, current: dict[str, Any], previous: dict[str, Any]) -> dict[str, Any]:
|
||
|
|
unit_by_key = {
|
||
|
|
"pendingCount": "单",
|
||
|
|
"pendingAmount": "元",
|
||
|
|
"avgSla": "h",
|
||
|
|
"autoPassRate": "%",
|
||
|
|
"riskCount": "单",
|
||
|
|
"slaRate": "%",
|
||
|
|
}
|
||
|
|
meta: dict[str, Any] = {}
|
||
|
|
for key, current_value in current.items():
|
||
|
|
previous_value = Decimal(str(previous.get(key, 0) or 0))
|
||
|
|
value = Decimal(str(current_value or 0))
|
||
|
|
diff = value - previous_value
|
||
|
|
change = self._change_percent(value, previous_value)
|
||
|
|
unit = unit_by_key.get(key, "")
|
||
|
|
meta[key] = {
|
||
|
|
"changeText": f"{'+' if change >= 0 else ''}{change:.1f}%",
|
||
|
|
"delta": f"较上一周期 {'+' if diff >= 0 else ''}{self._format_delta(diff, unit)}",
|
||
|
|
"trend": "up" if diff >= 0 else "down",
|
||
|
|
}
|
||
|
|
return meta
|
||
|
|
|
||
|
|
def _trend(
|
||
|
|
self,
|
||
|
|
labels: list[str],
|
||
|
|
claims: list[ExpenseClaim],
|
||
|
|
now: datetime,
|
||
|
|
) -> dict[str, Any]:
|
||
|
|
applications = [0 for _ in labels]
|
||
|
|
approved = [0 for _ in labels]
|
||
|
|
hours: list[list[Decimal]] = [[] for _ in labels]
|
||
|
|
index = {label: idx for idx, label in enumerate(labels)}
|
||
|
|
|
||
|
|
for claim in claims:
|
||
|
|
if self._status(claim) == "draft":
|
||
|
|
continue
|
||
|
|
label = self._date_label(self._claim_time(claim).date())
|
||
|
|
if label not in index:
|
||
|
|
continue
|
||
|
|
bucket = index[label]
|
||
|
|
applications[bucket] += 1
|
||
|
|
if self._status(claim) in SUCCESS_STATUSES:
|
||
|
|
approved[bucket] += 1
|
||
|
|
if claim.submitted_at:
|
||
|
|
hours[bucket].append(self._claim_sla_hours(claim, now))
|
||
|
|
|
||
|
|
return {
|
||
|
|
"labels": labels,
|
||
|
|
"applications": applications,
|
||
|
|
"approved": approved,
|
||
|
|
"avgHours": [self._decimal_number(self._average(row)) for row in hours],
|
||
|
|
}
|
||
|
|
|
||
|
|
def _spend_by_category(self, claims: list[ExpenseClaim]) -> list[dict[str, Any]]:
|
||
|
|
buckets: dict[str, Decimal] = defaultdict(Decimal)
|
||
|
|
for claim in claims:
|
||
|
|
if self._status(claim) in EXCLUDED_SPEND_STATUSES:
|
||
|
|
continue
|
||
|
|
label = EXPENSE_TYPE_LABELS.get(str(claim.expense_type or "").strip(), claim.expense_type)
|
||
|
|
buckets[str(label or "其他费用")] += self._claim_amount(claim)
|
||
|
|
|
||
|
|
rows = [
|
||
|
|
{"name": name, "value": self._decimal_number(value), "color": CHART_COLORS[index % len(CHART_COLORS)]}
|
||
|
|
for index, (name, value) in enumerate(sorted(buckets.items(), key=lambda item: item[1], reverse=True)[:6])
|
||
|
|
]
|
||
|
|
return rows or EMPTY_DONUT
|
||
|
|
|
||
|
|
def _exception_mix(
|
||
|
|
self,
|
||
|
|
claims: list[ExpenseClaim],
|
||
|
|
observations: list[RiskObservation],
|
||
|
|
) -> list[dict[str, Any]]:
|
||
|
|
buckets: dict[str, int] = defaultdict(int)
|
||
|
|
|
||
|
|
for observation in observations:
|
||
|
|
key = str(observation.risk_signal or observation.risk_type or "").strip()
|
||
|
|
buckets[RISK_SIGNAL_LABELS.get(key, key.replace("_", " ") or "风险观察")] += 1
|
||
|
|
|
||
|
|
if not buckets:
|
||
|
|
for claim in claims:
|
||
|
|
if self._status(claim) in {"draft", "deleted"}:
|
||
|
|
continue
|
||
|
|
for label in self._claim_risk_labels(claim):
|
||
|
|
buckets[label] += 1
|
||
|
|
|
||
|
|
rows = [
|
||
|
|
{"name": name, "value": count, "color": CHART_COLORS[index % len(CHART_COLORS)]}
|
||
|
|
for index, (name, count) in enumerate(sorted(buckets.items(), key=lambda item: item[1], reverse=True)[:6])
|
||
|
|
]
|
||
|
|
return rows or EMPTY_DONUT
|
||
|
|
|
||
|
|
def _department_ranking(self, claims: list[ExpenseClaim]) -> list[dict[str, Any]]:
|
||
|
|
buckets: dict[str, Decimal] = defaultdict(Decimal)
|
||
|
|
for claim in claims:
|
||
|
|
if self._status(claim) not in PENDING_STATUSES:
|
||
|
|
continue
|
||
|
|
buckets[str(claim.department_name or "未归属部门")] += self._claim_amount(claim)
|
||
|
|
|
||
|
|
rows = [
|
||
|
|
{
|
||
|
|
"name": name,
|
||
|
|
"amount": self._decimal_number(amount),
|
||
|
|
"value": self._decimal_number(amount),
|
||
|
|
"color": CHART_COLORS[index % len(CHART_COLORS)],
|
||
|
|
}
|
||
|
|
for index, (name, amount) in enumerate(sorted(buckets.items(), key=lambda item: item[1], reverse=True)[:5])
|
||
|
|
]
|
||
|
|
return rows
|
||
|
|
|
||
|
|
def _bottlenecks(self, claims: list[ExpenseClaim], now: datetime) -> list[dict[str, Any]]:
|
||
|
|
buckets: dict[str, list[Decimal]] = defaultdict(list)
|
||
|
|
for claim in claims:
|
||
|
|
if self._status(claim) not in PENDING_STATUSES:
|
||
|
|
continue
|
||
|
|
stage = self._stage_label(claim)
|
||
|
|
buckets[stage].append(self._claim_sla_hours(claim, now))
|
||
|
|
|
||
|
|
rows: list[dict[str, Any]] = []
|
||
|
|
for index, (stage, values) in enumerate(sorted(buckets.items(), key=lambda item: self._average(item[1]), reverse=True)[:3]):
|
||
|
|
avg_hours = self._average(values)
|
||
|
|
rows.append(
|
||
|
|
{
|
||
|
|
"name": stage,
|
||
|
|
"role": "审批节点",
|
||
|
|
"duration": f"{self._decimal_number(avg_hours):.1f} h",
|
||
|
|
"status": self._duration_status(avg_hours),
|
||
|
|
"tone": self._duration_tone(avg_hours),
|
||
|
|
"avatar": stage[:1] or str(index + 1),
|
||
|
|
}
|
||
|
|
)
|
||
|
|
return rows
|
||
|
|
|
||
|
|
def _budget_summary(self, fiscal_year: int) -> dict[str, Any]:
|
||
|
|
allocations = self._fetch_budget_allocations(fiscal_year)
|
||
|
|
total = Decimal("0.00")
|
||
|
|
used = Decimal("0.00")
|
||
|
|
available = Decimal("0.00")
|
||
|
|
|
||
|
|
for allocation in allocations:
|
||
|
|
balance = self.get_balance(allocation)
|
||
|
|
total += balance.total_amount
|
||
|
|
used += balance.reserved_amount + balance.consumed_amount
|
||
|
|
available += balance.available_amount
|
||
|
|
|
||
|
|
ratio = Decimal("0.00")
|
||
|
|
if total > Decimal("0.00"):
|
||
|
|
ratio = (used / total) * Decimal("100")
|
||
|
|
|
||
|
|
return {
|
||
|
|
"ratio": self._decimal_number(ratio),
|
||
|
|
"total": self._currency(total),
|
||
|
|
"used": self._currency(used),
|
||
|
|
"left": self._currency(available),
|
||
|
|
}
|
||
|
|
|
||
|
|
def _claim_time(self, claim: ExpenseClaim) -> datetime:
|
||
|
|
return self._as_utc(claim.submitted_at or claim.occurred_at or claim.created_at)
|
||
|
|
|
||
|
|
def _claim_sla_hours(self, claim: ExpenseClaim, now: datetime) -> Decimal:
|
||
|
|
start = self._as_utc(claim.submitted_at or claim.created_at or claim.occurred_at)
|
||
|
|
end = now
|
||
|
|
if self._status(claim) in SUCCESS_STATUSES | {"rejected", "returned"} and claim.updated_at:
|
||
|
|
end = self._as_utc(claim.updated_at)
|
||
|
|
hours = Decimal(str(max((end - start).total_seconds(), 0))) / Decimal("3600")
|
||
|
|
return hours.quantize(Decimal("0.1"))
|
||
|
|
|
||
|
|
def _claim_amount(self, claim: ExpenseClaim) -> Decimal:
|
||
|
|
return Decimal(str(claim.amount or 0))
|
||
|
|
|
||
|
|
def _claim_key(self, claim: ExpenseClaim) -> str:
|
||
|
|
return str(claim.claim_no or claim.id or "").strip()
|
||
|
|
|
||
|
|
def _has_claim_risk(self, claim: ExpenseClaim) -> bool:
|
||
|
|
return bool(claim.hermes_risk_flag or self._risk_flags(claim))
|
||
|
|
|
||
|
|
def _claim_risk_labels(self, claim: ExpenseClaim) -> list[str]:
|
||
|
|
labels: list[str] = []
|
||
|
|
if claim.hermes_risk_flag:
|
||
|
|
labels.append("风险扫描命中")
|
||
|
|
for flag in self._risk_flags(claim):
|
||
|
|
if isinstance(flag, dict):
|
||
|
|
label = str(flag.get("label") or flag.get("message") or flag.get("type") or "").strip()
|
||
|
|
else:
|
||
|
|
label = str(flag or "").strip()
|
||
|
|
labels.append(label or "规则异常")
|
||
|
|
return labels
|
||
|
|
|
||
|
|
def _risk_flags(self, claim: ExpenseClaim) -> list[Any]:
|
||
|
|
flags = claim.risk_flags_json or []
|
||
|
|
return flags if isinstance(flags, list) else []
|
||
|
|
|
||
|
|
def _stage_label(self, claim: ExpenseClaim) -> str:
|
||
|
|
stage = str(claim.approval_stage or self._status(claim) or "").strip().lower()
|
||
|
|
return STAGE_LABELS.get(stage, stage.replace("_", " ").strip() or "待审批")
|
||
|
|
|
||
|
|
def _status(self, claim: ExpenseClaim) -> str:
|
||
|
|
return str(claim.status or "").strip().lower()
|
||
|
|
|
||
|
|
def _as_utc(self, value: datetime | None) -> datetime:
|
||
|
|
if value is None:
|
||
|
|
return datetime.now(UTC)
|
||
|
|
if value.tzinfo is None:
|
||
|
|
return value.replace(tzinfo=UTC)
|
||
|
|
return value.astimezone(UTC)
|
||
|
|
|
||
|
|
def _day_start(self, value: date) -> datetime:
|
||
|
|
return datetime.combine(value, time.min, tzinfo=UTC)
|
||
|
|
|
||
|
|
def _day_after(self, value: date) -> datetime:
|
||
|
|
return datetime.combine(value + timedelta(days=1), time.min, tzinfo=UTC)
|
||
|
|
|
||
|
|
def _date_label(self, value: date) -> str:
|
||
|
|
return value.strftime("%m-%d")
|
||
|
|
|
||
|
|
def _days_from_label(self, value: str, *, default: int) -> int:
|
||
|
|
match = re.search(r"\d+", str(value or ""))
|
||
|
|
if not match:
|
||
|
|
return default
|
||
|
|
return max(1, min(int(match.group(0)), 90))
|
||
|
|
|
||
|
|
def _duration_status(self, hours: Decimal) -> str:
|
||
|
|
if hours >= Decimal("12"):
|
||
|
|
return "较慢"
|
||
|
|
if hours >= SLA_TARGET_HOURS:
|
||
|
|
return "偏慢"
|
||
|
|
return "正常"
|
||
|
|
|
||
|
|
def _duration_tone(self, hours: Decimal) -> str:
|
||
|
|
if hours >= Decimal("12"):
|
||
|
|
return "danger"
|
||
|
|
if hours >= SLA_TARGET_HOURS:
|
||
|
|
return "warning"
|
||
|
|
return "success"
|
||
|
|
|
||
|
|
def _average(self, values: list[Decimal]) -> Decimal:
|
||
|
|
if not values:
|
||
|
|
return Decimal("0.00")
|
||
|
|
return sum(values, Decimal("0.00")) / Decimal(str(len(values)))
|
||
|
|
|
||
|
|
def _percent(self, part: int | Decimal, total: int | Decimal) -> float:
|
||
|
|
total_decimal = Decimal(str(total or 0))
|
||
|
|
if total_decimal <= Decimal("0"):
|
||
|
|
return 0.0
|
||
|
|
return self._decimal_number((Decimal(str(part or 0)) / total_decimal) * Decimal("100"))
|
||
|
|
|
||
|
|
def _change_percent(self, current: Decimal, previous: Decimal) -> float:
|
||
|
|
if previous == Decimal("0"):
|
||
|
|
return 0.0 if current == Decimal("0") else 100.0
|
||
|
|
return self._decimal_number(((current - previous) / previous) * Decimal("100"))
|
||
|
|
|
||
|
|
def _decimal_number(self, value: Decimal) -> float:
|
||
|
|
return float(value.quantize(Decimal("0.1")))
|
||
|
|
|
||
|
|
def _format_delta(self, value: Decimal, unit: str) -> str:
|
||
|
|
if unit == "元":
|
||
|
|
return self._currency(value)
|
||
|
|
if unit == "h":
|
||
|
|
return f"{self._decimal_number(value):.1f}h"
|
||
|
|
if unit == "%":
|
||
|
|
return f"{self._decimal_number(value):.1f}%"
|
||
|
|
return f"{int(value)}{unit}"
|
||
|
|
|
||
|
|
def _currency(self, value: Decimal) -> str:
|
||
|
|
prefix = "-¥" if value < Decimal("0") else "¥"
|
||
|
|
amount = abs(value)
|
||
|
|
return f"{prefix}{amount:,.0f}"
|