-
Notifications
You must be signed in to change notification settings - Fork 0
/
homework.py
164 lines (130 loc) · 5.1 KB
/
homework.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
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
import logging
import os
import sys
import time
from dotenv import load_dotenv
import requests
import telegram
load_dotenv()
PRACTICUM_TOKEN = os.getenv('YA_P_TOKEN')
TELEGRAM_TOKEN = os.getenv('TG_TOKEN')
TELEGRAM_CHAT_ID = os.getenv('CHAT_ID')
RETRY_TIME = 600
ERROR_COUNT_LIMIT = 10
ENDPOINT = 'https://practicum.yandex.ru/api/user_api/homework_statuses/'
HEADERS = {'Authorization': f'OAuth {PRACTICUM_TOKEN}'}
HOMEWORK_VERDICTS = {
'approved': 'Работа проверена: ревьюеру всё понравилось. Ура!',
'reviewing': 'Работа взята на проверку ревьюером.',
'rejected': 'Работа проверена: у ревьюера есть замечания.'
}
logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(
logging.Formatter('%(asctime)s %(levelname)s %(message)s')
)
logger.addHandler(handler)
logger.setLevel(logging.INFO)
cache = {}
errors = {}
class SendMessageError(Exception):
"""Custom error for catching."""
pass
def cache_errors(message):
"""
Put errors in dict with counter.
Returns True if error occurs first time or error counter overfilled.
"""
if message not in errors:
errors[message] = 1
logger.info('error was added to cache and sent to telegram')
return True
errors[message] += 1
if errors[message] >= ERROR_COUNT_LIMIT:
del errors[message]
def send_message(bot, message):
"""Send message to bot."""
try:
bot.send_message(TELEGRAM_CHAT_ID, message)
logger.info(f'sent message:\n{message}')
except telegram.error.TelegramError as error:
logger.error(f'error when sending message:\n{error}')
raise SendMessageError('error while sending message')
def get_api_answer(current_timestamp):
"""Get JSON from API and return it as a python dict."""
params = {'from_date': current_timestamp}
try:
response = requests.get(ENDPOINT, params, headers=HEADERS)
except Exception as error:
logger.error(f'error getting ENDPOINT:\n{error}')
raise Exception('get_api func broke')
if response.status_code != 200:
raise Exception(
f'response status code not 200 but {response.status_code}'
)
return response.json()
def check_response(response):
"""If JSON correct returns list of homeworks."""
if not isinstance(response, dict):
logger.error('response is not dict')
raise TypeError('response is not dict')
if 'homeworks' not in response:
logger.error('homework not in response')
raise KeyError('homework not in response')
if not isinstance(response['homeworks'], list):
logger.error('homework is not list')
raise TypeError('homework is not list')
return response.get('homeworks')
def parse_status(homework):
"""
Compare current homework status with cache.
If there is a difference - return message.
"""
for key in ('homework_name', 'status'):
if key not in homework:
raise KeyError('There is no correct keys in homework')
homework_name = homework['homework_name']
homework_status = homework['status']
if homework_status not in HOMEWORK_VERDICTS:
raise Exception('status is different')
verdict = HOMEWORK_VERDICTS.get(homework_status)
if homework_name not in cache:
logger.info('adding new homework to cache')
if cache.get(homework_name) != homework_status:
cache[homework_name] = homework_status
logger.info(f'new status for {homework_name} = {homework_status}')
return f'Изменился статус проверки работы "{homework_name}". {verdict}'
def check_tokens():
"""Check that tokens exists. If not - stop programm."""
return all((PRACTICUM_TOKEN, TELEGRAM_TOKEN, TELEGRAM_CHAT_ID))
def main():
"""Основная логика работы бота."""
if not check_tokens():
logger.critical('TOKEN NOT FOUND. ALARM.')
sys.exit()
bot = telegram.Bot(token=TELEGRAM_TOKEN)
current_timestamp = int(time.time())
while True:
try:
logger.debug(f'new iteration on {current_timestamp}')
response_from_api = get_api_answer(current_timestamp)
current_timestamp = response_from_api.get('current_date')
response = check_response(response_from_api)
for homework in response:
message = parse_status(homework)
if message:
send_message(bot, message)
except SendMessageError as error:
logger.error(f'message was not sent. {error}')
except Exception as error:
message = f'Сбой в работе программы: {error}'
logger.error(f'error in main():\n{current_timestamp}\n{error}')
if cache_errors(message):
# How to catch this one?
# Adding new try/except results in
# main() too complex
send_message(bot, message)
finally:
time.sleep(RETRY_TIME)
if __name__ == '__main__':
main()