【Scripts】乐青映射-签到（根据API）
一、API文档地址
二、获取refresh_token
1. 后台创建应用
重定向地址填写
http://127.0.0.1:3000/oauth/callback
2. 创建本地oauth服务（可选操作）
from flask import Flask, request
from flask_cors import CORS
app = Flask(__name__)
CORS(app) # 可选：仅在需要跨域时启用
@app.route("/oauth/callback")
def oauth_callback():
refresh_token = request.args.get("refresh_token")
error = request.args.get("error")
state = request.args.get("state") # 可选：用于防御 CSRF
# 可选：验证 state 参数（需与授权请求时生成的随机值匹配）
# if state != expected_state:
# return "非法请求: State 不匹配", 400
if refresh_token:
print(f"[SUCCESS] 刷新令牌: {refresh_token}")
return "授权成功！", 200
elif error:
print(f"[ERROR] 错误类型: {error}")
return f"授权失败: {error}", 400
else:
return "非法请求", 400
if __name__ == "__main__":
app.run(host="0.0.0.0", port=3000, debug=True)
3. 获取refresh_token
创建好本地oauth服务后，访问以下地址：
https://dashboard.locyanfrp.cn/auth/oauth/authorize?app_id=99&scopes=Sign,User&redirect_url=http://127.0.0.1:3000/oauth/callback
注意：app_id根据自己实际情况填写
http://127.0.0.1:3000/oauth/callback?refresh_token=xxxxx
三、签到代码
send_message函数请自行编写
import configparser
import os
import requests
import logging
from datetime import datetime, timedelta
from typing import Optional, Dict, Any
from tenacity import retry, stop_after_attempt, wait_exponential
# 日志配置
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[logging.StreamHandler()]
)
logger = logging.getLogger("LocyanSign")
logger.setLevel(logging.DEBUG)
class LocyanOAuthError(Exception):
"""自定义认证异常"""
pass
class LocyanSign:
REQUIRED_SCOPES = {'Sign.Info', 'Sign.Sign', 'User.Info'}
def __init__(self):
self.msg = []
# 初始化配置校验
self._validate_env_vars()
self._init_api_client()
self._acquire_initial_token()
def send_message(self, msg: str, msg_type: str = "info"):
pass
def _validate_env_vars(self):
"""环境变量校验"""
self.user_id = "用户id，自己抓包浏览器获取"
self.refresh_token = "刚刚获取的refresh_token"
self.id = "官网创建应用的id"
if not all([self.user_id, self.refresh_token]):
raise ValueError("LOCYAN_USER_ID 和 LOCYAN_REFRESH_TOKEN 必须配置")
if len(self.refresh_token) != 64 or not self.refresh_token.isalnum():
raise ValueError("Refresh Token格式无效")
try:
self.user_id = int(self.user_id) # 转换为整数
except ValueError:
raise ValueError("User ID必须是数字")
def _init_api_client(self):
"""初始化API客户端"""
self.api_endpoints = [
"https://api.locyanfrp.cn/v2",
"https://backup.api.locyanfrp.cn/v2"
]
self.session = requests.Session()
self.session.headers.update({
"User-Agent": "LocyanSign/3.0",
"Accept": "application/json",
"Content-Type": "application/x-www-form-urlencoded"
})
self.access_token = None
self.token_expiry = None
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, max=10))
def _request(self, method: str, endpoint: str, ** kwargs) -> Optional[Dict[str, Any]]:
"""增强型请求方法"""
for base_url in self.api_endpoints:
try:
url = f"{base_url}{endpoint}"
logger.debug(f"请求 {method} {url}")
# 动态添加认证头
headers = kwargs.get('headers', {})
headers.setdefault("Authorization", f"Bearer {self.access_token}")
kwargs['headers'] = headers
response = self.session.request(
method, url,
timeout=(3.05, 15),
** kwargs
)
logger.debug(f"响应状态: {response.status_code}")
logger.debug(f"响应内容: {response.text[:200]}...")
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
logger.warning(f"{base_url} 请求失败: {str(e)}")
continue
logger.error("所有API端点请求失败")
raise LocyanOAuthError("API服务不可用")
def _acquire_initial_token(self):
"""获取初始访问令牌"""
logger.info("正在初始化访问令牌...")
data = {
"app_id": self.id,
"refresh_token": self.refresh_token
}
response = self._request("POST", "/auth/oauth/access-token", data=data)
if not response or response.get("status") != 200:
raise LocyanOAuthError("令牌获取失败")
token_data = response.get("data", {})
self.access_token = token_data.get("access_token")
if not self.access_token:
raise LocyanOAuthError("无效的令牌响应")
self.token_expiry = datetime.now() + timedelta(hours=1)
logger.info(f"令牌获取成功，有效期至 {self.token_expiry.strftime('%Y-%m-%d %H:%M:%S')}")
def _ensure_valid_token(self):
"""令牌有效性保障"""
if datetime.now() >= self.token_expiry:
logger.warning("检测到令牌过期，正在刷新...")
self._acquire_initial_token()
@staticmethod
def format_traffic(traffic: str) -> str:
"""精确的流量格式化 (基于API返回的MB单位)"""
try:
mb_value = int(traffic)
units = [
('TB', 1024 ** 2), # 1TB = 1024 GB = 1024^2 MB
('GB', 1024), # 1GB = 1024 MB
('MB', 1)
]
for unit, divisor in units:
if mb_value >= divisor:
value = mb_value / divisor
# 保留三位有效数字
if value >= 100:
return f"{value:.0f} {unit}"
else:
return f"{value:.2f} {unit}".rstrip('0').rstrip('.')
return f"{mb_value} MB"
except (ValueError, TypeError) as e:
logger.error(f"流量解析错误: {str(e)}")
return "未知格式"
def check_sign_status(self) -> bool:
"""查询签到状态"""
self._ensure_valid_token()
try:
response = self._request("GET", "/sign", params={"user_id": self.user_id})
return response.get("data", {}).get("status", False)
except Exception as e:
logger.error(f"状态查询失败: {str(e)}")
return False
def perform_sign(self) -> Dict[str, Any]:
"""执行签到操作"""
self._ensure_valid_token()
try:
response = self._request("POST", "/sign", data={"user_id": self.user_id})
return response.get("data", {})
except Exception as e:
logger.error(f"签到失败: {str(e)}")
return {}
def get_user_info(self) -> Dict[str, Any]:
"""获取用户信息"""
self._ensure_valid_token()
try:
response = self._request("GET", "/user/info", params={"user_id": self.user_id})
return response.get("data", {})
except Exception as e:
logger.error(f"信息获取失败: {str(e)}")
return {}
def execute(self):
"""主业务流程"""
try:
if self.check_sign_status():
info = self.get_user_info()
message = (
"✅ 今日已签到\r"
f"👤 用户：{info.get('username', '未知')}\r"
f"📊 剩余流量：{self.format_traffic(info.get('traffic', '0'))}"
)
logger.info(message.replace("\r", " - "))
self.send_message(message)
return
result = self.perform_sign()
info = self.get_user_info()
message = (
"🎉 签到成功\r"
f"👤 用户：{info.get('username', '未知')}\r"
f"📥 本次获得：{result.get('get_traffic', 0)}GB\r"
f"📆 累计签到：{result.get('sign_count', 0)}次\r"
f"📊 总流量：{self.format_traffic(info.get('traffic', '0'))}"
)
logger.info(message.replace("\r", " - "))
self.send_message(message)
except Exception as e:
error_msg = f"❌ 签到失败\r错误详情：{str(e)}"
logger.exception(error_msg)
self.send_message(error_msg, "error")
raise
if __name__ == "__main__":
try:
client = LocyanSign()
client.execute()
except Exception as e:
logger.critical(f"系统故障: {str(e)}")
exit(1)
评论区