【Python】乐青内穿账号自动签到

Cristy · 发表于 2024-11-11 10:57

本帖最后由 Cristy 于 2025-1-6 00:59 编辑

aliases:
tags:
author: Aura Service
created: 2024-11-11T10:20:43
updated: 2024-11-11T10:54:09
[b]updated: 2024-12-26T16:17:21 已经更新，更新兼容官方的新token获取方式，请从下往上看

免责声明： 本程序仅用于学习目的，严禁用于任何商业或非法用途。

隐私保护提示： 文章中已对敏感信息进行脱敏处理。如果您发现任何潜在的隐私泄露问题，请不吝告知，以便我及时进行修改或删除。感谢您的理解和支持！

乐青内网穿透自动签到脚本介绍

1 概述

本文将介绍一个用于乐青内网穿透服务的自动签到脚本。该脚本使用 Python 编写，能够自动登录用户账号并进行签到，同时支持流量监控和推送通知功能。通过此脚本，用户可以轻松管理自己的乐青账号，避免手动签到的繁琐。

2 功能特点

自动登录：脚本能够自动登录用户的乐青账号。
签到功能：支持自动签到，获取签到奖励流量。
流量监控：可以设置流量阈值，超过阈值后自动跳过签到。
推送通知：通过 PushPlus 发送签到结果通知。

2.1 效果展示

推送信息：

3 使用说明

3.1 环境准备

确保已安装 Python 3.x。
安装 requests 库，可以使用以下命令安装：
```
pip install requests
```
3.1.1 注册pushplus

Push Plus是一个微信通知工具，使用公众号的形式给你推送信息。
请自行前往官网了解相关信息。
官网： https://www.pushplus.plus/

3.2 配置账号信息

在脚本中配置你的账号信息。找到以下代码段并填写你的用户名、密码和流量限制：

accounts = [
    {
        'username': '用户名',
        'password': '密码',
        'traffic_limit': '流量总量限制阈值，填入数字'
    }
    # 添加更多账号...
]

同时，配置 PushPlus 的 token 和 topic：

token = '你的pushplus token'
topic = 'push plus （没有填空）对应组的topic'

3.3 代码结构

脚本的主要功能分为几个部分：

登录功能：通过 login 函数实现用户登录。
签到功能：通过 sign 函数实现签到操作。
流量检查：通过 check_sign 函数检查签到状态。
推送通知：通过 send_pushplus_message 函数发送签到结果通知。

3.4 主程序

主程序在 main 函数中实现，循环遍历所有账号，执行登录、签到和结果处理。最后，调用 send_pushplus_message 函数发送通知。

def main():
    results = []
    for account in accounts:
        # 登录、签到逻辑...
    result_str = '\n'.join(results)
    print(result_str)
    return result_str

3.5 运行脚本

在命令行中运行脚本：

python locyan_auto_sign.py

3.6 部署和使用

青龙部署：
- 环境变量配置（使用环境变量的话）：
- 代码上传和修改

这两处的代码选用前者为读取环境变量中的账号数据，选择下面的则为直接使用脚本内置账号数据

上传记录文件名和路径

配置定时任务，推荐每天执行两次（以防官方更新，服务器断网等情况）

用linux自带的定时任务完成
```
[root@linux ~]# crontab -l
# locyan
0 13,5 * * * python3.9 /home/locyan_auto_sign.py
```
- 如果需要使用环境变量配置，与青龙部署步骤相同，需要修改对应代码并配置环境变量

4 总结

通过使用这个自动签到脚本，用户可以轻松管理乐青内网穿透服务的签到操作，节省时间并确保流量的有效使用。希望本文能帮助你快速上手并充分利用这个工具！

5 更新

5.1 2024-11 旧版本

locyan_auto_sign.py

# -*- coding: utf-8 -*-  

