This repository was archived by the owner on Mar 9, 2026. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 472
Expand file tree
/
Copy pathmain.py
More file actions
388 lines (342 loc) · 14 KB
/
main.py
File metadata and controls
388 lines (342 loc) · 14 KB
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
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
"""
cron: 0 */6 * * *
new Env("Linux.Do 签到")
"""
import os
import random
import time
import functools
from loguru import logger
from DrissionPage import ChromiumOptions, Chromium
from tabulate import tabulate
from curl_cffi import requests
from bs4 import BeautifulSoup
from notify import NotificationManager
def retry_decorator(retries=3, min_delay=5, max_delay=10):
def decorator(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
for attempt in range(retries):
try:
return func(*args, **kwargs)
except Exception as e:
if attempt == retries - 1: # 最后一次尝试
logger.error(f"函数 {func.__name__} 最终执行失败: {str(e)}")
logger.warning(
f"函数 {func.__name__} 第 {attempt + 1}/{retries} 次尝试失败: {str(e)}"
)
if attempt < retries - 1:
sleep_s = random.uniform(min_delay, max_delay)
logger.info(
f"将在 {sleep_s:.2f}s 后重试 ({min_delay}-{max_delay}s 随机延迟)"
)
time.sleep(sleep_s)
return None
return wrapper
return decorator
os.environ.pop("DISPLAY", None)
os.environ.pop("DYLD_LIBRARY_PATH", None)
USERNAME = os.environ.get("LINUXDO_USERNAME")
PASSWORD = os.environ.get("LINUXDO_PASSWORD")
COOKIES = os.environ.get("LINUXDO_COOKIES", "").strip() # 手动设置的 Cookie 字符串,优先使用
BROWSE_ENABLED = os.environ.get("BROWSE_ENABLED", "true").strip().lower() not in [
"false",
"0",
"off",
]
if not USERNAME:
USERNAME = os.environ.get("USERNAME")
if not PASSWORD:
PASSWORD = os.environ.get("PASSWORD")
HOME_URL = "https://linux.do/"
LOGIN_URL = "https://linux.do/login"
SESSION_URL = "https://linux.do/session"
CSRF_URL = "https://linux.do/session/csrf"
class LinuxDoBrowser:
def __init__(self) -> None:
from sys import platform
if platform == "linux" or platform == "linux2":
platformIdentifier = "X11; Linux x86_64"
elif platform == "darwin":
platformIdentifier = "Macintosh; Intel Mac OS X 10_15_7"
elif platform == "win32":
platformIdentifier = "Windows NT 10.0; Win64; x64"
else:
platformIdentifier = "X11; Linux x86_64"
co = (
ChromiumOptions()
.headless(True)
.incognito(True)
.set_argument("--no-sandbox")
)
co.set_user_agent(
f"Mozilla/5.0 ({platformIdentifier}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36"
)
self.browser = Chromium(co)
self.page = self.browser.new_tab()
self.session = requests.Session()
self.session.headers.update(
{
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36 Edg/142.0.0.0",
"Accept": "application/json, text/javascript, */*; q=0.01",
"Accept-Language": "zh-CN,zh;q=0.9",
}
)
# 初始化通知管理器
self.notifier = NotificationManager()
@staticmethod
def parse_cookie_string(cookie_str: str) -> list[dict]:
"""
解析浏览器复制的 Cookie 字符串格式: "name1=value1; name2=value2"
返回 DrissionPage 所需的 cookie 列表格式。
"""
cookies = []
for part in cookie_str.strip().split(";"):
part = part.strip()
if "=" in part:
name, _, value = part.partition("=")
cookies.append(
{
"name": name.strip(),
"value": value.strip(),
"domain": ".linux.do",
"path": "/",
}
)
return cookies
def login_with_cookies(self, cookie_str: str) -> bool:
"""使用手动设置的 Cookie 直接登录,跳过账号密码流程"""
logger.info("检测到手动 Cookie,尝试 Cookie 登录...")
dp_cookies = self.parse_cookie_string(cookie_str)
if not dp_cookies:
logger.error("Cookie 解析失败或为空,无法使用 Cookie 登录")
return False
logger.info(f"成功解析 {len(dp_cookies)} 个 Cookie 条目")
# 同步到 requests.Session,以便后续 API 请求(如 print_connect_info)使用
for ck in dp_cookies:
self.session.cookies.set(ck["name"], ck["value"], domain="linux.do")
# 同步到 DrissionPage
self.page.set.cookies(dp_cookies)
logger.info("Cookie 设置完成,导航至 linux.do...")
self.page.get(HOME_URL)
time.sleep(5)
# 验证登录状态
try:
user_ele = self.page.ele("@id=current-user")
except Exception as e:
logger.warning(f"Cookie 登录验证异常: {str(e)}")
return True
if not user_ele:
if "avatar" in self.page.html:
logger.info("Cookie 登录验证成功 (通过 avatar)")
return True
logger.error("Cookie 登录验证失败 (未找到 current-user),Cookie 可能已过期")
return False
else:
logger.info("Cookie 登录验证成功")
return True
def login(self):
logger.info("开始账号密码登录")
# Step 1: Get CSRF Token
logger.info("获取 CSRF token...")
headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36 Edg/142.0.0.0",
"Accept": "application/json, text/javascript, */*; q=0.01",
"Accept-Language": "zh-CN,zh;q=0.9",
"X-Requested-With": "XMLHttpRequest",
"Referer": LOGIN_URL,
}
resp_csrf = self.session.get(CSRF_URL, headers=headers, impersonate="firefox135")
if resp_csrf.status_code != 200:
logger.error(f"获取 CSRF token 失败: {resp_csrf.status_code}")
return False
csrf_data = resp_csrf.json()
csrf_token = csrf_data.get("csrf")
logger.info(f"CSRF Token obtained: {csrf_token[:10]}...")
# Step 2: Login
logger.info("正在登录...")
headers.update(
{
"X-CSRF-Token": csrf_token,
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
"Origin": "https://linux.do",
}
)
data = {
"login": USERNAME,
"password": PASSWORD,
"second_factor_method": "1",
"timezone": "Asia/Shanghai",
}
try:
resp_login = self.session.post(
SESSION_URL, data=data, impersonate="chrome136", headers=headers
)
if resp_login.status_code == 200:
response_json = resp_login.json()
if response_json.get("error"):
logger.error(f"登录失败: {response_json.get('error')}")
return False
logger.info("登录成功!")
else:
logger.error(f"登录失败,状态码: {resp_login.status_code}")
logger.error(resp_login.text)
return False
except Exception as e:
logger.error(f"登录请求异常: {e}")
return False
# Step 3: Pass cookies to DrissionPage
logger.info("同步 Cookie 到 DrissionPage...")
cookies_dict = self.session.cookies.get_dict()
dp_cookies = []
for name, value in cookies_dict.items():
dp_cookies.append(
{
"name": name,
"value": value,
"domain": ".linux.do",
"path": "/",
}
)
self.page.set.cookies(dp_cookies)
logger.info("Cookie 设置完成,导航至 linux.do...")
self.page.get(HOME_URL)
time.sleep(5)
try:
user_ele = self.page.ele("@id=current-user")
except Exception as e:
logger.warning(f"登录验证失败: {str(e)}")
return True
if not user_ele:
# Fallback check for avatar
if "avatar" in self.page.html:
logger.info("登录验证成功 (通过 avatar)")
return True
logger.error("登录验证失败 (未找到 current-user)")
return False
else:
logger.info("登录验证成功")
return True
def click_topic(self):
topic_list = self.page.ele("@id=list-area").eles(".:title")
if not topic_list:
logger.error("未找到主题帖")
return False
logger.info(f"发现 {len(topic_list)} 个主题帖,随机选择10个")
for topic in random.sample(topic_list, 10):
self.click_one_topic(topic.attr("href"))
return True
@retry_decorator()
def click_one_topic(self, topic_url):
new_page = self.browser.new_tab()
try:
new_page.get(topic_url)
if random.random() < 0.3: # 0.3 * 30 = 9
self.click_like(new_page)
self.browse_post(new_page)
finally:
try:
new_page.close()
except Exception:
pass
def browse_post(self, page):
prev_url = None
# 开始自动滚动,最多滚动10次
for _ in range(10):
# 随机滚动一段距离
scroll_distance = random.randint(550, 650) # 随机滚动 550-650 像素
logger.info(f"向下滚动 {scroll_distance} 像素...")
page.run_js(f"window.scrollBy(0, {scroll_distance})")
logger.info(f"已加载页面: {page.url}")
if random.random() < 0.03: # 33 * 4 = 132
logger.success("随机退出浏览")
break
# 检查是否到达页面底部
at_bottom = page.run_js(
"window.scrollY + window.innerHeight >= document.body.scrollHeight"
)
current_url = page.url
if current_url != prev_url:
prev_url = current_url
elif at_bottom and prev_url == current_url:
logger.success("已到达页面底部,退出浏览")
break
# 动态随机等待
wait_time = random.uniform(2, 4) # 随机等待 2-4 秒
logger.info(f"等待 {wait_time:.2f} 秒...")
time.sleep(wait_time)
def run(self):
try:
# 优先使用手动 Cookie 登录,没有再使用账号密码
if COOKIES:
login_res = self.login_with_cookies(COOKIES)
if not login_res:
logger.warning("Cookie 登录失败,尝试账号密码登录...")
login_res = self.login()
else:
login_res = self.login()
if not login_res: # 登录
logger.warning("登录验证失败")
if BROWSE_ENABLED:
click_topic_res = self.click_topic() # 点击主题
if not click_topic_res:
logger.error("点击主题失败,程序终止")
return
logger.info("完成浏览任务")
self.print_connect_info() # 打印连接信息
self.send_notifications(BROWSE_ENABLED) # 发送通知
finally:
try:
self.page.close()
except Exception:
pass
try:
self.browser.quit()
except Exception:
pass
def click_like(self, page):
try:
# 专门查找未点赞的按钮
like_button = page.ele(".discourse-reactions-reaction-button")
if like_button:
logger.info("找到未点赞的帖子,准备点赞")
like_button.click()
logger.info("点赞成功")
time.sleep(random.uniform(1, 2))
else:
logger.info("帖子可能已经点过赞了")
except Exception as e:
logger.error(f"点赞失败: {str(e)}")
def print_connect_info(self):
logger.info("获取连接信息")
headers = {
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
}
resp = self.session.get(
"https://connect.linux.do/", headers=headers, impersonate="chrome136"
)
soup = BeautifulSoup(resp.text, "html.parser")
rows = soup.select("table tr")
info = []
for row in rows:
cells = row.select("td")
if len(cells) >= 3:
project = cells[0].text.strip()
current = cells[1].text.strip() if cells[1].text.strip() else "0"
requirement = cells[2].text.strip() if cells[2].text.strip() else "0"
info.append([project, current, requirement])
logger.info("--------------Connect Info-----------------")
logger.info("\n" + tabulate(info, headers=["项目", "当前", "要求"], tablefmt="pretty"))
def send_notifications(self, browse_enabled):
"""发送签到通知"""
status_msg = f"✅每日登录成功: {USERNAME}"
if browse_enabled:
status_msg += " + 浏览任务完成"
# 使用通知管理器发送所有通知
self.notifier.send_all("LINUX DO", status_msg)
if __name__ == "__main__":
if not COOKIES and (not USERNAME or not PASSWORD):
print("请设置 LINUXDO_COOKIES(Cookie 登录),或同时设置 USERNAME 和 PASSWORD(账号密码登录)")
exit(1)
browser = LinuxDoBrowser()
browser.run()