from __future__ import annotations from typing import Any from app.core.errors import EngineError, UnsupportedEngineError from app.core.interface import EngineBackend from app.models import ( CapabilitySet, EngineKind, EngineState, RunMode, RuntimeRequest, RuntimeSnapshot, RuntimeStats, ) class NativeEngineBackend(EngineBackend): """Android implementation backed by the Flet service extension. The service returns immediately with ``starting``; native SharedPreferences is authoritative and is polled by ``get_state``. This class contains no Android lifecycle code and is therefore safe to keep out of desktop builds. """ def __init__(self, kind: EngineKind, bridge: Any) -> None: self.kind = kind self.bridge = bridge self.name = kind.value self.capabilities = self._capabilities(kind) self._request: RuntimeRequest | None = None self._state = RuntimeSnapshot(engine=kind) @staticmethod def _capabilities(kind: EngineKind) -> CapabilitySet: if kind is EngineKind.FREETURN: return CapabilitySet( supports_proxy=True, supports_socks=False, supports_udp=True, supports_mobile=True, supports_stats=True, ) return CapabilitySet( # The exact SignBox AAR/API is a release input. Keeping these false # prevents the UI from advertising a native path that is not bundled. supports_mobile=True, ) async def start(self, request: RuntimeRequest) -> RuntimeSnapshot: payload = request.model_dump(mode="json") existing = await self.bridge.get_runtime_state() existing_state = self._result_state(existing) existing_engine = self._result_engine(existing) if existing_state in {EngineState.STARTING, EngineState.RUNNING}: if existing_engine is not None and existing_engine is not request.engine: raise EngineError("another native engine is active; stop it before switching") existing_mode = self._result_mode(existing) if ( request.mode is RunMode.BOTH and existing_mode is RunMode.VPN and request.proxy.access.value == "hotspot" ): result = await self.bridge.start_hotspot_proxy(payload) self._request = request self._state = self._snapshot_from_result(result, request, EngineState.STARTING) return self._raise_if_terminal_error(self._state) self._request = request self._state = self._snapshot_from_result(existing, request, existing_state) return self._state self._request = request if request.mode is RunMode.PROXY: result = ( await self.bridge.start_proxy(payload) if request.proxy.access.value == "local" else await self.bridge.start_hotspot_proxy(payload) ) else: result = await self.bridge.start_vpn(payload) if request.mode is RunMode.BOTH and self._result_state(result) not in { EngineState.ERROR, EngineState.UNSUPPORTED, }: if request.proxy.access.value == "hotspot": await self.bridge.start_hotspot_proxy(payload) else: await self.bridge.start_proxy(payload) self._state = self._snapshot_from_result(result, request, EngineState.STARTING) return self._raise_if_terminal_error(self._state) async def stop(self) -> RuntimeSnapshot: await self.bridge.stop_vpn() await self.bridge.stop_proxy() request = self._request self._state = RuntimeSnapshot( state=EngineState.STOPPED, engine=self.kind, mode=request.mode if request else None, profile_id=request.profile_id if request else None, ) return self._state async def restart(self, request: RuntimeRequest) -> RuntimeSnapshot: await self.stop() return await self.start(request) async def get_state(self) -> RuntimeSnapshot: result = await self.bridge.get_runtime_state() request = self._request if isinstance(result, dict): if request is None: engine = self._result_engine(result) if engine is not self.kind: return self._state self._state = self._snapshot_from_result(result, None, self._state.state) else: self._state = self._snapshot_from_result(result, request, self._state.state) return self._state async def get_stats(self) -> RuntimeStats: return self._state.stats def _raise_if_terminal_error(self, snapshot: RuntimeSnapshot) -> RuntimeSnapshot: if snapshot.state is EngineState.ERROR: raise EngineError(snapshot.message or "native runtime failed") if snapshot.state is EngineState.UNSUPPORTED: raise UnsupportedEngineError(snapshot.message or "native engine is unsupported") return snapshot @staticmethod def _result_state(result: object) -> EngineState: if isinstance(result, dict): try: return EngineState(str(result.get("state", "starting"))) except ValueError: return EngineState.STARTING return EngineState.STARTING @staticmethod def _result_engine(result: object) -> EngineKind | None: if not isinstance(result, dict): return None value = result.get("engine") try: return EngineKind(str(value)) if value else None except ValueError: return None def _snapshot_from_result( self, result: object, request: RuntimeRequest | None, fallback: EngineState ) -> RuntimeSnapshot: state = self._result_state(result) if state is EngineState.STOPPED and fallback is EngineState.STARTING: state = fallback message = result.get("message") if isinstance(result, dict) else None engine = request.engine if request else self._result_engine(result) mode = request.mode if request else self._result_mode(result) profile_id = request.profile_id if request else self._result_string(result, "profile_id") payload = self._result_payload(result) return RuntimeSnapshot( state=state, engine=engine, mode=mode, profile_id=profile_id, message=str(message) if message else None, proxy_addresses=self._proxy_addresses(payload), captcha_url=payload.get("captcha_url") if isinstance(payload.get("captcha_url"), str) else None, revision=request.config_revision if request else 0, ) @staticmethod def _result_string(result: object, key: str) -> str | None: value = result.get(key) if isinstance(result, dict) else None return str(value) if value else None @staticmethod def _result_mode(result: object): from app.models import RunMode value = result.get("mode") if isinstance(result, dict) else None try: return RunMode(str(value)) if value else None except ValueError: return None @staticmethod def _result_payload(result: object) -> dict[str, object]: if not isinstance(result, dict): return {} payload = result.get("payload") if isinstance(payload, dict): return payload if isinstance(payload, str): try: import json decoded = json.loads(payload) return decoded if isinstance(decoded, dict) else {} except json.JSONDecodeError: return {} return {} @staticmethod def _proxy_addresses(payload: dict[str, object]) -> list[str]: addresses = payload.get("addresses") if isinstance(addresses, list): return [str(item) for item in addresses] return []