Private
Public Access
forked from FOSS/AutoShop-Djimbo-Simple
138 lines
4.3 KiB
Python
138 lines
4.3 KiB
Python
# - *- coding: utf- 8 - *-
|
|
from typing import Any, Dict, Optional
|
|
|
|
from sqlalchemy import Integer, String, Float
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
from sqlalchemy.ext.asyncio import AsyncConnection
|
|
|
|
from tgbot.database.core import Base
|
|
from tgbot.database.entities import Payments
|
|
from tgbot.database.repository import BaseRepository
|
|
|
|
|
|
class PaymentsModel(Base):
|
|
__tablename__ = "storage_payments"
|
|
|
|
id: Mapped[int] = mapped_column(Integer, primary_key=True, default=1)
|
|
cryptobot_token: Mapped[str] = mapped_column(String, nullable=False, default="None")
|
|
stars_course: Mapped[float] = mapped_column(Float, nullable=False, default=1.5)
|
|
status_cryptobot: Mapped[str] = mapped_column(
|
|
String(16), nullable=False, default="False"
|
|
)
|
|
status_stars: Mapped[str] = mapped_column(
|
|
String(16), nullable=False, default="False"
|
|
)
|
|
|
|
# Оплата через Lolzteam
|
|
lolzteam_token: Mapped[str] = mapped_column(String, nullable=False, default="None")
|
|
lolzteam_merchant_id: Mapped[Optional[int]] = mapped_column(
|
|
Integer, nullable=True, default=None
|
|
)
|
|
status_lolzteam: Mapped[str] = mapped_column(
|
|
String(16), nullable=False, default="False"
|
|
)
|
|
|
|
|
|
ModelBase = Payments
|
|
BaseModel = Payments
|
|
|
|
|
|
async def ensure_payments_schema(conn: AsyncConnection) -> None:
|
|
columns = (
|
|
await conn.exec_driver_sql("PRAGMA table_info(storage_payments)")
|
|
).mappings().all()
|
|
|
|
if not columns:
|
|
return
|
|
|
|
merchant_column = next(
|
|
(column for column in columns if column["name"] == "lolzteam_merchant_id"),
|
|
None,
|
|
)
|
|
|
|
if merchant_column is None or merchant_column["notnull"] == 0:
|
|
return
|
|
|
|
await conn.exec_driver_sql(
|
|
"ALTER TABLE storage_payments RENAME TO storage_payments__legacy"
|
|
)
|
|
await conn.exec_driver_sql(
|
|
"""
|
|
CREATE TABLE storage_payments (
|
|
id INTEGER NOT NULL,
|
|
cryptobot_token VARCHAR NOT NULL,
|
|
stars_course FLOAT NOT NULL,
|
|
status_cryptobot VARCHAR(16) NOT NULL,
|
|
status_stars VARCHAR(16) NOT NULL,
|
|
lolzteam_token VARCHAR NOT NULL,
|
|
lolzteam_merchant_id INTEGER,
|
|
status_lolzteam VARCHAR(16) NOT NULL,
|
|
PRIMARY KEY (id)
|
|
)
|
|
"""
|
|
)
|
|
await conn.exec_driver_sql(
|
|
"""
|
|
INSERT INTO storage_payments (
|
|
id,
|
|
cryptobot_token,
|
|
stars_course,
|
|
status_cryptobot,
|
|
status_stars,
|
|
lolzteam_token,
|
|
lolzteam_merchant_id,
|
|
status_lolzteam
|
|
)
|
|
SELECT
|
|
id,
|
|
cryptobot_token,
|
|
stars_course,
|
|
status_cryptobot,
|
|
status_stars,
|
|
lolzteam_token,
|
|
lolzteam_merchant_id,
|
|
status_lolzteam
|
|
FROM storage_payments__legacy
|
|
"""
|
|
)
|
|
await conn.exec_driver_sql("DROP TABLE storage_payments__legacy")
|
|
|
|
|
|
class Paymentsx(BaseRepository[PaymentsModel, Payments]):
|
|
# Подключение модели платежных настроек
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.table_model = PaymentsModel
|
|
self.entity_model = Payments
|
|
self.storage_name = PaymentsModel.__tablename__
|
|
|
|
# Создание дефолтной строки платежей
|
|
async def ensure_default(self) -> None:
|
|
payments = await BaseRepository.get(self, id=1)
|
|
|
|
if payments is None:
|
|
await self._insert(id=1)
|
|
|
|
# Получение платежных настроек
|
|
async def get(self, **kwargs) -> Payments:
|
|
if not kwargs:
|
|
kwargs = {"id": 1}
|
|
|
|
payments = await BaseRepository.get(self, **kwargs)
|
|
|
|
if payments is None and kwargs == {"id": 1}:
|
|
await self.ensure_default()
|
|
payments = await BaseRepository.get(self, id=1)
|
|
|
|
if payments is None:
|
|
raise RuntimeError("Настройки платежных систем по умолчанию не сохранились")
|
|
|
|
return payments
|
|
|
|
# Обновление платежных настроек
|
|
async def update(self, where: Optional[Dict[str, Any]] = None, **kwargs) -> int:
|
|
if where is None:
|
|
where = {"id": 1}
|
|
|
|
return await self._update(where=where, **kwargs)
|