Added base script

notify-updates.py

+#!/usr/bin/env python3
+
+import logging
+import sqlite3
+import sys
+from typing import IO, List
+
+import click
+import yaml
+
+from updates_notifier import EntriesDatabase, Entry
+from updates_notifier.fetchers import create_fetchers_from_dict
+from updates_notifier.fetchers.project_fetcher import ProjectFetcher
+
+logger = logging.getLogger("notify-updates")
+
+
+def _insert_entries(entries_db: EntriesDatabase, entries: List[Entry]) -> None:
+    with entries_db:
+        for entry in entries:
+            entries_db.insert(entry)
+
+
+def _get_new_updates(
+    entries_db: EntriesDatabase, fetchers: List[ProjectFetcher]
+) -> List[Entry]:
+    entries: List[Entry] = []
+    for fetcher in fetchers:
+        entries.extend(fetcher.fetch_entries())
+
+    return list(filter(lambda e: not entries_db.contains(e), entries))
+
+
+def _print_updates(entries: List[Entry]) -> None:
+    nb_updates = len(entries)
+    if nb_updates > 0:
+        logger.info("%s new updates detected.", nb_updates)
+    else:
+        logger.info("No new updates detected.")
+
+    for entry in entries:
+        logger.debug(
+            f"New update: id={entry.feed_id} name='{entry.name}' version='{entry.version}', url='{entry.url}'"
+        )
+
+
+@click.command()
+@click.option(
+    "--dry-run",
+    "-d",
+    is_flag=True,
+    default=False,
+    help="Search for new updates, but no db updates or notifications",
+)
+@click.option("--verbose", "-v", is_flag=True, help="Toggle debug output")
+@click.argument("feeds_cfg", type=click.File(mode="r"))
+@click.argument("db_path", type=click.Path(dir_okay=False))
+def main(feeds_cfg: IO[str], db_path: str, dry_run: str, verbose: bool) -> None:
+    """
+    FEEDS_CFG:  Path to the YAML file containing the feeds' parameters
+    DB_PATH: Path to the SQLite3 database file
+    """
+    if verbose:
+        logging_level = logging.DEBUG
+        logging.getLogger("urllib3").setLevel(logging.WARNING)
+    else:
+        logging_level = logging.INFO
+    logging.basicConfig(
+        format="%(asctime)s %(levelname)s:%(name)s:%(funcName)s:%(message)s",
+        level=logging_level,
+    )
+
+    try:
+        feeders_dict = yaml.safe_load(feeds_cfg)
+    except yaml.error.YAMLError as err:
+        logger.error("Could not parse YAML feeds configuration:%s", str(err))
+        sys.exit(1)
+
+    try:
+        entries_db = EntriesDatabase(db_path)
+    except sqlite3.Error as err:
+        logger.error(
+            "Could not load or create SQLite3 database at path %s: %s",
+            db_path,
+            str(err),
+        )
+        sys.exit(1)
+
+    try:
+        fetchers = create_fetchers_from_dict(feeders_dict)
+    except RuntimeError as err:
+        logger.error("Could not create fetchers:%s", str(err))
+        sys.exit(1)
+
+    entries = _get_new_updates(entries_db, fetchers)
+
+    _print_updates(entries)
+
+    if not dry_run:
+        _insert_entries(entries_db, entries)
+
+
+if __name__ == "__main__":
+    main()