Passed
Push — master ( fcab64...ad3b9e )
by Daniil
01:31
created

scheduler.Date.today()   A

Complexity

Conditions 1

Size

Total Lines 6
Code Lines 3

Duplication

Lines 0
Ratio 0 %

Importance

Changes 0
Metric Value
cc 1
eloc 3
nop 1
dl 0
loc 6
rs 10
c 0
b 0
f 0
1
import os
2
import re
3
import time
4
from datetime import datetime
5
from datetime import timedelta
6
from typing import Union
7
8
import schedule
9
import requests
10
from bs4 import BeautifulSoup
11
12
from bot import Bot
13
from logger import Logger
14
from database import Database
15
16
bot = Bot()
17
18
19
class Date:
20
    """
21
    Вспомогательный класс, содержащий строковые представления дат
22
    """
23
24
    @property
25
    def today(self) -> str:
26
        """
27
        Возвращает сегодняшнюю дату в формате ГГГГ-ММ-ДД
28
        """
29
        return datetime.today().strftime("%Y-%m-%d")
30
31
    @property
32
    def day_after_tomorrow(self) -> str:
33
        """
34
        Возвращает послезавтрашнюю дату в формате ГГГГ-ММ-ДД
35
        """
36
        return datetime.strftime(datetime.now() + timedelta(2), "%Y-%m-%d")
37
38
    @property
39
    def tomorrow(self) -> str:
40
        """
41
        Возвращает послезавтрашнюю дату в формате ГГГГ-ММ-ДД,
42
        если сегодня суббота, завтрашнюю в любой другой день
43
        """
44
        if datetime.today().weekday() == 5:
45
            return self.day_after_tomorrow
46
        return datetime.strftime(datetime.now() + timedelta(1), "%Y-%m-%d")
47
48
49
class Schedule:
50
51
    """
52
    Класс, переводящий расписание из сырой веб-страницы в читаемую строку
53
    """
54
55
    def __init__(self, date: str, gid: str = "324"):
56
        self.log = Logger()
57
        self.date = date
58
        self.gid = gid
59
60
    def is_exist(self) -> bool:
61
        """
62
        Проверяет наличие расписания, основываясь на присутствии плашек
63
        "Расписание отстутствует" и "Расписание составлено, но не опубликовано"
64
        """
65
        soup = self.get_raw()
66
        warn = soup.find_all("div", {"class": "msg warning"})
67
        err = soup.find_all("div", {"class": "msg error"})
68
        if warn or err:
69
            self.log.log.info("Расписание отсутствует.")
70
            return False
71
        return True
72
73
    def get_raw(self) -> BeautifulSoup:
74
        """
75
        Получает с сервера веб-страницу с расписанием и возвращает объект
76
        веб-скрапера
77
        """
78
        request = requests.get(
79
            f"http://rating.ivpek.ru/timetable/timetable/show?gid={self.gid}&date"
80
            f"={self.date}"
81
        )
82
        try:
83
            if request.status_code != 200:
84
                self.log.log.error(
85
                    "Подключение неудачно. Автоматический повтор через 5 секунд."
86
                )
87
                raise requests.exceptions.ConnectionError
88
        except requests.exceptions.ConnectionError as e:
89
            self.log.log.error(msg=e)
90
            self.get_raw()
91
        else:
92
            soup = BeautifulSoup(request.text, "lxml")
93
            return soup
94
95
    def parse(self) -> Union[str, bool]:
96
        """
97
        Форматирует расписание в читаемую строку
98
        """
99
        soup = self.get_raw()
100
        for span in soup.find_all("span", {"class": "ldur"}):
101
            span.decompose()
102
        for br in soup.find_all("br"):
103
            soup.br.insert_before(" ")
104
            br.decompose()
105
        schedule_html = soup.find_all("table", {"class": "tbl"})[1]
106
        sch = []
107
        rows = schedule_html.find_all("tr")
108
        for row in rows:
109
            cols = row.find_all("td")
110
            cols = [el.text.strip() for el in cols]
111
            sch.append([el for el in cols if el])
112
        sch = [el for el in sch if len(el) > 1]
113
114
        msg = ""
115
        replacements = {
116
            "Лекция": "(Л)",
117
            "Лабораторная работа": "(Л/Р)",
118
            "Зачет": "(З)",
119
            "Диф. зачет": "(ДифЗ)",
120
            "Диф.зачет": "(ДифЗ)",
121
            "Экзамен": "(Э)",
122
            "(английский язык)": "",
123
        }
124
        for i in range(len(sch)):
125
            for j in range(len(sch[i])):
126
                item = sch[i][j]
127
                for k, v in replacements.items():
128
                    item = re.sub(k, v, item)
129
                if re.findall("Иностранный язык", item):
130
                    lesson = re.compile(r"(?<=\()\D+(?=\))").findall(item)
131
                    item = (
132
                        f"Английский язык ({lesson[1]}) "
133
                        f"Кузнецова И.Н/Коротина М.А. 12/13а"
134
                    )
135
                    sch[i][j + 1] = ""
136
                msg += f"{item} "
137
            msg += "\n"
138
        if not msg:
139
            self.log.log.info("Расписание отсутствует.")
140
            return False
141
        date = datetime.strptime(self.date, "%Y-%m-%d").strftime("%d.%m.%Y")
142
        msg = f"Расписание на {date}:\n{msg}"
143
        return msg
144
145
    def send(self):
146
        """
147
        Отправляет расписание в активную беседу
148
        и в ЛС подписчикам рассылки "Расписание"
149
        """
150
        sch = self.parse()
151
        if sch:
152
            db = Database(os.environ["DATABASE_URL"])
153
            subscribers = db.fetch_subcribers("schedule")
154
            bot.send_message(msg=sch, pid=bot.cid)
155
            bot.send_mailing(ids=subscribers, msg=sch)
156
157
158
def listen():
159
    """
160
    Слушает сервер на предмет наличия расписания.
161
    Если находит - отправляет, иначе ждет 15 минут
162
    """
163
    d = Date()
164
    sch = Schedule(d.tomorrow)
165
    if sch.is_exist():
166
        if sch.parse():
167
            sch.send()
168
        else:
169
            self.log.log.info("Расписание отстутствует")
0 ignored issues
show
Comprehensibility Best Practice introduced by
The variable self does not seem to be defined.
Loading history...
170
            time.sleep(15 * 60)
171
    else:
172
        self.log.log.info("Расписание отстутствует")
173
        time.sleep(15 * 60)
174
175
176
if __name__ == "__main__":
177
    schedule.every().day.at("09:20").do(listen)
178
    while True:
179
        schedule.run_pending()
180