dash_app/database.py - master - VerP404/MosaicMap

dash_app/database.py
217 строк9 KB
Rodionov Dmitriy
feat: запуск проекта.
17 дек 2025, 18:06
17 дек 2025, 18:06add2b2f
"""
Настройка и утилиты для работы с базой данных
"""

import os
from flask import Flask, current_app
from sqlalchemy import inspect, text

from models import db, Address, Building, District


def _ensure_additional_columns():
    """
    Грубая миграция схемы для уже существующих SQLite-баз.

    Добавляет недостающие колонки, если база была создана старой версией моделей.
    Вызывать только внутри app_context, после db.create_all().
    """
    engine = db.engine
    inspector = inspect(engine)

    # Добавляем поле district_id в таблицу адресов (привязка адреса к участку)
    addr_cols = {col["name"] for col in inspector.get_columns("addresses")}
    if "district_id" not in addr_cols:
        try:
            # SQLAlchemy 2.x: выполняем DDL через соединение/сессию
            with engine.begin() as conn:
                conn.execute(text("ALTER TABLE addresses ADD COLUMN district_id INTEGER"))
            print("[MIGRATION] Added column addresses.district_id")
        except Exception as exc:
            print(f"[MIGRATION] Skip adding addresses.district_id: {exc}")

    # Добавляем поле building_id в таблицу участков (основной корпус участка)
    dist_cols = {col["name"] for col in inspector.get_columns("districts")}
    if "building_id" not in dist_cols:
        try:
            with engine.begin() as conn:
                conn.execute(text("ALTER TABLE districts ADD COLUMN building_id INTEGER"))
            print("[MIGRATION] Added column districts.building_id")
        except Exception as exc:
            print(f"[MIGRATION] Skip adding districts.building_id: {exc}")


def create_database(app: Flask):
    """Создание всех таблиц в базе данных"""
    with app.app_context():
        db.create_all()
        _ensure_additional_columns()
        print("[OK] Baza dannykh sozdana")


def init_database(app: Flask):
    """Инициализация базы данных примерными данными"""
    with app.app_context():
        # Создаем таблицы (если нужно) и добавляем недостающие колонки
        db.create_all()
        _ensure_additional_columns()

        # Проверяем, есть ли уже данные
        if District.query.count() == 0:
            print("[DATA] Sozdaem primer dannykh...")

            # Пример корпусов
            building1 = Building(
                name="Городская поликлиника №1",
                address="ул. Ленина, 15",
                latitude=51.677,
                longitude=39.205,
                capacity=500,
                phone="+7 (473) 123-45-67",
                description="Основная поликлиника центрального района"
            )
            building2 = Building(
                name="Детская поликлиника №2",
                address="пр. Революции, 34",
                latitude=51.669,
                longitude=39.205,
                capacity=300,
                phone="+7 (473) 234-56-78",
                description="Специализированная детская поликлиника"
            )
            db.session.add_all([building1, building2])
            db.session.flush()

            # Примеры участков, каждый привязан к одному корпусу
            district1 = District(
                name="Участок №1 (Центральный)",
                description="Центральный район города Воронеж",
                polygon=[[51.675, 39.200], [51.680, 39.200], [51.680, 39.210], [51.675, 39.210]],
                color="#FF5733",
                population=15000,
                area_sq_km=2.5,
                building_id=building1.id,
            )
            district2 = District(
                name="Участок №2 (Детский)",
                description="Детская служба",
                polygon=[[51.670, 39.210], [51.685, 39.210], [51.685, 39.220], [51.670, 39.220]],
                color="#33A1FD",
                population=8000,
                area_sq_km=1.2,
                building_id=building2.id,
            )
            db.session.add_all([district1, district2])

            db.session.commit()
            print("[OK] Primer dannykh sozdan")


def get_database_stats():
    """Получение статистики базы данных.

    Предполагается вызов внутри Flask app_context (в Dash callback'ах он уже есть).
    """
    stats = {
        'addresses': Address.query.count(),
        'buildings': Building.query.count(),
        'districts': District.query.count(),
        'addresses_with_buildings': Address.query.filter(Address.building_id.isnot(None)).count()
    }
    return stats


def clear_database():
    """Очистка всех данных (для тестирования).

    ВАЖНО: вызывать только внутри app_context.
    """
    Address.query.delete()
    Building.query.delete()
    District.query.delete()
    db.session.commit()
    print("[CLEAN] Baza dannykh ochishchena")


def export_database_to_csv(output_dir: str = "exports"):
    """Экспорт данных в CSV файлы. Вызывать внутри app_context."""
    import pandas as pd

    os.makedirs(output_dir, exist_ok=True)

    # Экспорт адресов
    addresses = Address.query.all()
    if addresses:
        addresses_data = [{
            'osm_id': addr.osm_id,
            'osm_type': addr.osm_type,
            'latitude': addr.latitude,
            'longitude': addr.longitude,
            'housenumber': addr.housenumber,
            'street': addr.street,
            'city': addr.city,
            'postcode': addr.postcode,
            'building_name': addr.building.name if addr.building else None
        } for addr in addresses]

        df = pd.DataFrame(addresses_data)
        df.to_csv(os.path.join(output_dir, 'addresses.csv'), index=False, encoding='utf-8')
        print(f"[EXPORT] Eksportirovano {len(addresses)} adresov")

    # Экспорт корпусов
    buildings = Building.query.all()
    if buildings:
        buildings_data = [{
            'id': b.id,
            'name': b.name,
            'address': b.address,
            'latitude': b.latitude,
            'longitude': b.longitude,
            'capacity': b.capacity,
            # Один корпус может иметь несколько участков; соберём их имена строкой.
            'districts': ", ".join(d.name for d in getattr(b, "districts", [])) or None,
        } for b in buildings]

        df = pd.DataFrame(buildings_data)
        df.to_csv(os.path.join(output_dir, 'buildings.csv'), index=False, encoding='utf-8')
        print(f"[EXPORT] Eksportirovano {len(buildings)} korpusov")

    # Экспорт участков
    districts = District.query.all()
    if districts:
        districts_data = [{
            'id': d.id,
            'name': d.name,
            'color': d.color,
            'population': d.population,
            'area_sq_km': d.area_sq_km,
            'building_id': d.building_id,
            'building_name': d.building.name if d.building else None,
            'addresses_count': len(d.addresses),
            'total_capacity': d.total_capacity,
        } for d in districts]

        df = pd.DataFrame(districts_data)
        df.to_csv(os.path.join(output_dir, 'districts.csv'), index=False, encoding='utf-8')
        print(f"[EXPORT] Eksportirovano {len(districts)} uchastkov")


def backup_database(backup_file: str = "backup.db"):
    """Создание резервной копии текущей базы данных.

    Использует путь из SQLALCHEMY_DATABASE_URI (поддерживается только sqlite:///).
    """
    import shutil

    uri = current_app.config.get("SQLALCHEMY_DATABASE_URI")
    if not uri or not uri.startswith("sqlite:///"):
        print("[ERROR] backup_database podderzhivaet tolko sqlite:/// URI")
        return

    db_path = uri.replace("sqlite:///", "", 1)

    if os.path.exists(db_path):
        shutil.copy2(db_path, backup_file)
        print(f"[BACKUP] Rezervnaya kopiya sozdana: {backup_file}")
    else:
        print(f"[ERROR] Fail bazy dannykh ne naiden: {db_path}")