115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
from core.errors.logger import logger
|
|
from core.Constants import Constants
|
|
from core.Helpers import write_atomically
|
|
|
|
#######################
|
|
|
|
from enum import Enum
|
|
from pydantic import BaseModel, field_serializer, field_validator, ConfigDict
|
|
from datetime import datetime
|
|
from zoneinfo import ZoneInfo
|
|
from typing import Optional, Self, ClassVar
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
class ConnectionChoice(str, Enum):
|
|
TOR = "tor"
|
|
SYSTEM = "system"
|
|
|
|
|
|
class Configuration(BaseModel):
|
|
model_config = ConfigDict(
|
|
extra='ignore',
|
|
exclude_none=True
|
|
)
|
|
|
|
_instance: ClassVar[Optional[Self]] = None
|
|
connection: Optional[ConnectionChoice] = None
|
|
auto_sync_enabled: Optional[bool] = None
|
|
endpoint_verification_enabled: Optional[bool] = False
|
|
last_synced_at: Optional[datetime] = None
|
|
firewall: Optional[bool] = False
|
|
dns: Optional[bool] = False
|
|
did_sudo_setup: Optional[bool] = False
|
|
singbox: Optional[str] = None
|
|
|
|
|
|
|
|
@field_validator('last_synced_at', mode='before')
|
|
@classmethod
|
|
def parse_datetime(cls, v):
|
|
if isinstance(v, str):
|
|
v = v.replace('Z', '+00:00') # Z → +00:00 for parsing
|
|
return v
|
|
|
|
@field_serializer('last_synced_at')
|
|
def serialize_datetime(self, value: datetime) -> str:
|
|
if value:
|
|
value = value.replace(tzinfo=ZoneInfo('UTC'))
|
|
return value.isoformat().replace('+00:00', 'Z') # +00:00 → Z for JSON
|
|
return None
|
|
|
|
@classmethod
|
|
def get(cls) -> Optional[Self]:
|
|
# Return the cached instance if it exists
|
|
if cls._instance is not None:
|
|
return cls._instance
|
|
|
|
try:
|
|
with open(f'{Constants.HV_CONFIG_HOME}/config.json', 'r') as f:
|
|
config_file_contents = f.read()
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
try:
|
|
configuration_dict = json.loads(config_file_contents)
|
|
except ValueError:
|
|
sys.exit(1)
|
|
|
|
# Create the instance and cache it before returning
|
|
cls._instance = cls(**configuration_dict)
|
|
return cls._instance
|
|
|
|
def save(self) -> None:
|
|
# model_dump_json reflects the current state of this specific instance
|
|
config_file_contents = f'{self.model_dump_json(indent=4)}\n'
|
|
os.makedirs(Constants.HOME, exist_ok=True)
|
|
|
|
config_file_path = f'{Constants.HV_CONFIG_HOME}/config.json'
|
|
write_atomically(config_file_path, config_file_contents)
|
|
|
|
|
|
def get_setting(looking_for: str):
|
|
# Use the singleton getter
|
|
config = Configuration.get()
|
|
|
|
if not config:
|
|
logger.error(f"[CONFIG] Can't load the entire config")
|
|
return None
|
|
|
|
# getattr because Pydantic models are objects, not dicts
|
|
result = getattr(config, looking_for, None)
|
|
|
|
if result is None:
|
|
logger.error(f"[CONFIG] What you want isn't in the config: {looking_for}")
|
|
return None
|
|
|
|
return result
|
|
|
|
|
|
def read_config():
|
|
"""
|
|
Returns it as a dict, not as a model.
|
|
"""
|
|
try:
|
|
config_file_contents = open(f'{Constants.HV_CONFIG_HOME}/config.json', 'r').read()
|
|
except FileNotFoundError:
|
|
return None
|
|
|
|
try:
|
|
configuration = json.loads(config_file_contents)
|
|
except ValueError:
|
|
logger.error(f"[CONFIG] Can't load JSON config")
|
|
|
|
return configuration
|