272 lines
9.0 KiB
Python
272 lines
9.0 KiB
Python
import json
|
|
import os
|
|
from datetime import datetime, time, timedelta
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo
|
|
|
|
INDONESIA_TIMEZONES = {
|
|
"Asia/Jakarta": "WIB (Asia/Jakarta)",
|
|
"Asia/Makassar": "WITA (Asia/Makassar)",
|
|
"Asia/Jayapura": "WIT (Asia/Jayapura)",
|
|
}
|
|
|
|
DEFAULT_LOCAL_TIMEZONE = "Asia/Jakarta"
|
|
DEFAULT_TUANG_CUTOFF_TIME = "17:00:00"
|
|
DEFAULT_MASUK_CUTOFF_TIME = "24:00:00"
|
|
COUNTER_SETTINGS_FILE_NAME = "counter_settings.json"
|
|
|
|
|
|
def default_counter_settings():
|
|
return {
|
|
"local_timezone": DEFAULT_LOCAL_TIMEZONE,
|
|
"tuang_cutoff_time": DEFAULT_TUANG_CUTOFF_TIME,
|
|
"masuk_cutoff_time": DEFAULT_MASUK_CUTOFF_TIME,
|
|
}
|
|
|
|
|
|
def get_settings_file_candidates(base_dir=None):
|
|
current_dir = Path(base_dir or __file__).resolve().parent
|
|
candidates = []
|
|
|
|
env_path = os.environ.get("COUNTER_SETTINGS_FILE")
|
|
if env_path:
|
|
candidates.append(Path(env_path).expanduser())
|
|
|
|
candidates.extend(
|
|
[
|
|
current_dir / COUNTER_SETTINGS_FILE_NAME,
|
|
current_dir.parent / COUNTER_SETTINGS_FILE_NAME,
|
|
current_dir.parent / "karung-web-admin" / COUNTER_SETTINGS_FILE_NAME,
|
|
current_dir.parent / "karung-web" / COUNTER_SETTINGS_FILE_NAME,
|
|
Path("/etc/frigate-counter/karung-web-admin") / COUNTER_SETTINGS_FILE_NAME,
|
|
Path("/etc/frigate-counter/karung-web") / COUNTER_SETTINGS_FILE_NAME,
|
|
]
|
|
)
|
|
|
|
unique_candidates = []
|
|
seen_paths = set()
|
|
for candidate in candidates:
|
|
resolved_candidate = candidate.resolve()
|
|
if resolved_candidate in seen_paths:
|
|
continue
|
|
unique_candidates.append(candidate)
|
|
seen_paths.add(resolved_candidate)
|
|
|
|
return unique_candidates
|
|
|
|
|
|
def get_default_settings_file(base_dir=None):
|
|
return get_settings_file_candidates(base_dir)[0]
|
|
|
|
|
|
def normalize_timezone(timezone_name):
|
|
timezone_name = (timezone_name or DEFAULT_LOCAL_TIMEZONE).strip()
|
|
if timezone_name not in INDONESIA_TIMEZONES:
|
|
allowed_timezones = ", ".join(INDONESIA_TIMEZONES)
|
|
raise ValueError(f"Timezone must be one of: {allowed_timezones}.")
|
|
return timezone_name
|
|
|
|
|
|
def normalize_cutoff_time(cutoff_time):
|
|
cutoff_time = (cutoff_time or "").strip()
|
|
parts = cutoff_time.split(":")
|
|
|
|
if len(parts) == 2:
|
|
parts.append("00")
|
|
|
|
if len(parts) != 3:
|
|
raise ValueError("Cutoff time must use HH:MM or HH:MM:SS format.")
|
|
|
|
try:
|
|
hour, minute, second = [int(part) for part in parts]
|
|
except ValueError as error:
|
|
raise ValueError("Cutoff time must contain only numbers.") from error
|
|
|
|
if hour == 24 and minute == 0 and second == 0:
|
|
return "24:00:00"
|
|
if hour < 0 or hour > 23:
|
|
raise ValueError("Cutoff hour must be between 00 and 24.")
|
|
if minute < 0 or minute > 59:
|
|
raise ValueError("Cutoff minute must be between 00 and 59.")
|
|
if second < 0 or second > 59:
|
|
raise ValueError("Cutoff second must be between 00 and 59.")
|
|
|
|
return f"{hour:02d}:{minute:02d}:{second:02d}"
|
|
|
|
|
|
def parse_cutoff_time(cutoff_time):
|
|
normalized_cutoff = normalize_cutoff_time(cutoff_time)
|
|
hour, minute, second = [int(part) for part in normalized_cutoff.split(":")]
|
|
return time(hour=hour, minute=minute, second=second)
|
|
|
|
|
|
def cutoff_time_to_seconds(cutoff_time):
|
|
normalized_cutoff = normalize_cutoff_time(cutoff_time)
|
|
hour, minute, second = [int(part) for part in normalized_cutoff.split(":")]
|
|
return (hour * 60 * 60) + (minute * 60) + second
|
|
|
|
|
|
def time_to_seconds(value):
|
|
return (value.hour * 60 * 60) + (value.minute * 60) + value.second
|
|
|
|
|
|
def get_cutoff_datetime(base_date, cutoff_time, timezone):
|
|
cutoff_seconds = cutoff_time_to_seconds(cutoff_time)
|
|
if cutoff_seconds == 24 * 60 * 60:
|
|
return datetime.combine(
|
|
base_date + timedelta(days=1),
|
|
time(0, 0, 0),
|
|
tzinfo=timezone,
|
|
)
|
|
cutoff = parse_cutoff_time(cutoff_time)
|
|
return datetime.combine(base_date, cutoff, tzinfo=timezone)
|
|
|
|
|
|
def normalize_counter_settings(settings):
|
|
normalized_settings = default_counter_settings()
|
|
normalized_settings.update(settings or {})
|
|
normalized_settings["local_timezone"] = normalize_timezone(
|
|
normalized_settings.get("local_timezone")
|
|
)
|
|
normalized_settings["tuang_cutoff_time"] = normalize_cutoff_time(
|
|
normalized_settings.get("tuang_cutoff_time")
|
|
)
|
|
normalized_settings["masuk_cutoff_time"] = normalize_cutoff_time(
|
|
normalized_settings.get("masuk_cutoff_time")
|
|
)
|
|
return normalized_settings
|
|
|
|
|
|
def load_counter_settings(base_dir=None, overrides=None):
|
|
settings = default_counter_settings()
|
|
|
|
for settings_file in get_settings_file_candidates(base_dir):
|
|
if not settings_file.exists():
|
|
continue
|
|
|
|
try:
|
|
with settings_file.open() as file_handle:
|
|
loaded_settings = json.load(file_handle)
|
|
except (OSError, json.JSONDecodeError):
|
|
continue
|
|
|
|
if isinstance(loaded_settings, dict):
|
|
settings.update(loaded_settings)
|
|
break
|
|
|
|
if overrides:
|
|
settings.update(overrides)
|
|
|
|
try:
|
|
return normalize_counter_settings(settings)
|
|
except ValueError:
|
|
return default_counter_settings()
|
|
|
|
|
|
def save_counter_settings(settings, settings_file):
|
|
normalized_settings = normalize_counter_settings(settings)
|
|
settings_file = Path(settings_file)
|
|
settings_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
with settings_file.open("w") as file_handle:
|
|
json.dump(normalized_settings, file_handle, indent=2)
|
|
file_handle.write("\n")
|
|
|
|
return normalized_settings
|
|
|
|
|
|
def get_timezone(settings):
|
|
return ZoneInfo(normalize_timezone(settings.get("local_timezone")))
|
|
|
|
|
|
def get_local_now(settings=None, now=None):
|
|
settings = normalize_counter_settings(settings)
|
|
timezone = get_timezone(settings)
|
|
|
|
if now is None:
|
|
return datetime.now(timezone)
|
|
if now.tzinfo is None:
|
|
return now.replace(tzinfo=timezone)
|
|
return now.astimezone(timezone)
|
|
|
|
|
|
def get_cutoff_value(settings, counter_type):
|
|
settings = normalize_counter_settings(settings)
|
|
if counter_type == "tuang":
|
|
return settings["tuang_cutoff_time"]
|
|
if counter_type == "masuk":
|
|
return settings["masuk_cutoff_time"]
|
|
raise ValueError("Counter type must be 'tuang' or 'masuk'.")
|
|
|
|
|
|
def get_business_date(counter_type, settings=None, now=None):
|
|
settings = normalize_counter_settings(settings)
|
|
local_now = get_local_now(settings, now)
|
|
cutoff_seconds = cutoff_time_to_seconds(get_cutoff_value(settings, counter_type))
|
|
|
|
if time_to_seconds(local_now.time()) >= cutoff_seconds:
|
|
return local_now.date() + timedelta(days=1)
|
|
return local_now.date()
|
|
|
|
|
|
def get_finished_business_date(counter_type, settings=None, now=None):
|
|
settings = normalize_counter_settings(settings)
|
|
local_now = get_local_now(settings, now)
|
|
cutoff_seconds = cutoff_time_to_seconds(get_cutoff_value(settings, counter_type))
|
|
active_business_date = get_business_date(counter_type, settings, local_now)
|
|
|
|
if time_to_seconds(local_now.time()) >= cutoff_seconds:
|
|
return active_business_date - timedelta(days=1)
|
|
return active_business_date
|
|
|
|
|
|
def get_period_window(counter_type, settings=None, business_date=None):
|
|
settings = normalize_counter_settings(settings)
|
|
timezone = get_timezone(settings)
|
|
cutoff = get_cutoff_value(settings, counter_type)
|
|
if business_date is None:
|
|
business_date = get_business_date(counter_type, settings)
|
|
|
|
period_start = get_cutoff_datetime(
|
|
business_date - timedelta(days=1),
|
|
cutoff,
|
|
timezone,
|
|
)
|
|
period_end = get_cutoff_datetime(business_date, cutoff, timezone) - timedelta(seconds=1)
|
|
return period_start, period_end
|
|
|
|
|
|
def get_next_cutoff_datetime(counter_type, settings=None, now=None):
|
|
settings = normalize_counter_settings(settings)
|
|
local_now = get_local_now(settings, now)
|
|
cutoff = get_cutoff_value(settings, counter_type)
|
|
next_cutoff = get_cutoff_datetime(local_now.date(), cutoff, local_now.tzinfo)
|
|
|
|
if local_now >= next_cutoff:
|
|
next_cutoff += timedelta(days=1)
|
|
|
|
return next_cutoff
|
|
|
|
|
|
def get_cutoff_metadata(settings=None, now=None):
|
|
settings = normalize_counter_settings(settings)
|
|
local_now = get_local_now(settings, now)
|
|
metadata = {
|
|
"timezone": settings["local_timezone"],
|
|
"timezone_label": INDONESIA_TIMEZONES[settings["local_timezone"]],
|
|
"now": local_now.isoformat(timespec="seconds"),
|
|
"counters": {},
|
|
}
|
|
|
|
for counter_type in ("tuang", "masuk"):
|
|
business_date = get_business_date(counter_type, settings, local_now)
|
|
period_start, period_end = get_period_window(counter_type, settings, business_date)
|
|
metadata["counters"][counter_type] = {
|
|
"cutoff_time": get_cutoff_value(settings, counter_type),
|
|
"business_date": business_date.isoformat(),
|
|
"period_start": period_start.isoformat(timespec="seconds"),
|
|
"period_end": period_end.isoformat(timespec="seconds"),
|
|
}
|
|
|
|
return metadata
|