link generator/ router and vless support/ selective server billing/ traffic usage counting/ cosmetic improvements and bug fixes

2025-09-30 19:17:05 +03:00
parent 8e8dcab230
commit 608b520b72
29 changed files with 1735 additions and 465 deletions
@@ -1,23 +1,28 @@
 from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession

-from database.models import Key, Payment
+from database.models import Key, Payment, User


 async def get_hot_leads(session: AsyncSession):
-    """
-    Возвращает пользователей, у которых есть успешные оплаты, но нет активных ключей.
-    """
-    subquery = select(Key.tg_id).where(Key.expiry_time > func.extract("epoch", func.now()) * 1000).distinct()
+    now_ms = func.extract("epoch", func.now()) * 1000
+
+    sub_active = (
+        select(Key.tg_id)
+        .where(Key.expiry_time > now_ms)
+        .distinct()
+    )

    stmt = (
        select(Payment.tg_id)
+        .join(User, User.tg_id == Payment.tg_id)
        .distinct()
+        .where(User.trial == 1)
        .where(Payment.amount > 0)
        .where(Payment.status == "success")
        .where(Payment.payment_system.notin_(["referral", "coupon", "cashback"]))
-        .where(~Payment.tg_id.in_(subquery))
+        .where(~Payment.tg_id.in_(sub_active))
    )

    result = await session.execute(stmt)
-    return [row.tg_id for row in result]
+    return result.scalars().all()
@@ -121,7 +121,7 @@ async def update_key_expiry(session: AsyncSession, client_id: str, new_expiry_ti
    await session.execute(
        update(Key)
        .where(Key.client_id == client_id)
-        .values(expiry_time=new_expiry_time, notified=False, notified_24h=False)
+        .values(expiry_time=new_expiry_time)
    )
    await session.commit()
    logger.info(f"Срок действия ключа {client_id} обновлён до {new_expiry_time}")
@@ -171,3 +171,32 @@ async def update_key_tariff(session: AsyncSession, client_id: str, tariff_id: in
    await session.execute(update(Key).where(Key.client_id == client_id).values(tariff_id=tariff_id))
    await session.commit()
    logger.info(f"Тариф ключа {client_id} обновлён на {tariff_id}")
+
+
+async def get_subscription_link(session: AsyncSession, email: str) -> str | None:
+    result = await session.execute(
+        select(func.coalesce(Key.key, Key.remnawave_link)).where(Key.email == email)
+    )
+    return result.scalar_one_or_none()
+
+
+async def update_key_client_id(session: AsyncSession, email: str, new_client_id: str):
+    await session.execute(
+        update(Key)
+        .where(Key.email == email)
+        .values(client_id=new_client_id)
+    )
+    await session.commit()
+    logger.info(f"client_id обновлён для {email} -> {new_client_id}")
+
+
+async def update_key_link(session: AsyncSession, email: str, link: str) -> bool:
+    q = (
+        update(Key)
+        .where(Key.email == email)
+        .values(key=link)
+        .returning(Key.client_id)
+    )
+    res = await session.execute(q)
+    await session.commit()
+    return res.scalar_one_or_none() is not None
@@ -3,9 +3,10 @@ import uuid

 from datetime import datetime

-from sqlalchemy import JSON, BigInteger, Boolean, Column, DateTime, Float, ForeignKey, Integer, Numeric, String, Text
+from sqlalchemy import JSON, BigInteger, Boolean, Column, DateTime, Float, ForeignKey, Integer, Numeric, String, Text, UniqueConstraint
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.orm import Mapped, declarative_base, mapped_column
+from sqlalchemy.orm import relationship


 Base = declarative_base()
@@ -80,6 +81,7 @@ class Tariff(DictLikeMixin, Base):
    updated_at = Column(DateTime, default=datetime.utcnow)
    subgroup_title = Column(String, nullable=True)
    sort_order = Column(Integer, nullable=True)
+    vless = Column(Boolean, default=False)


 class Server(DictLikeMixin, Base):
@@ -96,6 +98,23 @@ class Server(DictLikeMixin, Base):
    tariff_group = Column(String)
    enabled = Column(Boolean, default=True)

+    subgroups = relationship("ServerSubgroup", back_populates="server", cascade="all, delete-orphan")
+
+
+class ServerSubgroup(DictLikeMixin, Base):
+    __tablename__ = "server_subgroups"
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    server_id = Column(Integer, ForeignKey("servers.id", ondelete="CASCADE"), index=True, nullable=False)
+    group_code = Column(String, nullable=False)
+    subgroup_title = Column(String, nullable=False)
+
+    server = relationship("Server", back_populates="subgroups")
+
+    __table_args__ = (
+        UniqueConstraint("server_id", "subgroup_title", name="uq_server_subgroup"),
+    )
+

 class Payment(DictLikeMixin, Base):
    __tablename__ = "payments"
@@ -2,7 +2,8 @@ from sqlalchemy import delete, func, insert, select, update
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy.ext.asyncio import AsyncSession

-from database.models import Key, Server
+from database.models import Key, Server, Tariff, ServerSubgroup
+
 from logger import logger


@@ -49,6 +50,15 @@ async def get_servers(session: AsyncSession, include_enabled: bool = False) -> d
        result = await session.execute(stmt)
        servers = result.scalars().all()

+        ids = [s.id for s in servers]
+        subs_map = {}
+        if ids:
+            r = await session.execute(
+                select(ServerSubgroup.server_id, ServerSubgroup.subgroup_title).where(ServerSubgroup.server_id.in_(ids))
+            )
+            for sid, sg in r.all():
+                subs_map.setdefault(sid, []).append(sg)
+
        grouped = {}
        for s in servers:
            if not include_enabled and not s.enabled:
@@ -63,9 +73,9 @@ async def get_servers(session: AsyncSession, include_enabled: bool = False) -> d
                "enabled": s.enabled,
                "max_keys": s.max_keys,
                "tariff_group": s.tariff_group,
+                "tariff_subgroups": subs_map.get(s.id, []),
                "cluster_name": cluster,
            })
