Azus

Never be a sad song with nothing to say.

0%

bupt-gym-reserve

Posted on Edited on

解析Repository bupt-gym-reserve

Python函数注释：

函数注释作用是提高代码可读性，暗示传入参数及返回数据的类型。

语法

  • 参数注释：以冒号（:）标记，建议传入的参数类型
  • 返回值注释： 以 -> 标记，建议函数返回的类型
    1
    def examp(a: expression, b: expression = 5) -> expression:

    访问函数注释

    函数对象有一个名为 annotations 的属性，它是一个映射（dict），用于将每个参数名（key）映射到其相关的注释（value）。
    1
    2
    3
    4
    5
    >>> type(sum.__annotations__)
    <class 'dict'>
    >>>
    >>> sum.__annotations__
    {'c': 'The default value is 5', 'return': <class 'float'>, 'b': <class 'int'>}
    注意： 映射中有一个特殊的 key，叫做“return”，仅当为函数的返回值提供注释时，才会显示该 key。

    之所以选择 “return”，是因为它不会与任何参数名冲突。“return”是 Python 中的一个关键字，任何使用“return”作为参数名的尝试都会引发 SyntaxError。
    动态注释etc.

存根文件和类型检查

src
存根文件（stub file）
类型检查器读取stub file检查代码中数据类型是否正确

super().init()

super().init

感觉super是用于继承父类的方法 ，不用super的话，直接重写父类的方法会覆盖掉父类的方法，所以通过super来重写又能继承父类的方法

代码段

主函数 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

if __name__ == '__main__':
    print('********** {} **********'.format(
        timelib.strftime('%Y-%m-%d %H:%M:%S', timelib.localtime(timelib.time())))
    )
    #生成配置文件
    (json_loader, config) = load_config()

    print('正在摇D100... _(:з」∠)_')
    
    roll_result = roll_the_dice(config.chance)

    print('怎么又要干活o(￣ヘ￣o＃)' if roll_result else '好耶，是摸鱼时间！(๑•̀ㅂ•́)و✧')
    #概率运行
    if not roll_result:
        sys.exit()
    #检查服务器推送
    notifier = SeverChanNotifier(sckey=config.sckey)
    #开始会话（检查登陆等）
    session = GymSession(config=config)
    
    reserver = Reserver(config=config, session=session)
    reserves = list()
    try:
        reserves = reserver.get_reserves(index_cache=session.index_cache)
    except PageFormatException as e:
        sys.stderr.write(f'捕获到错误：{e.msg}\n')
        notifier.send_msg('预约出现错误', f'捕获到错误：{e.msg}')
        sys.exit()

    reservable = [_reserve for _reserve in reserves if _reserve.reservable]
    print(f'当前可预约有{len(reservable)} / {len(reserves)}个')
    if len(reservable) > 0:
        if config.blacklist:
            try:
                blacklist_pattern = re.compile(config.blacklist)
            except re.error:
                raise RegexException('无法识别的正则表达式')
            success_list, fail_list = reserver.reserve_all(reservable, blacklist_pattern)
        else:
            success_list, fail_list = reserver.reserve_all(reservable)
        if len(fail_list) != 0:
            print(f'失败{len(fail_list)}个，正在尝试重新预约')
            new_fail_list = list()
            for _reserve, _ in fail_list:
                new_fail_list.append(_reserve)
            sl, fl = reserver.reserve_all(new_fail_list, blacklist_pattern)
            success_list += sl
            fail_list = fl
        if config.notify_enabled and (len(success_list) < 0 or len(fail_list) > 0):
            title = f'成功预约{len(success_list)}个健身房时段，失败{len(fail_list)}个'
            content = '以下时段预约成功：'
            for suc in success_list:
                content += str(suc) + '   '
            content += '以下时段预约失败：'
            for failure, reason in fail_list:
                content += f'{str(failure)}+  失败原因：{reason}'
            if not notifier.send_msg(title, content):
                sys.stderr.write('推送消息至微信失败')
    else:
        print('无可用时段，退出中...')
    session.save()
    if config.dumpconf and json_loader.load_status:
        config.save()
        pass

会话部分（登陆）session.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
# 发送http请求
import requests

from ..config_loader import GymConfig

#读取加载json文件
import json

import sys

from .magics import req_config

__all__ = (
    'GymSession',
)

