grym
/
fedmon
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add custom ignores; improve error handling; refactor notifications

This commit is contained in:
grym 2022-01-25 12:09:17 -05:00
parent 4e00328786
commit 206772f760
1 changed files with 30 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

48
fedmon/cli.py
View File

@ -6,7 +6,7 @@ import time
from datetime import datetime
from enum import Enum
from io import StringIO
from typing import Sequence, Tuple
from typing import List, Optional, Sequence, Tuple
import cachetools.func
import httpx
@ -65,12 +65,20 @@ def parse_data(data: Tuple[str]) -> Tuple[FedSite]:
@functools.lru_cache()
def analyze_data(data: Tuple[FedSite]) -> Summary:
def analyze_data(data: Tuple[FedSite], to_ignore: List[str]) -> Summary:
if to_ignore is None:
to_ignore = []
usps = {x for x in data if "tools.usps.com" in x.page}
usps_visitors = max(x.active_visitors for x in usps)
hot = tuple(
sorted(
(x for x in data if x.active_visitors >= usps_visitors and x not in usps),
(
x
for x in data
if x.active_visitors >= usps_visitors
and x not in usps
and not any(i in x.page for i in to_ignore)
),
key=lambda x: x.active_visitors,
)
)
@ -101,6 +109,19 @@ def format_response(summary: Summary, format: AllowedFormat, count: int) -> str:
return tab(data, tablefmt=format.value)
def notify(summary: Summary, counter) -> None:
try:
subprocess.check_call(
[
"notify-send",
appname,
format_response(summary, AllowedFormat.json, next(counter)),
]
)
except subprocess.CalledProcessError as e:
typer.echo(e, err=True)
@app.command()
def main(
format: AllowedFormat = typer.Option(
@ -110,6 +131,7 @@ def main(
default=False,
help="If set, use `notify-send` to raise a desktop notification, if available.",
),
ignore: Optional[List[str]] = typer.Option(None),
):
prev = None
counter = itertools.count()
@ -117,25 +139,15 @@ def main(
try:
data = get_data(config.base_url)
data = parse_data(data)
summary = analyze_data(data)
summary = analyze_data(data, to_ignore=ignore)
if summary != prev:
if summary.sites:
typer.echo(format_response(summary, format, next(counter)))
c = next(counter)
typer.echo(format_response(summary, format, c))
if send_notification:
try:
subprocess.check_call(
[
"notify-send",
appname,
format_response(
summary, AllowedFormat.json, next(counter)
),
]
)
except subprocess.CalledProcessError as e:
typer.echo(e, err=True)
notify(summary, c)
prev = summary
except httpx.ConnectError as e:
except (httpx.ConnectError, httpx.ReadTimeout) as e:
typer.echo(e, err=True)
finally:
time.sleep(config.poll)