-
        return grouped
    except SQLAlchemyError as e:
        logger.error(f"Ошибка при получении серверов: {e}")
@@ -203,17 +213,88 @@ async def update_server_cluster(session: AsyncSession, server_name: str, new_clu
        result = await session.execute(stmt_new_cluster)
        new_tariff_group = result.scalar_one_or_none()

-        stmt_update = (
+        await session.execute(
            update(Server)
            .where(Server.server_name == server_name)
            .values(cluster_name=new_cluster, tariff_group=new_tariff_group)
        )
-        await session.execute(stmt_update)
-        await session.commit()

-        logger.info(f"✅ Сервер {server_name} перемещен в кластер {new_cluster} с обновлением тарифной группы")
+        if server_data.get("id") is None:
+            rid = await session.execute(select(Server.id).where(Server.server_name == server_name).limit(1))
+            server_id = rid.scalar_one_or_none()
+        else:
+            server_id = server_data["id"]
+
+        if server_id is not None and new_tariff_group is not None:
+            await session.execute(
+                update(ServerSubgroup)
+                .where(ServerSubgroup.server_id == server_id)
+                .values(group_code=new_tariff_group)
+            )
+
+        await session.commit()
+        logger.info(f"✅ Сервер {server_name} перемещен в кластер {new_cluster} с обновлением тарифной группы и привязок подгрупп")
        return True
    except SQLAlchemyError as e:
        logger.error(f"❌ Ошибка при обновлении кластера сервера {server_name}: {e}")
        await session.rollback()
        return False
+
+
+async def resolve_device_limit_from_group(session: AsyncSession, server_id: str) -> int | None:
+    r = await session.execute(select(Server.tariff_group).where(Server.server_name == server_id))
+    group = r.scalar_one_or_none()
+    if not group:
+        return None
+    q = await session.execute(
+        select(Tariff.device_limit)
+        .where(Tariff.group_code == group, Tariff.is_active.is_(True))
+        .order_by(Tariff.duration_days.desc())
+        .limit(1)
+    )
+    dl = q.scalar_one_or_none()
+    return int(dl) if dl is not None else None
+
+
+async def filter_cluster_by_subgroup(session: AsyncSession, cluster: list, target_subgroup: str, cluster_id: str) -> list:
+    names = [s.get("server_name") for s in cluster if s.get("server_name")]
+    if not names:
+        return []
+
+    q_allowed = await session.execute(
+        select(Server.server_name)
+        .join(ServerSubgroup, ServerSubgroup.server_id == Server.id)
+        .where(
+            Server.server_name.in_(names),
+            Server.enabled.is_(True),
+            ServerSubgroup.subgroup_title == target_subgroup,
+        )
+    )
+    allowed = {n for (n,) in q_allowed.all()}
+    if allowed:
+        return [s for s in cluster if s.get("server_name") in allowed]
+
+    total_for_subgroup = await session.scalar(
+        select(func.count())
+        .select_from(ServerSubgroup)
+        .where(ServerSubgroup.subgroup_title == target_subgroup)
+    )
+    if not total_for_subgroup:
+        logger.info(f"Для подгруппы {target_subgroup} нет ни одного сервера. Используем весь кластер {cluster_id}.")
+        return cluster
+
+    q_any = await session.execute(
+        select(Server.server_name)
+        .join(ServerSubgroup, ServerSubgroup.server_id == Server.id)
+        .where(
+            Server.server_name.in_(names),
+            Server.enabled.is_(True),
+        )
+    )
+    any_bound = {n for (n,) in q_any.all()}
+    if any_bound:
+        logger.warning(f"Нет серверов под подгруппу {target_subgroup} в кластере {cluster_id}. Продление пропущено.")
+        return []
+
+    logger.info(f"В кластере {cluster_id} нет привязок подгрупп. Продлеваем по всему кластеру.")
+    return cluster