database.py - master - aaaaa_pleskach/VK_bot

database.py
212 строк11 KB
aaaaapleskach
расширение базы данных
31 май 2026, 14:27
31 май 2026, 14:273258bc6
import sqlite3
from typing import Dict, List, Tuple, Any, Optional

DB_NAME: str = "railway_bot.db"

def init_db() -> None:
    with sqlite3.connect(DB_NAME) as conn:
        cursor = conn.cursor()
        
        # Таблица университетов
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS universities (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                short_name TEXT UNIQUE NOT NULL,
                full_name TEXT NOT NULL,
                site_url TEXT NOT NULL,
                vk_url TEXT NOT NULL
            )
        """)
        
        # Таблица факультетов
        cursor.execute("""
            CREATE TABLE IF NOT EXISTS faculties (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                university_id INTEGER NOT NULL,
                faculty_name TEXT NOT NULL,
                profession_desc TEXT NOT NULL,
                passing_score TEXT NOT NULL DEFAULT 'Не указан',
                phone TEXT NOT NULL DEFAULT 'Не указан',
                email TEXT NOT NULL DEFAULT 'Не указан',
                professions TEXT NOT NULL DEFAULT 'Не указано',
                departments TEXT NOT NULL DEFAULT 'Не указаны',
                FOREIGN KEY (university_id) REFERENCES universities (id) ON DELETE CASCADE,
                UNIQUE(university_id, faculty_name)
            )
        """)
        conn.commit()
    
    _insert_initial_data()

def get_all_universities() -> List[Tuple[str, str]]:
    with sqlite3.connect(DB_NAME) as conn:
        cursor = conn.cursor()
        cursor.execute("SELECT short_name, full_name FROM universities")
        return cursor.fetchall()

def get_university_info(short_name: str) -> Optional[Dict[str, Any]]:
    with sqlite3.connect(DB_NAME) as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT id, full_name, site_url, vk_url FROM universities WHERE short_name = ?", 
            (short_name,)
        )
        row = cursor.fetchone()
        if not row:
            return None
        return {"id": row[0], "name": row[1], "site": row[2], "vk_group": row[3]}

def get_faculties_by_uni(uni_id: int) -> List[str]:
    with sqlite3.connect(DB_NAME) as conn:
        cursor = conn.cursor()
        cursor.execute(
            "SELECT faculty_name FROM faculties WHERE university_id = ?", 
            (uni_id,)
        )
        return [r[0] for r in cursor.fetchall()]

def get_profession_by_faculty(faculty_name: str) -> Optional[Dict[str, str]]:
    """Возвращает структурированный словарь со всеми данными факультета."""
    with sqlite3.connect(DB_NAME) as conn:
        cursor = conn.cursor()
        cursor.execute("""
            SELECT profession_desc, passing_score, phone, email, professions, departments 
            FROM faculties 
            WHERE faculty_name = ?
        """, (faculty_name,))
        row = cursor.fetchone()
        if row:
            return {
                "vk_link": row[0],
                "score": row[1],
                "phone": row[2],
                "email": row[3],
                "professions": row[4],
                "departments": row[5]
            }
        return None

def _insert_initial_data() -> None:
    data = {
        "ИрГУПС": {
            "name": "Иркутский гос. университет путей сообщения (ИрГУПС, Иркутск)",
            "site": "https://irgups.ru",
            "vk": "https://vk.com/irgups",
            "facs": {
                "ФУТИИТ": {
                    "vk": "https://vk.com/dekanat_futiit_irgups",
                    "score": "от 200 баллов",
                    "phone": "+7 (3952) 638-376",
                    "structure": [
                        {
                            "dep": "Управление эксплуатационной работой",
                            "jobs": "Ты будешь управлять движением поездов и работой железнодорожных станций."
                        },
                        {
                            "dep": "Информационные системы и защита информации",
                            "jobs": "Ты станешь специалистом по кибербезопасности и защите данных."
                        }
                    ]
                },
                "ФСОТ": {
                    "vk": "https://vk.com/fs0t_irgups",
                    "score": "от 187 баллов",
                    "phone": "+7 (3952) 638-306",
                    "structure": [
                        {
                            "dep": "Автоматика, телемеханика и связь",
                            "jobs": "Ты станешь инженером, который настраивает системы автоматики, светофоры и цифровую связь поездов."
                        },
                        {
                            "dep": "Электроэнергетика транспорта",
                            "jobs": "Ты станешь инженером, который управляет высоковольтными сетями, подстанциями и подачей электричества для поездов."
                        },
                        {
                            "dep": "Управление качеством и инженерная графика",
                            "jobs": "Ты станешь инженером, который проверяет безопасность транспорта, улучшает бизнес-процессы и контролирует качество техники."
                        }
                    ]
                },
                "ФТС": {
                    "vk": "https://vk.com/fts_irgups",
                    "score": "от 185 баллов",
                    "phone": "+7 (3952) 638-321",
                    "structure": [
                        {
                            "dep": "Электроподвижной состав",
                            "jobs": "Ты станешь инженером путей сообщения, который управляет ремонтом, обслуживанием и устройством современных электровозов."
                        },
                        {
                            "dep": "Вагоны и вагонное хозяйство",
                            "jobs": "Ты станешь инженером, который руководит ремонтом, подготовкой к рейсам и техническим обслуживанием вагонов."
                        }
                    ]
                },
                "ФЭиУ": {
                    "vk": "https://vk.com/feiu_irgups",
                    "score": "от 190 баллов",
                    "phone": "+7 (3952) 638-346",
                    "structure": [
                        {
                            "dep": "Экономика и управление на железнодорожном транспорте",
                            "jobs": "Ты станешь экономистом-менеджером, который управляет бюджетом, расходами и прибылью транспортной компании."
                        },
                        {
                            "dep": "Таможенное дело и правоведение",
                            "jobs": "Ты станешь специалистом, который контролирует границы, оформляет международные грузы."
                        },
                        {
                            "dep": "Финансовый и стратегический менеджмент",
                            "jobs": "Ты станешь финансовым аналитиком, который рассчитывает инвестиции, управляет капиталом и развивает бизнес-стратегии."
                        }
                    ]
                },
                "ФСЖД": {
                    "vk": "https://vk.com/fszhd_irgups",
                    "score": "от 188 баллов",
                    "phone": "+7 (3952) 638-317",
                    "structure": [
                        {
                            "dep": "Путь и путевое хозяйство",
                            "jobs": "Ты станешь инженером, который проектирует железные дороги, руководит их строительством и ремонтом рельсов."
                        },
                        {
                            "dep": "Строительство железных дорог, мостов и тоннелей",
                            "jobs": "Ты станешь инженером-строителем, который проектирует и возводит железнодорожные пути, мосты и тоннели."
                        }
                    ]
                }
            }
        }
    }
    
    with sqlite3.connect(DB_NAME) as conn:
        cursor = conn.cursor()
        for s_name, info in data.items():
            cursor.execute("""
                INSERT OR IGNORE INTO universities (short_name, full_name, site_url, vk_url) 
                VALUES (?, ?, ?, ?)
            """, (s_name, info["name"], info["site"], info["vk"]))
            
            cursor.execute("SELECT id FROM universities WHERE short_name = ?", (s_name,))
            row = cursor.fetchone()
            if not row:
                continue
            uni_id = row[0]
            
            for f_name, f_info in info["facs"].items():
                deps_text = ""
                profs_text = ""
                
                for item in f_info["structure"]:
                    deps_text += f"• {item['dep']}\n"
                    profs_text += f"▪️ Направление: {item['dep']}\n  {item['jobs']}\n\n"
                
                deps_text = deps_text.strip()
                profs_text = profs_text.strip()
                
                cursor.execute("""
                    INSERT OR REPLACE INTO faculties (university_id, faculty_name, profession_desc, passing_score, phone, professions, departments) 
                    VALUES (?, ?, ?, ?, ?, ?, ?)
                """, (uni_id, f_name, f_info["vk"], f_info["score"], f_info["phone"], profs_text, deps_text))
        conn.commit()