forked from Support/sp-hydra-veil-gui
358 lines
16 KiB
Python
Executable file
358 lines
16 KiB
Python
Executable file
import os
|
|
from datetime import datetime, timezone
|
|
from typing import Any, Callable, Iterable, Mapping
|
|
|
|
from core.controllers.ConfigurationController import ConfigurationController
|
|
from core.controllers.ProfileController import ProfileController
|
|
from core.errors.logger import logger
|
|
from core.models.BaseProfile import BaseProfile
|
|
from core.models.session.SessionProfile import SessionProfile
|
|
|
|
from gui.v2.actions.database_health import GuiStorageDatabaseError
|
|
from gui.v2.actions.profile_order import normalize_profile_order
|
|
from gui.v2.actions.profile_status import is_profile_enabled_for_gui
|
|
from gui.v2.actions.settings_data import build_verification_view, truncate_key
|
|
|
|
|
|
StatusCallback = Callable[[str], None]
|
|
LocationLookup = Callable[[str], Any]
|
|
|
|
ENCRYPTED_PROTOCOLS = ("wireguard", "hysteria2", "vless")
|
|
|
|
|
|
def _report_load_error(error: Exception, storage_message: str, fallback_message: str, on_error: StatusCallback) -> None:
|
|
"""Report database failures consistently so every menu operation presents the same recovery advice."""
|
|
logger.error(f"[MENU] {error}")
|
|
message = storage_message if isinstance(error, GuiStorageDatabaseError) else fallback_message
|
|
on_error(message)
|
|
|
|
|
|
def load_profile(profile_id: int, on_error: StatusCallback) -> BaseProfile | None:
|
|
"""Load one profile here so UI methods do not duplicate controller error handling."""
|
|
try:
|
|
return ProfileController.get(int(profile_id))
|
|
except Exception as error:
|
|
_report_load_error(
|
|
error,
|
|
"Local storage database could not be read. Restart and recover storage.db.",
|
|
"Could not load profile data. Sync or restart and try again.",
|
|
on_error,
|
|
)
|
|
return None
|
|
|
|
|
|
def load_profiles(on_error: StatusCallback) -> dict[int, BaseProfile]:
|
|
"""Load all profiles here so list rendering is not responsible for controller failures."""
|
|
try:
|
|
return ProfileController.get_all()
|
|
except Exception as error:
|
|
_report_load_error(
|
|
error,
|
|
"Local storage database could not be read. Restart and recover storage.db.",
|
|
"Could not load profiles. Sync or restart and try again.",
|
|
on_error,
|
|
)
|
|
return {}
|
|
|
|
|
|
def connection_code(profile: BaseProfile | None) -> str:
|
|
"""Normalize missing connection data once so menu rules do not repeatedly guard model attributes."""
|
|
connection = getattr(profile, "connection", None)
|
|
return getattr(connection, "code", None) or "unknown"
|
|
|
|
|
|
def location_key(profile: BaseProfile | None) -> str:
|
|
"""Convert both legacy dictionaries and current location models into the asset key expected by the menu."""
|
|
location = getattr(profile, "location", None)
|
|
if isinstance(location, dict):
|
|
return f'{location.get("country_code") or "na"}_{location.get("code") or "na"}'
|
|
|
|
country_code = getattr(location, "country_code", None)
|
|
code = getattr(location, "code", None)
|
|
return f"{country_code}_{code}" if country_code and code else "na_na"
|
|
|
|
|
|
def browser_info(profile: BaseProfile | None) -> tuple[str, str, bool]:
|
|
"""Keep browser compatibility interpretation outside Qt because it depends only on core profile data."""
|
|
if not isinstance(profile, SessionProfile):
|
|
return "unknown browser", "", False
|
|
|
|
application_version = getattr(profile, "application_version", None)
|
|
if isinstance(application_version, dict):
|
|
return (
|
|
application_version.get("application_code") or "unknown browser",
|
|
application_version.get("version_number") or "",
|
|
False,
|
|
)
|
|
|
|
return (
|
|
getattr(application_version, "application_code", None) or "unknown browser",
|
|
getattr(application_version, "version_number", None) or "",
|
|
bool(getattr(application_version, "supported", False)),
|
|
)
|
|
|
|
|
|
def profile_incomplete_reason(profile: BaseProfile | None) -> str | None:
|
|
"""Centralize profile readiness rules so the menu and enable worker cannot disagree."""
|
|
if profile is None:
|
|
return "Profile data could not be loaded. Sync or restart and try again."
|
|
if connection_code(profile) == "unknown":
|
|
return "Profile connection data is incomplete. Sync the database and try again."
|
|
|
|
location = getattr(profile, "location", None)
|
|
if location is None or isinstance(location, dict):
|
|
return "Profile location data is incomplete. Sync the database and try again."
|
|
if not isinstance(profile, SessionProfile):
|
|
return None
|
|
|
|
application_version = getattr(profile, "application_version", None)
|
|
if isinstance(application_version, dict) or application_version is None:
|
|
return "Profile browser data is incomplete. Sync the database and try again."
|
|
if not getattr(application_version, "application_code", None) or not getattr(application_version, "version_number", None):
|
|
return "Profile browser data is incomplete. Sync the database and try again."
|
|
return None
|
|
|
|
|
|
def location_operator(profile: BaseProfile | None) -> Any | None:
|
|
"""Hide legacy location shapes so rendering only receives a usable operator object."""
|
|
location = getattr(profile, "location", None)
|
|
return None if isinstance(location, dict) or location is None else getattr(location, "operator", None)
|
|
|
|
|
|
def profile_is_session(profile: BaseProfile | None) -> bool:
|
|
"""Expose profile type as a plain decision so pages do not inspect core model classes."""
|
|
return isinstance(profile, SessionProfile)
|
|
|
|
|
|
def _matched_profile(profile_id: int, profile: BaseProfile) -> tuple[str, dict[str, Any]]:
|
|
"""Build one normalized menu record so ordering does not contain profile-shape branches."""
|
|
protocol = connection_code(profile)
|
|
display_protocol = protocol if protocol in ENCRYPTED_PROTOCOLS else "hidetor"
|
|
connection_type = "browser-only" if profile_is_session(profile) else "system-wide"
|
|
connection_labels = {
|
|
"tor": "tor",
|
|
"wireguard": connection_type,
|
|
"hysteria2": connection_type,
|
|
"vless": connection_type,
|
|
}
|
|
browser, browser_version, browser_supported = browser_info(profile)
|
|
return f"Profile_{profile_id}", {
|
|
"id": profile_id,
|
|
"location": location_key(profile),
|
|
"protocol": display_protocol,
|
|
"connection": connection_labels.get(protocol, "just proxy"),
|
|
"browser": browser,
|
|
"browser_version": browser_version,
|
|
"browser_supported": browser_supported,
|
|
"dimentions": getattr(profile, "resolution", "None"),
|
|
"name": getattr(profile, "name", None) or f"Profile {profile_id}",
|
|
"incomplete_reason": profile_incomplete_reason(profile),
|
|
}
|
|
|
|
|
|
def match_core_profiles(profiles: Mapping[int, BaseProfile], gui_config_file: str | None) -> dict[str, dict[str, Any]]:
|
|
"""Translate core profiles into stable menu data before any Qt widgets are created."""
|
|
profile_ids = normalize_profile_order(gui_config_file, profiles.keys())
|
|
matched_profiles = (
|
|
_matched_profile(profile_id, profiles[profile_id])
|
|
for profile_id in profile_ids
|
|
if profile_id in profiles
|
|
)
|
|
return dict(matched_profiles)
|
|
|
|
|
|
def _enabled_profile_kind(profile: BaseProfile, enabled: bool | None = None) -> str | None:
|
|
"""Collapse enabled-state and profile-type checks so menu loading stays declarative."""
|
|
try:
|
|
is_enabled = is_profile_enabled_for_gui(profile) if enabled is None else enabled
|
|
except Exception:
|
|
return None
|
|
if not is_enabled:
|
|
return None
|
|
if not profile_is_session(profile):
|
|
return "system"
|
|
return "session_tor" if connection_code(profile) == "tor" else "session"
|
|
|
|
|
|
def load_menu_profiles(
|
|
gui_config_file: str | None, on_error: StatusCallback,
|
|
selected_id: int | None = None, state_change: Mapping[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Provide one menu-loading action so the page does not separately fetch, order, and classify profiles."""
|
|
profiles = load_profiles(on_error)
|
|
state_change = state_change or {}
|
|
changed_id = state_change.get("profile_id")
|
|
enabled_kinds = {
|
|
profile_id: kind
|
|
for profile_id, profile in profiles.items()
|
|
if (kind := _enabled_profile_kind(
|
|
profile, state_change.get("is_enabled") if profile_id == changed_id else None)) is not None
|
|
}
|
|
|
|
summaries = match_core_profiles(profiles, gui_config_file)
|
|
selected = profiles.get(selected_id)
|
|
details = None if selected is None else _profile_detail_data(
|
|
selected, summaries[f"Profile_{selected_id}"], selected_id in enabled_kinds)
|
|
return {
|
|
"display_profiles": summaries,
|
|
"selected_details": details,
|
|
"background": "background_connected" if "system" in enabled_kinds.values() else "original",
|
|
"enabled_kinds": enabled_kinds,
|
|
}
|
|
|
|
|
|
def _profile_detail_data(profile: BaseProfile | None, summary: Mapping[str, Any], is_connected: bool) -> dict[str, Any]:
|
|
"""Resolve profile display choices once so Qt code only renders the selected presentation."""
|
|
protocol = str(summary.get("protocol", "")).lower()
|
|
operator = location_operator(profile)
|
|
operator_name = getattr(operator, "name", "") if operator else ""
|
|
connection = connection_code(profile)
|
|
show_verification = protocol == "wireguard" and is_connected and connection == "wireguard"
|
|
show_operator_details = operator_name != "Simplified Privacy" and protocol in ENCRYPTED_PROTOCOLS
|
|
detail_modes = {
|
|
(True, True): "verification",
|
|
(True, False): "verification",
|
|
(False, True): "operator_details",
|
|
(False, False): "protocol_image",
|
|
}
|
|
display_mode = detail_modes[(show_verification, show_operator_details)]
|
|
profile_location = getattr(profile, "location", None)
|
|
profile_location = None if isinstance(profile_location, dict) else profile_location
|
|
|
|
return {
|
|
**summary,
|
|
"profile_loaded": profile is not None,
|
|
"message": None if profile is None else profile_status_message(profile, True, not is_connected),
|
|
"use_marquee": is_connected,
|
|
"selection": {"profile_number": str(summary["id"]), "name": summary["name"], "protocol": protocol},
|
|
"verification_values": [
|
|
(label, value, value if label == "Operator Name" else truncate_key(value))
|
|
for label, value in verification_values(profile)
|
|
],
|
|
"location_text": f"{profile_location.country_name}, {profile_location.name}" if profile_location else "",
|
|
"background_asset": _detail_background(summary, display_mode, profile_is_session(profile)),
|
|
"primary_asset": f"{protocol}.png" if protocol in ("hysteria2", "vless") else f"{protocol}_{summary['location']}.png",
|
|
"location_overlay": protocol in ("hysteria2", "vless"),
|
|
"overlay_y": {"hysteria2": 0.35}.get(protocol, 0.30),
|
|
"raised_primary": protocol == "hidetor" or (protocol == "wireguard" and summary["connection"] == "browser-only"),
|
|
"operator_name": operator_name,
|
|
"nostr_public_key": getattr(operator, "nostr_public_key", "") if operator else "",
|
|
"hydraveil_public_key": getattr(operator, "public_key", "") if operator else "",
|
|
"is_session": profile_is_session(profile),
|
|
"is_connected": is_connected,
|
|
"display_mode": display_mode,
|
|
"show_verified_icon": (
|
|
protocol == "wireguard"
|
|
and is_connected
|
|
and connection == "wireguard"
|
|
and ConfigurationController.get_endpoint_verification_enabled()
|
|
),
|
|
}
|
|
|
|
|
|
def _detail_background(summary: Mapping[str, Any], mode: str, is_session: bool) -> str | None:
|
|
"""Select background assets using profile policy before Qt starts drawing."""
|
|
if mode == "verification":
|
|
return None
|
|
protocol = summary["protocol"]
|
|
connection = summary["connection"]
|
|
if protocol == "hidetor":
|
|
prefix = "icon" if connection == "just proxy" else "hdtor"
|
|
return f"{prefix}_{summary['location']}.png"
|
|
if protocol == "wireguard" and connection == "browser-only":
|
|
return "browser only.png" if summary["browser_supported"] else "unsupported_browser_only.png"
|
|
return "browser only.png" if mode == "operator_details" and is_session else None
|
|
|
|
|
|
def menu_icons(btn_path: str, summary: Mapping[str, Any]) -> list[dict[str, Any]]:
|
|
"""Prepare icon paths and compatibility warnings without exposing protocol rules to Qt."""
|
|
connection = summary["connection"]
|
|
warning = connection not in ("tor", "just proxy") and not summary["browser_supported"] and summary["browser"] != "unknown browser"
|
|
return [{"kind": kind, "path": menu_icon_path(btn_path, kind, summary, connection),
|
|
"warning": kind == "browser" and warning} for kind in ("protocol", "location", "browser")]
|
|
|
|
|
|
def get_profile_details(
|
|
profile_id: int,
|
|
summary: Mapping[str, Any],
|
|
is_connected: bool,
|
|
on_error: StatusCallback,
|
|
) -> dict[str, Any]:
|
|
"""Load and prepare profile details as one task so the page never coordinates core model helpers."""
|
|
profile = load_profile(profile_id, on_error)
|
|
current_summary = _matched_profile(profile_id, profile)[1] if profile is not None else summary
|
|
return _profile_detail_data(profile, current_summary, is_connected)
|
|
|
|
|
|
def verification_values(profile: BaseProfile | None) -> list[tuple[str, str]]:
|
|
"""Prepare full verification values once so copy controls and displayed text cannot diverge."""
|
|
verification = build_verification_view(profile)
|
|
values = {
|
|
"Operator Name": verification["operator_name"],
|
|
"Nostr Key": verification["nostr_public_key"],
|
|
"HydraVeil Key": verification["hydraveil_public_key"],
|
|
}
|
|
return [(label, value or "N/A") for label, value in values.items()]
|
|
|
|
|
|
def menu_icon_path(btn_path: str, label_name: str, value: Mapping[str, Any], connection_type: str) -> str:
|
|
"""Choose menu asset paths outside Qt so protocol mapping remains declarative and reusable."""
|
|
protocol = str(value.get("protocol", ""))
|
|
protocol_icons = {
|
|
"hysteria2": "hysteria2_mini.png",
|
|
"vless": "vless_mini.png",
|
|
"tor": "toricon_mini.png",
|
|
"just proxy": "just proxy_mini.png",
|
|
}
|
|
if label_name == "protocol":
|
|
filename = protocol_icons.get(protocol) or protocol_icons.get(connection_type) or "wireguard_mini.png"
|
|
return os.path.join(btn_path, filename)
|
|
if label_name == "browser" and connection_type == "system-wide":
|
|
return os.path.join(btn_path, "wireguard_system_wide.png")
|
|
if label_name == "browser":
|
|
return os.path.join(btn_path, f'{value.get(label_name, "unknown browser")} latest_mini.png')
|
|
return os.path.join(btn_path, f'icon_mini_{value.get(label_name, "na_na")}.png')
|
|
|
|
|
|
def profile_status_message(profile: BaseProfile, is_enabled: bool, idle: bool = False) -> str:
|
|
"""Generate status text outside Qt so workers and pages report identical subscription state."""
|
|
subscription = getattr(profile, "subscription", None)
|
|
profile_date = getattr(subscription, "expires_at", None)
|
|
if profile_date is None:
|
|
return "Offline. No subscription found."
|
|
|
|
expiration_date = profile_date.replace(tzinfo=timezone.utc)
|
|
time_left = expiration_date - datetime.now(timezone.utc)
|
|
days_left = time_left.days
|
|
hours_left, _ = divmod(time_left.seconds, 3600)
|
|
if expiration_date < datetime.now(timezone.utc):
|
|
return "Offline. Subscription has expired."
|
|
if idle:
|
|
return f"Offline. Expires in {days_left} days."
|
|
if not is_enabled:
|
|
return f"Profile {int(profile.id)} disabled"
|
|
|
|
formatted_expiration = expiration_date.strftime("%Y-%m-%d %H:%M:%S")
|
|
return f"Profile {int(profile.id)} enabled. Expires on {formatted_expiration}. Time left: {days_left} days, {hours_left} hours."
|
|
|
|
def find_verification_icon(profile_id: int, btn_path: str, location_lookup: LocationLookup, on_error: StatusCallback) -> str | None:
|
|
"""Find the matching verification asset before navigation so the page does not scan files or compare locations."""
|
|
profile = load_profile(profile_id, on_error)
|
|
profile_location = getattr(profile, "location", None)
|
|
if profile is None or profile_location is None or isinstance(profile_location, dict):
|
|
return None
|
|
|
|
target = location_lookup(f"{profile_location.country_code}_{profile_location.code}")
|
|
if target is None:
|
|
return None
|
|
|
|
icon_files: Iterable[str] = os.listdir(btn_path)
|
|
icon_names = [name.removeprefix("button_").removesuffix(".png") for name in icon_files if name.startswith("button_") and name.endswith(".png")]
|
|
matches = [
|
|
name
|
|
for name in icon_names
|
|
if (candidate := location_lookup(name))
|
|
and candidate.country_code == target.country_code
|
|
and candidate.code == target.code
|
|
]
|
|
return matches[0] if matches else None
|