Create logger.py

test_sound.py

@@ -0,0 +1,13 @@
+import subprocess
+from datetime import datetime, time
+from utils.sound_utils import play_alarm_sound, SOUNDS
+now = datetime.now().time()
+namaz_zohr_time = time(12, 9) < now < time(13, 20)
+namaz_maghrib_time = time(17, 24) < now < time(18, 45)
+if not namaz_zohr_time and not namaz_maghrib_time:
+    print("*** PLAYED ALARM SOUND ***")
+    play_alarm_sound(**SOUNDS["notif"])
+    do_alarm_later = False
+else:
+    print("*** It's namaz time so we'll play alarm sound later")
+    do_alarm_later = True

utils/logger.py

@@ -0,0 +1,343 @@
+from __future__ import annotations
+
+import logging
+import logging.handlers
+import os
+from pathlib import Path
+import sys
+
+
+def setup_logging(
+    *,
+    app_name: str | None = "script",
+    log_dir: str | Path = "logs",
+    console_level: int | None = None,
+    file_level: int | None = None,
+    root_level: int | None = None,
+    calendar: str | None = None,  # "gregorian" | "jalali" | None
+    json_logs: bool | None = None,
+    enable_file: bool | None = None,  # None -> auto based on docker + env overrides
+    max_bytes: int = 10 * 1024 * 1024,
+    backup_count: int = 10,
+    use_rich: bool = False,
+    include_app_name: bool = True,
+    include_name: bool = True,
+    microseconds: int = 3,  # 0, 3, or 6
+    include_pid: bool = True,
+    include_thread: bool = True,
+) -> None:
+    """
+    Script-friendly logging:
+    - Console always
+    - File only if enabled (default: enabled locally, disabled in Docker)
+    - Optional JSON logs
+
+    Formatter options:
+    - microseconds: 0 (none), 3 (milliseconds), 6 (microseconds)
+    - app_name: if None, it's omitted
+    - include_pid/include_thread toggles
+    """
+
+    if use_rich:
+        try:
+            from rich.logging import RichHandler
+        except Exception:
+            use_rich = False
+
+    if use_rich and not sys.stderr.isatty():
+        use_rich = False
+
+    # ---- Environment overrides ----
+    env_root = os.getenv("LOG_LEVEL")
+    env_console = os.getenv("LOG_LEVEL_CONSOLE")
+    env_file = os.getenv("LOG_LEVEL_FILE")
+
+    if root_level is None:
+        root_level = _level_from_env(env_root, default=logging.DEBUG)
+    if console_level is None:
+        console_level = _level_from_env(env_console, default=logging.INFO)
+    if file_level is None:
+        file_level = _level_from_env(env_file, default=logging.DEBUG)
+
+    if json_logs is None:
+        json_logs = _truthy_env(os.getenv("LOG_JSON"), default=False)
+
+    if enable_file is None:
+        env_log_to_file = os.getenv("LOG_TO_FILE")
+        if env_log_to_file is not None:
+            enable_file = _truthy_env(env_log_to_file, default=True)
+        else:
+            enable_file = not _looks_like_docker()
+
+    log_dir = Path(os.getenv("LOG_DIR", str(log_dir))).resolve()
+
+    # ---- Calendar selection ----
+    if calendar is None:
+        calendar = "gregorian"
+    calendar = calendar.strip().lower()
+
+    if calendar == "jalali":
+        try:
+            import jdatetime  # noqa: F401
+        except Exception:
+            calendar = "gregorian"
+
+    if calendar not in ("gregorian", "jalali"):
+        raise ValueError("calendar must be 'gregorian' or 'jalali'")
+
+    # ---- Validate formatter knobs ----
+    if microseconds not in (0, 3, 6):
+        raise ValueError("microseconds must be 0, 3, or 6")
+
+    # ---- Configure root ----
+    root = logging.getLogger()
+    root.setLevel(root_level)
+
+    for h in list(root.handlers):
+        root.removeHandler(h)
+
+    # ---- Formatter ----
+    if json_logs:
+        formatter: logging.Formatter = _JsonFormatter(
+            app_name=app_name,
+            calendar=calendar,
+            microseconds=microseconds,
+            include_app_name=include_app_name,
+            include_name=include_name,
+            include_pid=include_pid,
+            include_thread=include_thread,
+        )
+    else:
+        formatter = _TextFormatter(
+            app_name=app_name,
+            calendar=calendar,
+            microseconds=microseconds,
+            include_app_name=include_app_name,
+            include_name=include_name,
+            include_pid=include_pid,
+            include_thread=include_thread,
+        )
+
+    # ---- Console handler ----
+    if use_rich and not json_logs:
+        console = RichHandler(
+            level=console_level,
+            rich_tracebacks=True,
+            tracebacks_show_locals=True,
+            show_time=False,  # YOU already handle time
+            show_level=False,  # YOU already handle level
+            show_path=False,  # avoid noise
+            markup=False,
+        )
+        console.setFormatter(formatter)
+    else:
+        console = logging.StreamHandler()
+        console.setLevel(console_level)
+        console.setFormatter(formatter)
+
+    root.addHandler(console)
+
+    # ---- File handler (rotating) ----
+    if enable_file:
+        log_dir.mkdir(parents=True, exist_ok=True)
+        log_path = log_dir / f"{app_name or 'app'}.log"
+
+        file_handler = logging.handlers.RotatingFileHandler(
+            filename=log_path,
+            maxBytes=max_bytes,
+            backupCount=backup_count,
+            encoding="utf-8",
+            delay=True,
+        )
+        file_handler.setLevel(file_level)
+        file_handler.setFormatter(formatter)
+        root.addHandler(file_handler)
+
+
+def _looks_like_docker() -> bool:
+    if os.path.exists("/.dockerenv"):
+        return True
+    try:
+        cgroup = Path("/proc/1/cgroup")
+        if cgroup.exists():
+            txt = cgroup.read_text(errors="ignore")
+            if "docker" in txt or "kubepods" in txt or "containerd" in txt:
+                return True
+    except Exception:
+        pass
+    return False
+
+
+def _truthy_env(val: str | None, *, default: bool) -> bool:
+    if val is None:
+        return default
+    return val.strip().lower() in ("1", "true", "yes", "y", "on")
+
+
+def _level_from_env(val: str | None, *, default: int) -> int:
+    if not val:
+        return default
+    name = val.strip().upper()
+    return getattr(logging, name, default)
+
+
+def _format_timestamp(*, calendar: str, microseconds: int) -> str:
+    """
+    microseconds:
+      - 0: YYYY-mm-dd HH:MM:SS
+      - 3: YYYY-mm-dd HH:MM:SS.mmm
+      - 6: YYYY-mm-dd HH:MM:SS.ffffff
+    """
+    from datetime import datetime
+
+    if calendar == "gregorian":
+        now = datetime.now()
+        base = now.strftime("%Y-%m-%d %H:%M:%S")
+        if microseconds == 0:
+            return base
+        us = now.microsecond  # 0..999999
+        if microseconds == 3:
+            return f"{base}.{us // 1000:03d}"
+        return f"{base}.{us:06d}"
+
+    # Jalali
+    try:
+        import jdatetime
+    except Exception:
+        # fallback
+        now = datetime.now()
+        base = now.strftime("%Y-%m-%d %H:%M:%S")
+        if microseconds == 0:
+            return base + " (gregorian-fallback)"
+        us = now.microsecond
+        if microseconds == 3:
+            return f"{base}.{us // 1000:03d} (gregorian-fallback)"
+        return f"{base}.{us:06d} (gregorian-fallback)"
+
+    jnow = jdatetime.datetime.now()
+    base = jnow.strftime("%Y-%m-%d %H:%M:%S")
+    if microseconds == 0:
+        return base
+    us = jnow.microsecond
+    if microseconds == 3:
+        return f"{base}.{us // 1000:03d}"
+    return f"{base}.{us:06d}"
+
+
+def _build_context_suffix(record: logging.LogRecord, *, include_pid: bool, include_thread: bool) -> str:
+    parts: list[str] = []
+    if include_pid:
+        parts.append(str(record.process))
+    if include_thread:
+        parts.append(record.threadName)
+    if not parts:
+        return ""
+    return " [" + ":".join(parts) + "]"
+
+
+class _TextFormatter(logging.Formatter):
+    def __init__(
+        self,
+        *,
+        app_name: str | None,
+        calendar: str,
+        microseconds: int,
+        include_app_name: bool,
+        include_name: bool,
+        include_pid: bool,
+        include_thread: bool,
+    ):
+        super().__init__()
+        self.app_name = app_name
+        self.calendar = calendar
+        self.microseconds = microseconds
+        self.include_app_name = include_app_name
+        self.include_name = include_name
+        self.include_pid = include_pid
+        self.include_thread = include_thread
+
+    def format(self, record: logging.LogRecord) -> str:
+        ts = _format_timestamp(calendar=self.calendar, microseconds=self.microseconds)
+
+        # best practice: keep a stable, grep-friendly prefix
+        prefix_parts = [ts, record.levelname]
+
+        if self.include_app_name:
+            prefix_parts.append(self.app_name)
+
+        if self.include_name:
+            prefix_parts.append(record.name)
+
+        ctx = _build_context_suffix(record, include_pid=self.include_pid, include_thread=self.include_thread)
+
+        base = " ".join(prefix_parts) + ctx + " " + record.getMessage()
+
+        if record.exc_info:
+            base += "\n" + self.formatException(record.exc_info)
+
+        return base
+
+
+class _JsonFormatter(logging.Formatter):
+    def __init__(
+        self,
+        *,
+        app_name: str | None,
+        calendar: str,
+        microseconds: int,
+        include_app_name: bool,
+        include_name: bool,
+        include_pid: bool,
+        include_thread: bool,
+    ):
+        super().__init__()
+        self.app_name = app_name
+        self.calendar = calendar
+        self.microseconds = microseconds
+        self.include_app_name = include_app_name
+        self.include_name = include_name
+        self.include_pid = include_pid
+        self.include_thread = include_thread
+
+    def format(self, record: logging.LogRecord) -> str:
+        import json
+
+        payload: dict[str, object] = {
+            "ts": _format_timestamp(calendar=self.calendar, microseconds=self.microseconds),
+            "level": record.levelname,
+            "msg": record.getMessage(),
+        }
+
+        if self.include_app_name:
+            payload["app"] = self.app_name
+        if self.include_name:
+            payload["logger"] = record.name
+        if self.include_pid:
+            payload["process"] = record.process
+        if self.include_thread:
+            payload["thread"] = record.threadName
+        if record.exc_info:
+            payload["exc"] = self.formatException(record.exc_info)
+
+        return json.dumps(payload, ensure_ascii=False)
+
+
+if __name__ == "__main__":
+    setup_logging(
+        calendar="jalali",
+        include_app_name=False,
+        include_name=True,
+        include_pid=False,
+        include_thread=False,
+        console_level=logging.DEBUG,
+        use_rich=True,
+    )
+    logging.getLogger("elasticsearch").setLevel(logging.WARNING)
+    logging.getLogger("elastic_transport").setLevel(logging.WARNING)
+    logging.getLogger("urllib3").setLevel(logging.WARNING)
+    logging.getLogger("tortoise.db_client").setLevel(logging.WARNING)
+    logging.getLogger("asyncio").setLevel(logging.WARNING)
+    logging.getLogger("tortoise").setLevel(logging.WARNING)
+    log = logging.getLogger(__name__)
+    log.info("script_started")
+    # time.sleep(2)