"""  
@File       : locyan_auto_sign.py  
@Author     : AuraService
@Time       : 2023/8/29 17:39 
初版脚本，自动签到
@Update     : 2024/8/03 11:02 
官方出现因流量剩余过大变成负数的情况，添加阈值，流量过大则跳过签到
@Update     : 2024/10/04 09:55  
接口更新了，重新更新代码  
@Update     : 2024/10/27 13:27  
修改了登录接口的请求头和参数类型  
1. 从params=params 改为 data=params2. 去掉了无效参数，规避服务端更新后因body类型不为表单导致的错误
3. 请求头删除了跨域相关内容  
@Description:  

"""  
import requests  

# 配置账号信息  
# accounts = json.loads(os.getenv('LOCYAN_CHECKIN'))  

# 本地测试使用account  
accounts = [  
    {  
        'username': '用户名',  
        'password': '密码',  
        'traffic_limit': '流量总量限制阈值'  
    }
    # 添加更多账号...  
]  

token = '你的pushplus token'  
topic = 'push plus topic'  

# 下面不需要动
base_url = 'https://api-v2.locyanfrp.cn/api/v2'  
login_url = '/auth/login'  
user_info_url = '/user/info'  
check_sign_url = "/sign"  
sign_url = "/sign"  

def login(username, password):  
    url = f'{base_url}{login_url}'  
    headers = {  
        "accept": "application/json, text/plain, */*",  
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8",  
        "content-type": "application/x-www-form-urlencoded",  
        "priority": "u=1, i",  
        "sec-ch-ua": '"Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"',  
        "sec-ch-ua-mobile": "?0",  
        "sec-ch-ua-platform": '"Windows"',  
        "Referer": "https://preview.locyanfrp.cn/",  
        "Referrer-Policy": "strict-origin-when-cross-origin"  
    }  
    params = {  
        'username': username,  
        'password': password,  
    }  
    try:  
        response = requests.post(url, headers=headers, data=params)  
        return response.json()  
    except Exception as e:  
        return {'status': False, 'message': str(e)}  

def sign(token, username):  
    url = f'{base_url}{sign_url}?username={username}'  
    headers = {  
        'accept': 'application/json, text/plain, */*',  
        'accept-language': 'zh-CN,zh;q=0.9',  
        'authorization': f'Bearer {token}',  
        'content-type': 'application/x-www-form-urlencoded',  
        'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',  
        'sec-ch-ua-mobile': '?0',  
        'sec-ch-ua-platform': '"Windows"',  
        'sec-fetch-dest': 'empty',  
        'sec-fetch-mode': 'cors',  
        'sec-fetch-site': 'same-site',  
        'Referer': 'https://preview.locyanfrp.cn/',  
    }  
    data = {  
        'token': token,  
    }  
    try:  
        response = requests.post(url, headers=headers, data=data)  
        return response.json()  
    except Exception as e:  
        return {'status': False, 'message': str(e)}  

def check_sign(token):  
    url = f'{base_url}{check_sign_url}'  
    headers = {  
        'accept': 'application/json, text/plain, */*',  
        'accept-language': 'zh-CN,zh;q=0.9',  
        'authorization': f'Bearer {token}',  
        'content-type': 'application/x-www-form-urlencoded',  
        'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',  
        'sec-ch-ua-mobile': '?0',  
        'sec-ch-ua-platform': '"Windows"',  
        'sec-fetch-dest': 'empty',  
        'sec-fetch-mode': 'cors',  
        'sec-fetch-site': 'same-site',  
        'Referer': 'https://preview.locyanfrp.cn/',  
    }  
    data = {}  
    try:  
        response = requests.get(url, headers=headers, data=data)  
        return response.json()  
    except Exception as e:  
        return {'status': False, 'message': str(e)}  

def send_pushplus_message(token, title, content, topic=''):  
    url = 'http://www.pushplus.plus/send'  
    if topic == '':  
        data = {  
            'token': token,  
            'title': title,  
            'content': content  
        }  
    else:  
        data = {  
            'token': token,  
            'title': title,  
            'content': content,  
            'topic': topic  
        }  
    response = requests.post(url, json=data)  
    if response.status_code == 200:  
        print("消息发送成功")  
    else:  
        print("消息发送失败")  

