泉站大桶订水任务脚本

1.购买服务器

阿里云：

服务器购买地址

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

若失效，可用地址

https://www.aliyun.com/activity/wuying/dj?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.代码如下

"""小程序搜 泉站大桶订水桶装水同城送水
变量  authcode # authorization #备注     （没有备注 也可以运行）变量名 qztoken项目  泉站订水 
"""import osimport requestsfrom datetime import datetime, timezone, timedeltaimport jsonimport sysimport timeimport random  from io import StringIO
enable_notification =1  # 控制是否启用通知的变量   0 不发送   1 发
# 只有在需要发送通知时才尝试导入notify模块if enable_notification == 1:    try:        from notify import send    except ModuleNotFoundError:        print("警告：未找到notify.py模块。它不是一个依赖项，请勿错误安装。程序将退出。")        sys.exit(1)
#---------解--的简化0.2框架--------# 配置参数base_url = "https://microuser.quanzhan888.com"  # 实际的基础URLuser_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/107.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF WindowsWechat(0x6309080f)XWEB/8519"
def get_beijing_date():  # 获取北京日期的函数    beijing_time = datetime.now(timezone(timedelta(hours=8)))    return beijing_time.date()
def timestamp_to_beijing_time(timestamp):    utc_zone = timezone.utc    beijing_zone = timezone(timedelta(hours=8))    utc_time = datetime.fromtimestamp(timestamp, utc_zone)    beijing_time = utc_time.astimezone(beijing_zone)    return beijing_time.strftime("%Y-%m-%d %H:%M:%S")
def get_env_variable(var_name):    value = os.getenv(var_name)    if value is None:        print(f'环境变量{var_name}未设置，请检查。')        return None        accounts = value.strip().split('\n')  # 使用 \n 分割    num_accounts = len(accounts)    print(f'-----------本次账号运行数量：{num_accounts}-----------')    print(f'泉站大桶订水--QGh3amllamll  ')
    return accounts
#113.28824159502027#23.103660007697727def fz_hs(auth_code, authorization, user_agent, sign): #封装headers    return {        'Host': 'microuser.quanzhan888.com',        'Connection': 'keep-alive',        'Content-Length': '2',        'charset': 'utf-8',        'product': "shop",        'authcode': auth_code,        'authorization': authorization,        'user-agent': user_agent,        'sign': sign,        'Accept-Encoding': 'gzip,compress,br,deflate',        'platform': "wx",        'x-requested-with': 'xmlhttprequest',        'content-type': 'application/x-www-form-urlencoded',    }
def wdqbsj(auth_code, authorization):  # 个人信息/钱包    url = f"{base_url}/user/get-wallet-info"    headers = fz_hs(auth_code, authorization, user_agent, "99914b932bd37a50b983c5e7c90ae93b")    data = json.dumps({})  # 发送空的JSON数据    #print(url)
    try:        response = requests.post(url, headers=headers, data=data)        response.raise_for_status()        response_data = response.json()        #print("解析的JSON数据：", response_data)
        # 判断code并提取所需数据        if response_data.get('code') == 0:            user_id = response_data['data']['wallet_info'].get('user_id')            total_balance = response_data['data']['wallet_info'].get('total_balance')            today_income = response_data['data']['wallet_info'].get('today_income')            #print(f"用户ID: {user_id}, 总余额: {total_balance}, 今日收入: {today_income}")            print(f"🆔: {user_id}, 总💸: {total_balance}, 今日: {today_income}")
            # 判断今日收入是否大于0            if float(today_income) > 0:                print("今日已有收入，不需要签到")                #tj_sign(auth_code, authorization)#测试提交签到            else:                print("今日无收入，需要签到")                tj_sign(auth_code, authorization)               
        else:            print("响应代码不为0，完整响应体：", response_data)
    except ValueError:        print("响应不是有效的JSON格式。")    except requests.exceptions.RequestException as e:        print(f"请求失败: {e}")

