This repository has been archived by the owner on Nov 25, 2021. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 3
/
ck_tieba.py
119 lines (106 loc) · 4.12 KB
/
ck_tieba.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
# -*- coding: utf-8 -*-
"""
cron: 12 16 * * *
new Env('百度贴吧');
"""
import hashlib
import json
import re
import requests
from notify_mtr import send
from utils import get_data
class Tieba:
def __init__(self, check_items):
self.check_items = check_items
@staticmethod
def login_info(session):
return session.get(url="https://zhidao.baidu.com/api/loginInfo").json()
def valid(self, session):
try:
content = session.get(url="https://tieba.baidu.com/dc/common/tbs")
except Exception as e:
return False, f"登录验证异常,错误信息: {e}"
data = json.loads(content.text)
if data["is_login"] == 0:
return False, "登录失败,cookie 异常"
tbs = data["tbs"]
user_name = self.login_info(session=session)["userName"]
return tbs, user_name
@staticmethod
def tieba_list_more(session):
content = session.get(url="https://tieba.baidu.com/f/like/mylike?&pn=1",
timeout=(5, 20),
allow_redirects=False)
try:
pn = int(
re.match(
r".*/f/like/mylike\?&pn=(.*?)\">尾页.*",
content.text,
re.S | re.I).group(1)
)
except Exception as e:
pn = 1
next_page = 1
pattern = re.compile(r".*?<a href=\"/f\?kw=.*?title=\"(.*?)\">")
while next_page <= pn:
tbname = pattern.findall(content.text)
for x in tbname:
yield x
next_page += 1
content = session.get(
url=f"https://tieba.baidu.com/f/like/mylike?&pn={next_page}",
timeout=(5, 20),
allow_redirects=False)
def get_tieba_list(self, session):
tieba_list = list(self.tieba_list_more(session=session))
return tieba_list
@staticmethod
def sign(session, tb_name_list, tbs):
success_count, error_count, exist_count, shield_count = 0, 0, 0, 0
for tb_name in tb_name_list:
md5 = hashlib.md5(f"kw={tb_name}tbs={tbs}tiebaclient!!!".encode(
"utf-8")).hexdigest()
data = {"kw": tb_name, "tbs": tbs, "sign": md5}
try:
response = session.post(
url="https://c.tieba.baidu.com/c/c/forum/sign",
data=data).json()
if response["error_code"] == "0":
success_count += 1
elif response["error_code"] == "160002":
exist_count += 1
elif response["error_code"] == "340006":
shield_count += 1
else:
error_count += 1
except Exception as e:
print(f"贴吧 {tb_name} 签到异常,原因{str(e)}")
msg = f"贴吧总数: {len(tb_name_list)}\n签到成功: {success_count}\n已经签到: {exist_count}\n被屏蔽的: {shield_count}\n签到失败: {error_count}"
return msg
def main(self):
msg_all = ""
for check_item in self.check_items:
cookie = {
item.split("=")[0]: item.split("=")[1]
for item in check_item.get("cookie").split("; ")
}
session = requests.session()
requests.utils.add_dict_to_cookiejar(session.cookies, cookie)
session.headers.update({"Referer": "https://www.baidu.com/"})
tbs, user_name = self.valid(session=session)
if tbs:
tb_name_list = self.get_tieba_list(session=session)
msg = self.sign(session=session,
tb_name_list=tb_name_list,
tbs=tbs)
msg = f"帐号信息: {user_name}\n{msg}"
else:
msg = f"帐号信息: {user_name}\n签到状态: Cookie 可能过期"
msg_all += msg + "\n\n"
return msg_all
if __name__ == "__main__":
data = get_data()
_check_items = data.get("TIEBA", [])
res = Tieba(check_items=_check_items).main()
print(res)
send("百度贴吧", res)