class GymSession(requests.Session):
    def __init__(self, config: GymConfig) -> None:
        #继承父类__init__
        super().__init__()
        #传入配置文件到Gymsession
        self.config = config

        # 最近一次登录结果
        self._login_flag = False

        # 尝试登录的次数
        self._login_attemption = 0

        self.cookie_path = config.cookie_path

        # ？？
        self.headers.update(req_config['headers'])

        #加载cookie（若已经登陆） 或重新登陆
        self._recover_or_create_session()

        self.index_cache = None

    def save(self): 
        # 保存cookie
        print('正在保存session...')
        with open(self.cookie_path, 'w+') as foo:
            json.dump(self.cookies.get_dict(), foo)


    def has_login(self, force_request=False) -> bool:
        # 如果登录的flag为假 或者 用户目前没有尝试过登录
        # 则会请求访问index来检查是否登录
        # 否则返回缓存的login_flag
        if (self._login_attemption == 0 and not self._login_flag) or force_request:
            response_text = self.get(req_config['urls']['index']).text
            self._login_flag = '点击下方可用时间段进行预订' in response_text
            if self._login_flag:
                self.index_cache = response_text
        return self._login_flag

    def _recover_or_create_session(self):
        #加载cookie（若已经登陆） 或重新登陆
        try:
            with open(self.cookie_path) as foo:
                # 存在cookie，直接从文件中加载
                cookies = json.load(foo)
                for k in cookies:
                    self.cookies.set(k, cookies[k])
                print('存在session，检测是否已经登录...')

                if self.has_login():
                    print('已经登录！')
                else:
                    print('用户未登录，正在尝试登录...')
                    self.cookies.clear()
                    #调用登陆
                    self.login(self.config['username'], self.config['password'])
        except (json.decoder.JSONDecodeError, FileNotFoundError):
            # 不存在cookie，创建cookie并登录
            print('不存在session, 登录中...')
            self.login(self.config['username'], self.config['password'])

    # 返回登录成功状态
    def login(self, username: str, password: str) -> bool:
        if not username or not password:
            sys.stderr.write('用户名以及密码不能为空\n')
            sys.exit()
        payload = {
            'username': username,
            'password': password,
            # 'ld': ld
            'usertype': 0,
            'action': None
        }
        self.post(req_config['urls']['login'], data=payload)
        
        login_res = self.has_login()
        self._login_attemption += 1
        self._login_flag = login_res
        print('登录成功！'if login_res else '登录失败！请检查用户名/密码是否正确')
        return login_res

Crontab issue

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
********** 2021-09-21 22:00:01 **********
正在摇D100... _(:з」∠)_
怎么又要干活o(￣ヘ￣o＃)
存在session，检测是否已经登录...
已经登录！
当前可预约有2 / 15个
正在预约第1个...
时间：2021/9/26
时段：18:40 - 19:40
Traceback (most recent call last):
  File "/Users/azus/Documents/Code/Py/bupt-gym-reserve/main.py", line 71, in <module>
    success_list, fail_list = reserver.reserve_all(reservable)
  File "/Users/azus/Documents/Code/Py/bupt-gym-reserve/bupt_gym_reserve/http/reserve.py", line 94, in reserve_all
    suc = self.reserve(r)
  File "/Users/azus/Documents/Code/Py/bupt-gym-reserve/bupt_gym_reserve/http/reserve.py", line 75, in reserve
    payload = {'blob': self._get_blob(reservable.year, reservable.mon,
  File "/Users/azus/Documents/Code/Py/bupt-gym-reserve/bupt_gym_reserve/http/reserve.py", line 105, in _get_blob
    raw = execjs.eval('''JSON.stringify(
  File "/opt/homebrew/lib/python3.8/site-packages/execjs/__init__.py", line 51, in eval
    return get().eval(source, cwd)
  File "/opt/homebrew/lib/python3.8/site-packages/execjs/_abstract_runtime.py", line 25, in eval
    return self.compile('', cwd=cwd).eval(source)
  File "/opt/homebrew/lib/python3.8/site-packages/execjs/_abstract_runtime_context.py", line 27, in eval
    return self._eval(source)
  File "/opt/homebrew/lib/python3.8/site-packages/execjs/_external_runtime.py", line 78, in _eval
    return self.exec_(code)
  File "/opt/homebrew/lib/python3.8/site-packages/execjs/_abstract_runtime_context.py", line 18, in exec_
    return self._exec_(source)
  File "/opt/homebrew/lib/python3.8/site-packages/execjs/_external_runtime.py", line 85, in _exec_
    output = self._exec_with_tempfile(source)
  File "/opt/homebrew/lib/python3.8/site-packages/execjs/_external_runtime.py", line 127, in _exec_with_tempfile
    self._fail_on_non_zero_status(ret, stdoutdata, stderrdata)
  File "/opt/homebrew/lib/python3.8/site-packages/execjs/_external_runtime.py", line 134, in _fail_on_non_zero_status
    raise ProcessExitedWithNonZeroStatus(status=status, stdout=stdoutdata, stderr=stderrdata)
execjs._exceptions.ProcessExitedWithNonZeroStatus: (1, '', 'The operation couldn’t be completed. Unable to locate a Java Runtime.\nPlease visit http://www.java.com for information on installing Java.\n\n')