https://pastein.ru/t/R4h
Загрузка данных
import psycopg2
from psycopg2 import sql, OperationalError
import os
from dotenv import load_dotenv
from datetime import datetime
from typing import Optional, List, Dict, Any
# Загрузка переменных окружения
load_dotenv()
class DatabaseManager:
"""Класс для управления подключением к БД"""
def __init__(self):
self.conn = None
self.cursor = None
self.connect()
def connect(self) -> bool:
"""Установка соединения с БД"""
try:
self.conn = psycopg2.connect(
host=os.getenv('DB_HOST', 'localhost'),
port=os.getenv('DB_PORT', '5432'),
database=os.getenv('DB_NAME', 'samoletiki'),
user=os.getenv('DB_USER', 'postgres'),
password=os.getenv('DB_PASSWORD', '')
)
self.cursor = self.conn.cursor()
print("✓ Подключение к базе данных 'samoletiki' установлено")
return True
except OperationalError as e:
print(f"✗ Ошибка подключения к БД: {e}")
return False
def close(self):
"""Закрытие соединения"""
if self.cursor:
self.cursor.close()
if self.conn:
self.conn.close()
print("✓ Соединение с БД закрыто")
def commit(self):
"""Подтверждение транзакции"""
if self.conn:
self.conn.commit()
def rollback(self):
"""Откат транзакции"""
if self.conn:
self.conn.rollback()
class AviationSystem:
"""Основной класс системы управления авиаперевозками"""
def __init__(self):
self.db = DatabaseManager()
self.current_passenger = None
def passenger_login(self, passport_number: str) -> Optional[Dict[str, Any]]:
"""Авторизация пассажира по номеру паспорта"""
try:
query = """
SELECT num_pasp, fio, adress, phone
FROM passzhir
WHERE num_pasp = %s
"""
self.db.cursor.execute(query, (passport_number,))
passenger = self.db.cursor.fetchone()
if passenger:
return {
'passport': passenger[0],
'name': passenger[1],
'address': passenger[2],
'phone': passenger[3]
}
return None
except Exception as e:
print(f"Ошибка при авторизации: {e}")
return None
def get_passenger_flights(self, passport_number: str) -> List[Dict[str, Any]]:
"""Получение всех рейсов пассажира"""
try:
query = """
SELECT
r.num_reis,
r.date_time_vilet,
r.reis_otmen,
m.aer_vilet,
m.aer_pril,
m.price_bilet,
m.prodol_poleta,
s.model AS model_samoleta,
s.bort_num,
k.fio AS fio_komandira
FROM bilet b
JOIN reis r ON b.num_reis = r.num_reis
JOIN marshrut m ON r.num_mar = m.num_mar
JOIN samolet s ON r.bort_num = s.bort_num
JOIN komandir k ON s.lich_num_komand = k.lich_num
WHERE b.num_pasp = %s
ORDER BY r.date_time_vilet
"""
self.db.cursor.execute(query, (passport_number,))
flights = self.db.cursor.fetchall()
result = []
for flight in flights:
result.append({
'flight_number': flight[0],
'departure_time': flight[1],
'is_canceled': flight[2],
'departure_airport': flight[3],
'arrival_airport': flight[4],
'ticket_price': float(flight[5]),
'duration_min': flight[6],
'aircraft_model': flight[7],
'tail_number': flight[8],
'captain_name': flight[9]
})
return result
except Exception as e:
print(f"Ошибка при получении рейсов: {e}")
return []
def get_all_routes(self) -> List[tuple]:
"""Получение всех маршрутов"""
try:
self.db.cursor.execute("""
SELECT num_mar, aer_vilet, aer_pril, price_bilet, prodol_poleta
FROM marshrut
ORDER BY num_mar
""")
return self.db.cursor.fetchall()
except Exception as e:
print(f"Ошибка: {e}")
return []
def get_all_flights(self) -> List[tuple]:
"""Получение всех рейсов"""
try:
self.db.cursor.execute("""
SELECT r.num_reis, m.aer_vilet, m.aer_pril,
r.date_time_vilet, r.reis_otmen, s.model
FROM reis r
JOIN marshrut m ON r.num_mar = m.num_mar
JOIN samolet s ON r.bort_num = s.bort_num
ORDER BY r.date_time_vilet
""")
return self.db.cursor.fetchall()
except Exception as e:
print(f"Ошибка: {e}")
return []
def get_aircraft_info(self) -> List[tuple]:
"""Получение информации о самолетах"""
try:
self.db.cursor.execute("""
SELECT s.bort_num, s.model, s.date_izgot,
s.srok_ecspluat, s.gotovnosty, k.fio,
EXTRACT(YEAR FROM AGE(CURRENT_DATE, s.date_izgot)) as age
FROM samolet s
JOIN komandir k ON s.lich_num_komand = k.lich_num
ORDER BY s.bort_num
""")
return self.db.cursor.fetchall()
except Exception as e:
print(f"Ошибка: {e}")
return []
def get_flight_passengers(self, flight_number: int) -> List[tuple]:
"""Получение списка пассажиров на рейсе"""
try:
self.db.cursor.execute("""
SELECT p.num_pasp, p.fio, p.phone
FROM bilet b
JOIN passzhir p ON b.num_pasp = p.num_pasp
WHERE b.num_reis = %s
""", (flight_number,))
return self.db.cursor.fetchall()
except Exception as e:
print(f"Ошибка: {e}")
return []
def buy_ticket(self, passport_number: str, flight_number: int) -> bool:
"""Покупка билета на рейс"""
try:
# Проверяем, существует ли рейс и не отменен ли он
self.db.cursor.execute("""
SELECT reis_otmen FROM reis WHERE num_reis = %s
""", (flight_number,))
flight = self.db.cursor.fetchone()
if not flight:
print("✗ Рейс не найден")
return False
if flight[0]:
print("✗ Нельзя купить билет на отмененный рейс")
return False
# Проверяем, не куплен ли уже билет
self.db.cursor.execute("""
SELECT * FROM bilet WHERE num_pasp = %s AND num_reis = %s
""", (passport_number, flight_number))
if self.db.cursor.fetchone():
print("✗ Билет на этот рейс уже куплен")
return False
# Покупаем билет
self.db.cursor.execute("""
INSERT INTO bilet (num_pasp, num_reis) VALUES (%s, %s)
""", (passport_number, flight_number))
self.db.commit()
print("✓ Билет успешно куплен!")
return True
except Exception as e:
self.db.rollback()
print(f"Ошибка при покупке билета: {e}")
return False
class ConsoleApp:
"""Консольное приложение для работы с системой"""
def __init__(self):
self.system = AviationSystem()
def clear_screen(self):
"""Очистка экрана"""
os.system('cls' if os.name == 'nt' else 'clear')
def print_header(self, title: str):
"""Вывод заголовка"""
print("=" * 90)
print(f" {title}")
print("=" * 90)
def print_menu(self) -> str:
"""Вывод главного меню"""
self.clear_screen()
self.print_header("АВИАКОМПАНИЯ 'ПОЛЕТ' - Система управления рейсами")
if self.system.current_passenger:
print(f"\n Добро пожаловать, {self.system.current_passenger['name']}!")
print(f" Номер паспорта: {self.system.current_passenger['passport']}\n")
print(" ГЛАВНОЕ МЕНЮ:")
print(" 1. Мои рейсы")
print(" 2. Все маршруты")
print(" 3. Все рейсы")
print(" 4. Информация о самолетах")
print(" 5. Купить билет")
print(" 6. Сменить пассажира")
print(" 0. Выход")
else:
print("\n ГЛАВНОЕ МЕНЮ:")
print(" 1. Вход по номеру паспорта")
print(" 2. Все маршруты")
print(" 3. Все рейсы")
print(" 4. Информация о самолетах")
print(" 0. Выход")
print("\n" + "-" * 90)
choice = input(" Выберите пункт меню: ")
return choice
def display_flights_table(self, flights: List[Dict[str, Any]]):
"""Отображение рейсов в виде таблицы"""
if not flights:
print("\n ✗ Рейсы не найдены")
return
print("\n СПИСОК РЕЙСОВ:")
print(" " + "-" * 100)
print(f" {'№ рейса':<10} {'Откуда':<15} {'Куда':<15} {'Дата и время':<22} {'Статус':<12} {'Самолет':<15}")
print(" " + "-" * 100)
for flight in flights:
status = "ОТМЕНЕН" if flight['is_canceled'] else "ВЫПОЛНЯЕТСЯ"
if isinstance(flight['departure_time'], datetime):
dep_time = flight['departure_time'].strftime("%d.%m.%Y %H:%M")
else:
dep_time = str(flight['departure_time'])[:16]
print(f" {flight['flight_number']:<10} {flight['departure_airport']:<15} "
f"{flight['arrival_airport']:<15} {dep_time:<22} {status:<12} "
f"{flight['aircraft_model']:<15}")
print(" " + "-" * 100)
def display_detailed_flight_info(self, flights: List[Dict[str, Any]]):
"""Детальное отображение информации о рейсах"""
if not flights:
print("\n ✗ У вас нет купленных билетов на рейсы")
return
print("\n ПОДРОБНАЯ ИНФОРМАЦИЯ О ВАШИХ РЕЙСАХ:")
print(" " + "=" * 90)
for i, flight in enumerate(flights, 1):
print(f"\n Рейс #{i}")
print(f" {'─' * 86}")
print(f" Номер рейса: {flight['flight_number']}")
print(f" Маршрут: {flight['departure_airport']} → {flight['arrival_airport']}")
if isinstance(flight['departure_time'], datetime):
dep_time = flight['departure_time'].strftime('%d.%m.%Y %H:%M')
else:
dep_time = str(flight['departure_time'])
print(f" Дата и время вылета: {dep_time}")
print(f" Длительность полета: {flight['duration_min']} мин. ({flight['duration_min'] // 60} ч {flight['duration_min'] % 60} мин)")
print(f" Цена билета: {flight['ticket_price']:.2f} руб.")
print(f" Статус рейса: {'❌ ОТМЕНЕН' if flight['is_canceled'] else '✅ ВЫПОЛНЯЕТСЯ'}")
print(f" Самолет: {flight['aircraft_model']} (борт {flight['tail_number']})")
print(f" Командир корабля: {flight['captain_name']}")
print("\n" + "=" * 90)
def display_routes(self):
"""Отображение всех маршрутов"""
routes = self.system.get_all_routes()
if not routes:
print("\n ✗ Маршруты не найдены")
return
print("\n ДОСТУПНЫЕ МАРШРУТЫ:")
print(" " + "-" * 90)
print(f" {'№':<5} {'Откуда':<20} {'Куда':<20} {'Цена (руб)':<15} {'Длительность':<15}")
print(" " + "-" * 90)
for route in routes:
duration = f"{route[4]} мин ({route[4] // 60}ч {route[4] % 60}мин)"
print(f" {route[0]:<5} {route[1]:<20} {route[2]:<20} {route[3]:<15.2f} {duration:<15}")
print(" " + "-" * 90)
def display_all_flights(self):
"""Отображение всех рейсов"""
flights = self.system.get_all_flights()
if not flights:
print("\n ✗ Рейсы не найдены")
return
print("\n ВСЕ РЕЙСЫ:")
print(" " + "-" * 110)
print(f" {'№ рейса':<10} {'Откуда':<15} {'Куда':<15} {'Дата и время вылета':<22} {'Статус':<12} {'Самолет':<20}")
print(" " + "-" * 110)
for flight in flights:
status = "ОТМЕНЕН" if flight[4] else "ВЫПОЛНЯЕТСЯ"
if isinstance(flight[3], datetime):
dep_time = flight[3].strftime("%d.%m.%Y %H:%M")
else:
dep_time = str(flight[3])[:16]
print(f" {flight[0]:<10} {flight[1]:<15} {flight[2]:<15} {dep_time:<22} {status:<12} {flight[5]:<20}")
print(" " + "-" * 110)
# Предложение посмотреть пассажиров рейса
show = input("\n Показать пассажиров конкретного рейса? (да/нет): ").lower()
if show in ['да', 'yes', 'y', 'д']:
flight_num = input(" Введите номер рейса: ")
if flight_num.isdigit():
self.display_flight_passengers(int(flight_num))
def display_flight_passengers(self, flight_number: int):
"""Отображение пассажиров на рейсе"""
passengers = self.system.get_flight_passengers(flight_number)
if not passengers:
print(f"\n ✗ На рейс №{flight_number} нет пассажиров или рейс не найден")
return
print(f"\n ПАССАЖИРЫ НА РЕЙСЕ №{flight_number}:")
print(" " + "-" * 80)
print(f" {'Номер паспорта':<15} {'ФИО':<35} {'Телефон':<20}")
print(" " + "-" * 80)
for passenger in passengers:
print(f" {passenger[0]:<15} {passenger[1]:<35} {passenger[2]:<20}")
print(" " + "-" * 80)
def display_aircraft_info(self):
"""Отображение информации о самолетах"""
aircrafts = self.system.get_aircraft_info()
if not aircrafts:
print("\n ✗ Информация о самолетах не найдена")
return
print("\n ПАРК САМОЛЕТОВ:")
print(" " + "=" * 100)
for aircraft in aircrafts:
status = "ГОТОВ" if aircraft[4] else "НЕ ГОТОВ"
status_icon = "✅" if aircraft[4] else "❌"
age = int(aircraft[6]) if aircraft[6] else 0
print(f"\n ✈️ Бортовой номер: {aircraft[0]}")
print(f" Модель: {aircraft[1]}")
print(f" Год выпуска: {aircraft[2].year if isinstance(aircraft[2], datetime) else aircraft[2][:4]}")
print(f" Возраст: {age} лет")
print(f" Срок службы: {aircraft[3]} лет")
print(f" Осталось лет: {aircraft[3] - age} лет")
print(f" Состояние: {status_icon} {status}")
print(f" Командир: {aircraft[5]}")
print(" " + "-" * 60)
print("=" * 100)
def login(self):
"""Авторизация пассажира"""
self.clear_screen()
self.print_header("ВХОД В СИСТЕМУ")
print("\n Для входа в систему введите номер паспорта")
print(" (например: MP1234567)\n")
passport = input(" Номер паспорта: ").strip()
if not passport:
print("\n ✗ Номер паспорта не может быть пустым")
input("\n Нажмите Enter для продолжения...")
return False
passenger = self.system.passenger_login(passport)
if passenger:
self.system.current_passenger = passenger
print(f"\n ✓ Авторизация успешна!")
print(f" Добро пожаловать, {passenger['name']}!")
input("\n Нажмите Enter для продолжения...")
return True
else:
print(f"\n ✗ Пассажир с номером паспорта {passport} не найден")
print(" Возможно, вам нужно зарегистрироваться в системе.")
input("\n Нажмите Enter для продолжения...")
return False
def logout(self):
"""Смена текущего пассажира"""
self.system.current_passenger = None
print("\n ✓ Вы вышли из системы")
input("\n Нажмите Enter для продолжения...")
def show_my_flights(self):
"""Отображение рейсов текущего пассажира"""
if not self.system.current_passenger:
print("\n ✗ Вы не авторизованы!")
input("\n Нажмите Enter для продолжения...")
return
flights = self.system.get_passenger_flights(self.system.current_passenger['passport'])
if flights:
self.display_flights_table(flights)
show_details = input("\n Показать детальную информацию? (да/нет): ").lower()
if show_details in ['да', 'yes', 'y', 'д']:
self.display_detailed_flight_info(flights)
else:
print("\n ✗ У вас нет купленных билетов на рейсы")
input("\n Нажмите Enter для продолжения...")
def buy_ticket(self):
"""Покупка билета"""
if not self.system.current_passenger:
print("\n ✗ Вы не авторизованы!")
input("\n Нажмите Enter для продолжения...")
return
self.clear_screen()
self.print_header("ПОКУПКА БИЛЕТА")
print(f"\n Пассажир: {self.system.current_passenger['name']}")
print(f" Номер паспорта: {self.system.current_passenger['passport']}\n")
# Показываем доступные рейсы
flights = self.system.get_all_flights()
if not flights:
print(" ✗ Нет доступных рейсов")
input("\n Нажмите Enter для продолжения...")
return
print(" ДОСТУПНЫЕ РЕЙСЫ:")
print(" " + "-" * 100)
print(f" {'№ рейса':<10} {'Откуда':<15} {'Куда':<15} {'Дата и время':<22} {'Статус':<12}")
print(" " + "-" * 100)
for flight in flights:
if not flight[4]: # Только неотмененные рейсы
status = "ДОСТУПЕН"
dep_time = flight[3].strftime("%d.%m.%Y %H:%M") if isinstance(flight[3], datetime) else str(flight[3])[:16]
print(f" {flight[0]:<10} {flight[1]:<15} {flight[2]:<15} {dep_time:<22} {status:<12}")
print(" " + "-" * 100)
flight_num = input("\n Введите номер рейса для покупки: ")
if flight_num.isdigit():
if self.system.buy_ticket(self.system.current_passenger['passport'], int(flight_num)):
pass
else:
print(" ✗ Неверный номер рейса")
input("\n Нажмите Enter для продолжения...")
def run(self):
"""Запуск приложения"""
print("\n Загрузка приложения...")
if not self.system.db.conn:
print("\n ✗ Не удалось подключиться к базе данных 'samoletiki'")
print(" Проверьте настройки подключения в файле .env")
return
while True:
choice = self.print_menu()
if choice == '1':
if self.system.current_passenger:
self.show_my_flights()
else:
self.login()
elif choice == '2':
self.display_routes()
input("\n Нажмите Enter для продолжения...")
elif choice == '3':
self.display_all_flights()
input("\n Нажмите Enter для продолжения...")
elif choice == '4':
self.display_aircraft_info()
input("\n Нажмите Enter для продолжения...")
elif choice == '5' and self.system.current_passenger:
self.buy_ticket()
elif choice == '6' and self.system.current_passenger:
self.logout()
elif choice == '0':
self.clear_screen()
print("\n Спасибо за использование системы!")
print(" До свидания!\n")
break
else:
print("\n ✗ Неверный выбор. Попробуйте снова.")
input("\n Нажмите Enter для продолжения...")
# Точка входа
if __name__ == "__main__":
app = ConsoleApp()
app.run()