42 lines
1.2 KiB
Python
42 lines
1.2 KiB
Python
|
from typing import Generator, Type, Dict
|
||
|
|
|
||
|
|
from ..utils import config, error
|
||
|
|
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__}.")
|
||
|
|
|
||
|
|
|
||
|
|
def run(self):
|
||
|
|
pass
|