2025年8月27日星期三

PGFans任务脚本

1.购买服务器阿里云:服务器购买地址https://t.aliyun.com/U/W6Zbs4若失效,可用地址

1.购买服务器

阿里云:

服务器购买地址

https://t.aliyun.com/U/W6Zbs4

若失效,可用地址

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=201905

2.部署教程

2024年最新青龙面板跑脚本教程(一)持续更新中

3.代码如下

#!/usr/bin/env python3# -*- coding: utf-8 -*-"""PGFans 论坛自动签到脚本适用于青龙面板定时任务"""
import hashlibimport jsonimport osimport randomimport requestsimport timefrom datetime import datetime

def bj_time():    """获取北京时间"""    import pytz    return datetime.now(pytz.timezone('Asia/Shanghai'))

def fmt_now():    """格式化当前时间"""    return bj_time().strftime("%Y-%m-%d %H:%M:%S")

def push_plus(token, title, content):    """PushPlus 消息推送"""    if not token:        return
    try:        url = "http://www.pushplus.plus/send"        data = {            "token": token,            "title": title,            "content": content,            "template""html"        }        response = requests.post(url, json=data, timeout=10)        if response.status_code == 200:            result = response.json()            if result.get("code") == 200:                print(f"[推送] 消息推送成功")            else:                print(f"[推送] 消息推送失败: {result.get('msg')}")        else:            print(f"[推送] 推送请求失败,状态码: {response.status_code}")    except Exception as e:        print(f"[推送] 推送异常: {str(e)}")