utils/read_wirte_file_utils.py

@@ -8,18 +8,24 @@ def serialize_sets(obj):
     raise TypeError
 
 
-def write_to_file(content, file_path: Path, json_dumps=False, indent=None):
-    file_path.parent.mkdir(parents=True, exist_ok=True)
-    with open(file_path, mode="w") as f:
+def write_to_file(content, file_path: str, json_dumps=False, indent=None):
+    PWD = Path(__file__).resolve().parent.parent
+    file = PWD / file_path
+    file.parent.mkdir(parents=True, exist_ok=True)
+    with open(file, mode="w") as f:
         if json_dumps:
-            return f.write(
-                json.dumps(
-                    content, indent=indent, ensure_ascii=False, default=serialize_sets
-                )
-            )
+            return f.write(json.dumps(content, indent=indent, ensure_ascii=False, default=serialize_sets))
         return f.write(content)
 
 
+def write_to_json_file(content, file_path: str, indent: int | None = 4):
+    pwd = Path(__file__).resolve().parent.parent
+    file = pwd / file_path
+    file.parent.mkdir(parents=True, exist_ok=True)
+    with open(file, mode="w") as f:
+        return f.write(json.dumps(content, indent=indent, ensure_ascii=False, default=serialize_sets))
+
+
 def append_to_file(content: str, file_path: Path):
     file_path.parent.mkdir(parents=True, exist_ok=True)
     with open(file_path, mode="a") as f:

utils/sound_utils.py

@@ -0,0 +1,24 @@
+import subprocess
+
+SOUNDS = {
+    "alarm": {
+        "file_path": "/usr/share/sounds/freedesktop/stereo/alarm-clock-elapsed.oga",
+        "volume": "50%",
+    },
+    "notif": {"file_path": "utils/assets/error-08-206492.mp3", "volume": "85%"},
+}
+
+
+def play_alarm_sound(file_path, volume):
+    subprocess.run(
+        [
+            "amixer",
+            "-D",
+            "pulse",
+            "sset",
+            "Master",
+            volume,
+        ],
+        capture_output=True,
+    )
+    subprocess.run(["paplay", file_path], capture_output=True)

utils/timer_utils.py

@@ -1,19 +1,35 @@
-from time import time
+from datetime import datetime
 
-start_time_of_timers = dict()
+from rich.console import Console
+
+console = Console()
+
+timers_started_ats = dict()
 
 
-def start_timer(name=1, print_=True):
-    started_at = time()
-    start_time_of_timers[name] = started_at
-    if print_:
-        print(f"{name} started at: {started_at}")
+def start_timer(name=1, do_print=True):
+    started_at = datetime.now()
+    timers_started_ats[name] = started_at
+    if do_print:
+        console.print(f"{name:<{len(str(name)) + 4}} ", end="")
+        print(f"started at: {started_at.strftime('%H:%M:%S.%f')}")
     return name, started_at
 
 
-def stop_timer(name=1, print_=True):
-    ended_at = time()
-    took = round(ended_at - start_time_of_timers[name], 6)
-    if print_:
-        print(f"{name} ended at: {ended_at}\ttook: {took} second(s)")
-    return name, round(ended_at - start_time_of_timers[name], 6)
+def stop_timer(name=1, do_print=True):
+    ended_at = datetime.now()
+    took = ended_at - timers_started_ats[name]
+    if do_print:
+        console.print(f"{name:<{len(str(name)) + 4}} ", end="")
+        print(f"{'ended at:':>11} {ended_at.strftime('%H:%M:%S.%f')}    ", end="")
+        console.print(f"took: {took}")
+    return name, ended_at, took
+
+
+if __name__ == "__main__":
+    from time import sleep
+
+    start_timer()
+    sleep(1)
+    stop_timer()
+