models.py - master - woodoo/web_complex_manager

models.py
422 строки16 KB
woodoo
upload files
25 ноя 2025, 13:35
25 ноя 2025, 13:35d517a88
from flask import Flask, render_template, request, jsonify, send_file
import mysql.connector
from mysql.connector import Error
import json
from datetime import datetime, timedelta
import folium
import tempfile
import os
import math
import io

app = Flask(__name__)
app.secret_key = 'complex-maintenance-secret-key'

# Конфигурация БД
DB_CONFIG = {
    'host': 'localhost',
    'database': 'complex_manager',
    'user': 'root',
    'password': 'rus1111LAN',
    'auth_plugin': 'mysql_native_password'
}

class MaintenanceScheduler:
    def __init__(self):
        self.complexes = []
        self.schedule = {}
        self.maintenance_types = {
            'КОРДОН': ['ТО1 (еженедельное)', 'ТО2 (ежемесячное)'],
            'Остальные': ['ТО1 (ежемесячное)', 'ТО2 (ежеквартальное)', 'ТО3 (ежегодное)']
        }
        self.start_point = (54.777479, 32.063765)
    
    def get_db_connection(self):
        try:
            return mysql.connector.connect(**DB_CONFIG)
        except Error as e:
            print(f"Ошибка подключения к БД: {e}")
            return None
    
    def parse_coordinates(self, coord_str):
        """Парсинг координат из строки формата '54.7680,32.0126'"""
        try:
            if not coord_str or coord_str == 'None':
                return None
            
            # Убираем пробелы и переносы строк
            coord_str = coord_str.replace('\n', '').replace('\r', '').strip()
            
            # Разделяем по запятой
            if ',' in coord_str:
                parts = coord_str.split(',')
                if len(parts) == 2:
                    lat = float(parts[0].strip())
                    lon = float(parts[1].strip())
                    return (lat, lon)
            return None
        except Exception as e:
            print(f"Ошибка парсинга координат '{coord_str}': {e}")
            return None
    
    def load_complexes_from_db(self):
        try:
            connection = self.get_db_connection()
            if connection is None:
                return False
            
            cursor = connection.cursor(dictionary=True)
            
            # Запрос для получения активных комплексов (не архивных)
            query = """
            SELECT 
                id,
                number,
                name,
                location,
                coordinates,
                complex_type,
                is_archived
            FROM complexes 
            WHERE is_archived = 0
            ORDER BY number
            """
            
            cursor.execute(query)
            complexes_data = cursor.fetchall()
            
            self.complexes = []
            for row in complexes_data:
                coordinates = self.parse_coordinates(row['coordinates'])
                
                complex_data = {
                    'id': row['id'],
                    'number': row['number'],
                    'name': row['name'],
                    'location': row['location'],
                    'coordinates': coordinates,
                    'type': 'КОРДОН' if 'КОРДОН' in str(row['number']) else 'Остальные',
                    'maintenance_count': 0,
                    'is_archived': row['is_archived']
                }
                self.complexes.append(complex_data)
            
            # Загружаем историю обслуживаний для подсчета количества
            self.load_maintenance_history()
            
            cursor.close()
            connection.close()
            return True
            
        except Error as e:
            print(f"Ошибка загрузки данных: {e}")
            return False
    
    def load_maintenance_history(self):
        """Загружает историю обслуживаний для подсчета количества"""
        try:
            connection = self.get_db_connection()
            if connection is None:
                return
            
            cursor = connection.cursor()
            
            # Подсчитываем количество обслуживаний для каждого комплекса
            query = """
            SELECT complex_id, COUNT(*) as maintenance_count 
            FROM maintenance_schedule 
            GROUP BY complex_id
            """
            
            cursor.execute(query)
            maintenance_counts = cursor.fetchall()
            
            # Обновляем счетчики в комплексах
            for complex_id, count in maintenance_counts:
                for complex in self.complexes:
                    if complex['id'] == complex_id:
                        complex['maintenance_count'] = count
                        break
            
            cursor.close()
            connection.close()
            
        except Error as e:
            print(f"Ошибка загрузки истории обслуживаний: {e}")
    
    def save_maintenance_to_db(self, date_key, team_complexes, team_number):
        """Сохраняет обслуживание в базу данных"""
        try:
            connection = self.get_db_connection()
            if connection is None:
                return False
            
            cursor = connection.cursor()
            
            maintenance_date = datetime.strptime(date_key, '%Y-%m-%d').date()
            
            for complex_data in team_complexes:
                complex = self.complexes[complex_data['complex_index']]
                
                # Вставляем запись об обслуживании
                query = """
                INSERT INTO maintenance_schedule 
                (complex_id, maintenance_date, maintenance_type, description, created_at)
                VALUES (%s, %s, %s, %s, NOW())
                """
                
                description = f"Плановое обслуживание бригадой {team_number}"
                
                cursor.execute(query, (
                    complex['id'],
                    maintenance_date,
                    complex_data['maintenance_type'],
                    description
                ))
            
            connection.commit()
            cursor.close()
            connection.close()
            return True
            
        except Error as e:
            print(f"Ошибка сохранения в БД: {e}")
            return False

