Newer
Older
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
#!/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()