-
Notifications
You must be signed in to change notification settings - Fork 46
/
上海宝山.py
273 lines (244 loc) · 10.9 KB
/
上海宝山.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
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
"""
上海宝山
v2 修复评论
抓任意包请求头 token
变量名: SHBS_TOKEN
cron: 35 13 * * *
const $ = new Env("上海宝山");
"""
import os
import random
import re
import time
import requests
from bs4 import BeautifulSoup
from urllib3.exceptions import InsecureRequestWarning, InsecurePlatformWarning
from common import qianwen_messages, make_request, save_result_to_file
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
requests.packages.urllib3.disable_warnings(InsecurePlatformWarning)
class SHBS():
name = "上海宝山"
def __init__(self, account_info):
self.token = account_info.split('#')[0]
self.isComment = account_info.split('#')[1]
self.verify = False
self.headers = {
'Host': 'bsapi.shmedia.tech',
'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate, br',
'Accept-Language': 'zh-Hans-CN;q=1',
'token': self.token,
'Content-Type': 'application/json; charset=utf-8',
'deviceId': 'af223dabdc3b484c8eae7809f6da7ba6',
'User-Agent': 'StandardApplication/6.2.7 (iPhone; iOS 16.6; Scale/3.00)',
'Connection': 'keep-alive'
}
def login_score(self):
json_data = {}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/points/login/add'
response = make_request(url, json_data=json_data, method='post', headers=self.headers)
# print(response)
if response and response['code'] == 0:
print("登录任务执行成功")
def sign(self):
json_data = {}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/personal/score/sign'
response = make_request(url, json_data, 'post', self.headers)
# print(response)
if response and response['code'] == 0:
save_result_to_file("success", self.name)
print(f'✅{response["data"]["title"]}')
else:
save_result_to_file("error", self.name)
print(f'❌{response["msg"]}')
def total_score(self):
json_data = {}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/personal/score/total'
response = make_request(url, json_data, 'post', self.headers)
if response and response['code'] == 0:
print(f'✅当前总积分：{response["data"]["score"]}')
else:
print(f'❌总积分获取失败：{response}')
def today_score(self):
json_data = {}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/personal/score/info'
response = make_request(url, json_data, 'post', self.headers)
if response and response['code'] == 0:
print(f'✅今日新增积分：{response["data"]["todayIncreasePoint"]}')
# return response["data"]["jobs"]
else:
print(f'❌今日积分获取失败：{response}')
def task_list(self):
json_data = {}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/personal/score/info'
response = make_request(url, json_data, 'post', self.headers)
if response and response['code'] == 0:
print("-----------------------")
print(f'🐹🐹🐹任务列表🐹🐹🐹')
print("-----------------------")
for i in response['data']['jobs']:
if "完善个人资料" in i["title"] or "填写邀请码" in i["title"]:
continue
print(f'👻{i["title"]}: {"已完成" if i["status"] == "1" else "未完成"}')
def article_list(self):
json_data = {
'orderBy': 'release_desc',
'channel': {
'id': 'f4b2dd979fe2497281660f0143df8ec2',
},
'pageSize': 50,
'pageNo': 1,
}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/news/content/list'
response = make_request(url, json_data, 'post', self.headers)
return response["data"]["records"]
def article_read_points_add(self):
json_data = {}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/points/read/add'
make_request(url, json_data, 'post', self.headers)
def article_count_usage_desc(self, id):
json_data = {
'id': id,
'countType': 'contentRead',
}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/common/count/usage/inc'
make_request(url, json_data, 'post', self.headers)
def article_content(self, id):
json_data = {'id': id}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/news/content/get'
response = make_request(url, json_data, 'post', self.headers)
# print(f'article_content，{response}')
return response
def article_read(self, id):
response = self.article_content(id)
if response and response['code'] == 0:
self.article_read_points_add()
self.article_count_usage_desc(id)
print(f'✅文章阅读成功')
else:
print(f'❌阅读失败，{response}')
def article_favor(self, id):
json_data = {'id': id}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/news/content/favor'
response = make_request(url, json_data, 'post', self.headers)
if response and response['code'] == 0:
print(f'✅文章收藏成功')
else:
print(f'❌收藏失败，{response}')
def article_favor_task(self, id):
response_content = self.article_content(id)
if response_content and response_content['code'] == 0:
if response_content['data']['count']["favorite"] is False:
self.article_favor(id)
elif response_content['data']['count']["favorite"]:
print(f'已经收藏过了，不再重复收藏')
else:
print(f'❌收藏失败，{response_content}')
else:
print(f'❌获取文章失败，{response_content}')
def article_share(self, id):
json_data = {}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/points/share/add'
response = make_request(url, json_data, 'post', self.headers)
if response and response['code'] == 0:
print(f'✅文章分享成功')
else:
print(f'❌文章分享失败，{response}')
def video_view_task(self):
json_data = {}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/points/video/add'
response = requests.post(url, headers=self.headers, json=json_data, verify=self.verify).json()
if response and response['code'] == 0:
print(f'✅看片儿完成+1')
else:
print(f'❌看片儿失败：{response}')
def get_gpt_comment(self, id):
basic_news_question = '我需要你针对下面的文章，从一个民众的角度进行评论，我希望你的输出只有评论内容，没有别的无关紧要的词语，回复格式是：芝麻开门#你的评论#， 评论语气要尽可能生活化、日常化，字数一定要限制在5-15字之间，下面是我需要你发表评论的文章内容：'
article_concent = ''
response = self.article_content(id)
#print(f'响应内容：【{response.status_code}】')
comment = ''
commentCount = 0
if response and response['code'] == 0:
commentCount = response["data"]["count"]["commentCount"]
if commentCount <= 0:
content = response["data"]["txt"]
#print(f'content响应内容：【{content}】')
soup = BeautifulSoup(content, 'html.parser')
content_text = soup.get_text()
print(f'content响应内容：【{content_text}】')
if content_text:
message = qianwen_messages(basic_news_question, content_text)
comment = message
else:
return "加油"
return comment
def article_comment_add(self, id, content):
json_data = {
'displayResources': [],
'content': content,
'targetType': 'content',
'targetId': id,
}
url = 'https://bsapi.shmedia.tech/media-basic-port/api/app/common/comment/add'
response = requests.post(url, headers=self.headers, json=json_data).json()
if response and response["code"] == 0:
print(f'✅文章评论成功')
else:
print(f'❌文章评论失败，{response}')
def article_comment_task(self, id):
comment = self.get_gpt_comment(id)
if comment == '':
print(f'😢未知错误或者文章可能评论过，算了吧，下一个')
else:
parts = comment.split('#')
if len(parts) > 1:
comment = parts[1].strip()
print(f'🐌预评论内容：【{comment}】, 你没意见我就在20s后评论了哈......')
time.sleep(random.randint(20, 25))
self.article_comment_add(id, comment)
def gift_list(self):
# todo
print('--------------------')
print('🐹🐹🐹可兑换商品列表🐹🐹🐹')
print('--------------------')
print('😂积分太少啦，暂无商品可兑换')
def main(self):
self.sign()
# for j in range(5):
# self.video_view_task()
# time.sleep(random.randint(20, 30))
article_list = self.article_list()
for i in range(5):
article_id = random.choice(article_list)["id"]
print('--------------------------------------------------------------------')
print(f'🐹随机抓取到一篇文章{article_id}，开始做任务......')
self.article_read(article_id)
time.sleep(random.randint(20, 35))
self.article_share(article_id)
time.sleep(random.randint(10, 18))
if self.isComment == '1':
self.article_comment_task(article_id)
time.sleep(random.randint(20, 40))
else:
print("⛔️未开启自动评论, 如要开启，请更改环境变量配置")
time.sleep(random.randint(10, 25))
self.article_favor_task(article_id)
time.sleep(random.randint(10, 20))
self.task_list()
self.total_score()
self.today_score()
self.gift_list()
if __name__ == '__main__':
env_name = 'SHBS_TOKEN'
tokenStr = os.getenv(env_name)
if not tokenStr:
print(f'⛔️未获取到ck变量：请检查变量 {env_name} 是否填写')
exit(0)
tokens = re.split(r'&', tokenStr)
print(f"上海宝山共获取到{len(tokens)}个账号")
for i, account_info in enumerate(tokens, start=1):
print(f"\n======== ▷ 第 {i} 个账号 ◁ ========")
SHBS(account_info).main()
print("\n随机等待30-60s进行下一个账号")
time.sleep(random.randint(30, 60))