-
-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
87 lines (70 loc) · 3.12 KB
/
main.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
import logging, os, platform, time
from selenium import webdriver
from selenium.webdriver.chrome.service import Service as ChromiumService
from selenium.webdriver.chrome.service import Service as ChromeService
import modules.asken as asken
from prometheus_client import CollectorRegistry, Gauge, Info, start_http_server
log_format = '%(asctime)s[%(filename)s:%(lineno)d][%(levelname)s] %(message)s'
log_level = os.getenv("LOGLEVEL", logging.INFO)
logging.basicConfig(format=log_format, datefmt='%Y-%m-%d %H:%M:%S%z', level=log_level)
if __name__ == '__main__':
logging.info("initializing exporter...")
registry = CollectorRegistry()
start_http_server(int(os.environ.get('PORT', 8000)), registry=registry)
logging.info("initializing chromium options...")
options = webdriver.ChromeOptions()
options.add_argument('--disable-dev-shm-usage')
if platform.system() == 'Linux':
logging.info("initializing chromium...")
driver = webdriver.Chrome(service=ChromiumService(), options=options)
else:
logging.info("initializing chrome...")
driver = webdriver.Chrome(service=ChromeService(), options=options)
driver.implicitly_wait(10)
today = time.strftime("%Y-%m-%d")
# login
logging.info("login to asken...")
username = os.environ['ASKEN_USERNAME']
password = os.environ['ASKEN_PASSWORD']
ak_driver = asken.login(driver, username, password)
metrics = {
"daily": {},
"weekly": {},
"monthly": {},
}
advice_targets = [
{ 'name': 'daily', 'path': '/0', 'desc': '1日分' },
{ 'name': 'weekly', 'path': '/1', 'desc': '過去7日の平均' },
{ 'name': 'monthly', 'path': '/2', 'desc': '月平均' },
]
# premium?
logging.info("checking asken premium...")
premium = asken.is_premium(ak_driver)
# create metrics
logging.info("create metrics instances...")
score_gauge = Gauge("asken_score", "あすけん > 健康度", registry=registry)
for target in advice_targets:
metrics[target['name']]['advice'] = Info(
"asken_{n}_advice".format(n=target['name']),
"あすけん > アドバイス > {d}".format(d=target['desc']),
registry=registry
)
metrics[target['name']]['detail_advice'] = Info(
"asken_{n}_detail_advice".format(n=target['name']),
"あすけん > 栄養価アドバイス > {d}".format(d=target['desc']),
registry=registry
)
while True:
# get daily score
logging.info("getting daily score...")
daily_score = asken.get_latest_daily_score(ak_driver)
score_gauge.set(daily_score)
# get advice
logging.info("getting advices...")
for target in advice_targets:
advice = asken.get_advice(ak_driver, target['path'], premium)
metrics[target['name']]['advice'].info( { "advice": advice } )
detail_advice = asken.get_detail_advice(ak_driver)
metrics[target['name']]['detail_advice'].info( { "advice": detail_advice } )
logging.info("scraping account is successfully.")
time.sleep(3600*4)