class PGFansClient:    def __init__(self, mobile, password, pushplus_token=None):        """初始化 PGFans 客户端"""        self.mobile = mobile        self.password = password        self.pushplus_token = pushplus_token        self.session = requests.Session()        self.session.headers.update({            'User-Agent''Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:142.0) Gecko/20100101 Firefox/142.0',            'Accept''application/json, text/plain, */*',            'Accept-Language''zh-CN,zh;q=0.8,zh-TW;q=0.7,zh-HK;q=0.5,en-US;q=0.3,en;q=0.2',            'Accept-Encoding''gzip, deflate, br, zstd',            'Content-type''application/x-www-form-urlencoded; charset=UTF-8',            'Origin''https://www.pgfans.cn',            'DNT''1',            'Connection''keep-alive',            'Referer''https://www.pgfans.cn/',            'Sec-Fetch-Dest''empty',            'Sec-Fetch-Mode''cors',            'Sec-Fetch-Site''same-site',            'Priority''u=0'        })        self.user_id = None        self.sessionid = None
    def log(self, message):        """记录日志"""        print(f"[{fmt_now()}{message}")
    def generate_signature(self, timestamp, action="login", **kwargs):        """生成签名"""        # 根据 JavaScript 代码分析得出的签名算法        # Md5.hashStr('CYYbQyB7FdIS8xuBEwVwbBDMQKOZPMXK|' + timestamp + '|' + action)        secret_key = "CYYbQyB7FdIS8xuBEwVwbBDMQKOZPMXK"
        # 时间戳需要是10位秒级时间戳        if len(timestamp) > 10:            timestamp = timestamp[:10]
        # 构造签名字符串        sign_string = f"{secret_key}|{timestamp}|{action}"
        # 生成 MD5 签名        return hashlib.md5(sign_string.encode()).hexdigest()
    def login(self):        """登录 PGFans 论坛"""        try:            self.log("开始登录 PGFans 论坛...")
            # 生成时间戳(10位秒级)            timestamp = str(int(time.time()))
            # 生成签名            signature = self.generate_signature(timestamp, "login")
            # 登录请求数据            login_data = {                "timestamp": timestamp,                "signature": signature,                "mobile": self.mobile,                "user_pass": self.password            }
            # 发送登录请求            login_url = "https://admin.pgfans.cn/user/User/login"            response = self.session.post(login_url, json=login_data)
            if response.status_code == 200:                result = response.json()                if result.get("code") == "200":                    data = result.get("data", {})                    self.user_id = data.get("id")                    self.sessionid = data.get("sessionid")
                    self.log(f"登录成功,用户ID: {self.user_id}")
                    # 执行登录验证                    return self.check_login()                else:                    error_msg = result.get("message""登录失败")                    self.log(f"登录失败: {error_msg}")                    return False            else:                self.log(f"登录请求失败,状态码: {response.status_code}")                return False
        except Exception as e:            self.log(f"登录异常: {str(e)}")            return False
    def check_login(self):        """验证登录状态"""        try:            if not self.user_id or not self.sessionid:                self.log("缺少用户ID或会话ID,无法验证登录")                return False
            # 生成时间戳和签名            timestamp = str(int(time.time()))            signature = self.generate_signature(timestamp, "checklogin")
            # 验证请求数据            check_data = {                "timestamp": timestamp,                "signature": signature,                "user_id": self.user_id,                "sessionid": self.sessionid            }
            # 发送验证请求            check_url = "https://admin.pgfans.cn/user/user/checkLogin"            response = self.session.post(check_url, json=check_data)
            if response.status_code == 200:                result = response.json()                if result.get("code") == "200":                    data = result.get("data", {})                    login_status = data.get("login_status")                    if login_status == 1:                        self.log("登录验证成功")                        return True                    else:                        self.log("登录验证失败,状态异常")                        return False                else:                    error_msg = result.get("message""验证失败")                    self.log(f"登录验证失败: {error_msg}")                    return False            else:                self.log(f"验证请求失败,状态码: {response.status_code}")                return False
        except Exception as e:            self.log(f"登录验证异常: {str(e)}")            return False
    def get_user_info(self):        """获取用户信息,包括P豆数量"""        try:            if not self.user_id or not self.sessionid:                self.log("缺少用户ID或会话ID,无法获取用户信息")                return None
            # 生成时间戳和签名            timestamp = str(int(time.time()))            signature = self.generate_signature(timestamp, "getnewinfo")
            # 用户信息请求数据            info_data = {                "timestamp": timestamp,                "signature": signature,                "user_id": self.user_id,                "sessionid": self.sessionid            }
            # 发送用户信息请求            info_url = "https://admin.pgfans.cn/user/user/getNewInfo"            response = self.session.post(info_url, json=info_data)
            if response.status_code == 200:                result = response.json()                if result.get("code") == "200":                    data = result.get("data", {})                    pgdou = data.get("pgdou"0)                    self.log(f"当前P豆数量: {pgdou}")                    return pgdou                else:                    error_msg = result.get("message""获取用户信息失败")                    self.log(f"获取用户信息失败: {error_msg}")                    return None            else:                self.log(f"用户信息请求失败,状态码: {response.status_code}")                return None
        except Exception as e:            self.log(f"获取用户信息异常: {str(e)}")            return None
    def checkin(self):        """执行签到"""        try:            # 先确保已登录            if not self.login():                raise RuntimeError("登录失败")
            self.log("开始执行签到...")
            # 生成时间戳和签名            timestamp = str(int(time.time()))            signature = self.generate_signature(timestamp, "signin")
            # 签到请求数据            checkin_data = {                "timestamp": timestamp,                "signature": signature,                "user_id": self.user_id            }
            # 发送签到请求            checkin_url = "https://admin.pgfans.cn/user/pgdou/signIn"            response = self.session.post(checkin_url, json=checkin_data)
            if response.status_code == 200:                result = response.json()                if result.get("code") == "200":                    data = result.get("data", {})                    earned_pgdou = data.get("pgdou"0)
                    # 获取当前总P豆数量                    total_pgdou = self.get_user_info()
                    details = f"获得 {earned_pgdou} 个 PG豆"                    if total_pgdou is not None:                        details += f",当前总计: {total_pgdou} 个 PG豆"
                    return {                        "message""签到成功",                        "details": details                    }                else:                    error_msg = result.get("message""签到失败")                    # 检查是否已经签到                    if "已签到" in error_msg or "重复" in error_msg:                        # 即使已签到,也获取当前P豆数量                        total_pgdou = self.get_user_info()                        details = "今天已经签到过了"                        if total_pgdou is not None:                            details += f",当前总计: {total_pgdou} 个 PG豆"
                        return {                            "message""今日已签到",                            "details": details                        }                    else:                        return {                            "message""签到失败",                            "details": error_msg                        }            else:                return {                    "message""签到失败",                    "details"f"请求失败,状态码: {response.status_code}"                }
        except Exception as e:            self.log(f"签到失败: {str(e)}")            return {                "message""签到失败",                "details"f"签到异常: {str(e)}"            }
    def run_checkin(self):        """执行签到任务"""        self.log("=== 开始 PGFans 论坛签到任务 ===")
        try:            result = self.checkin()
            today = bj_time().strftime("%Y-%m-%d")            title = f"PGFans 论坛签到结果 - {today}"
            if isinstance(result, dict):                message = result.get("message""未知状态")                details = result.get("details""")
                if "成功" in message:                    content = f" {message}\n\n📝 详情:{details}"                    self.log("签到成功")                elif "已签到" in message:                    content = f" {message}\n\n📝 详情:{details}"                    self.log("今日已签到")                else:                    content = f" {message}\n\n📝 详情:{details}"                    self.log("签到失败")            else:                content = f" 签到成功:{result}"                self.log("签到成功")
            # 推送结果            if self.pushplus_token:                push_plus(self.pushplus_token, title, content)
            return result
        except Exception as e:            error_msg = f"签到任务执行失败: {str(e)}"            self.log(error_msg)
            today = bj_time().strftime("%Y-%m-%d")            title = f"PGFans 论坛签到结果 - {today}"            content = f" 签到失败\n\n 错误信息:{error_msg}"
            if self.pushplus_token:                push_plus(self.pushplus_token, title, content)
            return {                "message""签到失败",                "details": error_msg            }