def main():  
    results = []  

    for account in accounts:  
        username = account['username']  
        password = account['password']  
        traffic_limit = account.get('traffic_limit', 0)  

        # 登录  
        login_res = login(username, password)  
        if not login_res.get('status', 200) == 200:  
            results.append(f"账号 {username} 登录失败: {login_res.get('message', None)}")  
            continue  
        login_response = login_res.get('data', {})  
        token = login_response.get('token', None)  
        email = login_response.get('email', '')  
        avatar = login_response.get('avatar', '')  
        traffic = login_response.get('traffic', {})  
        inbound = login_response.get('inbound', {})  
        outbound = login_response.get('outbound', {})  
        print("---------")  
        print(f"username : {username}")  
        print(f"token : {token}")  
        print(f"email : {email}")  
        print(f"avatar : {avatar}")  
        print(f"traffic : {traffic}")  
        print(f"inbound : {inbound}")  
        print(f"outbound : {outbound}")  

        re = f"————————————————————\n" + \  
             f"★账号: {username}\n" + \  
             f"签到阈值: {traffic_limit} GiB\n" + \  
             f"当前剩余流量: {traffic} GiB\n" + \  
             f"上行速率: {inbound}Mbps\n" + \  
             f"下行速率: {outbound}Mbps\n"  

        if int(traffic) >= int(traffic_limit):  
            re += f'😎 账号 {username} 流量已超过阈值，跳过签到'  
            results.append(re)  
            continue  
        # 签到  
        sign_response = sign(token, username)  
        # 检查签到状态  
        # check_response = check_sign(token)  

        if sign_response.get('status', 0) == 200 or sign_response.get('status', 0) == 403:  
            re += f"😀 签到成功 {sign_response.get('message', None)}"  
            if sign_response.get('data', {}):  
                get_traffic = sign_response.get('data', {}).get('get_traffic', 0)  
                total_get_traffic = sign_response.get('data', {}).get('total_get_traffic', 0)  
                sign_count = sign_response.get('data', {}).get('sign_count', 0)  
                re += f"|签到次数:{sign_count}|本次签到获得流量:{get_traffic}|签到获得总流量:{total_get_traffic}"  
        else:  
            re += f"😢 签到失败 {sign_response.get('message', None)}"  
        results.append(re)  

    result_str = '\n'.join(results)  
    print(result_str)  
    return result_str  

if __name__ == '__main__':  
    send_pushplus_message(token, '乐青映射签到通知', main())

5.2 2024-12-26 (目前最新)

前往官网登录自己的账号
打开左侧OAuth2.0 应用创建一个应用

重定向地址填一个可以访问的如：

 https://www.awwwards.com/404.html

其他随意

创建完毕记录app id 数字
在浏览器打开下面网址（自行修改参数）
修改参数参数app_id的值【APPID 】和参数redirect_url的值【https://www.awwwards.com/404.html】为实际填值
```
https://dashboard.locyanfrp.cn/auth/oauth/authorize?app_id=APPID&scopes=User,Proxy,Sign,Node&redirect_url=https://www.awwwards.com/404.html  
```
访问后同意授权等待自动跳转
记录跳转后的地址栏 refresh_token的值注意不要泄露

源码：

这次更新配置也有所变化，因为不需要登录了所以去掉了密码改为refresh_token值

locyan_auto_sign.py


