from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.models.models import FormulaCoefficient from functools import lru_cache import math _coeff_cache: dict[str, float] = {} async def get_coefficient(db: AsyncSession, key: str, default: float = 0.0) -> float: global _coeff_cache if not _coeff_cache: result = await db.execute( select(FormulaCoefficient.key, FormulaCoefficient.value) .where(FormulaCoefficient.is_active == True) ) _coeff_cache = {row.key: float(row.value) for row in result.all()} return _coeff_cache.get(key, default) async def reload_coefficients(db: AsyncSession): global _coeff_cache _coeff_cache = {} await get_coefficient(db, "dummy") def fmt_money(amount: float) -> str: return f"{amount:,.0f}".replace(",", " ") + " ₽" def fmt_percent(value: float) -> str: return f"{value:.1f}%" def fmt_hours(hours: float) -> str: return f"{hours:.1f} ч" def badge_class(status: str, badge_type: str = "default") -> str: mapping = { "P1": "badge-critical", "P2": "badge-high", "P3": "badge-medium", "P4": "badge-low", "open": "badge-open", "in_progress": "badge-progress", "completed": "badge-done", "cancelled": "badge-cancelled", "active": "badge-active", "inactive": "badge-inactive", "draft": "badge-draft", "final": "badge-done", "Risk": "badge-risk", "Crisis": "badge-crisis", "Stable": "badge-stable", "Growth": "badge-growth", } return mapping.get(status, "badge-default") async def calc_risk_score(db: AsyncSession, no_archive: bool, no_power_backup: bool, no_regulations: bool, system_failures: bool, no_documentation: bool) -> float: return ( (await get_coefficient(db, "risk_score.no_archive", 25) if no_archive else 0) + (await get_coefficient(db, "risk_score.no_power_backup", 20) if no_power_backup else 0) + (await get_coefficient(db, "risk_score.no_regulations", 15) if no_regulations else 0) + (await get_coefficient(db, "risk_score.system_failures", 20) if system_failures else 0) + (await get_coefficient(db, "risk_score.no_documentation", 10) if no_documentation else 0) ) async def risk_label(db: AsyncSession, score: float) -> str: if score <= await get_coefficient(db, "risk_score.bound_low", 20): return "Низкий" if score <= await get_coefficient(db, "risk_score.bound_medium", 50): return "Средний" if score <= await get_coefficient(db, "risk_score.bound_high", 75): return "Высокий" return "Критический" async def risk_multiplier(db: AsyncSession, score: float) -> float: if score <= await get_coefficient(db, "risk_score.bound_low", 20): return await get_coefficient(db, "risk_multiplier.low", 1.0) if score <= await get_coefficient(db, "risk_score.bound_medium", 50): return await get_coefficient(db, "risk_multiplier.medium", 1.3) if score <= await get_coefficient(db, "risk_score.bound_high", 75): return await get_coefficient(db, "risk_multiplier.high", 1.6) return await get_coefficient(db, "risk_multiplier.critical", 2.0) async def calc_complexity(db: AsyncSession, cameras: int, access_points: int, fire_type: str, has_it: bool) -> float: if cameras <= 20: video = await get_coefficient(db, "complexity.video_0_20", 10) elif cameras <= 100: video = await get_coefficient(db, "complexity.video_20_100", 20) else: video = await get_coefficient(db, "complexity.video_100plus", 35) if access_points <= 5: access = await get_coefficient(db, "complexity.access_0_5", 10) elif access_points <= 20: access = await get_coefficient(db, "complexity.access_5_20", 20) else: access = await get_coefficient(db, "complexity.access_20plus", 30) if fire_type == "simple": fire = await get_coefficient(db, "complexity.fire_simple", 15) elif fire_type == "medium": fire = await get_coefficient(db, "complexity.fire_medium", 25) elif fire_type == "complex": fire = await get_coefficient(db, "complexity.fire_complex", 40) else: fire = 0 it = await get_coefficient(db, "complexity.it", 10) if has_it else 0 return video + access + fire + it def complexity_label(score: float) -> str: if score <= 30: return "Простой" if score <= 70: return "Средний" if score <= 120: return "Сложный" return "Enterprise" async def calc_infrastructure_load(db: AsyncSession, server_state: str, network_state: str, power_state: str) -> float: if server_state == "none": server = await get_coefficient(db, "infra.server_none", 15) elif server_state == "weak": server = await get_coefficient(db, "infra.server_weak", 10) else: server = 0 if network_state == "unstable": network = await get_coefficient(db, "infra.network_unstable", 20) elif network_state == "partial": network = await get_coefficient(db, "infra.network_partial", 10) else: network = 0 if power_state == "none": power = await get_coefficient(db, "infra.power_none", 20) elif power_state == "weak": power = await get_coefficient(db, "infra.power_weak", 10) else: power = 0 return server + network + power async def calc_service_history(db: AsyncSession, service_state: str) -> float: mapping = { "none": "history.none", "irregular": "history.irregular", "formal": "history.formal", "sla": "history.sla", } key = mapping.get(service_state, "history.none") return await get_coefficient(db, key, 30) async def calc_object_index(db: AsyncSession, risk: float, complexity: float, infra: float, history: float) -> float: return round( risk * await get_coefficient(db, "object_index.risk_weight", 0.4) + complexity * await get_coefficient(db, "object_index.complexity_weight", 0.3) + infra * await get_coefficient(db, "object_index.infra_weight", 0.2) + history * await get_coefficient(db, "object_index.history_weight", 0.1), 2, ) def object_class(index: float) -> str: if index <= 30: return "A" if index <= 60: return "B" if index <= 90: return "C" return "D" def object_class_label(index: float) -> str: cls = object_class(index) labels = { "A": "Лёгкий SLA", "B": "Стандарт SLA", "C": "Сложный SLA", "D": "Enterprise / Высокий риск", } return labels.get(cls, "") def calc_sla_price(base_cost: float, object_index: float, region_factor: float, risk_mult: float) -> float: return round(base_cost * object_index * region_factor * risk_mult, 2) async def calc_engineer_score(db: AsyncSession, sla_compliance: float, response_time_score: float, resolution_time_score: float, diagnosis_accuracy: float, reopen_rate_score: float, risk_coverage_score: float) -> float: return round( (await get_coefficient(db, "engineer.score.sla_weight", 0.25) * sla_compliance) + (await get_coefficient(db, "engineer.score.response_weight", 0.20) * response_time_score) + (await get_coefficient(db, "engineer.score.resolution_weight", 0.20) * resolution_time_score) + (await get_coefficient(db, "engineer.score.diagnosis_weight", 0.15) * diagnosis_accuracy) + (await get_coefficient(db, "engineer.score.reopen_weight", 0.10) * reopen_rate_score) + (await get_coefficient(db, "engineer.score.risk_coverage_weight", 0.10) * risk_coverage_score), 2, ) async def engineer_grade(db: AsyncSession, score: float) -> str: if score >= await get_coefficient(db, "engineer.grade.senior", 90): return "Senior" if score >= await get_coefficient(db, "engineer.grade.strong", 80): return "Strong" if score >= await get_coefficient(db, "engineer.grade.middle", 70): return "Middle" return "Junior" async def calc_engineer_control_score(db: AsyncSession, sla_control: float, task_dist: float, incident_red: float, team_perf: float, response_coord: float) -> float: return round( (await get_coefficient(db, "ecs.sla_weight", 0.30) * sla_control) + (await get_coefficient(db, "ecs.task_dist_weight", 0.25) * task_dist) + (await get_coefficient(db, "ecs.incident_red_weight", 0.20) * incident_red) + (await get_coefficient(db, "ecs.team_perf_weight", 0.15) * team_perf) + (await get_coefficient(db, "ecs.response_coord_weight", 0.10) * response_coord), 2, ) async def calc_shs(db: AsyncSession, sla_stability: float, revenue_stability: float, retention: float, engineer_performance: float, incident_stability: float, sales_flow: float, operational_efficiency: float) -> float: return round( (await get_coefficient(db, "shs.sla_stability_weight", 0.22) * sla_stability) + (await get_coefficient(db, "shs.revenue_stability_weight", 0.18) * revenue_stability) + (await get_coefficient(db, "shs.retention_weight", 0.18) * retention) + (await get_coefficient(db, "shs.engineer_perf_weight", 0.15) * engineer_performance) + (await get_coefficient(db, "shs.incident_stability_weight", 0.12) * incident_stability) + (await get_coefficient(db, "shs.sales_flow_weight", 0.10) * sales_flow) + (await get_coefficient(db, "shs.operational_eff_weight", 0.05) * operational_efficiency), 2, ) async def calc_ceo_shs(db: AsyncSession, mrr_growth: float, sla_compliance: float, retention: float, productivity: float, conversion: float, incident_stability: float) -> float: return round( (await get_coefficient(db, "ceo_shs.mrr_weight", 0.25) * mrr_growth) + (await get_coefficient(db, "ceo_shs.sla_weight", 0.20) * sla_compliance) + (await get_coefficient(db, "ceo_shs.retention_weight", 0.20) * retention) + (await get_coefficient(db, "ceo_shs.productivity_weight", 0.15) * productivity) + (await get_coefficient(db, "ceo_shs.conversion_weight", 0.10) * conversion) + (await get_coefficient(db, "ceo_shs.incident_weight", 0.10) * incident_stability), 2, ) async def shs_status(db: AsyncSession, score: float) -> str: if score >= await get_coefficient(db, "shs.zone_growth", 85): return "Growth" if score >= await get_coefficient(db, "shs.zone_stable", 70): return "Stable" if score >= await get_coefficient(db, "shs.zone_risk", 50): return "Risk" return "Crisis" async def shs_zone(db: AsyncSession, score: float) -> dict: if score >= await get_coefficient(db, "shs.zone_growth", 85): return {"zone": "green", "label": "Growth", "action": ""} if score >= await get_coefficient(db, "shs.zone_stable", 70): return {"zone": "yellow", "label": "Stable", "action": "Мониторинг"} if score >= await get_coefficient(db, "shs.zone_risk", 50): return {"zone": "yellow", "label": "Risk", "action": "Снизить продажи, запустить аудит цикла, пересмотреть назначения"} return {"zone": "red", "label": "Crisis", "action": "Немедленный аудит, приостановка новых продаж, пересмотр команды"} async def shs_delta(db: AsyncSession, current_score: float, previous_score: float | None) -> dict | None: if previous_score is None: return None delta = current_score - previous_score threshold_warn = await get_coefficient(db, "shs.delta_warning", 5) threshold_crit = await get_coefficient(db, "shs.delta_critical", 10) if delta <= -threshold_crit: return {"delta": round(delta, 1), "level": "critical", "message": f"SHS упал более чем на {threshold_crit} пунктов — кризис"} if delta <= -threshold_warn: return {"delta": round(delta, 1), "level": "warning", "message": f"SHS упал более чем на {threshold_warn} пунктов — требуется внимание"} return {"delta": round(delta, 1), "level": "ok", "message": ""} async def calc_sla_stability_index(db: AsyncSession, sla_compliance: float, breach_severity_weighted: float) -> float: ssi = sla_compliance * (1 - breach_severity_weighted) return round(max(0, min(100, ssi)), 2) async def calc_breach_severity(db: AsyncSession, breaches_p1: int, breaches_p2: int, breaches_p3: int) -> float: total = breaches_p1 + breaches_p2 + breaches_p3 if total == 0: return 0 weighted = ( breaches_p1 * await get_coefficient(db, "ssi.breach_severity_p1", 1.0) + breaches_p2 * await get_coefficient(db, "ssi.breach_severity_p2", 0.5) + breaches_p3 * await get_coefficient(db, "ssi.breach_severity_p3", 0.2) ) return min(1, weighted / total) def calc_revenue_stability_index(mrr_history: list[float]) -> float: count = len(mrr_history) if count < 2: return 100 mean = sum(mrr_history) / count if mean <= 0: return 100 variance = sum((mrr - mean) ** 2 for mrr in mrr_history) / count std_dev = math.sqrt(variance) cv = std_dev / mean return round(max(0, min(100, 100 * (1 - cv))), 2) def calc_sales_flow_index(audits: int, sla_contracts: int, lead_quality: float = 1.0) -> float: if audits <= 0: return 0 conversion = sla_contracts / audits return round(min(100, conversion * 100 * lead_quality), 2) def calc_operational_efficiency_index(revenue: float, engineer_hours: float, cost_per_hour: float) -> float: total = engineer_hours * cost_per_hour if total <= 0: return 100 return round(min(100, (revenue / total) * 100), 2) async def calc_incident_stability_index(db: AsyncSession, incidents_p1: int, incidents_p2: int, incidents_p3: int, object_count: int) -> float: if object_count <= 0: return 100 weighted = ( incidents_p1 * await get_coefficient(db, "isi.severity_p1", 1.0) + incidents_p2 * await get_coefficient(db, "isi.severity_p2", 0.5) + incidents_p3 * await get_coefficient(db, "isi.severity_p3", 0.2) ) rate = weighted / object_count return round(max(0, min(100, 100 - (rate * 20))), 2) async def check_automation_rules(db: AsyncSession, shs: float | None, sla_compliance: float | None, retention: float | None) -> list[dict]: alerts = [] if shs is not None and shs < await get_coefficient(db, "rules.shs_threshold", 70): alerts.append({ "rule": "SHS", "severity": "critical", "message": f"SHS ниже {await get_coefficient(db, 'rules.shs_threshold', 70)} — снизить продажи, запустить аудит, пересмотреть назначения инженеров.", }) if sla_compliance is not None and sla_compliance < await get_coefficient(db, "rules.sla_threshold", 90): alerts.append({ "rule": "SLA", "severity": "warning", "message": f"SLA Compliance ниже {await get_coefficient(db, 'rules.sla_threshold', 90)}% — заморозить неприоритетные проекты, увеличить частоту проверок.", }) if retention is not None and retention < await get_coefficient(db, "rules.retention_threshold", 90): alerts.append({ "rule": "Retention", "severity": "critical", "message": f"Удержание клиентов ниже {await get_coefficient(db, 'rules.retention_threshold', 90)}% — обязательный аудит клиентов, пересмотр назначений инженеров.", }) return alerts def calc_sla_compliance(closed_in_sla: int, total: int) -> float: if total <= 0: return 100 return round((closed_in_sla / total) * 100, 2) def calc_reopen_rate(reopened: int, total: int) -> float: if total <= 0: return 0 return round((reopened / total) * 100, 2) def calc_diagnosis_accuracy(confirmed: int, found: int) -> float: if found <= 0: return 100 return round((confirmed / found) * 100, 2) def calc_response_time_score(actual_hours: float, sla_hours: float) -> float: if sla_hours <= 0: return 100 ratio = actual_hours / sla_hours if ratio <= 1.0: return 100 if ratio <= 1.2: return 80 return max(0, 100 - (ratio - 1.2) * 50) def calc_resolution_time_score(actual_hours: float, norm_hours: float) -> float: if norm_hours <= 0: return 100 ratio = actual_hours / norm_hours if ratio <= 1.0: return 100 if ratio <= 1.5: return 70 return max(0, 100 - (ratio - 1.5) * 40) def calc_risk_coverage(checked_zones: int, total_zones: int = 5) -> float: if total_zones <= 0: return 0 return round((checked_zones / total_zones) * 100, 2) def calc_documentation_quality(full_reports: int, total_visits: int) -> float: if total_visits <= 0: return 100 return round((full_reports / total_visits) * 100, 2) def calc_tlb(tasks_per_engineer: list[float]) -> float: count = len(tasks_per_engineer) if count < 2: return 0 mean = sum(tasks_per_engineer) / count variance = sum((t - mean) ** 2 for t in tasks_per_engineer) / count return round(math.sqrt(variance), 2) def calc_mrr(total_sla_annual: float) -> float: return round(total_sla_annual / 12, 2) def calc_rpe(sla_revenue: float, engineer_count: int) -> float: if engineer_count <= 0: return 0 return round(sla_revenue / engineer_count, 2) def calc_cpo(total_cost: float, object_count: int) -> float: if object_count <= 0: return 0 return round(total_cost / object_count, 2) def calc_arpc(total_revenue: float, client_count: int) -> float: if client_count <= 0: return 0 return round(total_revenue / client_count, 2) def calc_utilization(working_hours: int, available_hours: int) -> float: if available_hours <= 0: return 0 return round((working_hours / available_hours) * 100, 2) def calc_autonomy(independent: int, total: int) -> float: if total <= 0: return 100 return round((independent / total) * 100, 2) def calc_escalation_rate(escalated: int, total: int) -> float: if total <= 0: return 0 return round((escalated / total) * 100, 2) def calc_uptime(uptime_hours: float, total_hours: float) -> float: if total_hours <= 0: return 100 return round((uptime_hours / total_hours) * 100, 2) def calc_retention(active_clients: int, total_clients: int) -> float: if total_clients <= 0: return 100 return round((active_clients / total_clients) * 100, 2) async def calc_retention_key_client(db: AsyncSession, base_retention: float, lost_key_clients: int) -> float: penalty = lost_key_clients * await get_coefficient(db, "retention.key_client_penalty", 0.1) return round(max(0, base_retention - penalty * 100), 2) def calc_csat(positive: int, total: int) -> float: if total <= 0: return 100 return round((positive / total) * 100, 2) def calc_incident_rate(incidents: int, object_count: int) -> float: if object_count <= 0: return 0 return round(incidents / object_count, 2) def calc_incident_reduction(previous: int, current: int) -> float: if previous <= 0: return 0 return round(((previous - current) / previous) * 100, 2) async def calc_bonus_percent(db: AsyncSession, engineer_score: float) -> float: if engineer_score >= await get_coefficient(db, "bonus.score_90plus", 90): return await get_coefficient(db, "bonus.premium_90plus", 20) if engineer_score >= await get_coefficient(db, "bonus.score_80_89", 80): return await get_coefficient(db, "bonus.premium_80_89", 10) return await get_coefficient(db, "bonus.premium_below_80", 0)