karung-masuk/counter_time.py

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 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 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 get_timezone(settings):
    return ZoneInfo(normalize_timezone(settings.get("local_timezone")))


def get_local_now(settings=None, now=None):
    settings = normalize_counter_settings(settings)
    local_timezone = get_timezone(settings)

    if now is None:
        return datetime.now(local_timezone)
    if now.tzinfo is None:
        return now.replace(tzinfo=local_timezone)
    return now.astimezone(local_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