# Глобальный объект планировщика
scheduler = MaintenanceScheduler()

@app.route('/')
def index():
    """Главная страница с календарем"""
    # Загружаем данные при первом посещении
    if not scheduler.complexes:
        success = scheduler.load_complexes_from_db()
        if not success:
            return "Ошибка загрузки данных из БД", 500
    
    # Текущий месяц и год
    now = datetime.now()
    current_month = now.strftime("%m.%Y")
    
    # Генерируем календарь на текущий месяц
    calendar_data = generate_calendar(now.year, now.month)
    
    return render_template('index.html', 
                         current_month=current_month,
                         calendar_data=calendar_data,
                         complexes_count=len(scheduler.complexes))

@app.route('/api/complexes')
def get_complexes():
    """API для получения списка комплексов"""
    return jsonify(scheduler.complexes)

@app.route('/api/schedule/<date>')
def get_schedule(date):
    """API для получения расписания на день"""
    return jsonify(scheduler.schedule.get(date, {'team1': [], 'team2': []}))

@app.route('/api/schedule/<date>', methods=['POST'])
def save_schedule(date):
    """API для сохранения расписания на день"""
    try:
        data = request.json
        scheduler.schedule[date] = data
        
        # Сохраняем в базу данных
        success1 = scheduler.save_maintenance_to_db(date, data.get('team1', []), 1)
        success2 = scheduler.save_maintenance_to_db(date, data.get('team2', []), 2)
        
        if success1 and success2:
            # Перезагружаем данные для обновления счетчиков
            scheduler.load_complexes_from_db()
            return jsonify({'status': 'success', 'message': 'Расписание сохранено в БД'})
        else:
            return jsonify({'status': 'error', 'message': 'Ошибка сохранения в БД'}), 500
            
    except Exception as e:
        return jsonify({'status': 'error', 'message': str(e)}), 500

@app.route('/day/<date>')
def day_planning(date):
    """Страница планирования на день"""
    return render_template('day_planning.html', date=date)

@app.route('/map/<date>')
def show_map(date):
    """Генерация карты маршрута"""
    if date not in scheduler.schedule:
        return "Нет данных для этого дня"
    
    map_html = generate_route_map(date)
    return map_html

@app.route('/export/schedule')
def export_schedule():
    """Экспорт графика в JSON"""
    export_data = {
        'schedule': scheduler.schedule,
        'export_date': datetime.now().isoformat(),
        'complexes_count': len(scheduler.complexes)
    }
    
    return jsonify(export_data)

def generate_calendar(year, month):
    """Генерация данных календаря"""
    first_day = datetime(year, month, 1)
    if month == 12:
        last_day = datetime(year+1, 1, 1) - timedelta(days=1)
    else:
        last_day = datetime(year, month+1, 1) - timedelta(days=1)
    
    calendar = []
    current_day = 1
    
    # Заполняем календарь
    for week in range(6):
        week_days = []
        for day in range(7):
            if (week == 0 and day < first_day.weekday()) or current_day > last_day.day:
                week_days.append(None)
            else:
                date_key = f"{year}-{month:02d}-{current_day:02d}"
                day_schedule = scheduler.schedule.get(date_key, {})
                team1_count = len(day_schedule.get('team1', []))
                team2_count = len(day_schedule.get('team2', []))
                total_count = team1_count + team2_count
                
                week_days.append({
                    'day': current_day,
                    'date_key': date_key,
                    'team1_count': team1_count,
                    'team2_count': team2_count,
                    'total_count': total_count
                })
                current_day += 1
        calendar.append(week_days)
    
    return calendar

