add script
This commit is contained in:
parent
f24257df32
commit
ae14aaa467
1 changed files with 120 additions and 0 deletions
120
ics-now.py
Normal file
120
ics-now.py
Normal file
|
@ -0,0 +1,120 @@
|
|||
#!/usr/bin/env python3
|
||||
|
||||
"""
|
||||
This script yields JSON info about the current event + the next one.
|
||||
|
||||
It depends on the nice ics-query https://github.com/niccokunzmann/ics-query
|
||||
"""
|
||||
|
||||
import datetime
|
||||
import json
|
||||
import sys
|
||||
from argparse import ArgumentParser
|
||||
from pathlib import Path
|
||||
from subprocess import check_output
|
||||
|
||||
from icalendar import Calendar, Event
|
||||
|
||||
|
||||
def parse_dt(s: str) -> datetime.datetime:
|
||||
return datetime.datetime.strptime(s, "%Y-%m-%dT%H:%M:%S")
|
||||
|
||||
|
||||
def parse_minutes(s: str) -> datetime.timedelta:
|
||||
return datetime.timedelta(minutes=int(s))
|
||||
|
||||
|
||||
class IcsNow:
|
||||
def __init__(self):
|
||||
self.parser = ArgumentParser()
|
||||
self.parser.add_argument("ics", type=Path)
|
||||
self.parser.add_argument("--tz", default="Europe/Rome")
|
||||
self.parser.add_argument(
|
||||
"--now",
|
||||
default=datetime.datetime.now(),
|
||||
type=parse_dt,
|
||||
help="Simulate different time. Mostly for debug",
|
||||
)
|
||||
self.parser.add_argument(
|
||||
"--until",
|
||||
metavar="DURATION",
|
||||
type=parse_minutes,
|
||||
default=parse_minutes(60),
|
||||
help='How many minute should "next" include?',
|
||||
)
|
||||
self.parser.add_argument(
|
||||
"--ics-query",
|
||||
type=Path,
|
||||
default=Path("/usr/local/bin/ics-query"),
|
||||
help="Location of ics-query",
|
||||
)
|
||||
|
||||
def load_from_icsquery(self, cmd, args) -> list[Event]:
|
||||
result: bytes = check_output(
|
||||
[
|
||||
self.args.ics_query,
|
||||
cmd,
|
||||
"--tz",
|
||||
self.args.tz,
|
||||
"--component",
|
||||
"VEVENT",
|
||||
*args,
|
||||
self.args.ics,
|
||||
]
|
||||
)
|
||||
if not result:
|
||||
return []
|
||||
return Calendar.from_ical(result, multiple=True)
|
||||
|
||||
def pick_most_important(self, events: list[Event]) -> Event | None:
|
||||
if not events:
|
||||
return None
|
||||
# FIXME: sort by priority
|
||||
return events[0]
|
||||
|
||||
def pick_next(self, events: list[Event], current) -> Event | None:
|
||||
events.remove(current)
|
||||
if not events:
|
||||
return None
|
||||
events.sort(key=lambda e: e['DTSTART'].dt)
|
||||
return events[0]
|
||||
|
||||
def run(self):
|
||||
self.args = self.parser.parse_args()
|
||||
ret = {}
|
||||
|
||||
current = self.pick_most_important(self.load_from_icsquery(
|
||||
"at", [self.args.now.strftime("%Y-%m-%dT%H:%M:%S")]
|
||||
))
|
||||
ret["now"] = self.serialize(current)
|
||||
|
||||
next_events = self.load_from_icsquery(
|
||||
"between",
|
||||
[
|
||||
self.args.now.strftime("%Y-%m-%dT%H:%M:%S"),
|
||||
(self.args.now + self.args.until).strftime("%Y-%m-%dT%H:%M:%S"),
|
||||
],
|
||||
)
|
||||
ret["next"] = self.serialize(self.pick_next(next_events, current))
|
||||
json.dump(ret, sys.stdout, indent=2)
|
||||
sys.stdout.write("\n")
|
||||
|
||||
def serialize(self, event: Event | None):
|
||||
if event is None:
|
||||
return None
|
||||
out = {
|
||||
"title": event["SUMMARY"],
|
||||
}
|
||||
for field in ["start", "end"]:
|
||||
calfield = f"dt{field}"
|
||||
if calfield in event:
|
||||
out[field] = event[calfield].dt.isoformat()
|
||||
for field in ["uid", "description", "url"]:
|
||||
if field in event:
|
||||
out[field] = event[field]
|
||||
return out
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
cli = IcsNow()
|
||||
cli.run()
|
Loading…
Reference in a new issue