diff --git a/database/servers.py b/database/servers.py index 7475e0f9..09301a74 100644 --- a/database/servers.py +++ b/database/servers.py @@ -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 diff --git a/handlers/admin/clusters/cluster_manage.py b/handlers/admin/clusters/cluster_manage.py index f780a4ca..9c4fc540 100644 --- a/handlers/admin/clusters/cluster_manage.py +++ b/handlers/admin/clusters/cluster_manage.py @@ -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] diff --git a/handlers/admin/clusters/cluster_sync.py b/handlers/admin/clusters/cluster_sync.py index d907d233..5bd747fc 100644 --- a/handlers/admin/clusters/cluster_sync.py +++ b/handlers/admin/clusters/cluster_sync.py @@ -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 diff --git a/handlers/admin/clusters/cluster_tariffs.py b/handlers/admin/clusters/cluster_tariffs.py index 48985e54..c7de23e5 100644 --- a/handlers/admin/clusters/cluster_tariffs.py +++ b/handlers/admin/clusters/cluster_tariffs.py @@ -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"⚠️ Обнаружены привязки старого формата\n\n" + f"Кластер {cluster_name} содержит привязки по названиям подгрупп.\n" + f"Для использования новой системы необходимо сбросить текущие привязки.\n\n" + f"После сброса вы сможете привязать тарифы по ID.", + 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"🗂 Выберите серверы в кластере {cluster_name} для назначения подгруппы тарифов:", + f"📋 Выберите серверы для привязки тарифов\n" + f"Кластер: {cluster_name}", 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"🗂 Выберите серверы в кластере {cluster_name} для назначения подгруппы тарифов:", + f"📋 Выберите серверы для привязки тарифов\n" + f"Кластер: {cluster_name}", 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"🗂 Выберите серверы в кластере {cluster_name} для назначения подгруппы тарифов:", + f"📋 Выберите серверы для привязки тарифов\n" + f"Кластер: {cluster_name}", 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"📚 Выберите подгруппу для {len(selected)} сервер(а/ов) кластера {cluster_name}:", - reply_markup=build_tariff_subgroup_selection_kb(cluster_name, subgroups), + f"📋 Выберите тарифы для {len(selected_servers)} сервер(а/ов)\n" + f"Кластер: {cluster_name}", + 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"📋 Выберите тарифы для {len(selected_servers)} сервер(а/ов)\n" + f"Кластер: {cluster_name}", + 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"🧩 Привязки тарифов • {cluster_name}"] - lines.append("Подгруппы:") - if sub_map: - subs_lines = [] - for k in sorted(sub_map): - servers_list = ", ".join(sorted(set(sub_map[k]))) - subs_lines.append(f"• {k}: {servers_list}") - lines.append("
\n" + "\n".join(subs_lines) + "\n
") + lines.append("\n📋 Тарифы:") + 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"{subgroup}") + 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("
" + "\n".join(tariff_lines) + "
") + 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("
" + "\n".join(tariff_lines) + "
") else: lines.append("
— нет привязок
") - lines.append("Спецгруппы:") + if legacy_map: + lines.append("\n⚠️ Старые привязки (по названию):") + legacy_lines = [] + for k in sorted(legacy_map): + servers_list = ", ".join(sorted(set(legacy_map[k]))) + legacy_lines.append(f"• {k}: {servers_list}") + lines.append("
" + "\n".join(legacy_lines) + "
") + lines.append("Рекомендуется сбросить и настроить заново") + + lines.append("\n🎁 Спецгруппы:") 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"• {k}: {', '.join(vals) if vals else '—'}") - lines.append("
\n" + "\n".join(spec_lines) + "\n
") + lines.append("
" + "\n".join(spec_lines) + "
") else: lines.append("
— нет привязок
") @@ -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), diff --git a/handlers/admin/clusters/cluster_wizard.py b/handlers/admin/clusters/cluster_wizard.py index aaf1c5cb..b52528a0 100644 --- a/handlers/admin/clusters/cluster_wizard.py +++ b/handlers/admin/clusters/cluster_wizard.py @@ -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) diff --git a/handlers/admin/clusters/keyboard.py b/handlers/admin/clusters/keyboard.py index ffb1fd54..ec88063c 100644 --- a/handlers/admin/clusters/keyboard.py +++ b/handlers/admin/clusters/keyboard.py @@ -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() diff --git a/handlers/admin/users/users_gifts.py b/handlers/admin/users/users_gifts.py index b91f9040..9d94cecf 100644 --- a/handlers/admin/users/users_gifts.py +++ b/handlers/admin/users/users_gifts.py @@ -124,7 +124,7 @@ async def handle_gift_delete( await callback.message.edit_text( text=( f"❓ Удалить подарок?\n\n" - f"📆 Длительность: {gift.selected_months} дн.\n" + f"📆 Длительность: {gift.selected_months} мес.\n" f"📅 Создан: {created_str}\n" f"📊 Статус: {status}\n\n" f"⚠️ Это действие необратимо!" diff --git a/handlers/keys/key_mode/key_country_mode.py b/handlers/keys/key_mode/key_country_mode.py index 2372c2a0..1544a2b3 100644 --- a/handlers/keys/key_mode/key_country_mode.py +++ b/handlers/keys/key_mode/key_country_mode.py @@ -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] diff --git a/handlers/keys/operations/aggregated_links.py b/handlers/keys/operations/aggregated_links.py index 9481c9b2..89ee2a9f 100644 --- a/handlers/keys/operations/aggregated_links.py +++ b/handlers/keys/operations/aggregated_links.py @@ -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 ) diff --git a/handlers/keys/operations/creation.py b/handlers/keys/operations/creation.py index 7afe83f5..82af9a17 100644 --- a/handlers/keys/operations/creation.py +++ b/handlers/keys/operations/creation.py @@ -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: diff --git a/handlers/keys/operations/renewal.py b/handlers/keys/operations/renewal.py index 9ce2aebc..e2a3682b 100644 --- a/handlers/keys/operations/renewal.py +++ b/handlers/keys/operations/renewal.py @@ -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 diff --git a/handlers/keys/operations/update.py b/handlers/keys/operations/update.py index 3af8cb8b..f8d6affa 100644 --- a/handlers/keys/operations/update.py +++ b/handlers/keys/operations/update.py @@ -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( diff --git a/handlers/notifications/special_notifications.py b/handlers/notifications/special_notifications.py index 37e25efb..77d74e00 100644 --- a/handlers/notifications/special_notifications.py +++ b/handlers/notifications/special_notifications.py @@ -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( diff --git a/handlers/tariffs/subgroup_migration.py b/handlers/tariffs/subgroup_migration.py index 143b478f..3d3c9e06 100644 --- a/handlers/tariffs/subgroup_migration.py +++ b/handlers/tariffs/subgroup_migration.py @@ -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) diff --git a/panels/remnawave.cpython-312-x86_64-linux-gnu.so b/panels/remnawave.cpython-312-x86_64-linux-gnu.so index 3d1f07e9..684a8ac2 100644 Binary files a/panels/remnawave.cpython-312-x86_64-linux-gnu.so and b/panels/remnawave.cpython-312-x86_64-linux-gnu.so differ