def tj_sign(auth_code, authorization):  # 提交签到    url = f"{base_url}/user/do-sign"    headers = fz_hs(auth_code, authorization, user_agent, "99914b932bd37a50b983c5e7c90ae93b")    data = json.dumps({})  # 发送空的JSON数据
    try:        response = requests.post(url, headers=headers, data=data)        response.raise_for_status()        response_data = response.json()        #print("解析的JSON数据：", response_data)
        # 提取所需数据并转换时间戳        if 'data' in response_data and len(response_data['data']) > 0:            for item in response_data['data']:                user_id = item.get('user_id')                sign_date = timestamp_to_beijing_time(item.get('sign_date'))                sign_time = timestamp_to_beijing_time(item.get('sign_time'))                #print(f"用户: {user_id}, 签名日期: {sign_date}, 签到时间: {sign_time}")                print(f" 签名日期: {sign_date}, 签到🎉: {sign_time}")        return response_data    except ValueError:        print("响应不是有效的JSON格式。")        return None    except requests.exceptions.RequestException as e:        print(f"请求失败: {e}")        return None
#------------通知开始-----------
class Tee:    def __init__(self, *files):        self.files = files
    def write(self, obj):        for file in self.files:            file.write(obj)            file.flush()  # 确保及时输出
    def flush(self):        for file in self.files:            file.flush()#------------通知结束-----------








def main():    string_io = StringIO()    original_stdout = sys.stdout
    try:        sys.stdout = Tee(sys.stdout, string_io)
        var_name = 'qztoken'  # 环境变量名        accounts = get_env_variable(var_name)        if not accounts:            return
        print(f'找到 {len(accounts)} 个账号的令牌。')        total_tokens = len(accounts)                for index, account in enumerate(accounts, start=1):            parts = account.split('#')            auth_code, authorization = parts[0], parts[1]            remark = None if len(parts) == 2 else parts[2]  # 检查是否有备注
            remark_info = f" --- 备注: {remark}" if remark else ""            print(f"------账号 {index}/{total_tokens}{remark_info} ------")
            wdqbsj(auth_code, authorization)  # 个人信息/钱包
            # 暂停3到5秒            time.sleep(random.randint(3, 5))
    finally:        sys.stdout = original_stdout        output_content = string_io.getvalue()
        if enable_notification:            send("-泉站大桶订水-通知", output_content)
if __name__ == "__main__":    main()
"""#本地测试用 os.environ['qztoken'] = '''authcode # authorization 
'''
def main():    var_name = 'qztoken'  # 环境变量名    accounts = get_env_variable(var_name)    if not accounts:        return
    print(f'找到 {len(accounts)} 个账号的令牌。')    total_tokens = len(accounts)        for index, account in enumerate(accounts, start=1):        parts = account.split('#')        auth_code, authorization = parts[0], parts[1]        remark = None if len(parts) == 2 else parts[2]  # 检查是否有备注
        remark_info = f" --- 备注: {remark}" if remark else ""        print(f"------账号 {index}/{total_tokens}{remark_info} ------")
        wdqbsj(auth_code, authorization)# 个人信息/钱包

if __name__ == "__main__":    main()"""

解析

这个脚本用来自动处理一个名为泉站大桶订水小程序的日常任务，比如签到和检查用户的钱包信息。脚本通过接口与小程序后端通信，执行相关操作，并且可以处理多个用户账号。

主要函数

1. get_beijing_date:

• 作用：获取当前北京时间的日期。

• 方法：利用datetime模块计算得到北京时间。

2. timestamp_to_beijing_time:

• 作用：将UNIX时间戳转换为北京时间。

• 方法：使用datetime模块处理时间转换，确保时间显示为东八区的时间。

3. get_env_variable:

• 作用：从环境变量中获取指定的变量值，并处理成账号列表。

• 方法：读取环境变量，分割字符串得到账号列表，并打印账号数量。

4. fz_hs:

• 作用：封装请求的头信息。

• 方法：接收授权码、用户代理等信息，并返回构成HTTP请求头的字典。

5. wdqbsj:

• 作用：请求并获取用户的钱包信息。

• 方法：构建请求发送到特定的URL，解析响应数据，打印用户ID、总余额和今日收入。根据今日收入判断是否需要签到。

6. tj_sign:

• 作用：提交签到请求。

• 方法：发送签到请求到服务器，解析响应并打印签到结果。

7. Tee:

• 类作用：创建一个写入器对象，可以同时向多个输出流写入数据。

• 方法：write方法同时向所有文件对象写入数据，flush确保数据及时输出。

8. main:

• 作用：主函数，执行程序的主要逻辑。

• 方法：读取环境变量中的账号信息，对每个账号执行信息查询和签到操作，处理多账号。最后，如果启用了通知，将输出内容通过通知发送。

这个脚本结构清晰，通过读取环境变量中的账号信息，逐个执行签到和查询钱包的操作，适用于自动化处理每日任务。

注意：

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