73 lines
2.1 KiB
Python
73 lines
2.1 KiB
Python
from typing import Generator, Type, Dict
|
|
import os.path
|
|
from pathlib import Path
|
|
|
|
import requests
|
|
from ics import Calendar
|
|
|
|
from ..utils import config, error, prompt, ICS_FILE
|
|
from ..feeds import *
|
|
|
|
|
|
NAME_TO_FEED: Dict[str, Type[Feed]] = {
|
|
"mastodon": MastodonFeed,
|
|
"twitter": TwitterFeed,
|
|
}
|
|
|
|
|
|
class Routine:
|
|
@staticmethod
|
|
def iter_feeds() -> Generator[Type[Feed], None, None]:
|
|
for feed in config.get_field("active_feeds", []):
|
|
if feed not in NAME_TO_FEED:
|
|
raise ValueError(f"Feed {feed} is not implemented.")
|
|
|
|
yield NAME_TO_FEED[feed]
|
|
|
|
@staticmethod
|
|
def init_feed(feed: Type[Feed]) -> Feed:
|
|
feed_config = config.get_field(feed.__config_name__, {})
|
|
|
|
try:
|
|
feed_instance = feed(**feed_config)
|
|
except (TypeError, error.InvalidCredential) as e:
|
|
print(feed_config)
|
|
raise e
|
|
config.set_field(feed.__name__, feed.prompt_auth(), update_dict=True)
|
|
feed_config = config.get_field(feed.__config_name__, {})
|
|
|
|
try:
|
|
return feed(**feed_config)
|
|
except (TypeError, error.InvalidCredential):
|
|
raise error.InvalidCredential(f"Invalid credentials for {feed.__name__}.")
|
|
|
|
@staticmethod
|
|
def get_ics(use_cached: bool = False) -> Path:
|
|
if use_cached and ICS_FILE.exists():
|
|
with ICS_FILE.open("r") as f:
|
|
return Calendar(f.read())
|
|
|
|
ics_url = config.get_field("ics_url")
|
|
if not len(ics_url):
|
|
ics_url = prompt.for_string("url or path to the ICS file")
|
|
config.set_field("ics_url", ics_url)
|
|
|
|
origin_path = Path(ics_url)
|
|
|
|
if origin_path.is_file():
|
|
with origin_path.open("rb") as f:
|
|
with ICS_FILE.open("wb") as t:
|
|
t.write(f.read())
|
|
else:
|
|
r = requests.get(ics_url)
|
|
with ICS_FILE.open("wb") as t:
|
|
t.write(r.content)
|
|
|
|
with ICS_FILE.open("r") as f:
|
|
return Calendar(f.read())
|
|
|
|
|
|
|
|
def run(self):
|
|
pass
|