148 lines
4 KiB
Python
148 lines
4 KiB
Python
from core.Errors import UnknownConnectionTypeError
|
|
from core.models.Configuration import Configuration, ConnectionChoice
|
|
from datetime import datetime, timezone
|
|
from typing import Optional
|
|
|
|
|
|
|
|
class ConfigurationController:
|
|
_config: Optional[Configuration] = None
|
|
|
|
@staticmethod
|
|
def get():
|
|
if ConfigurationController._config is None:
|
|
ConfigurationController._config = Configuration.get()
|
|
return ConfigurationController._config
|
|
|
|
# return Configuration.get()
|
|
|
|
@staticmethod
|
|
def reload_from_disk():
|
|
ConfigurationController._config = None
|
|
|
|
@staticmethod
|
|
def get_or_new():
|
|
|
|
configuration = ConfigurationController.get()
|
|
|
|
if configuration is None:
|
|
return Configuration()
|
|
|
|
return configuration
|
|
|
|
@staticmethod
|
|
def get_connection():
|
|
|
|
configuration = ConfigurationController.get()
|
|
return configuration.connection.value
|
|
|
|
@staticmethod
|
|
def get_connection_enum():
|
|
|
|
configuration = ConfigurationController.get()
|
|
return configuration.connection
|
|
|
|
@staticmethod
|
|
def set_connection(connection_string: Optional[str] = None):
|
|
|
|
configuration = ConfigurationController.get_or_new()
|
|
|
|
if connection_string == "tor":
|
|
connection = ConnectionChoice.TOR
|
|
elif connection_string == "system":
|
|
connection = ConnectionChoice.SYSTEM
|
|
else:
|
|
raise UnknownConnectionTypeError(f'The choice of {connection_string} is not valid.')
|
|
|
|
configuration.connection = connection
|
|
configuration.save()
|
|
|
|
@staticmethod
|
|
def get_auto_sync_enabled():
|
|
|
|
configuration = ConfigurationController.get()
|
|
|
|
if configuration is None:
|
|
return False
|
|
|
|
return configuration.auto_sync_enabled
|
|
|
|
@staticmethod
|
|
def set_auto_sync_enabled(enable_auto_sync: Optional[bool] = None):
|
|
|
|
configuration = ConfigurationController.get_or_new()
|
|
configuration.auto_sync_enabled = enable_auto_sync
|
|
configuration.save()
|
|
|
|
@staticmethod
|
|
def get_endpoint_verification_enabled():
|
|
|
|
configuration = ConfigurationController.get()
|
|
|
|
if configuration is None:
|
|
return True
|
|
|
|
return configuration.endpoint_verification_enabled
|
|
|
|
@staticmethod
|
|
def set_endpoint_verification_enabled(enable_endpoint_verification: Optional[bool] = None):
|
|
|
|
configuration = ConfigurationController.get_or_new()
|
|
configuration.endpoint_verification_enabled = enable_endpoint_verification
|
|
configuration.save()
|
|
|
|
@staticmethod
|
|
def get_last_synced_at():
|
|
|
|
configuration = ConfigurationController.get()
|
|
|
|
if configuration is None:
|
|
return None
|
|
|
|
return configuration.last_synced_at
|
|
|
|
@staticmethod
|
|
def update_last_synced_at():
|
|
|
|
configuration = ConfigurationController.get()
|
|
configuration.last_synced_at = datetime.now(timezone.utc)
|
|
configuration.save()
|
|
|
|
@staticmethod
|
|
def update_or_create(configuration):
|
|
configuration.save()
|
|
|
|
@staticmethod
|
|
def change_firewall(new_value):
|
|
configuration = ConfigurationController.get()
|
|
configuration.firewall = new_value
|
|
configuration.save()
|
|
|
|
@staticmethod
|
|
def get_firewall_setting():
|
|
configuration = ConfigurationController.get()
|
|
return configuration.firewall
|
|
|
|
|
|
@staticmethod
|
|
def change_dns(new_value):
|
|
configuration = ConfigurationController.get()
|
|
configuration.dns = new_value
|
|
configuration.save()
|
|
|
|
@staticmethod
|
|
def get_dns_setting():
|
|
configuration = ConfigurationController.get()
|
|
return configuration.dns
|
|
|
|
@staticmethod
|
|
def get_singbox_version():
|
|
configuration = ConfigurationController.get()
|
|
return configuration.singbox
|
|
|
|
@staticmethod
|
|
def update_singbox_version(new_value) -> bool:
|
|
configuration = ConfigurationController.get()
|
|
configuration.singbox = new_value
|
|
configuration.save()
|
|
return True
|