1.购买服务器阿里云：服务器购买地址https://t.aliyun.com/U/W9mv4W若失效，可用地址
阿里云：
服务器购买地址
https://t.aliyun.com/U/W9mv4W若失效，可用地址
https://www.aliyun.com/minisite/goods?source=5176.29345612&userCode=49hts92d腾讯云：
https://curl.qcloud.com/wJpWmSfU若失效，可用地址
https://cloud.tencent.com/act/cps/redirect?redirect=2446&cps_key=ad201ee2ef3b771157f72ee5464b1fea&from=console华为云
https://activity.huaweicloud.com/cps.html?fromacct=64b5cf7cc11b4840bb4ed2ea0b2f4468&utm_source=V1g3MDY4NTY=&utm_medium=cps&utm_campaign=2019052.部署教程
3.代码如下
#!/usr/bin/env python3# -*- coding: utf-8 -*-"""Kingbase 论坛适用于青龙面板定时任务"""import randomimport timeimport jsonimport osimport requestsimport pytzfrom datetime import datetimedef bj_time():"""获取北京时间"""return datetime.now(pytz.timezone('Asia/Shanghai'))def fmt_now():"""格式化当前时间"""return bj_time().strftime("%Y-%m-%d %H:%M:%S")class KingbaseClient:def __init__(self, user, pwd, article_id, pushplus_token=None):self.user = userself.pwd = pwdself.article_id = article_idself.token = Noneself.pushplus_token = pushplus_tokenself.user_name = userdef log(self, message, level='INFO'):"""日志输出"""timestamp = fmt_now()print(f"[{timestamp}] [{level}] {message}")def send_notification(self, title, content):"""PushPlus消息推送"""if not self.pushplus_token:self.log(" 未配置PushPlus Token，跳过消息推送")returnattempts = 3pushplus_url = "http://www.pushplus.plus/send"# 在标题和内容中加入用户名称title_with_user = "[{}] {}".format(self.user_name, title)content_with_user = " 账号: {}\n\n{}".format(self.user_name, content)for attempt in range(attempts):try:response = requests.post(pushplus_url,data=json.dumps({"token": self.pushplus_token,"title": title_with_user,"content": content_with_user}).encode('utf-8'),headers={'Content-Type': 'application/json'},timeout=10)response.raise_for_status()self.log(" PushPlus响应: {}".format(response.text))breakexcept requests.exceptions.RequestException as e:self.log(" PushPlus推送失败: {}".format(e), 'ERROR')if attempt < attempts - 1:sleep_time = random.randint(30, 60)self.log("将在 {} 秒后重试...".format(sleep_time))time.sleep(sleep_time)def login(self):"""登录 Kingbase 论坛"""try:self.log("尝试登录 Kingbase...")login_url = "https://bbs.kingbase.com.cn/web-api/web/system/user/loginWeb"login_data = {"username": self.user,"password": self.pwd,"code": None,"loginMethod": "account","phoneNumber": None,"email": None}response = requests.post(login_url, json=login_data)r = response.json()if r.get("code") != 200:# 重试一次response = requests.post(login_url, json=login_data)r = response.json()if r.get("code") != 200:raise RuntimeError(f"登录失败: {r.get('msg')}")self.token = r["data"]self.log("Kingbase 登录成功")return Trueexcept Exception as e:self.log(f"登录异常: {str(e)}", 'ERROR')raisedef reply(self):"""发表回帖"""if not self.token:self.login()self.log("登录后等待3秒...")time.sleep(3)try:view_url = f"https://bbs.kingbase.com.cn/forumDetail?articleId={self.article_id}"view_headers = {"Authorization": f"Bearer {self.token}","Web-Token": self.token,"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:139.0) Gecko/20100101 Firefox/139.0"}view_cookies = {"Authorization": self.token,"Web-Token": self.token}requests.get(view_url, headers=view_headers, cookies=view_cookies)self.log("打开帖子后等待5秒...")time.sleep(5)headers = {"Content-Type": "application/json;charset=utf-8","Authorization": f"Bearer {self.token}","Web-Token": self.token,"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:139.0) Gecko/20100101 Firefox/139.0","Referer": f"https://bbs.kingbase.com.cn/forumDetail?articleId={self.article_id}","Origin": "https://bbs.kingbase.com.cn"}cookies = {"Authorization": self.token,"Web-Token": self.token}body = {"articleId": self.article_id,"commentContent": "<p><img src=\"/UEditorPlus/dialogs/emotion/./custom_emotion/emotion_02.png\"/></p>"}url = "https://bbs.kingbase.com.cn/web-api/web/forum/comment"self.log("发送回帖内容...")response = requests.post(url, headers=headers, cookies=cookies, json=body)r = response.json()if r.get("code") != 200:raise RuntimeError(f"回帖失败: {r.get('msg')}")return r.get("msg", "success")except Exception as e:self.log(f"回帖失败: {str(e)}", 'ERROR')raisedef run_checkin(self, reply_count=5):"""执行签到任务"""self.log("=== 开始 Kingbase 论坛回帖任务 ===")results = []success_count = 0for idx in range(1, reply_count + 1):self.log(f"=== 开始第 {idx}/{reply_count} 次回帖 ===")try:msg = self.reply()log_msg = f"第{idx}/{reply_count}次回帖成功：{msg}"self.log(f"[成功] {log_msg}")results.append(f" {log_msg}")success_count += 1except Exception as e:log_msg = f"第{idx}次回帖失败：{str(e)}"self.log(f"[失败] {log_msg}", 'ERROR')results.append(f" {log_msg}")if idx < reply_count:random_wait = random.randint(10, 60)self.log(f"回帖后随机等待 {random_wait} 秒...")time.sleep(random_wait)today = bj_time().strftime("%Y-%m-%d")title = f"Kingbase 论坛回帖结果 - {today}"content_lines = [f" 回帖统计：成功 {success_count}/{reply_count} 次",""," 详细结果："]content_lines.extend(results)content = "\n".join(content_lines)self.log("=== 任务完成，准备推送结果 ===")self.send_notification(title, content)self.log(f"Kingbase 回帖任务完成，成功 {success_count}/{reply_count} 次")return {"success_count": success_count,"total_count": reply_count,"results": results}def random_delay():delay_minutes = random.randint(0, 60)delay_seconds = delay_minutes * 60if delay_minutes > 0:from datetime import timedeltacurrent_time = bj_time()estimated_start = current_time + timedelta(minutes=delay_minutes)print(f" 随机延迟 {delay_minutes} 分钟后开始执行任务...")print(f" 预计开始时间: {estimated_start.strftime('%H:%M:%S')}")time.sleep(delay_seconds)print(f" 延迟结束，开始执行 Kingbase 回帖任务")else:print(f" 无需延迟，立即开始执行 Kingbase 回帖任务")def main():"""主函数"""try:random_delay()kb_users = os.environ.get("KINGBASE_USER", "").split("#")kb_pwds = os.environ.get("KINGBASE_PWD", "").split("#")pushplus_token = os.environ.get("PUSH_PLUS_TOKEN")article_id = os.environ.get("KINGBASE_ARTICLE_ID")reply_count = int(os.environ.get("KINGBASE_REPLY_CNT", "5"))if not article_id:print(" 错误：未配置 KINGBASE_ARTICLE_ID 环境变量")returnif not kb_users or not kb_users[0]:print(" 错误：未配置 KINGBASE_USER 环境变量")returnif not kb_pwds or not kb_pwds[0]:print(" 错误：未配置 KINGBASE_PWD 环境变量")return# 处理多账号情况for user, pwd in zip(kb_users, kb_pwds):if not user or not pwd:continueprint(f"\n{'='*50}")print(f"开始处理账号: {user}")print(f"{'='*50}")client = KingbaseClient(user, pwd, article_id, pushplus_token)result = client.run_checkin(reply_count)print(f"\n账号 {user} 处理完成")print(f"成功回帖: {result['success_count']}/{result['total_count']} 次")except Exception as e:print(f" 程序执行异常: {str(e)}")# 如果有推送token，发送错误通知if 'pushplus_token' in locals() and pushplus_token:try:error_title = "Kingbase 回帖任务异常"error_content = f" 程序执行异常: {str(e)}"requests.post("http://www.pushplus.plus/send",data=json.dumps({"token": pushplus_token,"title": error_title,"content": error_content}).encode('utf-8'),headers={'Content-Type': 'application/json'},timeout=10)except:passif __name__ == "__main__":main()
解析
该脚本为 Kingbase 论坛自动回帖脚本 ，脚本主要作用：
自动登录 Kingbase 论坛并对指定帖子循环回帖（支持多账号、可配置回帖次数与随机间隔）。
可选 PushPlus 推送：把每个账号的回帖统计与明细推送到微信服务号。
适配 青龙面板 定时任务，读取环境变量批量执行。
关键配置
KINGBASE_USER、KINGBASE_PWD：账号与密码，多账号用#分隔。KINGBASE_ARTICLE_ID：要回帖的目标帖子 ID。KINGBASE_REPLY_CNT：每个账号回帖次数（默认 5）。PUSH_PLUS_TOKEN：PushPlus 推送 Token（可选）。
主要方法（KingbaseClient）
__init__(user, pwd, article_id, pushplus_token=None)
初始化账号信息、目标帖子 ID、会话状态和可选的 PushPlus Token。log(message, level='INFO')
简单日志打印（含北京时间）。send_notification(title, content)
使用 PushPlus 推送当前账号的回帖结果（带重试机制）。login()
调用https://bbs.kingbase.com.cn/web-api/web/system/user/loginWeb
进行账号密码登录；成功后保存 token。
reply()
依赖已登录的token：先访问帖子详情页（模拟打开页面）；
构造请求头与 Cookie；
调用评论接口
.../web/forum/comment发送固定表情内容完成回帖；返回接口提示信息，异常时抛错。
run_checkin(reply_count=5)
完整回帖任务调度：循环调用reply()共reply_count次；每次间随机等待 10–60 秒；统计成功次数与明细，并（如配置）推送结果；返回统计数据。
其他函数
bj_time()/fmt_now()：北京时间与时间格式化。random_delay()：任务启动前随机延迟 0–60 分钟（用于错峰执行）。main()：读取环境变量、支持多账号依次执行run_checkin()，并输出汇总结果；异常时（若配置了 PushPlus）推送错误通知。
提示：脚本会对同一帖子重复发表固定"表情"评论；如需自定义评论内容或频率，可在
reply()的body["commentContent"]和run_checkin()的次数/间隔处调整。
注意：
本文部分变量已做脱敏处理，仅用于测试和学习研究，禁止用于商业用途，不能保证其合法性，准确性，完整性和有效性，请根据情况自行判断。技术层面需要提供帮助，可以通过打赏的方式进行探讨。
没有评论:
发表评论