def random_delay():    """随机延迟"""    delay = random.randint(130)    print(f"[延迟] 随机等待 {delay} 秒...")    time.sleep(delay)

def main():    """主函数"""    try:        # random_delay()
        pgfans_users = os.environ.get("PGFANS_USER""").split("#")        pgfans_pwds = os.environ.get("PGFANS_PWD""").split("#")        pushplus_token = os.environ.get("PUSH_PLUS_TOKEN")
        if not pgfans_users or not pgfans_users[0]:            print(" 错误:未配置 PGFANS_USER 环境变量")            return
        if not pgfans_pwds or not pgfans_pwds[0]:            print(" 错误:未配置 PGFANS_PWD 环境变量")            return
        # 确保用户名和密码数量匹配        if len(pgfans_users) != len(pgfans_pwds):            print(" 错误:用户名和密码数量不匹配")            return
        # 处理多账号        for i, (user, pwd) in enumerate(zip(pgfans_users, pgfans_pwds)):            if not user or not pwd:                print(f" 跳过第 {i+1} 个账号:用户名或密码为空")                continue
            print(f"\n{'='*50}")            print(f"开始处理第 {i+1} 个账号: {user}")            print(f"{'='*50}")
            try:                client = PGFansClient(user, pwd, pushplus_token)                result = client.run_checkin()                print(f"第 {i+1} 个账号签到完成: {result}")            except Exception as e:                print(f" 第 {i+1} 个账号处理失败: {str(e)}")
            # 多账号间隔            if i < len(pgfans_users) - 1:                time.sleep(random.randint(38))
        print("\n 所有账号处理完成!")
    except Exception as e:        print(f" 程序执行失败: {str(e)}")

if __name__ == "__main__":    main()

解析

该脚本为PGFans 论坛自动签到脚本,脚本主要作用:

  • 自动登录 PGFans 论坛(按站点 JS 规则生成签名并调用登录与校验接口)。

  • 自动每日签到(调用 signIn 接口;已签到会做友好提示)。

  • 查询账户信息(获取当前 PG豆 数量,组合到结果里)。

  • 结果推送(可选) 通过 PushPlus 把签到结果通知到微信服务号。

  • 多账号支持PGFANS_USERPGFANS_PWD 用 # 分隔),适配 青龙面板 定时任务。

主要方法/函数作用

工具函数

  • bj_time():返回东八区(北京时间)的 datetime

  • fmt_now():格式化当前北京时间为字符串(日志用)。

  • push_plus(token, title, content):向 PushPlus 接口发送通知(未配置 token 则直接跳过)。

类:PGFansClient

  • __init__(mobile, password, pushplus_token=None):初始化会话、通用请求头与推送配置;保存账号标识。

  • log(message):简单带时间的日志输出。

  • generate_signature(timestamp, action="login", **kwargs):生成接口签名
    规则:MD5("CYYbQyB7FdIS8xuBEwVwbBDMQKOZPMXK|{timestamp}|{action}")(秒级 10 位时间戳)。

  • login():按登录接口要求组装 timestampsignature、手机号与密码,调用登录;成功后保存 user_id 与 sessionid,并调用 check_login() 进一步校验。

  • check_login():携带 user_idsessionid 和签名调用校验接口,确认会话有效。

  • get_user_info():携带 user_idsessionid 和签名调用用户信息接口,获取当前 PG豆 数量。

  • checkin()

    1. 确保已登录;

    2. 生成 signin 动作的签名并调用签到接口;

    3. 若成功,读取当天获得 PG豆并查询总 PG豆;

    4. 若返回"已签到/重复",同样查询并返回当前总 PG豆;

    5. 统一返回结构化结果(messagedetails)。

  • run_checkin():一次完整任务编排:执行 checkin() → 组织标题/内容 → 若配置了 PushPlus 则推送 → 返回结果。

运行辅助

  • random_delay():执行前 1–30 秒的随机等待(默认注释掉)。

  • main():读取环境变量 PGFANS_USERPGFANS_PWDPUSH_PLUS_TOKEN;对多账号逐一创建 PGFansClient 并调用 run_checkin(),账号间随机短暂停顿;最终输出整体执行结果。


注意

本文部分变量已做脱敏处理,仅用于测试和学习研究,禁止用于商业用途,不能保证其合法性,准确性,完整性和有效性,请根据情况自行判断。技术层面需要提供帮助,可以通过打赏的方式进行探讨。


历史脚本txt文件获取>>
服务器搭建,人工服务咨询>>


没有评论:

发表评论

YouTube爆火 3D纪录片完整教程(含详细提示词)新手也能一步步复制百万播放

要是你最近常刷YouTube,过去几周肯定看到过这类3D纪录片视频霸屏。 要是你最近常刷YouTube,过去几周肯定看到过这类3D纪录片视频霸屏。下面这个频道 ,只发了10个视频,频道就开始盈利了,订阅量超过22万,每次上传视频都能轻松获得超百万的播放量。 还有同领域的另一位...