initial

counter_time.py

@@ -0,0 +1,271 @@
+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