527 lines
24 KiB
Python
527 lines
24 KiB
Python
import pg8000
|
||
import os
|
||
from typing import Optional, Dict, List, Any
|
||
from datetime import datetime, timedelta
|
||
import logging
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class Database:
|
||
def __init__(self):
|
||
self.conn = None
|
||
self.cursor = None
|
||
|
||
async def connect(self):
|
||
try:
|
||
self.conn = pg8000.connect(
|
||
host=os.getenv('DB_HOST', 'localhost'),
|
||
port=int(os.getenv('DB_PORT', '5432')),
|
||
database=os.getenv('DB_NAME', 'max_bot'),
|
||
user=os.getenv('DB_USER', 'max_user'),
|
||
password=os.getenv('DB_PASSWORD', 'max_password')
|
||
)
|
||
self.cursor = self.conn.cursor()
|
||
logger.info("✅ Подключение к PostgreSQL установлено (pg8000)")
|
||
await self.init_tables()
|
||
await self.init_schools()
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка подключения к БД: {e}")
|
||
raise
|
||
|
||
async def disconnect(self):
|
||
if self.cursor:
|
||
self.cursor.close()
|
||
if self.conn:
|
||
self.conn.close()
|
||
logger.info("Соединение с PostgreSQL закрыто")
|
||
|
||
async def init_tables(self):
|
||
try:
|
||
sql_path = os.path.join(os.path.dirname(__file__), 'init.sql')
|
||
with open(sql_path, 'r') as f:
|
||
sql = f.read()
|
||
self.cursor.execute(sql)
|
||
self.conn.commit()
|
||
logger.info("✅ Таблицы созданы/проверены")
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка инициализации таблиц: {e}")
|
||
raise
|
||
|
||
async def init_schools(self):
|
||
try:
|
||
self.cursor.execute("SELECT COUNT(*) FROM schools")
|
||
count = self.cursor.fetchone()[0]
|
||
if count == 0:
|
||
schools = [
|
||
('101', 'Средняя общеобразовательная школа №101', 'ул. Ленина, 15', 'Москва'),
|
||
('102', 'Средняя общеобразовательная школа №102', 'ул. Гагарина, 10', 'Москва'),
|
||
('103', 'Средняя общеобразовательная школа №103', 'ул. Мира, 25', 'Москва'),
|
||
('104', 'Гимназия №104', 'ул. Садовая, 8', 'Москва'),
|
||
('105', 'Лицей №105', 'ул. Тверская, 12', 'Москва'),
|
||
('201', 'Гимназия №201', 'ул. Пушкина, 5', 'Санкт-Петербург'),
|
||
('202', 'Средняя школа №202', 'ул. Невский проспект, 20', 'Санкт-Петербург'),
|
||
('203', 'Средняя школа №203', 'ул. Московская, 15', 'Санкт-Петербург'),
|
||
('301', 'Лицей №301', 'ул. Советская, 8', 'Казань'),
|
||
('302', 'Средняя школа №302', 'ул. Ленина, 42', 'Казань'),
|
||
('401', 'Средняя школа №401', 'ул. Мира, 12', 'Новосибирск'),
|
||
('402', 'Гимназия №402', 'ул. Кирова, 7', 'Новосибирск'),
|
||
('501', 'Школа №501 с углубленным изучением математики', 'ул. Лермонтова, 3', 'Екатеринбург'),
|
||
('502', 'Гимназия №502', 'ул. Свердлова, 10', 'Екатеринбург'),
|
||
('601', 'Средняя общеобразовательная школа №601', 'ул. Садовая, 25', 'Нижний Новгород'),
|
||
('602', 'Лицей №602', 'ул. Горького, 8', 'Нижний Новгород'),
|
||
('701', 'Гимназия №701', 'ул. Чехова, 10', 'Ростов-на-Дону'),
|
||
('702', 'Средняя школа №702', 'ул. Пушкинская, 15', 'Ростов-на-Дону'),
|
||
]
|
||
for school in schools:
|
||
self.cursor.execute("""
|
||
INSERT INTO schools (school_number, school_name, address, city)
|
||
VALUES (%s, %s, %s, %s)
|
||
""", school)
|
||
self.conn.commit()
|
||
logger.info(f"✅ Добавлено {len(schools)} школ в справочник")
|
||
except Exception as e:
|
||
logger.error(f"❌ Ошибка инициализации школ: {e}")
|
||
|
||
async def get_parent(self, user_id: int) -> Optional[Dict]:
|
||
self.cursor.execute("SELECT * FROM parents WHERE user_id = %s", (user_id,))
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return None
|
||
|
||
async def create_parent(self, user_id: int, username: str) -> Dict:
|
||
self.cursor.execute("""
|
||
INSERT INTO parents (user_id, username)
|
||
VALUES (%s, %s)
|
||
ON CONFLICT (user_id) DO UPDATE
|
||
SET username = %s, updated_at = CURRENT_TIMESTAMP
|
||
RETURNING *
|
||
""", (user_id, username, username))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
|
||
async def get_all_schools(self) -> List[Dict]:
|
||
self.cursor.execute("SELECT * FROM schools ORDER BY school_number")
|
||
rows = self.cursor.fetchall()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
|
||
async def get_school_by_number(self, school_number: str) -> Optional[Dict]:
|
||
self.cursor.execute("SELECT * FROM schools WHERE school_number = %s", (school_number,))
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return None
|
||
|
||
async def get_school_by_id(self, school_id: int) -> Optional[Dict]:
|
||
self.cursor.execute("SELECT * FROM schools WHERE id = %s", (school_id,))
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return None
|
||
|
||
async def search_schools(self, query: str) -> List[Dict]:
|
||
self.cursor.execute("""
|
||
SELECT * FROM schools
|
||
WHERE school_number ILIKE %s OR school_name ILIKE %s
|
||
ORDER BY school_number
|
||
""", (f'%{query}%', f'%{query}%'))
|
||
rows = self.cursor.fetchall()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
|
||
async def get_children(self, parent_id: int) -> List[Dict]:
|
||
self.cursor.execute("""
|
||
SELECT
|
||
c.id, c.parent_id, c.card_number, c.student_name, c.school_id,
|
||
s.school_number, s.school_name, s.address, s.city,
|
||
c.created_at, c.updated_at
|
||
FROM children c
|
||
JOIN schools s ON c.school_id = s.id
|
||
WHERE c.parent_id = %s
|
||
ORDER BY c.id
|
||
""", (parent_id,))
|
||
rows = self.cursor.fetchall()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
|
||
async def get_child_by_card_number(self, card_number: str) -> Optional[Dict]:
|
||
self.cursor.execute("""
|
||
SELECT
|
||
c.id, c.parent_id, c.card_number, c.student_name, c.school_id,
|
||
s.school_number, s.school_name
|
||
FROM children c
|
||
JOIN schools s ON c.school_id = s.id
|
||
WHERE c.card_number = %s
|
||
""", (card_number,))
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return None
|
||
|
||
async def add_child(self, parent_id: int, card_number: str, student_name: str, school_number: str) -> Dict:
|
||
parent = await self.get_parent(parent_id)
|
||
if not parent:
|
||
raise ValueError("Родитель не найден. Сначала отправьте /start")
|
||
school = await self.get_school_by_number(school_number)
|
||
if not school:
|
||
raise ValueError(f"Школа с номером '{school_number}' не найдена в справочнике")
|
||
self.cursor.execute("""
|
||
INSERT INTO children (parent_id, card_number, student_name, school_id)
|
||
VALUES (%s, %s, %s, %s)
|
||
ON CONFLICT (parent_id, card_number) DO UPDATE
|
||
SET student_name = %s, school_id = %s, updated_at = CURRENT_TIMESTAMP
|
||
RETURNING *
|
||
""", (parent_id, card_number, student_name, school['id'], student_name, school['id']))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
child = dict(zip(columns, row))
|
||
child['school_number'] = school['school_number']
|
||
child['school_name'] = school['school_name']
|
||
return child
|
||
|
||
async def delete_child(self, parent_id: int, child_id: int) -> bool:
|
||
self.cursor.execute("DELETE FROM children WHERE parent_id = %s AND id = %s", (parent_id, child_id))
|
||
self.conn.commit()
|
||
return self.cursor.rowcount > 0
|
||
|
||
async def get_access_event_by_id(self, event_id: str) -> Optional[Dict]:
|
||
self.cursor.execute("SELECT * FROM access_events WHERE event_id = %s", (event_id,))
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return None
|
||
|
||
async def save_access_event(self, event_id: str, user_id: str, resource_number: int,
|
||
event_time: datetime, device_id: str = None,
|
||
access_zone_id1: str = None, access_zone_id2: str = None) -> Dict:
|
||
self.cursor.execute("""
|
||
INSERT INTO access_events
|
||
(event_id, user_id, device_id, resource_number, access_zone_id1, access_zone_id2, event_time)
|
||
VALUES (%s, %s, %s, %s, %s, %s, %s)
|
||
ON CONFLICT (event_id) DO NOTHING
|
||
RETURNING *
|
||
""", (event_id, user_id, device_id, resource_number, access_zone_id1, access_zone_id2, event_time))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
existing = await self.get_access_event_by_id(event_id)
|
||
return existing if existing else {}
|
||
|
||
async def mark_event_processed(self, event_id: int) -> bool:
|
||
self.cursor.execute("UPDATE access_events SET processed = TRUE WHERE id = %s", (event_id,))
|
||
self.conn.commit()
|
||
return self.cursor.rowcount > 0
|
||
|
||
async def save_notification(self, parent_id: int, child_id: int, message: str, event_type: str) -> Dict:
|
||
self.cursor.execute("""
|
||
INSERT INTO notifications (parent_id, child_id, message, event_type)
|
||
VALUES (%s, %s, %s, %s)
|
||
RETURNING *
|
||
""", (parent_id, child_id, message, event_type))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
|
||
async def get_unread_notifications(self, parent_id: int) -> List[Dict]:
|
||
self.cursor.execute("""
|
||
SELECT * FROM notifications
|
||
WHERE parent_id = %s AND is_read = FALSE
|
||
ORDER BY created_at DESC
|
||
""", (parent_id,))
|
||
rows = self.cursor.fetchall()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
|
||
async def mark_notification_as_read(self, notification_id: int) -> bool:
|
||
self.cursor.execute("UPDATE notifications SET is_read = TRUE WHERE id = %s", (notification_id,))
|
||
self.conn.commit()
|
||
return self.cursor.rowcount > 0
|
||
|
||
# --- ПОДПИСОЧНАЯ СИСТЕМА ---
|
||
async def get_subscription(self, parent_id: int) -> Optional[Dict]:
|
||
self.cursor.execute("SELECT * FROM subscriptions WHERE parent_id = %s", (parent_id,))
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return None
|
||
|
||
async def create_subscription(self, parent_id: int) -> Dict:
|
||
trial_end = datetime.now() + timedelta(days=7)
|
||
self.cursor.execute("""
|
||
INSERT INTO subscriptions (parent_id, status, trial_start, trial_end)
|
||
VALUES (%s, %s, %s, %s)
|
||
ON CONFLICT (parent_id) DO UPDATE
|
||
SET status = 'trial', trial_start = CURRENT_TIMESTAMP, trial_end = %s, updated_at = CURRENT_TIMESTAMP
|
||
RETURNING *
|
||
""", (parent_id, 'trial', datetime.now(), trial_end, trial_end))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
|
||
async def check_subscription_status(self, parent_id: int) -> Dict:
|
||
subscription = await self.get_subscription(parent_id)
|
||
if not subscription:
|
||
subscription = await self.create_subscription(parent_id)
|
||
now = datetime.now()
|
||
status = subscription.get('status')
|
||
if status == 'trial':
|
||
trial_end = subscription.get('trial_end')
|
||
if trial_end and trial_end < now:
|
||
await self.update_subscription_status(parent_id, 'expired')
|
||
subscription['status'] = 'expired'
|
||
elif status == 'active':
|
||
paid_until = subscription.get('paid_until')
|
||
if paid_until and paid_until < now:
|
||
await self.update_subscription_status(parent_id, 'expired')
|
||
subscription['status'] = 'expired'
|
||
return subscription
|
||
|
||
async def update_subscription_status(self, parent_id: int, status: str) -> bool:
|
||
self.cursor.execute("""
|
||
UPDATE subscriptions
|
||
SET status = %s, updated_at = CURRENT_TIMESTAMP
|
||
WHERE parent_id = %s
|
||
""", (status, parent_id))
|
||
self.conn.commit()
|
||
return self.cursor.rowcount > 0
|
||
|
||
async def extend_subscription(self, parent_id: int, months: int = 1) -> Dict:
|
||
try:
|
||
subscription = await self.get_subscription(parent_id)
|
||
if not subscription:
|
||
subscription = await self.create_subscription(parent_id)
|
||
current_end = subscription.get('paid_until')
|
||
if current_end and current_end > datetime.now():
|
||
paid_until = current_end + timedelta(days=30 * months)
|
||
else:
|
||
paid_until = datetime.now() + timedelta(days=30 * months)
|
||
self.cursor.execute("""
|
||
UPDATE subscriptions
|
||
SET status = 'active', paid_until = %s, last_payment_date = CURRENT_TIMESTAMP, updated_at = CURRENT_TIMESTAMP
|
||
WHERE parent_id = %s
|
||
RETURNING *
|
||
""", (paid_until, parent_id))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
except Exception as e:
|
||
self.conn.rollback()
|
||
logger.error(f"❌ Ошибка продления подписки: {e}")
|
||
raise
|
||
|
||
# --- ПЛАТЕЖИ ---
|
||
async def record_payment(self, parent_id: int, amount: float, transaction_id: str, status: str = 'pending') -> Dict:
|
||
"""Запись платежа с проверкой на дубликат"""
|
||
try:
|
||
# Сначала проверяем, существует ли платеж
|
||
existing = await self.get_payment_by_transaction_id(transaction_id)
|
||
if existing:
|
||
if existing.get('status') != status:
|
||
await self.update_payment_status(transaction_id, status)
|
||
return await self.get_payment_by_transaction_id(transaction_id)
|
||
return existing
|
||
|
||
self.cursor.execute("""
|
||
INSERT INTO payments (parent_id, amount, transaction_id, status)
|
||
VALUES (%s, %s, %s, %s)
|
||
RETURNING *
|
||
""", (parent_id, amount, transaction_id, status))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
except Exception as e:
|
||
# Если ошибка уникальности, пытаемся получить существующий платеж
|
||
if 'duplicate key' in str(e).lower():
|
||
self.conn.rollback()
|
||
existing = await self.get_payment_by_transaction_id(transaction_id)
|
||
if existing:
|
||
return existing
|
||
self.conn.rollback()
|
||
logger.error(f"❌ Ошибка записи платежа: {e}")
|
||
raise
|
||
|
||
async def get_payment_by_transaction_id(self, transaction_id: str) -> Optional[Dict]:
|
||
self.cursor.execute("SELECT * FROM payments WHERE transaction_id = %s", (transaction_id,))
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return None
|
||
|
||
async def update_payment_status(self, transaction_id: str, status: str) -> bool:
|
||
try:
|
||
self.cursor.execute("""
|
||
UPDATE payments
|
||
SET status = %s
|
||
WHERE transaction_id = %s
|
||
""", (status, transaction_id))
|
||
self.conn.commit()
|
||
return self.cursor.rowcount > 0
|
||
except Exception as e:
|
||
self.conn.rollback()
|
||
logger.error(f"❌ Ошибка обновления статуса платежа: {e}")
|
||
return False
|
||
|
||
async def get_expiring_subscriptions(self, days_threshold: int = 3) -> List[Dict]:
|
||
"""
|
||
Получение подписок, у которых до окончания осталось days_threshold дней или меньше.
|
||
"""
|
||
now = datetime.now()
|
||
threshold_date = now + timedelta(days=days_threshold)
|
||
|
||
self.cursor.execute("""
|
||
SELECT
|
||
s.*,
|
||
p.username,
|
||
p.user_id
|
||
FROM subscriptions s
|
||
JOIN parents p ON s.parent_id = p.user_id
|
||
WHERE s.status IN ('trial', 'active')
|
||
AND s.trial_end <= %s
|
||
AND s.trial_end > %s
|
||
UNION
|
||
SELECT
|
||
s.*,
|
||
p.username,
|
||
p.user_id
|
||
FROM subscriptions s
|
||
JOIN parents p ON s.parent_id = p.user_id
|
||
WHERE s.status = 'active'
|
||
AND s.paid_until <= %s
|
||
AND s.paid_until > %s
|
||
""", (threshold_date, now, threshold_date, now))
|
||
|
||
rows = self.cursor.fetchall()
|
||
if rows:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
return []
|
||
|
||
async def get_subscription_notification_sent(self, parent_id: int, days_left: int) -> bool:
|
||
"""Проверка, отправлялось ли уже уведомление об окончании подписки для данного количества дней"""
|
||
self.cursor.execute("""
|
||
SELECT 1 FROM subscription_notifications
|
||
WHERE parent_id = %s AND days_left = %s
|
||
""", (parent_id, days_left))
|
||
return self.cursor.fetchone() is not None
|
||
|
||
async def get_expiring_subscriptions(self, days_threshold: int = 3) -> List[Dict]:
|
||
"""
|
||
Получение подписок, у которых до окончания осталось days_threshold дней или меньше.
|
||
"""
|
||
now = datetime.now()
|
||
threshold_date = now + timedelta(days=days_threshold)
|
||
|
||
self.cursor.execute("""
|
||
SELECT
|
||
s.*,
|
||
p.username,
|
||
p.user_id
|
||
FROM subscriptions s
|
||
JOIN parents p ON s.parent_id = p.user_id
|
||
WHERE s.status IN ('trial', 'active')
|
||
AND s.trial_end <= %s
|
||
AND s.trial_end > %s
|
||
UNION
|
||
SELECT
|
||
s.*,
|
||
p.username,
|
||
p.user_id
|
||
FROM subscriptions s
|
||
JOIN parents p ON s.parent_id = p.user_id
|
||
WHERE s.status = 'active'
|
||
AND s.paid_until <= %s
|
||
AND s.paid_until > %s
|
||
""", (threshold_date, now, threshold_date, now))
|
||
|
||
rows = self.cursor.fetchall()
|
||
if rows:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
return []
|
||
|
||
async def get_subscription_notification_sent(self, parent_id: int, days_left: int) -> bool:
|
||
"""Проверка, отправлялось ли уже уведомление об окончании подписки для данного количества дней"""
|
||
self.cursor.execute("""
|
||
SELECT 1 FROM subscription_notifications
|
||
WHERE parent_id = %s AND days_left = %s
|
||
""", (parent_id, days_left))
|
||
return self.cursor.fetchone() is not None
|
||
|
||
async def mark_subscription_notification_sent(self, parent_id: int, days_left: int) -> Dict:
|
||
"""Отметить, что уведомление об окончании подписки отправлено"""
|
||
self.cursor.execute("""
|
||
INSERT INTO subscription_notifications (parent_id, days_left)
|
||
VALUES (%s, %s)
|
||
ON CONFLICT (parent_id, days_left) DO NOTHING
|
||
RETURNING *
|
||
""", (parent_id, days_left))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return {}
|
||
|
||
async def clean_old_notifications(self, days: int = 30):
|
||
"""Очистка старых записей уведомлений (старше days дней)"""
|
||
self.cursor.execute("""
|
||
DELETE FROM subscription_notifications
|
||
WHERE notified_at < CURRENT_DATE - INTERVAL '%s days'
|
||
""", (days,))
|
||
self.conn.commit()
|
||
return self.cursor.rowcount
|
||
|
||
async def mark_subscription_notification_sent(self, parent_id: int, days_left: int) -> Dict:
|
||
"""Отметить, что уведомление об окончании подписки отправлено"""
|
||
self.cursor.execute("""
|
||
INSERT INTO subscription_notifications (parent_id, days_left)
|
||
VALUES (%s, %s)
|
||
ON CONFLICT (parent_id, days_left) DO NOTHING
|
||
RETURNING *
|
||
""", (parent_id, days_left))
|
||
self.conn.commit()
|
||
row = self.cursor.fetchone()
|
||
if row:
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return dict(zip(columns, row))
|
||
return {}
|
||
|
||
async def clean_old_notifications(self, days: int = 30):
|
||
"""Очистка старых записей уведомлений (старше days дней)"""
|
||
self.cursor.execute("""
|
||
DELETE FROM subscription_notifications
|
||
WHERE notified_at < CURRENT_DATE - INTERVAL '%s days'
|
||
""", (days,))
|
||
self.conn.commit()
|
||
return self.cursor.rowcount
|
||
async def get_payment_history(self, parent_id: int, limit: int = 10) -> List[Dict]:
|
||
self.cursor.execute("""
|
||
SELECT * FROM payments
|
||
WHERE parent_id = %s
|
||
ORDER BY payment_date DESC
|
||
LIMIT %s
|
||
""", (parent_id, limit))
|
||
rows = self.cursor.fetchall()
|
||
columns = [desc[0] for desc in self.cursor.description]
|
||
return [dict(zip(columns, row)) for row in rows]
|
||
|
||
|
||
db = Database() |