195 lines
7.8 KiB
Python
195 lines
7.8 KiB
Python
from __future__ import annotations
|
||
|
||
from typing import TYPE_CHECKING
|
||
|
||
from PyQt6.QtCore import QTimer
|
||
|
||
from ..profiles.geoip import normalize_country
|
||
from ..importer.link_parser import parse_links_text, validate_node_outbound
|
||
|
||
if TYPE_CHECKING:
|
||
from .controller import AppController
|
||
|
||
|
||
def import_nodes_from_text(controller: AppController, text: str) -> tuple[int, list[str]]:
|
||
nodes, errors = parse_links_text(text)
|
||
if not nodes:
|
||
return 0, errors
|
||
|
||
existing_links = {node.link for node in controller.state.nodes if node.link}
|
||
max_order = max((node.sort_order for node in controller.state.nodes), default=0)
|
||
first_new_id: str | None = None
|
||
added = 0
|
||
for node in nodes:
|
||
problem = validate_node_outbound(node)
|
||
if problem:
|
||
errors.append(problem)
|
||
continue
|
||
if node.link and node.link in existing_links:
|
||
continue
|
||
max_order += 1
|
||
node.sort_order = max_order
|
||
controller.state.nodes.append(node)
|
||
if node.link:
|
||
existing_links.add(node.link)
|
||
if first_new_id is None:
|
||
first_new_id = node.id
|
||
added += 1
|
||
|
||
if first_new_id:
|
||
controller.state.selected_node_id = first_new_id
|
||
elif not controller.state.selected_node_id and controller.state.nodes:
|
||
controller.state.selected_node_id = controller.state.nodes[0].id
|
||
|
||
controller.nodes_changed.emit(controller.state.nodes)
|
||
controller.selection_changed.emit(controller.selected_node)
|
||
controller.save()
|
||
QTimer.singleShot(500, controller._start_country_ip_resolution)
|
||
|
||
if added:
|
||
controller._desired_connected = True
|
||
controller._request_transition("new node imported")
|
||
|
||
return added, errors
|
||
|
||
|
||
def remove_nodes(controller: AppController, node_ids: set[str]) -> None:
|
||
if not node_ids:
|
||
return
|
||
removed_selected = controller.state.selected_node_id in node_ids
|
||
should_reconcile = removed_selected and (controller.connected or controller._desired_connected)
|
||
controller.state.nodes = [node for node in controller.state.nodes if node.id not in node_ids]
|
||
if removed_selected:
|
||
controller.state.selected_node_id = controller.state.nodes[0].id if controller.state.nodes else None
|
||
controller._reset_auto_switch_state(reset_cooldown=True, reset_cycle=True)
|
||
controller.nodes_changed.emit(controller.state.nodes)
|
||
controller.selection_changed.emit(controller.selected_node)
|
||
controller.save()
|
||
if not should_reconcile:
|
||
return
|
||
if controller.state.selected_node_id is None:
|
||
if controller._can_connect_without_selected_node():
|
||
controller._request_transition("active node removed")
|
||
return
|
||
controller._desired_connected = False
|
||
controller._request_transition("active node removed")
|
||
return
|
||
controller._desired_connected = True
|
||
controller._request_transition("active node removed")
|
||
|
||
|
||
def update_node(controller: AppController, node_id: str, updates: dict) -> bool:
|
||
node = controller._get_node_by_id(node_id)
|
||
if not node:
|
||
return False
|
||
if "is_favorite" in updates:
|
||
node.is_favorite = bool(updates["is_favorite"])
|
||
if "country_override" in updates:
|
||
node.country_override = normalize_country(updates["country_override"])
|
||
node.country_code = node.country_override
|
||
controller._start_country_ip_resolution()
|
||
if "name" in updates:
|
||
node.name = updates["name"]
|
||
if "group" in updates:
|
||
node.group = updates["group"]
|
||
if "tags" in updates:
|
||
node.tags = list(updates["tags"])
|
||
controller.nodes_changed.emit(controller.state.nodes)
|
||
controller.save()
|
||
return True
|
||
|
||
|
||
def bulk_update_nodes(controller: AppController, node_ids: set[str], operations: dict) -> int:
|
||
group = operations.get("group", "")
|
||
add_tags = operations.get("add_tags", [])
|
||
remove_tags = set(operations.get("remove_tags", []))
|
||
updated = 0
|
||
for node in controller.state.nodes:
|
||
if node.id not in node_ids:
|
||
continue
|
||
if group:
|
||
node.group = group
|
||
if add_tags:
|
||
existing = set(node.tags)
|
||
for tag in add_tags:
|
||
if tag not in existing:
|
||
node.tags.append(tag)
|
||
if remove_tags:
|
||
node.tags = [tag for tag in node.tags if tag not in remove_tags]
|
||
updated += 1
|
||
if updated:
|
||
controller.nodes_changed.emit(controller.state.nodes)
|
||
controller.save()
|
||
return updated
|
||
|
||
|
||
def get_all_groups(controller: AppController) -> list[str]:
|
||
return sorted({node.group for node in controller.state.nodes if node.group})
|
||
|
||
|
||
def get_all_tags(controller: AppController) -> list[str]:
|
||
tags: set[str] = set()
|
||
for node in controller.state.nodes:
|
||
tags.update(node.tags)
|
||
return sorted(tags)
|
||
|
||
|
||
def reorder_nodes(controller: AppController, node_id: str, direction: str) -> None:
|
||
ordered = sorted(controller.state.nodes, key=lambda node: node.sort_order)
|
||
idx = next((i for i, node in enumerate(ordered) if node.id == node_id), None)
|
||
if idx is None:
|
||
return
|
||
if direction == "up" and idx > 0:
|
||
ordered[idx], ordered[idx - 1] = ordered[idx - 1], ordered[idx]
|
||
elif direction == "down" and idx < len(ordered) - 1:
|
||
ordered[idx], ordered[idx + 1] = ordered[idx + 1], ordered[idx]
|
||
elif direction == "top" and idx > 0:
|
||
node = ordered.pop(idx)
|
||
ordered.insert(0, node)
|
||
elif direction == "bottom" and idx < len(ordered) - 1:
|
||
node = ordered.pop(idx)
|
||
ordered.append(node)
|
||
else:
|
||
return
|
||
for index, node in enumerate(ordered):
|
||
node.sort_order = index + 1
|
||
controller.nodes_changed.emit(controller.state.nodes)
|
||
controller.save()
|
||
|
||
|
||
def set_selected_node(controller: AppController, node_id: str, *, reset_auto_switch: bool = True) -> None:
|
||
from .protocol_core import ProtocolCore, protocol_core
|
||
target = next((node for node in controller.state.nodes if node.id == node_id), None)
|
||
session = getattr(controller, "_active_session", None)
|
||
# Re-selecting the current or already requested row is idempotent.
|
||
pending = getattr(controller, "_pending_transport_node_id", None)
|
||
if node_id == (pending or controller.state.selected_node_id):
|
||
return
|
||
if controller.connected and target is not None and (
|
||
protocol_core(target) is ProtocolCore.AMNEZIA or
|
||
(session is not None and session.sidecar_kind == "amnezia")
|
||
):
|
||
controller._pending_transport_node_id = node_id
|
||
if reset_auto_switch:
|
||
controller._reset_auto_switch_state(reset_cooldown=True, reset_cycle=True)
|
||
controller._auto_switch_manual_hold = True
|
||
controller._desired_connected = True
|
||
controller._request_transition("transport node switched")
|
||
return
|
||
if controller.state.selected_node_id == node_id:
|
||
return
|
||
controller.state.selected_node_id = node_id
|
||
if reset_auto_switch:
|
||
# Ручной выбор сбрасывает cooldown/cycle авто-переключения; сам
|
||
# auto_switch_service выбирает ноду с reset_auto_switch=False, чтобы
|
||
# не обнулять свой учёт анти-дребезга (П4/A6). Заодно ручной выбор
|
||
# фиксирует сервер до явного повторного включения авто-переключения.
|
||
controller._reset_auto_switch_state(reset_cooldown=True, reset_cycle=True)
|
||
controller._auto_switch_manual_hold = True
|
||
controller.selection_changed.emit(controller.selected_node)
|
||
controller.schedule_save()
|
||
if controller.connected or controller._desired_connected:
|
||
controller._desired_connected = True
|
||
if controller.connected and controller._try_hot_switch_selected_node():
|
||
return
|
||
controller._request_transition("node switched")
|