-
Notifications
You must be signed in to change notification settings - Fork 0
/
web591.py
99 lines (74 loc) · 2.41 KB
/
web591.py
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
#!/usr/bin/env python3
"""Module for 591 crawler."""
import argparse
import logging
import typing
import toml
import fetcher
import store
import notify
logger = logging.getLogger("crawler")
log_formatter = logging.Formatter(
fmt="%(asctime)s: %(message)s", datefmt="%Y-%m-%d %H:%M:%S"
)
stream_handler = logging.StreamHandler()
stream_handler.setFormatter(log_formatter)
logger.setLevel(logging.DEBUG)
logger.addHandler(stream_handler)
class Settings:
core: typing.Dict[str, typing.Any]
query: fetcher.QueryOptions
storage: typing.Dict[str, typing.Any]
notify: typing.Dict[str, typing.Any]
def __init__(self, path):
settings = toml.load(path)
self.core = settings["core"]
self.query = settings["query"]
self.storage = settings["storage"]
self.notify = settings["notify"]
def scan(settings: Settings):
logger.warning("Scan houses from web site")
storage: fetcher.HouseStore = store.LiteHouseStore(
settings.storage["sqlite"]["path"]
)
notifier: fetcher.Notifier
if settings.core["should_notify"]:
notifier = notify.IftttNotifier(
logger,
fetcher.get_591_url,
settings.notify["ifttt"]["webhook_token"],
settings.notify["ifttt"]["event_name"],
)
else:
notifier = notify.EmptyNotifier()
agent = fetcher.Fetcher(logger, settings.core["page_delay"])
scanner = fetcher.Scanner(
logger, storage, notifier, agent, settings.core["batch_delay"]
)
scanner.scan_house(settings.query)
def clean(settings: Settings):
storage: fetcher.HouseStore = store.LiteHouseStore(
settings.storage["sqlite"]["path"]
)
logger.warning("Clean all records in storage")
for key in storage:
del storage[key]
methods: typing.Dict[str, typing.Callable[[Settings], None]] = {
"scan": scan,
"clean": clean,
}
def main() -> None:
parser = argparse.ArgumentParser(description="591 crawling tool")
parser.add_argument(
"-c", dest="config", required=True, help="The configuration file"
)
parser.add_argument("action", help="The action to perform: scan, clean")
args = vars(parser.parse_args())
settings = Settings(args["config"])
action = args["action"]
handler = methods.get(action)
if not handler:
raise RuntimeError("no handler function")
handler(settings)
if __name__ == "__main__":
main()