# -*- coding: utf-8 -*-
"""
@File       : locyan_auto_sign.py
@Author     : AuraService
@Time       : 2023/8/29 17:39
@Update     : 2024/10/04 09:55
接口更新了，重新更新代码
@Update     : 2024/10/27 13:27
修改了登录接口的请求头和参数类型
1. 从params=params 改为 data=params
2. 去掉了无效参数
3. 请求头删除了跨域相关内容
@Update    : 2024/12/12 13:00
添加随机请求头 user-agent 以防被屏蔽
@Update    : 2024/12/24 13:00
# 登录后创建OAuth2.0 APP名称，重定向地址随意填写可访问的地址  如 https://www.awwwards.com/404.html 同时记录appid
修改下面地址中的APPID和重定向地址redirect_url，并用浏览器访问，看地址栏跳转后得到获取足够权限的refresh_token参数值
https://dashboard.locyanfrp.cn/auth/oauth/authorize?app_id=APPID&scopes=User,Proxy,Sign,Node&redirect_url=https://www.awwwards.com/404.html
更新accesstoken获取方式
@Description:
"""
import json
import os
import requests
from fake_useragent import UserAgent
ua = UserAgent().chrome
# 本地测试使用account
# 青龙部署的时候请注释下面这一块声明代码
accounts = [
    {
        'username': 'username',
        'refresh_token': 'xxxxxxxxxxxxx',
        'app_id': 'app_id',
        'traffic_limit': '限制额度'
    }
]
# 配置账号信息  （青龙部署使用这一块读取环境变量）
# accounts = json.loads(os.getenv('LOCYAN_CHECKIN'))
token = 'pushplus token'
topic = 'pushplus topic'
base_url = 'https://api-v2.locyanfrp.cn/api/v2'
login_url = '/auth/login'
user_info_url = '/user/info'
check_sign_url = "/sign"
sign_url = "/sign"
get_access_token_url = '/auth/oauth/access-token'
def login(username, password):
    url = f'{base_url}{login_url}'
    headers = {
        "User-Agent": ua,
        "accept": "application/json, text/plain, */*",
        "accept-language": "zh-CN,zh;q=0.9,en;q=0.8",
        "content-type": "application/x-www-form-urlencoded",
        "priority": "u=1, i",
        "sec-ch-ua": '"Chromium";v="130", "Google Chrome";v="130", "Not?A_Brand";v="99"',
        "sec-ch-ua-mobile": "?0",
        "sec-ch-ua-platform": '"Windows"',
        "Referer": "https://preview.locyanfrp.cn/",
        "Referrer-Policy": "strict-origin-when-cross-origin"
    }
    params = {
        'username': username,
        'password': password,
    }
    try:
        response = requests.post(url, headers=headers, data=params)
        return response.json()
    except Exception as e:
        return {'status': False, 'message': str(e)}
def sign(token, user_id):
    url = f'{base_url}{sign_url}'
    headers = {
        "User-Agent": ua,
        'accept': 'application/json, text/plain, */*',
        'accept-language': 'zh-CN,zh;q=0.9',
        'authorization': f'Bearer {token}',
        'content-type': 'application/x-www-form-urlencoded',
        'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"Windows"',
        'sec-fetch-dest': 'empty',
        'sec-fetch-mode': 'cors',
        'sec-fetch-site': 'same-site',
        'Referer': 'https://preview.locyanfrp.cn/',
    }
    data = {
        'user_id': user_id,
    }
    try:
        response = requests.post(url, headers=headers, data=data)
        return response.json()
    except Exception as e:
        return {'status': False, 'message': str(e)}
def get_access_token(refresh_token="", app_id=""):
    """
        成功响应示例：
        {            "data": {                "access_token": "OA_ca77f5e448130b4458c176740b04d2b01b776e14c08b8b8b29df169e21xxxxx",                "user_id": 111100            },            "message": "OK",            "status": 200        }    """
    url = f'{base_url}{get_access_token_url}'
    headers = {
        "User-Agent": ua,
        'accept': 'application/json, text/plain, */*',
        'accept-language': 'zh-CN,zh;q=0.9',
        'content-type': 'application/x-www-form-urlencoded',
        'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"Windows"',
        'sec-fetch-dest': 'empty',
        'sec-fetch-mode': 'cors',
        'sec-fetch-site': 'same-site',
        'Referer': 'https://preview.locyanfrp.cn/',
    }
    data = {
        'refresh_token': refresh_token,
        'app_id': app_id
    }
    try:
        response = requests.post(url, headers=headers, data=data)
        return response.json()
    except Exception as e:
        return {'status': False, 'message': str(e)}
def get_user_info(user_id="", token=""):
    url = f'{base_url}{user_info_url}?user_id={user_id}'
    headers = {
        "User-Agent": ua,
        'accept': 'application/json, text/plain, */*',
        'accept-language': 'zh-CN,zh;q=0.9',
        'authorization': f'Bearer {token}',
        'content-type': 'application/x-www-form-urlencoded',
        'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"Windows"',
        'sec-fetch-dest': 'empty',
        'sec-fetch-mode': 'cors',
        'sec-fetch-site': 'same-site',
        'Referer': 'https://preview.locyanfrp.cn/',
    }
    payload = {}
    try:
        response = requests.get(url, headers=headers, data=payload)
        return response.json()
    except Exception as e:
        return {'status': False, 'message': str(e)}