def generate_route_map(date):
    """Генерация карты с маршрутами"""
    day_schedule = scheduler.schedule[date]
    
    m = folium.Map(location=scheduler.start_point, zoom_start=12)
    
    # Начальная точка
    folium.Marker(
        scheduler.start_point,
        popup="Начало маршрута",
        tooltip="Начальная точка",
        icon=folium.Icon(color='green', icon='home')
    ).add_to(m)
    
    colors = ['red', 'blue']
    
    for team_num in [1, 2]:
        team_key = f'team{team_num}'
        if team_key in day_schedule and day_schedule[team_key]:
            team_complexes = day_schedule[team_key]
            coordinates = [scheduler.start_point]
            
            for complex_data in team_complexes:
                complex = scheduler.complexes[complex_data['complex_index']]
                if complex['coordinates']:
                    coordinates.append(complex['coordinates'])
                    
                    folium.Marker(
                        complex['coordinates'],
                        popup=f"""
                        <b>{complex['number']}</b><br>
                        {complex['location']}<br>
                        <b>ТО:</b> {complex_data['maintenance_type']}<br>
                        <b>Тип:</b> {complex['type']}
                        """,
                        tooltip=complex['number'],
                        icon=folium.Icon(color=colors[team_num-1], icon='wrench')
                    ).add_to(m)
            
            if len(coordinates) > 1:
                coordinates.append(scheduler.start_point)
                road_path = generate_road_path(coordinates)
                
                folium.PolyLine(
                    road_path,
                    color=colors[team_num-1],
                    weight=6,
                    opacity=0.7,
                    popup=f"Маршрут бригады {team_num} ({len(team_complexes)} комплексов)"
                ).add_to(m)
    
    # Сохраняем карту во временный файл
    map_dir = os.path.join('static', 'maps')
    os.makedirs(map_dir, exist_ok=True)
    map_file = os.path.join(map_dir, f"maintenance_map_{date}.html")
    m.save(map_file)
    
    return f'''
    <html>
    <head>
        <title>Карта маршрута {date}</title>
        <link href="https://cdn.jsdelivr.net/npm/bootstrap@5.1.3/dist/css/bootstrap.min.css" rel="stylesheet">
    </head>
    <body>
        <div class="container mt-4">
            <h2>Карта маршрутов обслуживания</h2>
            <p>Дата: {date}</p>
            <iframe src="/static/maps/maintenance_map_{date}.html" width="100%" height="600" frameborder="0"></iframe>
            <div class="mt-3">
                <a href="/" class="btn btn-secondary">Вернуться к календарю</a>
            </div>
        </div>
    </body>
    </html>
    '''

def generate_road_path(coordinates):
    """Генерация пути по дорогам"""
    if len(coordinates) < 2:
        return coordinates
    
    path = [coordinates[0]]
    for i in range(1, len(coordinates)):
        start = coordinates[i-1]
        end = coordinates[i]
        
        num_intermediate = max(2, int(calculate_distance(start, end) * 100))
        for j in range(1, num_intermediate):
            fraction = j / num_intermediate
            intermediate_lat = start[0] + (end[0] - start[0]) * fraction
            intermediate_lon = start[1] + (end[1] - start[1]) * fraction
            
            deviation = 0.001
            intermediate_lat += (math.sin(fraction * math.pi * 2) * deviation)
            intermediate_lon += (math.cos(fraction * math.pi * 2) * deviation)
            
            path.append((intermediate_lat, intermediate_lon))
        
        path.append(end)
    
    return path

def calculate_distance(coord1, coord2):
    """Вычисление расстояния между координатами"""
    lat1, lon1 = coord1
    lat2, lon2 = coord2
    return math.sqrt((lat2 - lat1)**2 + (lon2 - lon1)**2)

if __name__ == '__main__':
    # Создаем необходимые папки
    os.makedirs('templates', exist_ok=True)
    os.makedirs('static/maps', exist_ok=True)
    
    print("🚀 Запуск веб-приложения...")
    print("📊 Загрузка данных из БД...")
    
    # Предварительная загрузка данных
    scheduler.load_complexes_from_db()
    print(f"✅ Загружено {len(scheduler.complexes)} комплексов")
    
    app.run(debug=True, host='0.0.0.0', port=5000)