forked from HSE_team/BetterCallPraskovia
94 lines
3.1 KiB
Python
94 lines
3.1 KiB
Python
import os
|
|
import sys
|
|
from sqlalchemy import create_engine, inspect
|
|
from sqlalchemy.orm import declarative_base, Session
|
|
from sqlalchemy import Column, String, DateTime, Boolean, Integer, Text
|
|
import uuid
|
|
from datetime import datetime
|
|
|
|
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
DB_PATH = os.path.join(BASE_DIR, 'data', 'bot.db')
|
|
DATABASE_URL = f"sqlite:///{DB_PATH}"
|
|
|
|
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
|
|
|
|
if os.path.exists(DB_PATH):
|
|
try:
|
|
temp_engine = create_engine(DATABASE_URL)
|
|
inspector = inspect(temp_engine)
|
|
tables = inspector.get_table_names()
|
|
if tables:
|
|
sys.exit(0)
|
|
except:
|
|
pass
|
|
|
|
choice = input("Перезаписать БД? (y/N): ")
|
|
if choice.lower() != 'y':
|
|
sys.exit(0)
|
|
|
|
engine = create_engine(DATABASE_URL, echo=False)
|
|
Base = declarative_base()
|
|
|
|
class UserModel(Base):
|
|
__tablename__ = "users"
|
|
|
|
user_id = Column("user_id", String(36), primary_key=True, default=lambda: str(uuid.uuid4()))
|
|
telegram_id = Column("telegram_id", String(100), nullable=False, unique=True)
|
|
created_at = Column("created_at", DateTime, default=datetime.utcnow, nullable=False)
|
|
role = Column("role", String(20), default="user", nullable=False)
|
|
is_premium = Column(Boolean, default=False, nullable=False)
|
|
premium_until = Column(DateTime, nullable=True)
|
|
questions_used = Column(Integer, default=0, nullable=False)
|
|
username = Column(String(100), nullable=True)
|
|
first_name = Column(String(100), nullable=True)
|
|
last_name = Column(String(100), nullable=True)
|
|
|
|
class PaymentModel(Base):
|
|
__tablename__ = "payments"
|
|
|
|
id = Column(Integer, primary_key=True, autoincrement=True)
|
|
payment_id = Column(String(36), default=lambda: str(uuid.uuid4()), nullable=False, unique=True)
|
|
user_id = Column(Integer, nullable=False)
|
|
amount = Column(String(20), nullable=False)
|
|
currency = Column(String(3), default="RUB", nullable=False)
|
|
status = Column(String(20), default="pending", nullable=False)
|
|
created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
|
|
yookassa_payment_id = Column(String(100), unique=True, nullable=True)
|
|
description = Column(Text, nullable=True)
|
|
|
|
try:
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
session = Session(bind=engine)
|
|
|
|
existing = session.query(UserModel).filter_by(telegram_id="123456789").first()
|
|
if not existing:
|
|
test_user = UserModel(
|
|
telegram_id="123456789",
|
|
username="test_user",
|
|
first_name="Test",
|
|
last_name="User",
|
|
is_premium=True
|
|
)
|
|
session.add(test_user)
|
|
|
|
existing_payment = session.query(PaymentModel).filter_by(yookassa_payment_id="test_yoo_001").first()
|
|
if not existing_payment:
|
|
test_payment = PaymentModel(
|
|
user_id=123456789,
|
|
amount="500.00",
|
|
status="succeeded",
|
|
description="Test payment",
|
|
yookassa_payment_id="test_yoo_001"
|
|
)
|
|
session.add(test_payment)
|
|
|
|
session.commit()
|
|
session.close()
|
|
|
|
except Exception as e:
|
|
print(f"Ошибка: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
|