internet_use_control/windows/webrestrict/overlay.py

"""The child's warning window (AGENTS.md FR-9).

A small always-on-top Tkinter window: what is blocked, why, and until when. It
is an *alert*, not a defence in depth — the child can close it, and on Windows
the enforcement is the process block itself (FR-3) plus the presence and
offline reports that reach Sam (FR-6, FR-13).

Tkinter wants one thread that owns the interpreter, so the window runs on its
own daemon thread and the agent talks to it through a queue. If Tk is missing,
or there is no interactive desktop (a service session, a headless test), the
overlay is simply unavailable and blocking continues.
"""

from __future__ import annotations

import logging
import queue
import threading
from dataclasses import dataclass
from typing import Sequence

log = logging.getLogger("webrestrict.overlay")

TITLE = "Screen time"


@dataclass(frozen=True)
class Notice:
    """What the window shows while a block is active."""

    reason: str
    until: str
    names: Sequence[str]

    @property
    def blocked_text(self) -> str:
        if not self.names:
            return "This computer is blocked while the window is active."
        shown = ", ".join(self.names[:8])
        if len(self.names) > 8:
            shown += f", and {len(self.names) - 8} more"
        return f"Blocked now: {shown}"


class Overlay:
    """A queue-fed Tkinter window with a tiny, testable interface."""

    def __init__(self, enabled: bool = True):
        self.enabled = enabled
        self.available = False
        self._queue: "queue.Queue[tuple[str, object]]" = queue.Queue()
        self._thread: threading.Thread | None = None
        self._shown: Notice | None = None

    def start(self) -> None:
        if not self.enabled:
            log.info("overlay disabled by configuration")
            return
        self._thread = threading.Thread(target=self._run, name="webrestrict-overlay", daemon=True)
        self._thread.start()

    def update(self, notice: Notice | None) -> None:
        """Show the notice, hide on ``None``. Repeat calls with the same content are dropped."""
        if notice == self._shown:
            return
        self._shown = notice
        if notice is None:
            self._queue.put(("hide", None))
        else:
            self._queue.put(("show", notice))

    def stop(self) -> None:
        self._queue.put(("stop", None))

    # ---------------------------------------------------------------------
    def _run(self) -> None:  # pragma: no cover - needs a desktop
        try:
            import tkinter as tk
        except ImportError as exc:
            log.warning("tkinter is unavailable, no warning window: %s", exc)
            return

        try:
            root = tk.Tk()
        except Exception as exc:  # noqa: BLE001 - no interactive desktop
            log.warning("the warning window is unavailable: %s", exc)
            return

        root.withdraw()
        root.title(TITLE)
        root.attributes("-topmost", True)
        root.resizable(False, False)

        frame = tk.Frame(root, padx=18, pady=16)
        frame.pack(fill="both", expand=True)
        heading = tk.Label(frame, text="Screen time is over", font=("Segoe UI", 14, "bold"), anchor="w")
        heading.pack(fill="x")
        reason_label = tk.Label(frame, text="", wraplength=380, justify="left", anchor="w")
        reason_label.pack(fill="x", pady=(8, 0))
        until_label = tk.Label(frame, text="", anchor="w", font=("Segoe UI", 9))
        until_label.pack(fill="x", pady=(6, 0))
        blocked_label = tk.Label(frame, text="", wraplength=380, justify="left", anchor="w")
        blocked_label.pack(fill="x", pady=(8, 0))
        tk.Button(frame, text="Close warning", command=root.withdraw).pack(anchor="e", pady=(12, 0))
        root.geometry("420x260")

        self.available = True
        log.info("warning window is available")

        def drain() -> None:  # pragma: no cover - needs a desktop
            stop = False
            while True:
                try:
                    action, payload = self._queue.get_nowait()
                except queue.Empty:
                    break
                if action == "show" and isinstance(payload, Notice):
                    reason_label.config(text=f"The rules say: {payload.reason}.")
                    until_label.config(text=f"Until {payload.until}." if payload.until else "No end time is set.")
                    blocked_label.config(text=payload.blocked_text)
                    root.deiconify()
                    root.lift()
                    root.attributes("-topmost", True)
                    place_bottom_right(root)
                elif action == "hide":
                    root.withdraw()
                elif action == "stop":
                    stop = True
            if stop:
                root.destroy()
                return
            root.after(200, drain)

        root.after(200, drain)
        try:
            root.mainloop()
        except Exception as exc:  # noqa: BLE001 - nothing left to do
            log.debug("warning window loop ended: %s", exc)


def place_bottom_right(root) -> None:  # pragma: no cover - needs a desktop
    """Park the window bottom-right, out of the way of what it is covering."""
    root.update_idletasks()
    width = root.winfo_width() or 420
    height = root.winfo_height() or 260
    screen = root.winfo_screenwidth(), root.winfo_screenheight()
    root.geometry(f"{width}x{height}+{max(screen[0] - width - 24, 0)}+{max(screen[1] - height - 72, 0)}")