timka123
77b26757bf
- Refactored TimerPage component to improve layout and user experience. - Updated theme configuration to utilize design tokens for better maintainability. - Introduced new tokens for colors and typography to align with design principles. - Simplified state management and improved timer functionality. - Enhanced accessibility and responsiveness of UI elements.
214 lines
7.4 KiB
Python
214 lines
7.4 KiB
Python
"""Blocking policy.
|
|
|
|
The backend decides *whether* blocking should be on right now and *what* is on
|
|
the list; the desktop app is the thing that actually enforces it (it is the only
|
|
piece running on your machine with the privileges to rewrite `/etc/hosts` or
|
|
kill a process). Keeping the policy here means the web UI can edit the rules,
|
|
and a second machine running the desktop app obeys the same list.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import date, datetime
|
|
from fnmatch import fnmatch
|
|
from typing import Any
|
|
|
|
from sqlalchemy import func, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import BlockEvent
|
|
from app.services import pomodoro
|
|
from app.settings_schema import AppSettings
|
|
from app.timeutil import ensure_utc, in_window, local_date, local_time
|
|
|
|
|
|
def is_allowed(host: str, allow_masks: list[str]) -> bool:
|
|
"""Whether an allow-list entry (plain or masked) covers this hostname."""
|
|
return any(fnmatch(host, mask) for mask in allow_masks)
|
|
|
|
|
|
def expand_entry(raw: str, subdomains: list[str]) -> tuple[list[str], str | None]:
|
|
"""Turn one blocklist entry into concrete hostnames.
|
|
|
|
A hosts file has no wildcards, so a mask has to be expanded into real names:
|
|
|
|
* `youtube.com` → `youtube.com`, `www.youtube.com`, `m.youtube.com`
|
|
* `*.youtube.com` → the bare domain plus every configured subdomain
|
|
* `*gram*` → nothing expandable; returned as a leftover pattern so the
|
|
UI can say so instead of silently dropping it
|
|
|
|
Returns `(hostnames, unexpandable_pattern)`.
|
|
"""
|
|
entry = raw.strip().lower().lstrip(".")
|
|
if not entry:
|
|
return [], None
|
|
|
|
if entry.startswith("*."):
|
|
base = entry[2:]
|
|
if not base or "*" in base:
|
|
return [], entry
|
|
return [base, *[f"{prefix}.{base}" for prefix in subdomains]], None
|
|
|
|
if "*" in entry:
|
|
return [], entry
|
|
|
|
return [entry, f"www.{entry}", f"m.{entry}"], None
|
|
|
|
|
|
def expand_domains(settings: AppSettings) -> list[str]:
|
|
"""Every hostname the desktop app should point at the redirect IP.
|
|
|
|
`youtube.com` means "that site", so `www.` and `m.` come along; masks are
|
|
expanded as described above; and anything the allow list covers — itself
|
|
maskable — is taken back out, which is how `*.google.com` blocked plus
|
|
`*.docs.google.com` allowed works.
|
|
"""
|
|
blocker = settings.blocker
|
|
allow_masks = [
|
|
entry.strip().lower().lstrip(".") for entry in blocker.allowed_domains if entry.strip()
|
|
]
|
|
subdomains = [part.strip().lower() for part in blocker.subdomain_expansions if part.strip()]
|
|
|
|
hosts: list[str] = []
|
|
for raw in blocker.blocked_domains:
|
|
expanded, _ = expand_entry(raw, subdomains)
|
|
for host in expanded:
|
|
if host not in hosts and not is_allowed(host, allow_masks):
|
|
hosts.append(host)
|
|
return hosts
|
|
|
|
|
|
def unexpandable_masks(settings: AppSettings) -> list[str]:
|
|
"""Blocklist entries a hosts file cannot express, so the UI can flag them."""
|
|
subdomains = [part.strip().lower() for part in settings.blocker.subdomain_expansions if part.strip()]
|
|
leftovers: list[str] = []
|
|
for raw in settings.blocker.blocked_domains:
|
|
_, pattern = expand_entry(raw, subdomains)
|
|
if pattern and pattern not in leftovers:
|
|
leftovers.append(pattern)
|
|
return leftovers
|
|
|
|
|
|
def schedule_active(settings: AppSettings, moment: datetime | None = None) -> bool:
|
|
now_time = local_time(settings.timezone, moment)
|
|
weekday = (moment or datetime.now()).weekday()
|
|
for window in settings.blocker.schedule:
|
|
if weekday in window.days and in_window(now_time, window.start, window.end):
|
|
return True
|
|
return False
|
|
|
|
|
|
def policy(session: Session, settings: AppSettings) -> dict[str, Any]:
|
|
"""What the desktop app polls: should I block, and what."""
|
|
blocker = settings.blocker
|
|
current = pomodoro.current_session(session)
|
|
active = False
|
|
reason = "disabled"
|
|
|
|
if blocker.enabled:
|
|
if blocker.mode == "always":
|
|
active, reason = True, "always on"
|
|
elif blocker.mode == "scheduled":
|
|
active = schedule_active(settings)
|
|
reason = "inside a blocking window" if active else "outside every window"
|
|
else: # during_work
|
|
if current is not None and current.kind == "work" and current.status in ("running", "paused"):
|
|
active, reason = True, "a work interval is running"
|
|
elif (
|
|
current is not None
|
|
and current.kind != "work"
|
|
and blocker.block_during_breaks
|
|
and current.status in ("running", "paused")
|
|
):
|
|
active, reason = True, "blocking stays on through breaks"
|
|
else:
|
|
reason = "no work interval is running"
|
|
|
|
return {
|
|
"active": active,
|
|
"reason": reason,
|
|
"mode": blocker.mode,
|
|
"enabled": blocker.enabled,
|
|
"strict": blocker.strict,
|
|
"redirect_ip": blocker.redirect_ip,
|
|
"domains": expand_domains(settings) if active else [],
|
|
"apps": [app.strip() for app in blocker.blocked_apps if app.strip()] if active else [],
|
|
"unexpandable": unexpandable_masks(settings),
|
|
"kill_apps": blocker.kill_blocked_apps,
|
|
"notify_on_block": blocker.notify_on_block,
|
|
"poll_seconds": blocker.app_poll_seconds,
|
|
"session_id": current.id if current else None,
|
|
"all_domains": expand_domains(settings),
|
|
"all_apps": [app.strip() for app in blocker.blocked_apps if app.strip()],
|
|
}
|
|
|
|
|
|
def record_event(
|
|
session: Session,
|
|
settings: AppSettings,
|
|
*,
|
|
kind: str,
|
|
target: str,
|
|
action: str = "blocked",
|
|
details: str | None = None,
|
|
) -> BlockEvent:
|
|
current = pomodoro.current_session(session)
|
|
event = BlockEvent(
|
|
kind=kind,
|
|
target=target.strip().lower(),
|
|
action=action,
|
|
details=details,
|
|
local_date=local_date(settings.timezone),
|
|
session_id=current.id if current else None,
|
|
)
|
|
session.add(event)
|
|
session.commit()
|
|
return event
|
|
|
|
|
|
def events_for_day(session: Session, settings: AppSettings, day: date | None = None) -> dict[str, Any]:
|
|
day = day or local_date(settings.timezone)
|
|
rows = list(session.scalars(select(BlockEvent).where(BlockEvent.local_date == day)))
|
|
by_target = dict(
|
|
session.execute(
|
|
select(BlockEvent.target, func.count(BlockEvent.id))
|
|
.where(BlockEvent.local_date == day)
|
|
.group_by(BlockEvent.target)
|
|
.order_by(func.count(BlockEvent.id).desc())
|
|
).all()
|
|
)
|
|
return {
|
|
"date": day.isoformat(),
|
|
"total": len(rows),
|
|
"by_target": [{"target": target, "count": count} for target, count in by_target.items()],
|
|
"events": [
|
|
{
|
|
"id": row.id,
|
|
"kind": row.kind,
|
|
"target": row.target,
|
|
"action": row.action,
|
|
"at": ensure_utc(row.at).isoformat(),
|
|
"details": row.details,
|
|
}
|
|
for row in sorted(rows, key=lambda row: ensure_utc(row.at), reverse=True)[:100]
|
|
],
|
|
}
|
|
|
|
|
|
DEFAULT_DOMAINS = [
|
|
"*.youtube.com",
|
|
"twitter.com",
|
|
"x.com",
|
|
"*.instagram.com",
|
|
"tiktok.com",
|
|
"reddit.com",
|
|
"facebook.com",
|
|
"vk.com",
|
|
"twitch.tv",
|
|
"netflix.com",
|
|
"9gag.com",
|
|
"pikabu.ru",
|
|
]
|
|
|
|
DEFAULT_APPS = ["steam*", "discord", "telegram-desktop", "spotify"]
|