def check_sign(token):
    url = f'{base_url}{check_sign_url}'
    headers = {
        "User-Agent": ua,
        'accept': 'application/json, text/plain, */*',
        'accept-language': 'zh-CN,zh;q=0.9',
        'authorization': f'Bearer {token}',
        'content-type': 'application/x-www-form-urlencoded',
        'sec-ch-ua': '"Chromium";v="116", "Not)A;Brand";v="24", "Google Chrome";v="116"',
        'sec-ch-ua-mobile': '?0',
        'sec-ch-ua-platform': '"Windows"',
        'sec-fetch-dest': 'empty',
        'sec-fetch-mode': 'cors',
        'sec-fetch-site': 'same-site',
        'Referer': 'https://preview.locyanfrp.cn/',
    }
    data = {}
    try:
        response = requests.get(url, headers=headers, data=data)
        return response.json()
    except Exception as e:
        return {'status': False, 'message': str(e)}
def send_pushplus_message(token, title, content, topic=''):
    url = 'http://www.pushplus.plus/send'
    if topic == '':
        data = {
            'token': token,
            'title': title,
            'content': content
        }
    else:
        data = {
            'token': token,
            'title': title,
            'content': content,
            'topic': topic
        }
    response = requests.post(url, json=data)
    if response.status_code == 200:
        print("消息发送成功")
    else:
        print("消息发送失败")
def main():
    results = []
    for account in accounts:
        username = account['username']
        app_id = account['app_id']
        refresh_token = account['refresh_token']
        traffic_limit = account.get('traffic_limit', 0)
        # 登录
        login_res = get_access_token(refresh_token, app_id)
        if not login_res.get('status', 200) == 200:
            results.append(f"账号 {username} 登录失败: {login_res.get('message', None)}")
            continue
        login_response = login_res.get('data', {})
        user_id = login_response.get('user_id', None)
        token = login_response.get('access_token', None)
        user_info = get_user_info(user_id, token).get('data', {})
        email = user_info.get('email', '')
        avatar = user_info.get('avatar', '')
        traffic = user_info.get('traffic', {})
        inbound = user_info.get('inbound', {})
        outbound = user_info.get('outbound', {})
        print("---------")
        print(f"username : {username}")
        print(f"user_id : {user_id}")
        print(f"token : {token}")
        print(f"email : {email}")
        print(f"avatar : {avatar}")
        print(f"traffic : {traffic}")
        print(f"inbound : {inbound}")
        print(f"outbound : {outbound}")
        re = f"————————————————————\n" + \
             f"★账号: {username}\n" + \
             f"UserId: {user_id}\n" + \
             f"签到阈值: {traffic_limit} GiB\n" + \
             f"当前剩余流量: {traffic} GiB\n" + \
             f"上行速率: {inbound}Mbps\n" + \
             f"下行速率: {outbound}Mbps\n"
        if int(traffic) >= int(traffic_limit):
            re += f'😎 账号 {username} 流量已超过阈值，跳过签到'
            results.append(re)
            continue
        # 签到
        sign_response = sign(token, user_id)
        # 检查签到状态
        # check_response = check_sign(token)
        if sign_response.get('status', 0) == 200 or sign_response.get('status', 0) == 403:
            re += f"😀 签到成功 {sign_response.get('message', None)}"
            if sign_response.get('data', {}):
                get_traffic = sign_response.get('data', {}).get('get_traffic', 0)
                total_get_traffic = sign_response.get('data', {}).get('total_get_traffic', 0)
                sign_count = sign_response.get('data', {}).get('sign_count', 0)
                re += f"|签到次数:{sign_count}|本次签到获得流量:{get_traffic}|签到获得总流量:{total_get_traffic}"
        else:
            re += f"😢 签到失败 {sign_response.get('message', None)}"
        results.append(re)
    result_str = '\n'.join(results)
    print(result_str)
    return result_str
