100 lines
2.8 KiB
Python
100 lines
2.8 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
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
|
|
class ConnectionChoice(str, Enum):
|
|
TOR = "tor"
|
|
SYSTEM = "system"
|
|
|
|
class Configuration(BaseModel):
|
|
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
|
|
|
|
|
|
model_config = ConfigDict(
|
|
extra='ignore', # Ignore unknown fields in JSON
|
|
exclude_none=True # Don't serialize None values
|
|
)
|
|
|
|
@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
|
|
|
|
|
|
@staticmethod
|
|
def get():
|
|
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)
|
|
|
|
return Configuration(**configuration_dict) # Pydantic validates on init
|
|
|
|
|
|
def save(self: Self):
|
|
config_file_contents = f'{self.model_dump_json(indent=4)}\n'
|
|
os.makedirs(Constants.HV_CONFIG_HOME, exist_ok=True)
|
|
|
|
config_file_path = f'{Constants.HV_CONFIG_HOME}/config.json'
|
|
write_atomically(config_file_path, config_file_contents)
|
|
|
|
|
|
def read_config():
|
|
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
|
|
|
|
def get_setting(looking_for):
|
|
config = read_config()
|
|
if not config:
|
|
logger.error(f"[CONFIG] Can't load the entire config")
|
|
return None
|
|
|
|
if looking_for not in config:
|
|
logger.error(f"[CONFIG] What you want isn't in the config")
|
|
return None
|
|
|
|
result = config[looking_for]
|
|
|
|
return result
|