deystvuy-kak-senior-full-sta/app.py

732 lines
27 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""
BI Dashboard для оценки эффективности автобаз
Корпоративный сервис с авторизацией, RBAC и расчётом KPI
"""
import streamlit as st
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
from datetime import datetime, timedelta
from typing import Dict, Optional
import os
from dotenv import load_dotenv
from data_fetcher import MockDataFetcher, GoogleSheetsFetcher, DataFetcher
# Загрузка переменных окружения
load_dotenv()
# ============================================================================
# КОНФИГУРАЦИЯ СТРАНИЦЫ
# ============================================================================
st.set_page_config(
page_title="BI-дашборд Автобазы",
page_icon="🚛",
layout="wide",
initial_sidebar_state="expanded"
)
# ============================================================================
# МЕНЕДЖЕР СЕССИИ И АВТОРИЗАЦИЯ
# ============================================================================
def init_session_state():
"""Инициализация состояния сессии"""
if 'authenticated' not in st.session_state:
st.session_state.authenticated = False
if 'user_role' not in st.session_state:
st.session_state.user_role = None
if 'username' not in st.session_state:
st.session_state.username = None
if 'data_fetcher' not in st.session_state:
st.session_state.data_fetcher = None
if 'cache_timestamp' not in st.session_state:
st.session_state.cache_timestamp = None
def check_credentials(username: str, password: str) -> Optional[Dict]:
"""
Проверка учётных данных.
В продакшене заменить на проверку через БД/LDAP.
Returns:
Dict с ролью пользователя или None
"""
# Хардкод для демо (в продакшене - БД)
users = {
os.getenv('ADMIN_LOGIN', 'admin'): {
'password': os.getenv('ADMIN_PASSWORD', 'admin123'),
'role': 'admin',
'name': 'Администратор системы',
'permissions': ['view_all', 'export', 'manage_users', 'view_finance']
},
os.getenv('MANAGER_LOGIN', 'manager'): {
'password': os.getenv('MANAGER_PASSWORD', 'manager123'),
'role': 'manager',
'name': 'Руководитель автобазы',
'permissions': ['view_all', 'export', 'view_finance']
},
'driver': {
'password': 'driver123',
'role': 'driver',
'name': 'Водитель',
'permissions': ['view_own']
},
'accountant': {
'password': 'accountant123',
'role': 'accountant',
'name': 'Бухгалтер',
'permissions': ['view_finance', 'export']
}
}
user = users.get(username)
if user and user['password'] == password:
return user
return None
def login_page():
"""Страница авторизации"""
st.markdown("""
<style>
.login-box {
background: #f8f9fa;
padding: 2rem;
border-radius: 10px;
box-shadow: 0 4px 6px rgba(0,0,0,0.1);
max-width: 400px;
margin: 2rem auto;
}
.stButton>button {
width: 100%;
}
</style>
""", unsafe_allow_html=True)
st.markdown("<h1 style='text-align: center;'>🚛 BI-дашборд Автобазы</h1>", unsafe_allow_html=True)
st.markdown("<p style='text-align: center; color: #666;'>Система мониторинга эффективности транспортных подразделений</p>", unsafe_allow_html=True)
with st.form("login_form"):
username = st.text_input("Логин", placeholder="admin")
password = st.text_input("Пароль", type="password", placeholder="••••••••")
submitted = st.form_submit_button("Войти", use_container_width=True)
if submitted:
if not username or not password:
st.error("Введите логин и пароль")
else:
user = check_credentials(username, password)
if user:
st.session_state.authenticated = True
st.session_state.user_role = user['role']
st.session_state.username = user['name']
st.session_state.permissions = user['permissions']
st.rerun()
else:
st.error("Неверный логин или пароль")
# Демонстрационные учётные данные
with st.expander("📋 Тестовые учётные данные"):
st.markdown("""
| Логин | Пароль | Роль | Права |
|-------|--------|------|-------|
| admin | admin123 | Администратор | Полный доступ |
| manager | manager123 | Руководитель | Просмотр + экспорт |
| accountant | accountant123 | Бухгалтер | Финансы + экспорт |
| driver | driver123 | Водитель | Только свои данные |
""")
# ============================================================================
# БИЗНЕС-ЛОГИКА: РАСЧЁТ KPI
# ============================================================================
class KPICalculator:
"""Калькулятор ключевых показателей эффективности"""
def __init__(self, data_fetcher: DataFetcher):
self.fetcher = data_fetcher
self._cache = {}
def load_all_data(self):
"""Загрузить все данные в кэш"""
self._cache = {
'fleet': self.fetcher.get_fleet_data(),
'requests': self.fetcher.get_requests_data(),
'accidents': self.fetcher.get_accidents_data(),
'maintenance': self.fetcher.get_maintenance_data(),
'fuel': self.fetcher.get_fuel_data(),
'employees': self.fetcher.get_employees_data()
}
return self._cache
def calc_request_fulfillment_rate(self, department: Optional[str] = None) -> float:
"""
Коэффициент исполнения заявок.
Formula: Исполнено / Всего заявок * 100%
"""
df = self._cache['requests']
if department:
df = df[df['department'] == department]
if len(df) == 0:
return 0.0
return round(df['fulfilled'].mean() * 100, 2)
def calc_fleet_utilization(self, department: Optional[str] = None) -> float:
"""
Коэффициент использования парка (КИП).
Formula: (Всего ТС - В ремонте) / Всего ТС * 100%
"""
df = self._cache['fleet']
if department:
df = df[df['department'] == department]
if len(df) == 0:
return 0.0
active = len(df[df['status'] != 'В ремонте'])
total = len(df)
return round((active / total) * 100, 2)
def calc_safety_score(self, department: Optional[str] = None) -> float:
"""
Показатель безопасности.
Formula: 100 - (ДТП * 5 + Нарушения * 2) / Всего ТС * 10
"""
df_fleet = self._cache['fleet']
if department:
df_fleet = df_fleet[df_fleet['department'] == department]
df_accidents = self._cache['accidents']
if department:
# Нужно связать ТС с подразделением
vehicle_ids = df_fleet['vehicle_id'].tolist()
df_accidents = df_accidents[df_accidents['vehicle_id'].isin(vehicle_ids)]
total_vehicles = len(df_fleet)
if total_vehicles == 0:
return 100.0
accidents_count = len(df_accidents[df_accidents['type'] == 'ДТП'])
violations_count = len(df_accidents[df_accidents['type'] == 'Нарушение ПДД'])
score = 100 - ((accidents_count * 5 + violations_count * 2) / total_vehicles * 10)
return round(max(0, min(100, score)), 2)
def calc_maintenance_score(self, department: Optional[str] = None) -> float:
"""
Показатель эксплуатации и ремонтов.
Formula: 100 - (Простой дней / Всего ТС) * 2
"""
df_fleet = self._cache['fleet']
if department:
df_fleet = df_fleet[df_fleet['department'] == department]
df_maint = self._cache['maintenance']
if department:
vehicle_ids = df_fleet['vehicle_id'].tolist()
df_maint = df_maint[df_maint['vehicle_id'].isin(vehicle_ids)]
total_vehicles = len(df_fleet)
if total_vehicles == 0:
return 100.0
total_downtime = df_maint['downtime_days'].sum()
score = 100 - (total_downtime / total_vehicles * 2)
return round(max(0, min(100, score)), 2)
def calc_fuel_efficiency(self, department: Optional[str] = None) -> Dict:
"""
Показатели финансов и ГСМ.
Returns: dict с перерасходом и стоимостью 1 км
"""
df_fleet = self._cache['fleet']
if department:
df_fleet = df_fleet[df_fleet['department'] == department]
df_fuel = self._cache['fuel']
if department:
vehicle_ids = df_fleet['vehicle_id'].tolist()
df_fuel = df_fuel[df_fuel['vehicle_id'].isin(vehicle_ids)]
if len(df_fuel) == 0:
return {'overrun_percent': 0.0, 'cost_per_km': 0.0}
planned = df_fuel['planned_cost'].sum()
actual = df_fuel['actual_cost'].sum()
total_mileage = df_fleet['mileage'].sum()
overrun = ((actual - planned) / planned * 100) if planned > 0 else 0
cost_per_km = (actual / total_mileage * 1000) if total_mileage > 0 else 0
return {
'overrun_percent': round(overrun, 2),
'cost_per_km': round(cost_per_km, 2)
}
def calc_department_rating(self) -> pd.DataFrame:
"""
Рассчитать сводный рейтинг подразделений.
Weighted score:
- Исполнение заявок: 25%
- КИП: 20%
- Безопасность: 20%
- Эксплуатация: 15%
- Финансы: 20%
"""
departments = self._cache['fleet']['department'].unique()
ratings = []
for dept in departments:
kpi1 = self.calc_request_fulfillment_rate(dept)
kpi2 = self.calc_fleet_utilization(dept)
kpi3 = self.calc_safety_score(dept)
kpi4 = self.calc_maintenance_score(dept)
kpi5 = self.calc_fuel_efficiency(dept)
# Нормализуем KPI5 (перерасход) в score (0-100)
fuel_score = max(0, 100 - abs(kpi5['overrun_percent']) * 2)
# Взвешенная оценка
weighted_score = (
kpi1 * 0.25 +
kpi2 * 0.20 +
kpi3 * 0.20 +
kpi4 * 0.15 +
fuel_score * 0.20
)
ratings.append({
'Подразделение': dept,
'Исполнение заявок': kpi1,
'КИП': kpi2,
'Безопасность': kpi3,
'Эксплуатация': kpi4,
'ГСМ перерасход %': kpi5['overrun_percent'],
'Стоимость 1 км': kpi5['cost_per_km'],
'Сводный рейтинг': round(weighted_score, 2)
})
df_rating = pd.DataFrame(ratings)
df_rating = df_rating.sort_values('Сводный рейтинг', ascending=False)
df_rating['Место'] = range(1, len(df_rating) + 1)
return df_rating
def get_budget_report(self, year: int = 2027) -> Dict:
"""
Сформировать отчёт для обоснования бюджета.
"""
df_fleet = self._cache['fleet']
df_maint = self._cache['maintenance']
df_fuel = self._cache['fuel']
df_requests = self._cache['requests']
# Текущие расходы
total_maintenance = df_maint['cost'].sum()
total_fuel = df_fuel['actual_cost'].sum()
total_fines = self._cache['accidents']['fine_amount'].sum()
# Прогноз на 2027
vehicles_count = len(df_fleet)
avg_age = datetime.now().year - df_fleet['year'].median()
# Коэффициенты индексации
inflation = 1.12 # 12% инфляция
wear_factor = 1 + (avg_age * 0.05) # Износ
forecast = {
'Текущие расходы ТО': total_maintenance,
'Текущие расходы ГСМ': total_fuel,
'Штрафы и инциденты': total_fines,
'Всего за 2024': total_maintenance + total_fuel + total_fines,
'Прогноз ТО 2027': int(total_maintenance * inflation * wear_factor),
'Прогноз ГСМ 2027': int(total_fuel * inflation),
'Рекомендуемый бюджет': int((total_maintenance + total_fuel) * inflation * wear_factor),
'Автопарк': vehicles_count,
'Средний возраст': round(avg_age, 1),
'Потребность в обновлении': int(vehicles_count * 0.15) # 15% парка
}
return forecast
# ============================================================================
# ВИЗУАЛИЗАЦИЯ
# ============================================================================
def render_rating_table(rating_df: pd.DataFrame):
"""Таблица рейтинга с цветовым кодированием"""
def color_score(val):
if val >= 80:
color = '#d4edda' # Зелёный
elif val >= 60:
color = '#fff3cd' # Жёлтый
else:
color = '#f8d7da' # Красный
return f'background-color: {color}'
styled = rating_df.style.applymap(
color_score,
subset=['Исполнение заявок', 'КИП', 'Безопасность', 'Эксплуатация', 'Сводный рейтинг']
)
st.dataframe(styled, use_container_width=True, hide_index=True)
def render_kpi_cards(calculator: KPICalculator):
"""Карточки общих KPI"""
cols = st.columns(5)
kpi1 = calculator.calc_request_fulfillment_rate()
kpi2 = calculator.calc_fleet_utilization()
kpi3 = calculator.calc_safety_score()
kpi4 = calculator.calc_maintenance_score()
kpi5 = calculator.calc_fuel_efficiency()
with cols[0]:
st.metric(
label="📋 Исполнение заявок",
value=f"{kpi1}%",
delta=f"{kpi1 - 85:.1f}%",
delta_color="normal"
)
with cols[1]:
st.metric(
label="🚛 КИП",
value=f"{kpi2}%",
delta=f"{kpi2 - 80:.1f}%",
delta_color="normal"
)
with cols[2]:
st.metric(
label="🛡 Безопасность",
value=f"{kpi3}",
delta=f"{kpi3 - 75:.1f}",
delta_color="normal"
)
with cols[3]:
st.metric(
label="🔧 Эксплуатация",
value=f"{kpi4}",
delta=f"{kpi4 - 80:.1f}",
delta_color="normal"
)
with cols[4]:
overrun = kpi5['overrun_percent']
st.metric(
label="⛽ Перерасход ГСМ",
value=f"{overrun}%",
delta=f"{overrun:.1f}%",
delta_color="inverse" if overrun > 0 else "normal"
)
def render_charts(calculator: KPICalculator):
"""Графики и диаграммы"""
rating_df = calculator.calc_department_rating()
col1, col2 = st.columns(2)
with col1:
# Рейтинг подразделений
fig_bar = px.bar(
rating_df.sort_values('Сводный рейтинг'),
x='Сводный рейтинг',
y='Подразделение',
orientation='h',
title='📊 Сводный рейтинг подразделений',
color='Сводный рейтинг',
color_continuous_scale='RdYlGn'
)
fig_bar.update_layout(height=400)
st.plotly_chart(fig_bar, use_container_width=True)
with col2:
# Радарная диаграмма KPI
departments = rating_df['Подразделение'].tolist()
fig_radar = go.Figure()
for dept in departments:
dept_data = rating_df[rating_df['Подразделение'] == dept].iloc[0]
fig_radar.add_trace(go.Scatterpolar(
r=[
dept_data['Исполнение заявок'],
dept_data['КИП'],
dept_data['Безопасность'],
dept_data['Эксплуатация'],
max(0, 100 - abs(dept_data['ГСМ перерасход %']) * 2)
],
theta=['Исполнение', 'КИП', 'Безопасность', 'Эксплуатация', 'ГСМ'],
fill='toself',
name=dept
))
fig_radar.update_layout(
polar=dict(radialaxis=dict(visible=True, range=[0, 100])),
showlegend=True,
title='🎯 Сравнение KPI подразделений',
height=400
)
st.plotly_chart(fig_radar, use_container_width=True)
# Динамика по месяцам
col3, col4 = st.columns(2)
with col3:
# Структура расходов
df_fuel = calculator._cache['fuel']
monthly_fuel = df_fuel.groupby(df_fuel['date'].str[:7])['actual_cost'].sum()
fig_line = px.line(
x=monthly_fuel.index,
y=monthly_fuel.values,
title='💰 Динамика расходов на ГСМ по месяцам',
markers=True
)
fig_line.update_layout(height=350)
st.plotly_chart(fig_line, use_container_width=True)
with col4:
# Распределение типов ТС
df_fleet = calculator._cache['fleet']
vehicle_types = df_fleet['vehicle_type'].value_counts()
fig_pie = px.pie(
values=vehicle_types.values,
names=vehicle_types.index,
title='🚚 Структура автопарка',
hole=0.4
)
fig_pie.update_layout(height=350)
st.plotly_chart(fig_pie, use_container_width=True)
def render_budget_report(calculator: KPICalculator):
"""Отчёт для обоснования бюджета 2027"""
st.header("📈 Отчёт для обоснования бюджета на 2027 год")
forecast = calculator.get_budget_report(2027)
# Карточки с цифрами
col1, col2, col3, col4 = st.columns(4)
with col1:
st.metric(
label="Всего за 2024",
value=f"{forecast['Всего за 2024']:,.0f} ₽"
)
with col2:
st.metric(
label="Рекомендуемый бюджет 2027",
value=f"{forecast['Рекомендуемый бюджет']:,.0f} ₽"
)
with col3:
st.metric(
label="Автопарк",
value=f"{forecast['Автопарк']} ТС"
)
with col4:
st.metric(
label="Обновление парка",
value=f"{forecast['Потребность в обновлении']} ТС"
)
# Детализация
st.subheader("Структура расходов")
budget_data = pd.DataFrame({
'Статья расходов': [
'ТО и ремонты (прогноз)',
'ГСМ (прогноз)',
'Резерв на инциденты',
'Обновление парка'
],
'Сумма (₽)': [
forecast['Прогноз ТО 2027'],
forecast['Прогноз ГСМ 2027'],
int(forecast['Штрафы и инциденты'] * 1.12),
forecast['Потребность в обновлении'] * 2500000
]
})
st.dataframe(budget_data, use_container_width=True, hide_index=True)
# Экспорт
csv = budget_data.to_csv(index=False, sep=';', decimal=',').encode('utf-8-sig')
st.download_button(
label="📥 Скачать отчёт (CSV)",
data=csv,
file_name=f"budget_forecast_2027_{datetime.now().strftime('%Y%m%d')}.csv",
mime="text/csv"
)
def render_footer():
"""Подвал с политикой конфиденциальности"""
st.markdown("---")
st.markdown("""
<div style='
background: #fff3cd;
padding: 1rem;
border-radius: 5px;
border-left: 4px solid #ffc107;
margin-top: 2rem;
font-size: 0.85rem;
'>
<strong>⚠️ Внимание:</strong> информация носит строго конфиденциальный характер.
Я передаю данные для обработки, но требую хранить их в безопасности и не передавать третьим лицам
согласно закону. Запрещено размещение в открытом доступе.
</div>
""", unsafe_allow_html=True)
st.markdown("""
<div style='text-align: center; color: #666; font-size: 0.8rem; margin-top: 1rem;'>
BI-дашборд Автобазы v1.0 | Внутренний корпоративный сервис | 2024
</div>
""", unsafe_allow_html=True)
# ============================================================================
# ГЛАВНОЕ ПРИЛОЖЕНИЕ
# ============================================================================
def main():
init_session_state()
# Проверка авторизации
if not st.session_state.authenticated:
login_page()
render_footer()
return
# Боковая панель
with st.sidebar:
st.markdown(f"👤 **{st.session_state.username}**")
st.markdown(f"Роль: {st.session_state.user_role}")
st.markdown("---")
if st.button("🚪 Выйти"):
st.session_state.authenticated = False
st.session_state.user_role = None
st.session_state.username = None
st.rerun()
st.markdown("---")
st.markdown("### 📊 Разделы")
# Инициализация данных
if st.session_state.data_fetcher is None:
# Используем mock-данные для демо
st.session_state.data_fetcher = MockDataFetcher()
st.session_state.calculator = KPICalculator(st.session_state.data_fetcher)
st.session_state.calculator.load_all_data()
calculator = st.session_state.calculator
# Навигация
pages = {
"📈 Дашборд": "dashboard",
"🏆 Рейтинг": "rating",
"💰 Бюджет 2027": "budget"
}
# Фильтрация меню по ролям
if 'view_finance' not in st.session_state.permissions:
pages.pop("💰 Бюджет 2027", None)
selected = st.radio("Перейти к:", list(pages.keys()), index=0, label_visibility="collapsed")
current_page = pages[selected]
st.markdown("---")
st.markdown("### 🔧 Настройки")
# Выбор источника данных
data_source = st.selectbox(
"Источник данных:",
["Mock (демо)", "Google Sheets", "API (ERP)"],
index=0
)
if data_source == "Google Sheets":
st.warning("Требуется настройка service_account.json")
st.markdown("---")
st.markdown(f"Последнее обновление: {datetime.now().strftime('%d.%m.%Y %H:%M')}")
# Основной контент
if current_page == "dashboard":
st.title("📈 Дашборд эффективности автобаз")
# KPI карточки
render_kpi_cards(calculator)
st.markdown("---")
# Графики
render_charts(calculator)
# Детальные данные
with st.expander("📋 Детальные данные по автопарку"):
st.dataframe(calculator._cache['fleet'], use_container_width=True)
with st.expander("📊 Статистика по заявкам"):
st.dataframe(calculator._cache['requests'], use_container_width=True)
elif current_page == "rating":
st.title("🏆 Рейтинг подразделений")
# Сводная таблица
rating_df = calculator.calc_department_rating()
render_rating_table(rating_df)
st.markdown("---")
# Детализация по выбранному подразделению
st.subheader("🔍 Детализация по подразделению")
selected_dept = st.selectbox(
"Выберите подразделение:",
rating_df['Подразделение'].tolist()
)
dept_data = rating_df[rating_df['Подразделение'] == selected_dept].iloc[0]
col1, col2, col3 = st.columns(3)
with col1:
st.metric("Исполнение заявок", f"{dept_data['Исполнение заявок']}%")
st.metric("КИП", f"{dept_data['КИП']}%")
with col2:
st.metric("Безопасность", f"{dept_data['Безопасность']}")
st.metric("Эксплуатация", f"{dept_data['Эксплуатация']}")
with col3:
st.metric("Перерасход ГСМ", f"{dept_data['ГСМ перерасход %']}%")
st.metric("Стоимость 1 км", f"{dept_data['Стоимость 1 км']} ₽")
elif current_page == "budget":
if 'view_finance' not in st.session_state.permissions:
st.error("⛔ Недостаточно прав для просмотра финансового раздела")
return
render_budget_report(calculator)
# Подвал на всех страницах
render_footer()
if __name__ == "__main__":
main()