if __name__ == '__main__':
    send_pushplus_message(token, '乐青映射签到通知', main())

09琪琪 · 发表于 2024-11-12 04:47

zxy20014 发表于 2024-11-11 12:03
青龙提醒“unexpected character after line continuation character”

re = f"————————————————————\n" + \
         f"★账号: {username}\n" + \
         f"签到阈值: {traffic_limit} GiB\n" + \
         f"当前剩余流量: {traffic} GiB\n" + \
         f"上行速率: {inbound}Mbps\n" + \
         f"下行速率: {outbound}Mbps\n"

已解决，这段代码后面的\有空格，删除\后重新敲打\保证没空格

gxlu1997 · 发表于 2025-1-5 20:34

Cristy 发表于 2025-1-4 20:31
我的意思是我更新了，再复制一次试试

可以了，新代码send_pushplus_message(token, '乐青映射签到通知', main())``` 最后多了三个``` 删掉以后能正常推送了，谢谢大佬

Cristy · 发表于 2025-3-18 09:31

MuskaAmi 发表于 2025-3-18 03:20
提醒一下楼主，我们这边 api-v2.locyanfrp.cn 准备弃用了，建议换到新的

感谢官方团队对广大网友提供的公益服务。

zxy20014 · 发表于 2024-11-11 12:03

青龙提醒“unexpected character after line continuation character”

Jansen · 发表于 2024-11-11 12:29

感谢分享

GTO · 发表于 2024-11-11 13:23

他们家签到给的流量好多

peter08 · 发表于 2024-11-11 14:44

请问这个乐青内网穿透有官网吗？

anning666 · 发表于 2024-11-11 15:49

学习楼主的脚本,以后其他类似的情况可以参考,tks~~~~

kings0b · 发表于 2024-11-11 16:49

学到了感觉分享

09琪琪 · 发表于 2024-11-12 04:36

zxy20014 发表于 2024-11-11 12:03
青龙提醒“unexpected character after line continuation character”

我也是。怎么弄

Cristy · 发表于 2024-11-12 09:13

zxy20014 发表于 2024-11-11 12:03
青龙提醒“unexpected character after line continuation character”

有没有完整提示看看是哪一行出的错

帐号		自动登录	找回密码
密码			注册[Register]

[Python 原创] 【Python】乐青内穿账号自动签到

aliases:
tags:
author: Aura Service
created: 2024-11-11T10:20:43
updated: 2024-11-11T10:54:09
[b]updated: 2024-12-26T16:17:21 已经更新，更新兼容官方的新token获取方式，请从下往上看

乐青内网穿透自动签到脚本介绍

1 概述

2 功能特点

2.1 效果展示

3 使用说明

3.1 环境准备

3.1.1 注册pushplus

3.2 配置账号信息

3.3 代码结构

3.4 主程序

3.5 运行脚本

3.6 部署和使用

4 总结

5 更新

5.1 2024-11 旧版本

5.2 2024-12-26 (目前最新)

免费评分

免费评分

免费评分

[Python 原创] 【Python】乐青内穿账号自动签到

aliases: tags: author: Aura Service created: 2024-11-11T10:20:43 updated: 2024-11-11T10:54:09 [b]updated: 2024-12-26T16:17:21 已经更新，更新兼容官方的新token获取方式，请从下往上看

乐青内网穿透自动签到脚本介绍

1 概述

2 功能特点

2.1 效果展示

3 使用说明

3.1 环境准备

3.1.1 注册pushplus

3.2 配置账号信息

3.3 代码结构

3.4 主程序

3.5 运行脚本

3.6 部署和使用

4 总结

5 更新

5.1 2024-11 旧版本

5.2 2024-12-26 (目前最新)

免费评分

免费评分

免费评分

aliases:
tags:
author: Aura Service
created: 2024-11-11T10:20:43
updated: 2024-11-11T10:54:09
[b]updated: 2024-12-26T16:17:21 已经更新，更新兼容官方的新token获取方式，请从下往上看