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

Add nasty formatting hacks for table displays to look nice

This commit is contained in:
grym 2022-01-21 22:11:47 -05:00
parent 60a23b6c81
commit fd4652ded6
1 changed files with 20 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
fedmon/cli.py
View File

@ -1,20 +1,21 @@
import csv import csv
import functools import functools
import itertools
import time import time
from datetime import datetime from datetime import datetime
from enum import Enum from enum import Enum
from io import StringIO from io import StringIO
from typing import List, Tuple from typing import Tuple
import cachetools.func import cachetools.func
import httpx import httpx
import tabulate
import typer import typer
from pydantic import BaseModel, BaseSettings, Field from pydantic import BaseModel, BaseSettings, Field
from tabulate import tabulate as tab from tabulate import tabulate as tab
from tabulate import tabulate_formats
app = typer.Typer() app = typer.Typer()
tabulate_formats = set(tabulate.tabulate_formats) tabulate_formats = set(tabulate_formats)
AllowedFormat = Enum( AllowedFormat = Enum(
"AllowedFormat", {"json": "json", "csv": "csv", **{v: v for v in tabulate_formats}} "AllowedFormat", {"json": "json", "csv": "csv", **{v: v for v in tabulate_formats}}
@ -61,7 +62,7 @@ def parse_data(data: Tuple[str]) -> Tuple[FedSite]:
@functools.lru_cache() @functools.lru_cache()
def analyze_data(data: Tuple[FedSite]) -> Tuple[FedSite]: def analyze_data(data: Tuple[FedSite]) -> Summary:
usps = {x for x in data if "tools.usps.com" in x.page} usps = {x for x in data if "tools.usps.com" in x.page}
usps_visitors = max(x.active_visitors for x in usps) usps_visitors = max(x.active_visitors for x in usps)
hot = tuple( hot = tuple(
@ -73,7 +74,7 @@ def analyze_data(data: Tuple[FedSite]) -> Tuple[FedSite]:
return Summary(sites=hot) return Summary(sites=hot)
def format_response(summary: Summary, format: AllowedFormat) -> str: def format_response(summary: Summary, format: AllowedFormat, count: int) -> str:
if format == AllowedFormat.json: if format == AllowedFormat.json:
return summary.json(indent=4) return summary.json(indent=4)
elif format == AllowedFormat.csv: elif format == AllowedFormat.csv:
@ -81,22 +82,32 @@ def format_response(summary: Summary, format: AllowedFormat) -> str:
to_write = summary.dict()["sites"] to_write = summary.dict()["sites"]
headers = FedSite.schema()["properties"].keys() headers = FedSite.schema()["properties"].keys()
writer = csv.DictWriter(data, fieldnames=headers) writer = csv.DictWriter(data, fieldnames=headers)
writer.writeheader() if count == 0:
writer.writerows(summary.dict()["sites"]) writer.writeheader()
writer.writerows(to_write)
return data.getvalue() return data.getvalue()
else: else:
return tab(summary.dict()["sites"], tablefmt=format.value, headers="keys") data = summary.dict()["sites"]
for site in data:
site["active_visitors"] = site["active_visitors"].rjust(19)
site["page_title"] = site["page_title"].rjust(25)
site["page"] = site["page"].rjust(26)
if count == 0:
return tab(data, tablefmt=format.value, headers="keys")
else:
return tab(data, tablefmt=format.value)
@app.command() @app.command()
def main(format: AllowedFormat = typer.Option(default="json", show_choices=True)): def main(format: AllowedFormat = typer.Option(default="json", show_choices=True)):
prev = None prev = None
counter = itertools.count()
while True: while True:
data = get_data(config.base_url) data = get_data(config.base_url)
data = parse_data(data) data = parse_data(data)
summary = analyze_data(data) summary = analyze_data(data)
if summary != prev: if summary != prev:
if summary: if summary:
typer.echo(format_response(summary, format)) typer.echo(format_response(summary, format, next(counter)))
prev = summary prev = summary
time.sleep(config.poll) time.sleep(config.poll)