zapret-kvn/xray_fluent/models.py
loop-uh a7cf61a05e
All checks were successful
Windows project source guards / test (push) Successful in 36s
Prepare v0.4.79: unified theming, responsive layout, real flags
- Single source of truth for theme: theme.py tokens, FluentDialog base
  for all dialogs, setCustomStyleSheet migration, theme applied before
  window creation, SystemThemeListener for auto mode
- Responsive layout: ScrollablePage base with as-needed horizontal
  scrolling, word wrap on dashboard labels, adaptive dashboard grid,
  realistic 860x560 window minimum
- Real country flags (flagcdn PNG set) with themed border and fallback
- Active server row marked by accent stripe only (no synthesized bold)
- Subscription polish from parallel work (QR utils, http, diagnostics)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-08-11 23:21:28 +03:00

508 lines
22 KiB
Python

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Any
import uuid
from .constants import ROUTING_RULE, STATE_SCHEMA_VERSION
def utc_now_iso() -> str:
return datetime.now(timezone.utc).isoformat()
@dataclass(slots=True)
class Node:
id: str = field(default_factory=lambda: str(uuid.uuid4()))
name: str = ""
scheme: str = ""
server: str = ""
port: int = 0
link: str = ""
outbound: dict[str, Any] = field(default_factory=dict)
group: str = "Default"
tags: list[str] = field(default_factory=list)
ping_ms: int | None = None
last_used_at: str | None = None
created_at: str = field(default_factory=utc_now_iso)
country_code: str = ""
speed_mbps: float | None = None
is_alive: bool | None = None
ping_history: list[tuple[str, int | None]] = field(default_factory=list)
speed_history: list[tuple[str, float | None]] = field(default_factory=list)
sort_order: int = 0
subscription_id: str | None = None
source_key: str = ""
provider_name: str = ""
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"scheme": self.scheme,
"server": self.server,
"port": self.port,
"link": self.link,
"outbound": self.outbound,
"group": self.group,
"tags": list(self.tags),
"ping_ms": self.ping_ms,
"last_used_at": self.last_used_at,
"created_at": self.created_at,
"country_code": self.country_code,
"speed_mbps": self.speed_mbps,
"is_alive": self.is_alive,
"ping_history": self.ping_history,
"speed_history": self.speed_history,
"sort_order": self.sort_order,
"subscription_id": self.subscription_id,
"source_key": self.source_key,
"provider_name": self.provider_name,
}
@staticmethod
def from_dict(data: dict[str, Any]) -> "Node":
return Node(
id=str(data.get("id") or uuid.uuid4()),
name=str(data.get("name") or ""),
scheme=str(data.get("scheme") or ""),
server=str(data.get("server") or ""),
port=int(data.get("port") or 0),
link=str(data.get("link") or ""),
outbound=dict(data.get("outbound") or {}),
group=str(data.get("group") or "Default"),
tags=list(data.get("tags") or []),
ping_ms=data.get("ping_ms"),
last_used_at=data.get("last_used_at"),
created_at=str(data.get("created_at") or utc_now_iso()),
country_code=str(data.get("country_code") or ""),
speed_mbps=data.get("speed_mbps"),
is_alive=data.get("is_alive"),
ping_history=data.get("ping_history", []),
speed_history=data.get("speed_history", []),
sort_order=int(data.get("sort_order", 0)),
subscription_id=(str(data.get("subscription_id")) if data.get("subscription_id") else None),
source_key=str(data.get("source_key") or ""),
provider_name=str(data.get("provider_name") or data.get("name") or ""),
)
@dataclass(slots=True)
class SubscriptionInfo:
upload: int = 0
download: int = 0
total: int = 0
expire: int = 0
web_page_url: str = ""
support_url: str = ""
@property
def used(self) -> int:
return max(0, self.upload) + max(0, self.download)
def to_dict(self) -> dict[str, Any]:
return {
"upload": self.upload,
"download": self.download,
"total": self.total,
"expire": self.expire,
"web_page_url": self.web_page_url,
"support_url": self.support_url,
}
@staticmethod
def from_dict(data: dict[str, Any]) -> "SubscriptionInfo":
def _number(key: str) -> int:
try:
return max(0, int(data.get(key) or 0))
except (TypeError, ValueError):
return 0
return SubscriptionInfo(
upload=_number("upload"),
download=_number("download"),
total=_number("total"),
expire=_number("expire"),
web_page_url=str(data.get("web_page_url") or ""),
support_url=str(data.get("support_url") or ""),
)
@dataclass(slots=True)
class Subscription:
id: str = field(default_factory=lambda: str(uuid.uuid4()))
name: str = ""
url: str = ""
user_agent: str = ""
client_profile: str = "zapret"
send_hwid: bool = False
hwid: str = ""
auto_update: bool = True
update_interval_hours: int | None = None
provider_interval_hours: int | None = None
include_pattern: str = ""
exclude_pattern: str = ""
etag: str = ""
last_modified: str = ""
last_checked_at: str | None = None
last_success_at: str | None = None
last_error: str = ""
failure_count: int = 0
backoff_until: str | None = None
info: SubscriptionInfo = field(default_factory=SubscriptionInfo)
pending_url: str = ""
hidden_source_keys: list[str] = field(default_factory=list)
created_at: str = field(default_factory=utc_now_iso)
sort_order: int = 0
@property
def effective_interval_hours(self) -> int:
value = self.update_interval_hours or self.provider_interval_hours or 24
return max(1, int(value))
def to_dict(self) -> dict[str, Any]:
return {
"id": self.id,
"name": self.name,
"url": self.url,
"user_agent": self.user_agent,
"client_profile": self.client_profile,
"send_hwid": self.send_hwid,
"hwid": self.hwid,
"auto_update": self.auto_update,
"update_interval_hours": self.update_interval_hours,
"provider_interval_hours": self.provider_interval_hours,
"include_pattern": self.include_pattern,
"exclude_pattern": self.exclude_pattern,
"etag": self.etag,
"last_modified": self.last_modified,
"last_checked_at": self.last_checked_at,
"last_success_at": self.last_success_at,
"last_error": self.last_error,
"failure_count": self.failure_count,
"backoff_until": self.backoff_until,
"info": self.info.to_dict(),
"pending_url": self.pending_url,
"hidden_source_keys": list(self.hidden_source_keys),
"created_at": self.created_at,
"sort_order": self.sort_order,
}
@staticmethod
def from_dict(data: dict[str, Any]) -> "Subscription":
def _optional_positive_int(key: str) -> int | None:
try:
value = int(data.get(key) or 0)
except (TypeError, ValueError):
return None
return value if value > 0 else None
def _non_negative_int(key: str) -> int:
try:
return max(0, int(data.get(key) or 0))
except (TypeError, ValueError):
return 0
return Subscription(
id=str(data.get("id") or uuid.uuid4()),
name=str(data.get("name") or ""),
url=str(data.get("url") or ""),
user_agent=str(data.get("user_agent") or ""),
client_profile=str(data.get("client_profile") or "zapret"),
send_hwid=bool(data.get("send_hwid", False)),
hwid=str(data.get("hwid") or ""),
auto_update=bool(data.get("auto_update", True)),
update_interval_hours=_optional_positive_int("update_interval_hours"),
provider_interval_hours=_optional_positive_int("provider_interval_hours"),
include_pattern=str(data.get("include_pattern") or ""),
exclude_pattern=str(data.get("exclude_pattern") or ""),
etag=str(data.get("etag") or ""),
last_modified=str(data.get("last_modified") or ""),
last_checked_at=data.get("last_checked_at"),
last_success_at=data.get("last_success_at"),
last_error=str(data.get("last_error") or ""),
failure_count=_non_negative_int("failure_count"),
backoff_until=data.get("backoff_until"),
info=SubscriptionInfo.from_dict(dict(data.get("info") or {})),
pending_url=str(data.get("pending_url") or ""),
hidden_source_keys=[str(item) for item in (data.get("hidden_source_keys") or [])],
created_at=str(data.get("created_at") or utc_now_iso()),
sort_order=_non_negative_int("sort_order"),
)
@dataclass(slots=True)
class SubscriptionUpdateResult:
subscription_id: str
success: bool
message: str = ""
added: int = 0
updated: int = 0
removed: int = 0
skipped: int = 0
warnings: list[str] = field(default_factory=list)
not_modified: bool = False
reconnect_required: bool = False
@dataclass(slots=True)
class RoutingSettings:
mode: str = ROUTING_RULE
bypass_lan: bool = True
direct_domains: list[str] = field(default_factory=list)
proxy_domains: list[str] = field(default_factory=list)
block_domains: list[str] = field(default_factory=list)
dns_mode: str = "system" # system | builtin
dns_bootstrap_server: str = "1.1.1.1" # DNS for direct traffic
dns_bootstrap_type: str = "udp" # udp | tcp | tls | https
dns_proxy_server: str = "8.8.8.8" # DNS for proxy traffic
dns_proxy_type: str = "tcp" # tcp | tls | https
process_rules: list[dict[str, str]] = field(default_factory=list) # [{"process": "chrome.exe", "action": "direct|proxy|block"}]
process_preset_routes: dict[str, str] = field(default_factory=dict) # {"telegram": "proxy", "windows_system": "direct"}
service_routes: dict[str, str] = field(default_factory=dict) # {"youtube": "proxy", "steam": "direct", ...}
tun_default_outbound: str = "direct" # "proxy" | "direct"
def to_dict(self) -> dict[str, Any]:
return {
"mode": self.mode,
"bypass_lan": self.bypass_lan,
"direct_domains": list(self.direct_domains),
"proxy_domains": list(self.proxy_domains),
"block_domains": list(self.block_domains),
"dns_mode": self.dns_mode,
"dns_bootstrap_server": self.dns_bootstrap_server,
"dns_bootstrap_type": self.dns_bootstrap_type,
"dns_proxy_server": self.dns_proxy_server,
"dns_proxy_type": self.dns_proxy_type,
"process_rules": list(self.process_rules),
"process_preset_routes": dict(self.process_preset_routes),
"service_routes": dict(self.service_routes),
"tun_default_outbound": self.tun_default_outbound,
}
@staticmethod
def from_dict(data: dict[str, Any]) -> "RoutingSettings":
return RoutingSettings(
mode=str(data.get("mode") or ROUTING_RULE),
bypass_lan=bool(data.get("bypass_lan", True)),
direct_domains=list(data.get("direct_domains") or []),
proxy_domains=list(data.get("proxy_domains") or []),
block_domains=list(data.get("block_domains") or []),
dns_mode=str(data.get("dns_mode") or "system"),
dns_bootstrap_server=str(data.get("dns_bootstrap_server") or "1.1.1.1"),
dns_bootstrap_type=str(data.get("dns_bootstrap_type") or "udp"),
dns_proxy_server=str(data.get("dns_proxy_server") or "8.8.8.8"),
dns_proxy_type=str(data.get("dns_proxy_type") or "tcp"),
process_rules=list(data.get("process_rules") or []),
process_preset_routes=dict(data.get("process_preset_routes") or {}),
service_routes=dict(data.get("service_routes") or {}),
tun_default_outbound=str(data.get("tun_default_outbound") or "direct"),
)
@dataclass(slots=True)
class SecuritySettings:
enabled: bool = False
password_hash: str = ""
salt: str = ""
auto_lock_minutes: int = 15
def to_dict(self) -> dict[str, Any]:
return {
"enabled": self.enabled,
"password_hash": self.password_hash,
"salt": self.salt,
"auto_lock_minutes": self.auto_lock_minutes,
}
@staticmethod
def from_dict(data: dict[str, Any]) -> "SecuritySettings":
return SecuritySettings(
enabled=bool(data.get("enabled", False)),
password_hash=str(data.get("password_hash") or ""),
salt=str(data.get("salt") or ""),
auto_lock_minutes=int(data.get("auto_lock_minutes") or 15),
)
@dataclass(slots=True)
class AppSettings:
theme: str = "system" # system | light | dark
accent_color: str = "#0078D4"
auto_connect_last: bool = True
start_minimized: bool = False
enable_system_proxy: bool = True
system_proxy_bypass_lan: bool = True
launch_on_startup: bool = False
reconnect_on_network_change: bool = True
xray_path: str = ""
log_level: str = "warning"
check_updates: bool = True
allow_updates: bool = True
release_channel: str = "stable" # stable | beta | nightly
update_feed_url: str = ""
xray_release_channel: str = "stable" # stable | beta | nightly
xray_update_feed_url: str = ""
xray_auto_update: bool = False
tun_mode: bool = False
proxy_engine: str = "singbox" # "singbox" | "xray"
tun_engine: str = "singbox" # "singbox" | "xray" | "tun2socks"
xray_config_file: str = ""
xray_template_file: str = ""
singbox_path: str = ""
singbox_config_file: str = ""
singbox_template_file: str = ""
window_width: int = 1000
window_height: int = 720
window_x: int = -1
window_y: int = -1
zapret_preset: str = ""
zapret_autostart: bool = False
auto_switch_enabled: bool = True
auto_switch_threshold_kbps: int = 50
auto_switch_delay_sec: int = 30
auto_switch_cooldown_sec: int = 60
# Server list view preferences (stable english keys; empty = defaults)
nodes_sort_key: str = "manual" # manual | name | group | type | ping | speed | last_used
nodes_sort_desc: bool = False
nodes_group_filter: str = ""
nodes_tag_filter: str = ""
nodes_source_filter: str = ""
nodes_visible_columns: list[str] = field(default_factory=list) # empty = default set
def to_dict(self) -> dict[str, Any]:
return {
"theme": self.theme,
"accent_color": self.accent_color,
"auto_connect_last": self.auto_connect_last,
"start_minimized": self.start_minimized,
"enable_system_proxy": self.enable_system_proxy,
"system_proxy_bypass_lan": self.system_proxy_bypass_lan,
"launch_on_startup": self.launch_on_startup,
"reconnect_on_network_change": self.reconnect_on_network_change,
"xray_path": self.xray_path,
"log_level": self.log_level,
"check_updates": self.check_updates,
"allow_updates": self.allow_updates,
"release_channel": self.release_channel,
"update_feed_url": self.update_feed_url,
"xray_release_channel": self.xray_release_channel,
"xray_update_feed_url": self.xray_update_feed_url,
"xray_auto_update": self.xray_auto_update,
"tun_mode": self.tun_mode,
"proxy_engine": self.proxy_engine,
"tun_engine": self.tun_engine,
"xray_config_file": self.xray_config_file,
"xray_template_file": self.xray_template_file,
"singbox_path": self.singbox_path,
"singbox_config_file": self.singbox_config_file,
"singbox_template_file": self.singbox_template_file,
"window_width": self.window_width,
"window_height": self.window_height,
"window_x": self.window_x,
"window_y": self.window_y,
"zapret_preset": self.zapret_preset,
"zapret_autostart": self.zapret_autostart,
"auto_switch_enabled": self.auto_switch_enabled,
"auto_switch_threshold_kbps": self.auto_switch_threshold_kbps,
"auto_switch_delay_sec": self.auto_switch_delay_sec,
"auto_switch_cooldown_sec": self.auto_switch_cooldown_sec,
"nodes_sort_key": self.nodes_sort_key,
"nodes_sort_desc": self.nodes_sort_desc,
"nodes_group_filter": self.nodes_group_filter,
"nodes_tag_filter": self.nodes_tag_filter,
"nodes_source_filter": self.nodes_source_filter,
"nodes_visible_columns": list(self.nodes_visible_columns),
}
@staticmethod
def from_dict(data: dict[str, Any]) -> "AppSettings":
return AppSettings(
theme=str(data.get("theme") or "system"),
accent_color=str(data.get("accent_color") or "#0078D4"),
auto_connect_last=bool(data.get("auto_connect_last", True)),
start_minimized=bool(data.get("start_minimized", False)),
enable_system_proxy=bool(data.get("enable_system_proxy", True)),
system_proxy_bypass_lan=bool(data.get("system_proxy_bypass_lan", True)),
launch_on_startup=bool(data.get("launch_on_startup", False)),
reconnect_on_network_change=bool(data.get("reconnect_on_network_change", True)),
xray_path=str(data.get("xray_path") or ""),
log_level=str(data.get("log_level") or "warning"),
check_updates=bool(data.get("check_updates", True)),
allow_updates=bool(data.get("allow_updates", True)),
release_channel=str(data.get("release_channel") or "stable"),
update_feed_url=str(data.get("update_feed_url") or ""),
xray_release_channel=str(data.get("xray_release_channel") or "stable"),
xray_update_feed_url=str(data.get("xray_update_feed_url") or ""),
xray_auto_update=bool(data.get("xray_auto_update", False)),
tun_mode=bool(data.get("tun_mode", False)),
proxy_engine=str(data.get("proxy_engine") or "singbox"),
tun_engine=str(data.get("tun_engine") or "singbox"),
xray_config_file=str(data.get("xray_config_file") or ""),
xray_template_file=str(data.get("xray_template_file") or ""),
singbox_path=str(data.get("singbox_path") or ""),
singbox_config_file=str(data.get("singbox_config_file") or ""),
singbox_template_file=str(data.get("singbox_template_file") or ""),
window_width=int(data.get("window_width") or 1000),
window_height=int(data.get("window_height") or 720),
window_x=int(data.get("window_x", -1)),
window_y=int(data.get("window_y", -1)),
zapret_preset=str(data.get("zapret_preset") or ""),
zapret_autostart=bool(data.get("zapret_autostart", False)),
auto_switch_enabled=bool(data.get("auto_switch_enabled", True)),
auto_switch_threshold_kbps=int(data.get("auto_switch_threshold_kbps") or 50),
auto_switch_delay_sec=int(data.get("auto_switch_delay_sec") or 30),
auto_switch_cooldown_sec=int(data.get("auto_switch_cooldown_sec") or 60),
nodes_sort_key=str(data.get("nodes_sort_key") or "manual"),
nodes_sort_desc=bool(data.get("nodes_sort_desc", False)),
nodes_group_filter=str(data.get("nodes_group_filter") or ""),
nodes_tag_filter=str(data.get("nodes_tag_filter") or ""),
nodes_source_filter=str(data.get("nodes_source_filter") or ""),
nodes_visible_columns=[str(item) for item in (data.get("nodes_visible_columns") or [])],
)
@dataclass(slots=True)
class AppState:
schema_version: int = STATE_SCHEMA_VERSION
selected_node_id: str | None = None
subscription_device_id: str = field(default_factory=lambda: str(uuid.uuid4()).upper())
nodes: list[Node] = field(default_factory=list)
subscriptions: list[Subscription] = field(default_factory=list)
routing: RoutingSettings = field(default_factory=RoutingSettings)
settings: AppSettings = field(default_factory=AppSettings)
security: SecuritySettings = field(default_factory=SecuritySettings)
def to_dict(self) -> dict[str, Any]:
return {
"schema_version": self.schema_version,
"selected_node_id": self.selected_node_id,
"subscription_device_id": self.subscription_device_id,
"nodes": [node.to_dict() for node in self.nodes],
"subscriptions": [subscription.to_dict() for subscription in self.subscriptions],
"routing": self.routing.to_dict(),
"settings": self.settings.to_dict(),
"security": self.security.to_dict(),
}
@staticmethod
def from_dict(data: dict[str, Any]) -> "AppState":
nodes_raw = data.get("nodes") or []
nodes = [Node.from_dict(item) for item in nodes_raw if isinstance(item, dict)]
subscriptions_raw = data.get("subscriptions") or []
subscriptions = [
Subscription.from_dict(item) for item in subscriptions_raw if isinstance(item, dict)
]
return AppState(
schema_version=int(data.get("schema_version") or STATE_SCHEMA_VERSION),
selected_node_id=data.get("selected_node_id"),
subscription_device_id=str(
data.get("subscription_device_id") or uuid.uuid4()
).upper(),
nodes=nodes,
subscriptions=subscriptions,
routing=RoutingSettings.from_dict(dict(data.get("routing") or {})),
settings=AppSettings.from_dict(dict(data.get("settings") or {})),
security=SecuritySettings.from_dict(dict(data.get("security") or {})),
)