houseassassin/Solo_bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor special notifications / Refactor server tariff bindings

Browse Source
This commit is contained in:
Capybara-z
2026-01-08 23:18:33 +03:00
parent 0fa5c6b682
commit 562c0f8d25
15 changed files with 488 additions and 128 deletions
Download Patch File Download Diff File Expand all files Collapse all files
database/servers.py
+98 -8
View File
@@ -53,12 +53,16 @@ async def get_servers(session: AsyncSession, include_enabled: bool = False) -> d
ids = [s.id for s in servers]
subs_map = {}
tariffs_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)
if sg and sg.isdigit():
tariffs_map.setdefault(sid, []).append(int(sg))
else:
subs_map.setdefault(sid, []).append(sg)
groups_map = {}
if ids:
@@ -88,8 +92,10 @@ async def get_servers(session: AsyncSession, include_enabled: bool = False) -> d
"max_keys": s.max_keys,
"tariff_group": s.tariff_group,
"tariff_subgroups": subs_map.get(s.id, []),
"tariff_ids": tariffs_map.get(s.id, []),
"special_groups": special,
"cluster_name": cluster,
"server_id": s.id,
})
return grouped
except SQLAlchemyError as e:
@@ -272,12 +278,32 @@ async def resolve_device_limit_from_group(session: AsyncSession, server_id: str)
async def filter_cluster_by_subgroup(
session: AsyncSession, cluster: list, target_subgroup: str, cluster_id: str
session: AsyncSession,
cluster: list,
target_subgroup: str,
cluster_id: str,
tariff_id: int | None = None,
) -> list:
names = [s.get("server_name") for s in cluster if s.get("server_name")]
if not names:
return []
if tariff_id:
tariff_id_str = str(tariff_id)
q_by_tariff = 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 == tariff_id_str,
)
)
allowed_by_tariff = {n for (n,) in q_by_tariff.all()}
if allowed_by_tariff:
logger.debug(f"Найдены серверы по tariff_id={tariff_id}: {allowed_by_tariff}")
return [s for s in cluster if s.get("server_name") in allowed_by_tariff]
q_allowed = await session.execute(
select(Server.server_name)
.join(ServerSubgroup, ServerSubgroup.server_id == Server.id)
@@ -291,11 +317,15 @@ async def filter_cluster_by_subgroup(
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)
check_values = [target_subgroup]
if tariff_id:
check_values.append(str(tariff_id))
total_bindings = await session.scalar(
select(func.count()).select_from(ServerSubgroup).where(ServerSubgroup.subgroup_title.in_(check_values))
)
if not total_for_subgroup:
logger.info(f"Для подгруппы {target_subgroup} нет ни одного сервера. Используем весь кластер {cluster_id}.")
if not total_bindings:
logger.info(f"Для подгруппы/тарифа нет привязок. Используем весь кластер {cluster_id}.")
return cluster
q_any = await session.execute(
@@ -308,8 +338,68 @@ async def filter_cluster_by_subgroup(
)
any_bound = {n for (n,) in q_any.all()}
if any_bound:
logger.warning(f"Нет серверов под подгруппу {target_subgroup} в кластере {cluster_id}. Продление пропущено.")
logger.warning(f"Нет серверов под подгруппу {target_subgroup} в кластере {cluster_id}.")
return []
logger.info(f"В кластере {cluster_id} нет привязок подгрупп. Продлеваем по всему кластеру.")
logger.info(f"В кластере {cluster_id} нет привязок. Используем весь кластер.")
return cluster
async def filter_cluster_by_tariff(
session: AsyncSession, cluster: list, tariff_id: int, cluster_id: str
) -> list:
names = [s.get("server_name") for s in cluster if s.get("server_name")]
if not names:
return []
tariff_id_str = str(tariff_id)
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 == tariff_id_str,
)
)
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_tariff = await session.scalar(
select(func.count()).select_from(ServerSubgroup).where(ServerSubgroup.subgroup_title == tariff_id_str)
)
if not total_for_tariff:
logger.info(f"Для тарифа {tariff_id} нет привязок серверов. Используем весь кластер {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"Нет серверов под тариф {tariff_id} в кластере {cluster_id}.")
return []
logger.info(f"В кластере {cluster_id} нет привязок тарифов. Используем весь кластер.")
return cluster
async def has_legacy_subgroup_bindings(session: AsyncSession, server_ids: list[int]) -> bool:
if not server_ids:
return False
result = await session.execute(
select(ServerSubgroup.subgroup_title)
.where(ServerSubgroup.server_id.in_(server_ids))
)
for (title,) in result.all():
if title and not title.isdigit():
return True
return False
handlers/admin/clusters/cluster_manage.py
+7 -1
View File
@@ -79,8 +79,14 @@ async def handle_cluster_servers(callback: CallbackQuery, session: AsyncSession)
allowed = set(ALLOWED_GROUP_CODES)
lines = []
for s in cluster_servers:
tids = s.get("tariff_ids") or []
subs = s.get("tariff_subgroups") or []
subs_str = ", ".join(sorted(subs)) if subs else "—"
if tids:
subs_str = f"{len(tids)} тариф(ов)"
elif subs:
subs_str = ", ".join(sorted(subs))
else:
subs_str = "—"
grps = s.get("special_groups") or []
grps = [g for g in grps if g in allowed]
handlers/admin/clusters/cluster_sync.py
+12 -5
View File
@@ -288,9 +288,13 @@ async def handle_sync_server(
tariff = tariffs_cache.get(key["tariff_id"]) if key["tariff_id"] else None
if tariff and server_info:
if tariff.get("subgroup_title") and tariff.get("subgroup_title") not in server_info.get(
"tariff_subgroups", []
):
subgroup = tariff.get("subgroup_title")
tid = key["tariff_id"]
has_new_binding = tid and tid in (server_info.get("tariff_ids") or [])
has_old_binding = subgroup and subgroup in (server_info.get("tariff_subgroups") or [])
has_any_binding = bool(server_info.get("tariff_ids") or server_info.get("tariff_subgroups"))
if has_any_binding and subgroup and not has_new_binding and not has_old_binding:
continue
if tariff.get("group_code") and tariff.get("group_code").lower() in ALLOWED_GROUP_CODES:
@@ -558,9 +562,12 @@ async def handle_sync_cluster(
remna = RemnawaveAPI(cluster_servers[0]["api_url"])
filtered_servers = cluster_servers
if subgroup_title:
if subgroup_title or (tariff and tariff.get("id")):
tid = tariff.get("id") if tariff else None
filtered_servers = [
s for s in cluster_servers if subgroup_title in s.get("tariff_subgroups", [])
s for s in cluster_servers
if (tid and tid in (s.get("tariff_ids") or []))
or (subgroup_title and subgroup_title in (s.get("tariff_subgroups") or []))
]
if not filtered_servers:
filtered_servers = cluster_servers
handlers/admin/clusters/cluster_tariffs.py
+228 -80
View File
@@ -6,6 +6,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from database import get_servers
from database.models import Server, ServerSpecialgroup, ServerSubgroup, Tariff
from database.servers import has_legacy_subgroup_bindings
from filters.admin import IsAdminFilter
from handlers.utils import ALLOWED_GROUP_CODES
from logger import logger
@@ -14,11 +15,13 @@ from .base import router
from .keyboard import (
AdminClusterCallback,
build_attach_tariff_kb,
build_legacy_reset_kb,
build_manage_cluster_kb,
build_select_group_servers_kb,
build_select_subgroup_servers_kb,
build_tariff_group_selection_for_servers_kb,
build_tariff_group_selection_kb,
build_tariff_selection_kb,
build_tariff_subgroup_selection_kb,
)
@@ -76,22 +79,35 @@ async def apply_tariff_group(callback: CallbackQuery, callback_data: AdminCluste
@router.callback_query(AdminClusterCallback.filter(F.action == "set_subgroup"))
async def show_servers_for_subgroup(
async def show_servers_for_tariffs(
callback: CallbackQuery, callback_data: AdminClusterCallback, session: AsyncSession, state: FSMContext
):
cluster_name = callback_data.data
servers = await get_servers(session=session, include_enabled=True)
cluster_servers = servers.get(cluster_name, [])
server_ids = [s.get("server_id") for s in cluster_servers if s.get("server_id")]
if server_ids and await has_legacy_subgroup_bindings(session, server_ids):
await callback.message.edit_text(
f"<b>⚠️ Обнаружены привязки старого формата</b>\n\n"
f"Кластер <code>{cluster_name}</code> содержит привязки по названиям подгрупп.\n"
f"Для использования новой системы необходимо сбросить текущие привязки.\n\n"
f"<i>После сброса вы сможете привязать тарифы по ID.</i>",
reply_markup=build_legacy_reset_kb(cluster_name),
)
return
data = await state.get_data()
selected = set(data.get(f"subgrp_sel:{cluster_name}", []))
await callback.message.edit_text(
f"<b>🗂 Выберите серверы в кластере <code>{cluster_name}</code> для назначения подгруппы тарифов:</b>",
f"<b>📋 Выберите серверы для привязки тарифов</b>\n"
f"<i>Кластер: {cluster_name}</i>",
reply_markup=build_select_subgroup_servers_kb(cluster_name, cluster_servers, selected),
)
@router.callback_query(AdminClusterCallback.filter(F.action == "toggle_server_subgroup"))
async def toggle_server_for_subgroup(
async def toggle_server_for_tariffs(
callback: CallbackQuery, callback_data: AdminClusterCallback, session: AsyncSession, state: FSMContext
):
cluster_name, idx_str = callback_data.data.split("|", 1)
@@ -119,34 +135,39 @@ async def toggle_server_for_subgroup(
selected.add(server_name)
await state.update_data({key: list(selected)})
await callback.message.edit_text(
f"<b>🗂 Выберите серверы в кластере <code>{cluster_name}</code> для назначения подгруппы тарифов:</b>",
f"<b>📋 Выберите серверы для привязки тарифов</b>\n"
f"<i>Кластер: {cluster_name}</i>",
reply_markup=build_select_subgroup_servers_kb(cluster_name, cluster_servers, selected),
)
@router.callback_query(AdminClusterCallback.filter(F.action == "reset_subgroup_selection"))
async def reset_subgroup_selection(
async def reset_tariff_selection(
callback: CallbackQuery, callback_data: AdminClusterCallback, session: AsyncSession, state: FSMContext
):
cluster_name = callback_data.data
servers = await get_servers(session=session, include_enabled=True)
cluster_servers = servers.get(cluster_name, [])
await state.update_data({f"subgrp_sel:{cluster_name}": []})
await state.update_data({
f"subgrp_sel:{cluster_name}": [],
f"tariff_sel:{cluster_name}": [],
})
await callback.message.edit_text(
f"<b>🗂 Выберите серверы в кластере <code>{cluster_name}</code> для назначения подгруппы тарифов:</b>",
f"<b>📋 Выберите серверы для привязки тарифов</b>\n"
f"<i>Кластер: {cluster_name}</i>",
reply_markup=build_select_subgroup_servers_kb(cluster_name, cluster_servers, set()),
)
@router.callback_query(AdminClusterCallback.filter(F.action == "choose_subgroup"))
async def choose_subgroup(
async def choose_tariffs(
callback: CallbackQuery, callback_data: AdminClusterCallback, session: AsyncSession, state: FSMContext
):
cluster_name = callback_data.data
key = f"subgrp_sel:{cluster_name}"
data = await state.get_data()
selected = set(data.get(key, []))
if not selected:
selected_servers = set(data.get(key, []))
if not selected_servers:
await callback.answer("Сначала выберите хотя бы один сервер", show_alert=True)
return
@@ -158,83 +179,148 @@ async def choose_subgroup(
group_code = group_codes[0]
res2 = await session.execute(
select(func.distinct(Tariff.subgroup_title))
.where(Tariff.group_code == group_code)
.where(Tariff.subgroup_title.isnot(None))
.order_by(Tariff.subgroup_title.asc())
result = await session.execute(
select(Tariff)
.where(Tariff.group_code == group_code, Tariff.is_active.is_(True))
.order_by(Tariff.subgroup_title.nulls_last(), Tariff.sort_order, Tariff.id)
)
subgroups = [r[0] for r in res2.fetchall()]
if not subgroups:
await callback.message.edit_text("❌ Для этой группы нет доступных подгрупп.")
tariffs = result.scalars().all()
if not tariffs:
await callback.message.edit_text("❌ Для этой группы нет доступных тарифов.")
return
selected_tariffs = set(data.get(f"tariff_sel:{cluster_name}", []))
await callback.message.edit_text(
f"<b>📚 Выберите подгруппу для {len(selected)} сервер(а/ов) кластера <code>{cluster_name}</code>:</b>",
reply_markup=build_tariff_subgroup_selection_kb(cluster_name, subgroups),
f"<b>📋 Выберите тарифы для {len(selected_servers)} сервер(а/ов)</b>\n"
f"<i>Кластер: {cluster_name}</i>",
reply_markup=build_tariff_selection_kb(cluster_name, tariffs, selected_tariffs),
)
@router.callback_query(AdminClusterCallback.filter(F.action == "apply_tariff_subgroup"))
async def apply_tariff_subgroup(
@router.callback_query(AdminClusterCallback.filter(F.action == "toggle_tariff"))
async def toggle_tariff_selection(
callback: CallbackQuery, callback_data: AdminClusterCallback, session: AsyncSession, state: FSMContext
):
cluster_name, tariff_id_str = callback_data.data.split("|", 1)
tariff_id = int(tariff_id_str)
key = f"tariff_sel:{cluster_name}"
data = await state.get_data()
selected_tariffs = set(data.get(key, []))
if tariff_id in selected_tariffs:
selected_tariffs.remove(tariff_id)
else:
selected_tariffs.add(tariff_id)
await state.update_data({key: list(selected_tariffs)})
res = await session.execute(select(Server.tariff_group).where(Server.cluster_name == cluster_name).distinct())
group_codes = [r[0] for r in res.fetchall() if r[0]]
if not group_codes:
return
result = await session.execute(
select(Tariff)
.where(Tariff.group_code == group_codes[0], Tariff.is_active.is_(True))
.order_by(Tariff.subgroup_title.nulls_last(), Tariff.sort_order, Tariff.id)
)
tariffs = result.scalars().all()
selected_servers = set(data.get(f"subgrp_sel:{cluster_name}", []))
await callback.message.edit_text(
f"<b>📋 Выберите тарифы для {len(selected_servers)} сервер(а/ов)</b>\n"
f"<i>Кластер: {cluster_name}</i>",
reply_markup=build_tariff_selection_kb(cluster_name, tariffs, selected_tariffs),
)
@router.callback_query(AdminClusterCallback.filter(F.action == "apply_tariffs"))
async def apply_tariffs(
callback: CallbackQuery, callback_data: AdminClusterCallback, session: AsyncSession, state: FSMContext
):
try:
cluster_name, idx_str = callback_data.data.split("|", 1)
i = int(idx_str)
res = await session.execute(select(Server.tariff_group).where(Server.cluster_name == cluster_name).distinct())
group_codes = [r[0] for r in res.fetchall() if r[0]]
if not group_codes:
await callback.answer("Не найдена тарифная группа кластера", show_alert=True)
return
group_code = group_codes[0]
res2 = await session.execute(
select(func.distinct(Tariff.subgroup_title))
.where(Tariff.group_code == group_code)
.where(Tariff.subgroup_title.isnot(None))
.order_by(Tariff.subgroup_title.asc())
)
subgroups = [r[0] for r in res2.fetchall()]
if i < 0 or i >= len(subgroups):
await callback.answer("Подгруппа не найдена", show_alert=True)
return
subgroup_title = subgroups[i]
key = f"subgrp_sel:{cluster_name}"
cluster_name = callback_data.data
data = await state.get_data()
selected = set(data.get(key, []))
if not selected:
await callback.message.edit_text("❌ Не выбраны серверы для назначения подгруппы.")
selected_servers = set(data.get(f"subgrp_sel:{cluster_name}", []))
selected_tariffs = set(data.get(f"tariff_sel:{cluster_name}", []))
if not selected_servers:
await callback.answer("Не выбраны серверы", show_alert=True)
return
if not selected_tariffs:
await callback.answer("Не выбраны тарифы", show_alert=True)
return
servers_q = await session.execute(select(Server.id, Server.server_name).where(Server.server_name.in_(selected)))
id_by_name = {name: sid for sid, name in servers_q.fetchall()}
missing_ids = [id_by_name[n] for n in selected if n in id_by_name]
if not missing_ids:
servers_q = await session.execute(
select(Server.id, Server.server_name, Server.tariff_group)
.where(Server.server_name.in_(selected_servers))
)
servers_data = servers_q.fetchall()
server_ids = [row[0] for row in servers_data]
group_code = servers_data[0][2] if servers_data else "standard"
if not server_ids:
await callback.answer("Серверы не найдены", show_alert=True)
return
existing_q = await session.execute(
select(ServerSubgroup.server_id)
.where(ServerSubgroup.server_id.in_(missing_ids))
.where(ServerSubgroup.subgroup_title == subgroup_title)
selected_tariff_strs = {str(tid) for tid in selected_tariffs}
await session.execute(
delete(ServerSubgroup)
.where(ServerSubgroup.server_id.in_(server_ids))
.where(ServerSubgroup.subgroup_title.regexp_match(r'^\d+$'))
.where(ServerSubgroup.subgroup_title.notin_(selected_tariff_strs))
)
already = {r[0] for r in existing_q.fetchall()}
to_insert = [sid for sid in missing_ids if sid not in already]
if to_insert:
session.add_all([
ServerSubgroup(server_id=sid, group_code=group_code, subgroup_title=subgroup_title) for sid in to_insert
])
await session.commit()
for tariff_id in selected_tariffs:
tariff_id_str = str(tariff_id)
existing_q = await session.execute(
select(ServerSubgroup.server_id)
.where(ServerSubgroup.server_id.in_(server_ids))
.where(ServerSubgroup.subgroup_title == tariff_id_str)
)
already = {r[0] for r in existing_q.fetchall()}
to_insert = [sid for sid in server_ids if sid not in already]
if to_insert:
session.add_all([
ServerSubgroup(server_id=sid, group_code=group_code, subgroup_title=tariff_id_str)
for sid in to_insert
])
await session.commit()
await state.update_data({key: []})
await state.update_data({
f"subgrp_sel:{cluster_name}": [],
f"tariff_sel:{cluster_name}": [],
})
servers = await get_servers(session, include_enabled=True)
cluster_servers = servers.get(cluster_name, [])
text = render_attach_tariff_menu_text(cluster_name, cluster_servers)
all_tariff_ids = set()
for s in cluster_servers:
all_tariff_ids.update(s.get("tariff_ids") or [])
tariffs_cache = {}
if all_tariff_ids:
result = await session.execute(select(Tariff).where(Tariff.id.in_(all_tariff_ids)))
for t in result.scalars().all():
tariffs_cache[t.id] = {
"id": t.id,
"name": t.name,
"subgroup_title": t.subgroup_title,
"group_code": t.group_code,
}
text = render_attach_tariff_menu_text(cluster_name, cluster_servers, tariffs_cache)
await callback.message.edit_text(
text=text,
reply_markup=build_attach_tariff_kb(cluster_name),
@@ -242,8 +328,8 @@ async def apply_tariff_subgroup(
)
except Exception as e:
logger.error(f"Ошибка при применении подгруппы тарифов: {e}")
await callback.message.edit_text("❌ Произошла ошибка при назначении подгруппы.")
logger.error(f"Ошибка при применении тарифов: {e}")
await callback.message.edit_text("❌ Произошла ошибка при назначении тарифов.")
@router.callback_query(AdminClusterCallback.filter(F.action == "reset_cluster_subgroups"))
@@ -272,11 +358,23 @@ async def reset_cluster_subgroups(callback: CallbackQuery, callback_data: AdminC
await callback.message.edit_text("❌ Не удалось сбросить подгруппы.")
def render_attach_tariff_menu_text(cluster_name: str, cluster_servers: list[dict]) -> str:
sub_map: dict[str, list[str]] = {}
def render_attach_tariff_menu_text(
cluster_name: str,
cluster_servers: list[dict],
tariffs_cache: dict[int, dict] | None = None
) -> str:
tariff_map: dict[int, list[str]] = {}
legacy_map: dict[str, list[str]] = {}
for s in cluster_servers:
server_name = s["server_name"]
for tid in s.get("tariff_ids") or []:
tariff_map.setdefault(tid, []).append(server_name)
for sg in s.get("tariff_subgroups") or []:
sub_map.setdefault(sg, []).append(s["server_name"])
legacy_map.setdefault(sg, []).append(server_name)
allowed = tuple(ALLOWED_GROUP_CODES)
spec_map: dict[str, list[str]] = {k: [] for k in allowed}
@@ -287,24 +385,57 @@ def render_attach_tariff_menu_text(cluster_name: str, cluster_servers: list[dict
lines = [f"<b>🧩 Привязки тарифов • {cluster_name}</b>"]
lines.append("<b>Подгруппы:</b>")
if sub_map:
subs_lines = []
for k in sorted(sub_map):
servers_list = ", ".join(sorted(set(sub_map[k])))
subs_lines.append(f"• <b>{k}</b>: {servers_list}")
lines.append("<blockquote>\n" + "\n".join(subs_lines) + "\n</blockquote>")
lines.append("\n<b>📋 Тарифы:</b>")
if tariff_map and tariffs_cache:
grouped: dict[str | None, list[tuple[int, str, list[str]]]] = {}
for tid, servers in tariff_map.items():
tariff = tariffs_cache.get(tid, {})
subgroup = tariff.get("subgroup_title")
name = tariff.get("name", f"ID:{tid}")
grouped.setdefault(subgroup, []).append((tid, name, servers))
tariff_lines = []
subgroups_sorted = sorted(grouped.keys(), key=lambda x: (x is None, x or ""))
for subgroup in subgroups_sorted:
tariffs_list = grouped[subgroup]
if subgroup:
tariff_lines.append(f"<b>{subgroup}</b>")
for tid, name, servers in sorted(tariffs_list, key=lambda x: x[1]):
servers_str = ", ".join(sorted(set(servers)))
tariff_lines.append(f" └ {name}: {servers_str}")
else:
for tid, name, servers in sorted(tariffs_list, key=lambda x: x[1]):
servers_str = ", ".join(sorted(set(servers)))
tariff_lines.append(f"• {name}: {servers_str}")
lines.append("<blockquote>" + "\n".join(tariff_lines) + "</blockquote>")
elif tariff_map:
tariff_lines = []
for tid, servers in sorted(tariff_map.items()):
servers_str = ", ".join(sorted(set(servers)))
tariff_lines.append(f"• ID:{tid}: {servers_str}")
lines.append("<blockquote>" + "\n".join(tariff_lines) + "</blockquote>")
else:
lines.append("<blockquote>— нет привязок</blockquote>")
lines.append("<b>Спецгруппы:</b>")
if legacy_map:
lines.append("\n<b>⚠️ Старые привязки (по названию):</b>")
legacy_lines = []
for k in sorted(legacy_map):
servers_list = ", ".join(sorted(set(legacy_map[k])))
legacy_lines.append(f"• <b>{k}</b>: {servers_list}")
lines.append("<blockquote>" + "\n".join(legacy_lines) + "</blockquote>")
lines.append("<i>Рекомендуется сбросить и настроить заново</i>")
lines.append("\n<b>🎁 Спецгруппы:</b>")
has_spec = any(spec_map[k] for k in allowed)
if has_spec:
spec_lines = []
for k in allowed:
vals = sorted(set(spec_map[k]))
spec_lines.append(f"• <b>{k}</b>: {', '.join(vals) if vals else '—'}")
lines.append("<blockquote>\n" + "\n".join(spec_lines) + "\n</blockquote>")
lines.append("<blockquote>" + "\n".join(spec_lines) + "</blockquote>")
else:
lines.append("<blockquote>— нет привязок</blockquote>")
@@ -319,7 +450,24 @@ async def handle_attach_tariff_menu(callback: CallbackQuery, session: AsyncSessi
servers = await get_servers(session, include_enabled=True)
cluster_servers = servers.get(cluster_name, [])
text = render_attach_tariff_menu_text(cluster_name, cluster_servers)
all_tariff_ids = set()
for s in cluster_servers:
all_tariff_ids.update(s.get("tariff_ids") or [])
tariffs_cache = {}
if all_tariff_ids:
result = await session.execute(
select(Tariff).where(Tariff.id.in_(all_tariff_ids))
)
for t in result.scalars().all():
tariffs_cache[t.id] = {
"id": t.id,
"name": t.name,
"subgroup_title": t.subgroup_title,
"group_code": t.group_code,
}
text = render_attach_tariff_menu_text(cluster_name, cluster_servers, tariffs_cache)
await callback.message.edit_text(
text=text,
reply_markup=build_attach_tariff_kb(cluster_name),
handlers/admin/clusters/cluster_wizard.py
+4 -2
View File
@@ -41,9 +41,11 @@ async def handle_servers(callback_query: CallbackQuery, session: AsyncSession):
if message and message.text:
await message.edit_text(text=text, reply_markup=markup)
elif message and message.caption:
await message.edit_caption(caption=text, reply_markup=markup)
else:
try:
await message.delete()
except Exception:
pass
await message.answer(text=text, reply_markup=markup)
handlers/admin/clusters/keyboard.py
+74 -8
View File
@@ -75,6 +75,16 @@ def build_manage_cluster_kb(cluster_servers: list, cluster_name: str) -> InlineK
def build_attach_tariff_kb(cluster_name: str) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
builder.row(
InlineKeyboardButton(
text="📋 Привязать тарифы",
callback_data=AdminClusterCallback(action="set_subgroup", data=cluster_name).pack(),
),
InlineKeyboardButton(
text="🧹 Сбросить",
callback_data=AdminClusterCallback(action="reset_cluster_subgroups", data=cluster_name).pack(),
),
)
builder.row(
InlineKeyboardButton(
text="🗂 Спецгруппы",
@@ -87,18 +97,25 @@ def build_attach_tariff_kb(cluster_name: str) -> InlineKeyboardMarkup:
)
builder.row(
InlineKeyboardButton(
text="🗂 Подгруппы",
callback_data=AdminClusterCallback(action="set_subgroup", data=cluster_name).pack(),
),
text="🔙 Назад",
callback_data=AdminClusterCallback(action="manage", data=cluster_name).pack(),
)
)
return builder.as_markup()
def build_legacy_reset_kb(cluster_name: str) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
builder.row(
InlineKeyboardButton(
text="🧹 Сбросить",
text="🧹 Сбросить привязки",
callback_data=AdminClusterCallback(action="reset_cluster_subgroups", data=cluster_name).pack(),
),
)
)
builder.row(
InlineKeyboardButton(
text="🔙 Назад",
callback_data=AdminClusterCallback(action="manage", data=cluster_name).pack(),
callback_data=AdminClusterCallback(action="attach_tariff_menu", data=cluster_name).pack(),
)
)
return builder.as_markup()
@@ -128,7 +145,7 @@ def build_select_subgroup_servers_kb(
builder.row(
InlineKeyboardButton(
text="📚 Выбрать подгруппу",
text="📋 Выбрать тарифы",
callback_data=AdminClusterCallback(action="choose_subgroup", data=cluster_name).pack(),
)
)
@@ -141,7 +158,7 @@ def build_select_subgroup_servers_kb(
builder.row(
InlineKeyboardButton(
text="🔙 Назад",
callback_data=AdminClusterCallback(action="manage", data=cluster_name).pack(),
callback_data=AdminClusterCallback(action="attach_tariff_menu", data=cluster_name).pack(),
)
)
@@ -165,6 +182,55 @@ def build_tariff_subgroup_selection_kb(cluster_name: str, subgroups: list[str])
return builder.as_markup()
def build_tariff_selection_kb(cluster_name: str, tariffs: list, selected: set[int]) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
grouped: dict[str | None, list] = {}
for t in tariffs:
subgroup = t.subgroup_title
grouped.setdefault(subgroup, []).append(t)
subgroups_sorted = sorted(grouped.keys(), key=lambda x: (x is None, x or ""))
for subgroup in subgroups_sorted:
tariffs_list = grouped[subgroup]
if subgroup:
builder.row(
InlineKeyboardButton(
text=f"━━ {subgroup} ━━",
callback_data="noop",
)
)
for t in tariffs_list:
mark = "✅" if t.id in selected else "⬜️"
builder.row(
InlineKeyboardButton(
text=f"{mark} {t.name}",
callback_data=AdminClusterCallback(
action="toggle_tariff",
data=f"{cluster_name}|{t.id}"
).pack(),
)
)
builder.row(
InlineKeyboardButton(
text="✅ Применить",
callback_data=AdminClusterCallback(action="apply_tariffs", data=cluster_name).pack(),
)
)
builder.row(
InlineKeyboardButton(
text="⬅️ Назад к выбору серверов",
callback_data=AdminClusterCallback(action="set_subgroup", data=cluster_name).pack(),
)
)
return builder.as_markup()
def build_cluster_management_kb(cluster_name: str) -> InlineKeyboardMarkup:
builder = InlineKeyboardBuilder()
handlers/admin/users/users_gifts.py
+1 -1
View File
@@ -124,7 +124,7 @@ async def handle_gift_delete(
await callback.message.edit_text(
text=(
f"❓ <b>Удалить подарок?</b>\n\n"
f"📆 Длительность: {gift.selected_months} дн.\n"
f"📆 Длительность: {gift.selected_months} мес.\n"
f"📅 Создан: {created_str}\n"
f"📊 Статус: {status}\n\n"
f"⚠️ Это действие необратимо!"
handlers/keys/key_mode/key_country_mode.py
+4 -1
View File
@@ -173,7 +173,9 @@ async def key_country_mode(
server["special_groups"] = [g for g in groups_map.get(server["id"], []) if g in ALLOWED_GROUP_CODES]
if subgroup_title:
servers = await filter_cluster_by_subgroup(session, servers, subgroup_title, least_loaded_cluster)
servers = await filter_cluster_by_subgroup(
session, servers, subgroup_title, least_loaded_cluster, tariff_id=plan
)
if not servers:
text = "❌ Нет доступных серверов в выбранном кластере."
if safe_to_edit:
@@ -340,6 +342,7 @@ async def change_location_callback(callback_query: CallbackQuery, session: Any):
available_servers_dict,
subgroup_title.strip(),
cluster_name,
tariff_id=key_tariff_id,
)
if filtered_servers:
available_servers = [s["server_name"] for s in filtered_servers]
handlers/keys/operations/aggregated_links.py
+1 -1
View File
@@ -103,7 +103,7 @@ async def make_aggregated_link(
legacy_links_enabled = bool(MODES_CONFIG.get("LEGACY_LINKS_ENABLED", LEGACY_LINKS))
servers = (
await filter_cluster_by_subgroup(session, cluster_all, subgroup_code, cluster_id)
await filter_cluster_by_subgroup(session, cluster_all, subgroup_code, cluster_id, tariff_id=plan)
if subgroup_code
else cluster_all
)
handlers/keys/operations/creation.py
+4 -2
View File
@@ -6,7 +6,7 @@ from sqlalchemy import update
from sqlalchemy.ext.asyncio import AsyncSession
from config import PUBLIC_LINK, REMNAWAVE_LOGIN, REMNAWAVE_PASSWORD, SUPERNODE
from database import get_servers, get_tariff_by_id, store_key
from database import filter_cluster_by_subgroup, get_servers, get_tariff_by_id, store_key
from database.models import User
from handlers.utils import ALLOWED_GROUP_CODES, check_server_key_limit
from hooks.processors import process_extract_cryptolink_from_result
@@ -97,7 +97,9 @@ async def create_key_on_cluster(
device_limit_value = int(hwid_limit or 0)
if subgroup_title:
subgroup_servers = [s for s in enabled_servers if subgroup_title in s.get("tariff_subgroups", [])]
subgroup_servers = await filter_cluster_by_subgroup(
session, enabled_servers, subgroup_title, cluster_id, tariff_id=plan
)
if subgroup_servers:
enabled_servers = subgroup_servers
else:
handlers/keys/operations/renewal.py
+4 -1
View File
@@ -252,6 +252,7 @@ async def renew_key_in_cluster(
old_subgroup=old_subgroup,
target_subgroup=target_subgroup,
external_squad_uuid=external_squad_uuid,
tariff_id=plan,
)
await update_key_expiry(session, new_client_id or client_id, new_expiry_time)
@@ -281,7 +282,9 @@ async def renew_key_in_cluster(
cluster_scope = [single_server]
else:
if target_subgroup:
target = await filter_cluster_by_subgroup(session, cluster, target_subgroup, cluster_id)
target = await filter_cluster_by_subgroup(
session, cluster, target_subgroup, cluster_id, tariff_id=plan
)
cluster_scope = target if target else cluster
else:
cluster_scope = cluster
handlers/keys/operations/update.py
+8 -2
View File
@@ -33,6 +33,7 @@ async def update_key_on_cluster(
device_limit: int = None,
remnawave_link: str = None,
subgroup_code: str | None = None,
tariff_id: int | None = None,
):
try:
servers = await get_servers(session)
@@ -50,7 +51,9 @@ async def update_key_on_cluster(
raise ValueError(f"Кластер или сервер с ID/именем {cluster_id} не найден.")
if subgroup_code:
filtered = await filter_cluster_by_subgroup(session, cluster, subgroup_code, cluster_id)
filtered = await filter_cluster_by_subgroup(
session, cluster, subgroup_code, cluster_id, tariff_id=tariff_id
)
if not filtered:
logger.warning(f"[Update] Нет серверов для подгруппы {subgroup_code} в кластере {cluster_id}.")
return client_id, remnawave_link
@@ -238,7 +241,9 @@ async def update_subscription(
cluster_servers = []
if subgroup_code:
prefiltered = await filter_cluster_by_subgroup(session, cluster_servers, subgroup_code, new_cluster_id)
prefiltered = await filter_cluster_by_subgroup(
session, cluster_servers, subgroup_code, new_cluster_id, tariff_id=tariff_id
)
if not prefiltered:
logger.warning(
f"[Update] Пересоздание пропущено: нет серверов под подгруппу {subgroup_code} в {new_cluster_id}."
@@ -273,6 +278,7 @@ async def update_subscription(
device_limit=device_limit,
remnawave_link=remnawave_link,
subgroup_code=subgroup_code,
tariff_id=tariff_id,
)
aggregated = await make_aggregated_link(
handlers/notifications/special_notifications.py
+39 -15
View File
@@ -5,6 +5,7 @@ import pytz
from aiogram import Bot, Router, types
from aiogram.types import InlineKeyboardButton, WebAppInfo
from aiogram.utils.keyboard import InlineKeyboardBuilder
from sqlalchemy import update
from sqlalchemy.ext.asyncio import AsyncSession
from config import (
@@ -15,12 +16,8 @@ from config import (
SUPPORT_CHAT_URL,
)
from core.bootstrap import MODES_CONFIG, NOTIFICATIONS_CONFIG
from database import (
add_notification,
check_notifications_bulk,
mark_trial_extended,
update_key_notified,
)
from database import add_notification, check_notifications_bulk
from database.models import Key, User
from database.tariffs import get_tariffs
from handlers.buttons import CONNECT_DEVICE, MAIN_MENU, SUPPORT, TRIAL_BONUS
from handlers.keys.operations import get_user_traffic
@@ -52,7 +49,10 @@ async def notify_inactive_trial_users(bot: Bot, session: AsyncSession):
users = await check_notifications_bulk(session, "inactive_trial", inactive_hours)
logger.info(f"Найдено {len(users)} неактивных пользователей для уведомления.")
messages = []
if not users:
logger.info("Проверка пользователей с неактивным пробным периодом завершена.")
return
trial_tariffs = await get_tariffs(session, group_code="trial")
if not trial_tariffs:
@@ -60,6 +60,8 @@ async def notify_inactive_trial_users(bot: Bot, session: AsyncSession):
return
trial_days = trial_tariffs[0]["duration_days"]
messages = []
users_to_extend = []
for user in users:
tg_id = user["tg_id"]
@@ -82,7 +84,7 @@ async def notify_inactive_trial_users(bot: Bot, session: AsyncSession):
extra_days_formatted=format_days(extra_days),
total_days_formatted=format_days(total_days),
)
await mark_trial_extended(tg_id, session)
users_to_extend.append(tg_id)
else:
message = TRIAL_INACTIVE_FIRST_MSG.format(
display_name=display_name,
@@ -104,19 +106,31 @@ async def notify_inactive_trial_users(bot: Bot, session: AsyncSession):
source_file="special_notifications",
messages_per_second=25,
)
sent_count = 0
sent_tg_ids = []
for msg, result in zip(messages, results, strict=False):
if result:
await add_notification(session, msg["tg_id"], msg["notification_id"])
sent_count += 1
logger.info(f"Отправлено {sent_count} уведомлений неактивным пользователям.")
sent_tg_ids.append(msg["tg_id"])
if sent_tg_ids:
for tg_id in sent_tg_ids:
await add_notification(session, tg_id, "inactive_trial")
logger.info(f"Отправлено {len(sent_tg_ids)} уведомлений неактивным пользователям.")
extend_ids = [tg_id for tg_id in users_to_extend if tg_id in sent_tg_ids]
if extend_ids:
await session.execute(
update(User).where(User.tg_id.in_(extend_ids)).values(trial_extended=True)
)
await session.commit()
logger.info(f"Bulk: отмечено {len(extend_ids)} пользователей с расширенным триалом")
logger.info("Проверка пользователей с неактивным пробным периодом завершена.")
async def notify_users_no_traffic(bot: Bot, session: AsyncSession, current_time: int, keys: list):
logger.info("Проверка пользователей с нулевым трафиком...")
current_dt = datetime.fromtimestamp(current_time / 1000, tz=moscow_tz)
messages = []
inactive_traffic_hours = int(NOTIFICATIONS_CONFIG.get("INACTIVE_TRAFFIC_ENABLED", NOTIFY_INACTIVE_TRAFFIC))
if inactive_traffic_hours <= 0:
@@ -124,6 +138,9 @@ async def notify_users_no_traffic(bot: Bot, session: AsyncSession, current_time:
return
remnawave_webapp_enabled = bool(MODES_CONFIG.get("REMNAWAVE_WEBAPP_ENABLED", REMNAWAVE_WEBAPP))
messages = []
keys_to_mark_notified = []
for key in keys:
tg_id = key.tg_id
@@ -145,6 +162,8 @@ async def notify_users_no_traffic(bot: Bot, session: AsyncSession, current_time:
if current_dt > expiry_dt:
continue
keys_to_mark_notified.append(client_id)
try:
traffic_data = await get_user_traffic(session, tg_id, email)
except Exception as error:
@@ -201,10 +220,15 @@ async def notify_users_no_traffic(bot: Bot, session: AsyncSession, current_time:
"client_id": client_id,
})
if keys_to_mark_notified:
try:
await update_key_notified(session, tg_id, client_id)
await session.execute(
update(Key).where(Key.client_id.in_(keys_to_mark_notified)).values(notified=True)
)
await session.commit()
logger.info(f"Bulk: отмечено {len(keys_to_mark_notified)} ключей как notified")
except Exception as error:
logger.error(f"Ошибка обновления notified для {tg_id} ({client_id}): {error}")
logger.error(f"Ошибка bulk-обновления notified: {error}")
if messages:
results = await send_messages_with_limit(
handlers/tariffs/subgroup_migration.py
+4 -1
View File
@@ -242,8 +242,11 @@ async def migrate_between_subgroups(
old_subgroup: str,
target_subgroup: str,
external_squad_uuid: str | None = None,
tariff_id: int | None = None,
) -> tuple[str, str | None]:
target = await filter_cluster_by_subgroup(session, cluster_all, target_subgroup, cluster_id)
target = await filter_cluster_by_subgroup(
session, cluster_all, target_subgroup, cluster_id, tariff_id=tariff_id
)
xui_tgt, remna_tgt = split_by_panel(target)
old_set = await filter_cluster_by_subgroup(session, cluster_all, old_subgroup, cluster_id)
panels/remnawave.cpython-312-x86_64-linux-gnu.so
BIN
View File
Binary file not shown.