570 lines
9.8 KiB
Python
570 lines
9.8 KiB
Python
from __future__ import annotations
|
|
|
|
import copy
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import threading
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import yaml
|
|
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parent.parent
|
|
CONFIG_DIR = PROJECT_ROOT / "config"
|
|
|
|
USERS_FILE = CONFIG_DIR / "users.yaml"
|
|
STATE_FILE = CONFIG_DIR / "state.json"
|
|
|
|
DAY_NAMES = (
|
|
"sunday",
|
|
"monday",
|
|
"tuesday",
|
|
"wednesday",
|
|
"thursday",
|
|
"friday",
|
|
"saturday",
|
|
)
|
|
|
|
_config_lock = threading.RLock()
|
|
|
|
|
|
DEFAULT_USERS_CONFIG = {
|
|
"users": []
|
|
}
|
|
|
|
DEFAULT_STATE = {
|
|
"usage": {},
|
|
"temporary_grants": [],
|
|
}
|
|
|
|
|
|
def _ensure_config_dir() -> None:
|
|
CONFIG_DIR.mkdir(
|
|
parents=True,
|
|
exist_ok=True,
|
|
)
|
|
|
|
|
|
def _atomic_write(
|
|
path: Path,
|
|
content: str,
|
|
) -> None:
|
|
_ensure_config_dir()
|
|
|
|
directory = path.parent
|
|
|
|
fd, temporary_name = tempfile.mkstemp(
|
|
prefix=f".{path.name}.",
|
|
suffix=".tmp",
|
|
dir=directory,
|
|
text=True,
|
|
)
|
|
|
|
temporary_path = Path(temporary_name)
|
|
|
|
try:
|
|
with os.fdopen(
|
|
fd,
|
|
"w",
|
|
encoding="utf-8",
|
|
) as file:
|
|
file.write(content)
|
|
file.flush()
|
|
os.fsync(file.fileno())
|
|
|
|
os.replace(
|
|
temporary_path,
|
|
path,
|
|
)
|
|
|
|
finally:
|
|
try:
|
|
temporary_path.unlink()
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def _normalize_users_config(
|
|
data: Any,
|
|
) -> dict[str, Any]:
|
|
if not isinstance(data, dict):
|
|
return copy.deepcopy(
|
|
DEFAULT_USERS_CONFIG
|
|
)
|
|
|
|
users = data.get("users")
|
|
|
|
if not isinstance(users, list):
|
|
users = []
|
|
|
|
normalized_users = []
|
|
|
|
for user in users:
|
|
if not isinstance(user, dict):
|
|
continue
|
|
|
|
normalized = dict(user)
|
|
|
|
normalized.setdefault(
|
|
"enabled",
|
|
True,
|
|
)
|
|
|
|
normalized.setdefault(
|
|
"daily_allowance",
|
|
{},
|
|
)
|
|
|
|
normalized.setdefault(
|
|
"access_windows",
|
|
{},
|
|
)
|
|
|
|
normalized_users.append(
|
|
normalized
|
|
)
|
|
|
|
return {
|
|
"users": normalized_users
|
|
}
|
|
|
|
|
|
def _normalize_state(
|
|
data: Any,
|
|
) -> dict[str, Any]:
|
|
if not isinstance(data, dict):
|
|
return copy.deepcopy(
|
|
DEFAULT_STATE
|
|
)
|
|
|
|
state = dict(data)
|
|
|
|
if not isinstance(
|
|
state.get("usage"),
|
|
dict,
|
|
):
|
|
state["usage"] = {}
|
|
|
|
if not isinstance(
|
|
state.get("temporary_grants"),
|
|
list,
|
|
):
|
|
state["temporary_grants"] = []
|
|
|
|
return state
|
|
|
|
|
|
def load_users_config() -> dict[str, Any]:
|
|
with _config_lock:
|
|
_ensure_config_dir()
|
|
|
|
if not USERS_FILE.exists():
|
|
save_users_config(
|
|
DEFAULT_USERS_CONFIG
|
|
)
|
|
|
|
try:
|
|
with USERS_FILE.open(
|
|
"r",
|
|
encoding="utf-8",
|
|
) as file:
|
|
data = yaml.safe_load(file)
|
|
|
|
except (
|
|
OSError,
|
|
yaml.YAMLError,
|
|
):
|
|
raise RuntimeError(
|
|
f"Unable to read {USERS_FILE}"
|
|
)
|
|
|
|
return _normalize_users_config(
|
|
data
|
|
)
|
|
|
|
|
|
def save_users_config(
|
|
data: dict[str, Any],
|
|
) -> None:
|
|
with _config_lock:
|
|
normalized = _normalize_users_config(
|
|
data
|
|
)
|
|
|
|
content = yaml.safe_dump(
|
|
normalized,
|
|
sort_keys=False,
|
|
allow_unicode=True,
|
|
default_flow_style=False,
|
|
)
|
|
|
|
_atomic_write(
|
|
USERS_FILE,
|
|
content,
|
|
)
|
|
|
|
|
|
def load_state() -> dict[str, Any]:
|
|
with _config_lock:
|
|
_ensure_config_dir()
|
|
|
|
if not STATE_FILE.exists():
|
|
save_state(
|
|
DEFAULT_STATE
|
|
)
|
|
|
|
try:
|
|
with STATE_FILE.open(
|
|
"r",
|
|
encoding="utf-8",
|
|
) as file:
|
|
data = json.load(file)
|
|
|
|
except (
|
|
OSError,
|
|
json.JSONDecodeError,
|
|
):
|
|
raise RuntimeError(
|
|
f"Unable to read {STATE_FILE}"
|
|
)
|
|
|
|
return _normalize_state(
|
|
data
|
|
)
|
|
|
|
|
|
def save_state(
|
|
data: dict[str, Any],
|
|
) -> None:
|
|
with _config_lock:
|
|
normalized = _normalize_state(
|
|
data
|
|
)
|
|
|
|
content = json.dumps(
|
|
normalized,
|
|
indent=2,
|
|
ensure_ascii=False,
|
|
)
|
|
|
|
_atomic_write(
|
|
STATE_FILE,
|
|
content,
|
|
)
|
|
|
|
|
|
def next_user_id(
|
|
users: list[dict[str, Any]],
|
|
) -> int:
|
|
highest = 0
|
|
|
|
for user in users:
|
|
try:
|
|
user_id = int(
|
|
user.get("id", 0)
|
|
)
|
|
except (
|
|
TypeError,
|
|
ValueError,
|
|
):
|
|
continue
|
|
|
|
highest = max(
|
|
highest,
|
|
user_id,
|
|
)
|
|
|
|
return highest + 1
|
|
|
|
|
|
def find_user(
|
|
user_id: int,
|
|
) -> dict[str, Any] | None:
|
|
config = load_users_config()
|
|
|
|
try:
|
|
requested_id = int(user_id)
|
|
except (
|
|
TypeError,
|
|
ValueError,
|
|
):
|
|
return None
|
|
|
|
for user in config["users"]:
|
|
try:
|
|
current_id = int(
|
|
user.get("id")
|
|
)
|
|
except (
|
|
TypeError,
|
|
ValueError,
|
|
):
|
|
continue
|
|
|
|
if current_id == requested_id:
|
|
return user
|
|
|
|
return None
|
|
|
|
|
|
def find_user_by_username(
|
|
username: str,
|
|
) -> dict[str, Any] | None:
|
|
config = load_users_config()
|
|
|
|
for user in config["users"]:
|
|
if (
|
|
str(
|
|
user.get("username", "")
|
|
)
|
|
== username
|
|
):
|
|
return user
|
|
|
|
return None
|
|
|
|
|
|
def get_day_name(
|
|
weekday: int,
|
|
) -> str:
|
|
weekday = int(weekday)
|
|
|
|
if not 0 <= weekday <= 6:
|
|
raise ValueError(
|
|
"weekday must be between 0 and 6"
|
|
)
|
|
|
|
return DAY_NAMES[weekday]
|
|
|
|
|
|
def normalize_day_name(
|
|
day: str,
|
|
) -> str:
|
|
value = str(day).strip().lower()
|
|
|
|
aliases = {
|
|
"sun": "sunday",
|
|
"mon": "monday",
|
|
"tue": "tuesday",
|
|
"tues": "tuesday",
|
|
"wed": "wednesday",
|
|
"thu": "thursday",
|
|
"thur": "thursday",
|
|
"thurs": "thursday",
|
|
"fri": "friday",
|
|
"sat": "saturday",
|
|
}
|
|
|
|
return aliases.get(
|
|
value,
|
|
value,
|
|
)
|
|
|
|
|
|
def python_weekday_to_name(
|
|
python_weekday: int,
|
|
) -> str:
|
|
python_weekday = int(
|
|
python_weekday
|
|
)
|
|
|
|
if not 0 <= python_weekday <= 6:
|
|
raise ValueError(
|
|
"weekday must be between 0 and 6"
|
|
)
|
|
|
|
return DAY_NAMES[
|
|
(python_weekday + 1) % 7
|
|
]
|
|
|
|
|
|
def parse_time(
|
|
value: Any,
|
|
) -> int:
|
|
if isinstance(value, int):
|
|
return max(
|
|
0,
|
|
min(
|
|
value,
|
|
1439,
|
|
),
|
|
)
|
|
|
|
text = str(value).strip()
|
|
|
|
if ":" not in text:
|
|
return max(
|
|
0,
|
|
min(
|
|
int(text),
|
|
1439,
|
|
),
|
|
)
|
|
|
|
hours_text, minutes_text = (
|
|
text.split(
|
|
":",
|
|
1,
|
|
)
|
|
)
|
|
|
|
hours = int(
|
|
hours_text
|
|
)
|
|
|
|
minutes = int(
|
|
minutes_text
|
|
)
|
|
|
|
if not 0 <= hours <= 23:
|
|
raise ValueError(
|
|
f"Invalid hour: {hours}"
|
|
)
|
|
|
|
if not 0 <= minutes <= 59:
|
|
raise ValueError(
|
|
f"Invalid minute: {minutes}"
|
|
)
|
|
|
|
return (
|
|
hours * 60
|
|
+ minutes
|
|
)
|
|
|
|
|
|
def format_time(
|
|
minute: int,
|
|
) -> str:
|
|
minute = int(minute)
|
|
|
|
minute = max(
|
|
0,
|
|
min(
|
|
minute,
|
|
1439,
|
|
),
|
|
)
|
|
|
|
hours = minute // 60
|
|
minutes = minute % 60
|
|
|
|
return (
|
|
f"{hours:02d}:"
|
|
f"{minutes:02d}"
|
|
)
|
|
|
|
|
|
def normalize_allowances(
|
|
allowances: Any,
|
|
) -> dict[str, int]:
|
|
if not isinstance(
|
|
allowances,
|
|
dict,
|
|
):
|
|
allowances = {}
|
|
|
|
result = {}
|
|
|
|
for day in DAY_NAMES:
|
|
value = allowances.get(
|
|
day,
|
|
0,
|
|
)
|
|
|
|
try:
|
|
seconds = int(value)
|
|
except (
|
|
TypeError,
|
|
ValueError,
|
|
):
|
|
seconds = 0
|
|
|
|
result[day] = max(
|
|
0,
|
|
seconds,
|
|
)
|
|
|
|
return result
|
|
|
|
|
|
def normalize_access_windows(
|
|
windows: Any,
|
|
) -> dict[str, list[dict[str, str]]]:
|
|
if not isinstance(
|
|
windows,
|
|
dict,
|
|
):
|
|
windows = {}
|
|
|
|
result = {}
|
|
|
|
for day in DAY_NAMES:
|
|
day_windows = windows.get(
|
|
day,
|
|
[],
|
|
)
|
|
|
|
if not isinstance(
|
|
day_windows,
|
|
list,
|
|
):
|
|
continue
|
|
|
|
normalized = []
|
|
|
|
for window in day_windows:
|
|
if isinstance(
|
|
window,
|
|
dict,
|
|
):
|
|
start = window.get(
|
|
"start"
|
|
)
|
|
end = window.get(
|
|
"end"
|
|
)
|
|
|
|
elif (
|
|
isinstance(
|
|
window,
|
|
(list, tuple),
|
|
)
|
|
and len(window) >= 2
|
|
):
|
|
start = window[0]
|
|
end = window[1]
|
|
|
|
else:
|
|
continue
|
|
|
|
try:
|
|
start_minute = parse_time(
|
|
start
|
|
)
|
|
end_minute = parse_time(
|
|
end
|
|
)
|
|
except (
|
|
TypeError,
|
|
ValueError,
|
|
):
|
|
continue
|
|
|
|
normalized.append(
|
|
{
|
|
"start": format_time(
|
|
start_minute
|
|
),
|
|
"end": format_time(
|
|
end_minute
|
|
),
|
|
}
|
|
)
|
|
|
|
if normalized:
|
|
result[day] = normalized
|
|
|
|
return result
|