mirror of
https://github.com/smallfawn/QLScriptPublic.git
synced 2026-11-05 22:36:53 +08:00
Move scripts to daily and add erke sign-in
Rename 360.py and 51dalili.py into the daily/ folder and add a one-line Env comment to daily/51dalili.py. Add a new daily/erke.py implementing Hongxing Erke sign-in: environment-driven ERKE_CONF support, signature calculation, ErkeAPI class with get_integral and sign_in methods, logging, multi-account handling, and a summary output. Remove an old QingLong-specific Erke script file (non-ASCII filename).
This commit is contained in:
smallfawn committed
1 parent
77851275ab
commit
c9f6a965d0
4 files changed
+188
-239
No files matched your search
File renamed without changes.
@@ -1,4 +1,5 @@
|
||||
#51代理每日签到
|
||||
#new Env("51代理自动签到")
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
import time
|
||||
+187
@@ -0,0 +1,187 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
new Env('鸿星尔克签到');
|
||||
cron: 1 10 * * *
|
||||
"""
|
||||
# =================================================================
|
||||
# 使用说明：
|
||||
# 1. 环境变量名称：ERKE_CONF
|
||||
# 2. 环境变量格式：member_id#enterprise_id#unionid#openid#wx_openid#备注
|
||||
# 3. 多账号支持：多个账号请换行，或者使用 & 符号连接
|
||||
# 示例：
|
||||
# 12345#67890#unionid123#openid123#wx_openid123#我的账号
|
||||
# 54321#09876#unionid456#openid456#wx_openid456#大号
|
||||
# =================================================================
|
||||
# 配置日志
|
||||
import os
|
||||
import requests
|
||||
import hashlib
|
||||
import random
|
||||
import logging
|
||||
from typing import Dict
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
||||
logger = logging.getLogger("ErkeCheckIn")
|
||||
|
||||
def get_gmt8_time() -> str:
|
||||
"""获取北京时间字符串"""
|
||||
tz = timezone(timedelta(hours=8))
|
||||
return datetime.now(tz).strftime('%Y-%m-%d %H:%M:%S')
|
||||
|
||||
def calculate_sign(appid: str, member_id: str, timestamp: str = None) -> dict:
|
||||
"""计算请求签名逻辑"""
|
||||
if timestamp is None:
|
||||
timestamp = get_gmt8_time()
|
||||
|
||||
random_num = random.randint(0, 9999999)
|
||||
trans_id = appid + timestamp
|
||||
|
||||
# 按照JS逻辑拼接字符串
|
||||
sign_str = f"timestamp={timestamp}transId={trans_id}secret=damogic8888random={random_num}memberId={member_id}"
|
||||
sign = hashlib.md5(sign_str.encode()).hexdigest()
|
||||
|
||||
return {
|
||||
'sign': sign,
|
||||
'timestamp': timestamp,
|
||||
'transId': trans_id,
|
||||
'random': random_num,
|
||||
'appid': appid,
|
||||
'memberId': member_id
|
||||
}
|
||||
|
||||
class ErkeAPI:
|
||||
def __init__(self, member_id, enterprise_id, unionid, openid, wx_openid, account_name):
|
||||
self.member_id = member_id
|
||||
self.enterprise_id = enterprise_id
|
||||
self.unionid = unionid
|
||||
self.openid = openid
|
||||
self.wx_openid = wx_openid
|
||||
self.account_name = account_name
|
||||
self.appid = "wxa1f1fa3785a47c7d"
|
||||
self.base_url = 'https://hope.demogic.com/gic-wx-app'
|
||||
self.ua = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36 MicroMessenger/7.0.20.1781(0x6700143B) NetType/WIFI MiniProgramEnv/Windows WindowsWechat/WMPF'
|
||||
|
||||
def get_headers(self, sign_val: str) -> Dict[str, str]:
|
||||
return {
|
||||
'User-Agent': self.ua,
|
||||
'Content-Type': 'application/x-www-form-urlencoded',
|
||||
'sign': sign_val, # 注意：原逻辑此处传入的是enterprise_id
|
||||
'channelEntrance': 'wx_app',
|
||||
'Referer': 'https://servicewechat.com/wxa1f1fa3785a47c7d/85/page-frame.html',
|
||||
}
|
||||
|
||||
def get_integral(self):
|
||||
"""查询积分明细"""
|
||||
sign_data = calculate_sign(self.appid, self.member_id)
|
||||
data = {
|
||||
'currentPage': 1,
|
||||
'pageSize': 20,
|
||||
'memberId': self.member_id,
|
||||
'cliqueId': '-1',
|
||||
'enterpriseId': self.enterprise_id,
|
||||
'unionid': self.unionid,
|
||||
'openid': self.openid,
|
||||
'wxOpenid': self.wx_openid,
|
||||
'random': sign_data['random'],
|
||||
'appid': self.appid,
|
||||
'transId': sign_data['transId'],
|
||||
'sign': sign_data['sign'],
|
||||
'timestamp': sign_data['timestamp'],
|
||||
'gicWxaVersion': '3.9.56'
|
||||
}
|
||||
|
||||
try:
|
||||
res = requests.post(f"{self.base_url}/integral_record.json",
|
||||
headers=self.get_headers(self.enterprise_id),
|
||||
data=data, timeout=10).json()
|
||||
response_obj = res.get('response', {})
|
||||
accumulate = response_obj.get('accumulatPoints', 0)
|
||||
logger.info(f"[{self.account_name}] 当前累计积分: {accumulate}")
|
||||
return accumulate
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.account_name}] 积分查询异常: {e}")
|
||||
return "未知"
|
||||
|
||||
def sign_in(self):
|
||||
"""执行签到"""
|
||||
sign_data = calculate_sign(self.appid, self.member_id)
|
||||
payload = {
|
||||
'source': 'wxapp',
|
||||
'memberId': self.member_id,
|
||||
'cliqueId': '-1',
|
||||
'enterpriseId': self.enterprise_id,
|
||||
'unionid': self.unionid,
|
||||
'openid': self.openid,
|
||||
'wxOpenid': self.wx_openid,
|
||||
'sign': sign_data['sign'],
|
||||
'random': sign_data['random'],
|
||||
'appid': self.appid,
|
||||
'transId': sign_data['transId'],
|
||||
'timestamp': sign_data['timestamp'],
|
||||
'gicWxaVersion': '3.9.56'
|
||||
}
|
||||
|
||||
headers = self.get_headers(self.enterprise_id)
|
||||
headers['Content-Type'] = 'application/json;charset=UTF-8'
|
||||
|
||||
try:
|
||||
res = requests.post(f"{self.base_url}/sign/member_sign.json",
|
||||
headers=headers, json=payload, timeout=10).json()
|
||||
msg = res.get('message', '无返回信息')
|
||||
code = res.get('code')
|
||||
if code in ['0000', '1001', 0]:
|
||||
logger.info(f"[{self.account_name}] 签到结果: {msg}")
|
||||
return True, msg
|
||||
else:
|
||||
logger.warning(f"[{self.account_name}] 签到失败: {msg}")
|
||||
return False, msg
|
||||
except Exception as e:
|
||||
logger.error(f"[{self.account_name}] 签到请求异常: {e}")
|
||||
return False, str(e)
|
||||
|
||||
def main():
|
||||
# 尝试从环境变量读取配置
|
||||
conf_str = os.getenv("ERKE_CONF", "")
|
||||
if not conf_str:
|
||||
logger.error("未找到环境变量 ERKE_CONF，请检查配置！")
|
||||
return
|
||||
# 处理多账号，支持换行或 & 分隔
|
||||
accounts_raw = conf_str.replace("&", "\n").splitlines()
|
||||
summary = []
|
||||
|
||||
for line in accounts_raw:
|
||||
if not line.strip(): continue
|
||||
|
||||
params = line.strip().split("#")
|
||||
if len(params) < 5:
|
||||
logger.error(f"账号配置格式错误 (需5个#分隔): {line}")
|
||||
continue
|
||||
|
||||
# 提取参数
|
||||
m_id, e_id, uid, oid, wx_oid = params[:5]
|
||||
name = params[5] if len(params) > 5 else f"账号_{m_id[-4:]}"
|
||||
|
||||
api = ErkeAPI(m_id, e_id, uid, oid, wx_oid, name)
|
||||
|
||||
# 执行流程
|
||||
points_before = api.get_integral()
|
||||
success, msg = api.sign_in()
|
||||
points_after = api.get_integral()
|
||||
|
||||
status_emoji = "✅" if success else "❌"
|
||||
summary.append(f"{status_emoji} 【{name}】: {msg} (积分: {points_before} -> {points_after})")
|
||||
|
||||
# 打印最终统计
|
||||
print("\n" + "="*30)
|
||||
print(" 鸿星尔克任务简报")
|
||||
print("="*30)
|
||||
print("\n".join(summary))
|
||||
|
||||
# 如果有通知推送逻辑，可以在此处调用
|
||||
# send_notification("鸿星尔克签到", "\n".join(summary))
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -1,239 +0,0 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
import json
|
||||
import logging
|
||||
import sys
|
||||
from typing import List, Dict, Any
|
||||
from pathlib import Path
|
||||
from api import ErkeAPI
|
||||
|
||||
# === 青龙面板专用配置 ===
|
||||
# 1. 脚本需放置在 /ql/scripts 目录
|
||||
# 2. 配置文件需放在 /ql/config/token.json
|
||||
# 3. 依赖模块需在 /ql/lib 目录（api.py/notification.py）
|
||||
# ======================
|
||||
|
||||
# 青龙环境标准路径
|
||||
project_root = Path("/ql")
|
||||
sys.path.insert(0, str(project_root / "lib"))
|
||||
|
||||
# 配置日志（青龙默认使用INFO级别）
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s - %(levelname)s - %(message)s",
|
||||
datefmt="%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
logger = logging.getLogger("erke_sign")
|
||||
|
||||
class ErkeTasks:
|
||||
"""鸿星尔克签到任务自动化执行类（青龙面板专用）"""
|
||||
def __init__(self, config_path: str = None):
|
||||
"""初始化任务执行器
|
||||
Args:
|
||||
config_path: 配置文件路径（青龙默认使用 /ql/config/token.json）
|
||||
"""
|
||||
# 青龙标准配置路径
|
||||
self.config_path = Path("/ql/config/token.json") if config_path is None else Path(config_path)
|
||||
self.accounts: List[Dict[str, Any]] = []
|
||||
self.account_results: List[Dict[str, Any]] = []
|
||||
self._init_accounts()
|
||||
|
||||
def _init_accounts(self):
|
||||
"""从青龙配置文件读取账号信息"""
|
||||
if not self.config_path.exists():
|
||||
logger.error(f"配置文件不存在: {self.config_path}")
|
||||
raise FileNotFoundError(f"请在 /ql/config/ 创建 token.json 配置文件")
|
||||
|
||||
try:
|
||||
with open(self.config_path, 'r', encoding='utf-8') as f:
|
||||
config_data = json.load(f)
|
||||
erke_config = config_data.get('erke', {})
|
||||
self.accounts = erke_config.get('accounts', [])
|
||||
|
||||
if not self.accounts:
|
||||
logger.warning("配置文件中没有找到 erke 账号信息")
|
||||
else:
|
||||
logger.info(f"成功加载 {len(self.accounts)} 个账号配置")
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"配置文件JSON解析失败: {e}")
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"读取配置文件失败: {e}")
|
||||
raise
|
||||
|
||||
def process_account(self, account: Dict[str, Any]) -> Dict[str, Any]:
|
||||
"""处理单个账号任务"""
|
||||
account_name = account.get('account_name', '未命名账号')
|
||||
logger.info(f"\n{'='*50}")
|
||||
logger.info(f"开始处理账号: {account_name}")
|
||||
logger.info(f"{'='*50}")
|
||||
|
||||
result = {
|
||||
'account_name': account_name,
|
||||
'success': False,
|
||||
'integral_info': None,
|
||||
'sign_info': None,
|
||||
'error': None
|
||||
}
|
||||
|
||||
try:
|
||||
# 初始化API
|
||||
api = ErkeAPI(
|
||||
member_id=account.get('member_id', ''),
|
||||
enterprise_id=account.get('enterprise_id', ''),
|
||||
unionid=account.get('unionid', ''),
|
||||
openid=account.get('openid', ''),
|
||||
wx_openid=account.get('wx_openid', ''),
|
||||
user_agent=account.get('user_agent')
|
||||
)
|
||||
|
||||
# 1. 查询积分明细
|
||||
logger.info(f"[{account_name}] 查询积分明细...")
|
||||
integral_result = api.get_integral_record(current_page=1, page_size=5)
|
||||
|
||||
if integral_result['success']:
|
||||
result['integral_info'] = integral_result['result']
|
||||
logger.info(f"[{account_name}] 积分明细查询成功")
|
||||
|
||||
# 解析积分信息
|
||||
if integral_result['result'] and isinstance(integral_result['result'], dict):
|
||||
response = integral_result['result'].get('response', {})
|
||||
if isinstance(response, dict):
|
||||
accumulate_points = response.get('accumulatPoints', 0)
|
||||
frozen_points = response.get('frozenPoints', 0)
|
||||
available_points = accumulate_points - frozen_points
|
||||
|
||||
logger.info(f"[{account_name}] 累计积分: {accumulate_points}")
|
||||
logger.info(f"[{account_name}] 冻结积分: {frozen_points}")
|
||||
logger.info(f"[{account_name}] 可用积分: {available_points}")
|
||||
|
||||
page_data = response.get('page', {})
|
||||
if page_data:
|
||||
total_count = page_data.get('totalCount', 0)
|
||||
logger.info(f"[{account_name}] 积分记录数: {total_count}")
|
||||
else:
|
||||
logger.warning(f"[{account_name}] 积分明细查询失败: {integral_result['error']}")
|
||||
|
||||
# 2. 执行签到
|
||||
logger.info(f"[{account_name}] 执行签到...")
|
||||
sign_result = api.member_sign()
|
||||
|
||||
if sign_result['success']:
|
||||
result['sign_info'] = sign_result['result']
|
||||
if sign_result['result'] and isinstance(sign_result['result'], dict):
|
||||
code = sign_result['result'].get('code', '')
|
||||
message = sign_result['result'].get('message', '')
|
||||
|
||||
# code: 1001 表示已签到，0000 表示签到成功
|
||||
if code == '0000':
|
||||
result['success'] = True
|
||||
logger.info(f"[{account_name}] 签到成功: {message}")
|
||||
elif code == '1001':
|
||||
result['success'] = True
|
||||
logger.info(f"[{account_name}] {message}")
|
||||
else:
|
||||
result['success'] = False
|
||||
result['error'] = message
|
||||
logger.warning(f"[{account_name}] 签到返回: {message}")
|
||||
else:
|
||||
result['success'] = True
|
||||
logger.info(f"[{account_name}] 签到完成")
|
||||
else:
|
||||
result['error'] = sign_result['error']
|
||||
logger.error(f"[{account_name}] 签到失败: {sign_result['error']}")
|
||||
|
||||
except Exception as e:
|
||||
error_msg = f"处理账号时发生异常: {str(e)}"
|
||||
logger.error(f"[{account_name}] {error_msg}")
|
||||
result['error'] = error_msg
|
||||
|
||||
return result
|
||||
|
||||
def run(self):
|
||||
"""执行所有账号的签到任务"""
|
||||
logger.info("="*60)
|
||||
logger.info("鸿星尔克签到任务开始执行")
|
||||
logger.info("="*60)
|
||||
|
||||
if not self.accounts:
|
||||
logger.error("没有可处理的账号")
|
||||
return
|
||||
|
||||
# 处理每个账号
|
||||
for account in self.accounts:
|
||||
result = self.process_account(account)
|
||||
self.account_results.append(result)
|
||||
|
||||
# 输出统计信息
|
||||
self._print_summary()
|
||||
# 发送通知
|
||||
self._send_notification()
|
||||
|
||||
def _print_summary(self):
|
||||
"""输出执行结果统计"""
|
||||
logger.info("\n" + "="*60)
|
||||
logger.info("执行结果统计")
|
||||
logger.info("="*60)
|
||||
|
||||
success_count = sum(1 for r in self.account_results if r['success'])
|
||||
fail_count = len(self.account_results) - success_count
|
||||
|
||||
logger.info(f"总账号数: {len(self.account_results)}")
|
||||
logger.info(f"成功: {success_count}")
|
||||
logger.info(f"失败: {fail_count}")
|
||||
|
||||
if fail_count > 0:
|
||||
logger.info("\n失败账号详情:")
|
||||
for result in self.account_results:
|
||||
if not result['success']:
|
||||
logger.info(f" - {result['account_name']}: {result['error']}")
|
||||
|
||||
def _send_notification(self):
|
||||
"""发送执行结果通知"""
|
||||
try:
|
||||
success_count = sum(1 for r in self.account_results if r['success'])
|
||||
total_count = len(self.account_results)
|
||||
|
||||
# 构建通知内容
|
||||
content_lines = [
|
||||
f"📊 执行统计:",
|
||||
f" - 总账号数: {total_count}",
|
||||
f" - 成功: {success_count}",
|
||||
f" - 失败: {total_count - success_count}",
|
||||
]
|
||||
|
||||
content_lines.append("\n📋 账号详情:")
|
||||
for result in self.account_results:
|
||||
status = "✅" if result['success'] else "❌"
|
||||
content_lines.append(f" {status} {result['account_name']}")
|
||||
if result['success'] and result['sign_info']:
|
||||
if isinstance(result['sign_info'], dict):
|
||||
message = result['sign_info'].get('message', '')
|
||||
if message:
|
||||
content_lines.append(f" └─ {message}")
|
||||
|
||||
content = "\n".join(content_lines)
|
||||
|
||||
# 发送通知（青龙支持的推送方式）
|
||||
send_notification(
|
||||
title="鸿星尔克签到任务完成",
|
||||
content=content,
|
||||
sound=NotificationSound.BIRDSONG
|
||||
)
|
||||
logger.info("通知发送成功")
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"发送通知失败: {str(e)}")
|
||||
|
||||
def main():
|
||||
"""主函数（青龙面板执行入口）"""
|
||||
try:
|
||||
tasks = ErkeTasks()
|
||||
tasks.run()
|
||||
except Exception as e:
|
||||
logger.error(f"程序执行失败: {str(e)}")
|
||||
sys.exit(1)
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in new issue
Block a user