写个查询话费的青龙面板的脚本（Python脚本）

首先感谢作者https://github.com/Cp0204/ChinaTelecomMonitor.git

由于原作者只支持单个电信号码查询的任务脚本，为此我在他的基础上改写了代码，使得支持多个手机号码查询，本人是源代码的受益者，再次感谢原作者Cp0204

先上代码

# !/usr/bin/env python3 # -*- coding: utf-8 -*- # Repo: https://github.com/Cp0204/ChinaTelecomMonitor # ConfigFile: telecom_config.json # Modify: 2025-06-22 """ 任务名称 name: 电信套餐用量监控 定时规则 cron: 0 20 * * * """ import os import sys import json from datetime import datetime # 兼容青龙 try: from telecom_class import Telecom except: print("正在尝试自动安装依赖...") os.system("pip3 install pycryptodome requests &> /dev/null") from telecom_class import Telecom CONFIG_DATA = {} NOTIFYS = [] CONFIG_PATH = sys.argv[1] if len(sys.argv) > 1 else "telecom_config.json" # 发送通知消息 def send_notify(title, body): try: # 导入通知模块 import notify # 如未配置 push_config 则使用青龙环境通知设置 if CONFIG_DATA.get("push_config"): notify.push_config.update(CONFIG_DATA["push_config"]) notify.push_config["CONSOLE"] = notify.push_config.get("CONSOLE", True) notify.send(title, body) except Exception as e: if e: print("发送通知消息失败！") # 添加消息 def add_notify(text): global NOTIFYS NOTIFYS.append(text) print("📢", text) return text def process_account(phonenum, password): """处理单个账号的主要逻辑""" telecom = Telecom() account_config = CONFIG_DATA.get("accounts", {}).get(phonenum, {}) def auto_login(): print(f"自动登录：{phonenum}") # 记录登录失败次数，避免风控 login_fail_time = account_config.get("loginFailTime", 0) if login_fail_time < 5: data = telecom.do_login(phonenum, password) if data.get("responseData").get("resultCode") == "0000": print(f"自动登录：成功") login_info = data["responseData"]["data"]["loginSuccessResult"] login_info["phonenum"] = phonenum login_info["createTime"] = datetime.now().strftime("%Y-%m-%d %H:%M:%S") account_config["login_info"] = login_info account_config["loginFailTime"] = 0 telecom.set_login_info(login_info) return True else: login_fail_time = int( data.get("responseData", {}) .get("data", {}) .get("loginFailResult", {}) .get("loginFailTime", login_fail_time + 1) ) account_config["loginFailTime"] = login_fail_time update_config() add_notify(f"自动登录：{phonenum} 已连续失败{login_fail_time}次") return False else: print( f"自动登录：{phonenum} 已连续失败{login_fail_time}次，为避免风控不再执行；修正登录信息后，如需重新登录请删除程序目录下(.json)配置中的 loginFailTime 键值" ) return False # 读取缓存Token login_info = account_config.get("login_info", {}) if login_info and login_info.get("phonenum"): print(f"尝试使用缓存登录：{login_info['phonenum']}") telecom.set_login_info(login_info) # 验证缓存登录是否有效 important_data = telecom.qry_important_data() if important_data.get("responseData"): print(f"缓存登录验证成功") else: print(f"缓存登录验证失败，尝试重新登录") if not auto_login(): return important_data = telecom.qry_important_data() else: if not auto_login(): return important_data = telecom.qry_important_data() # 获取主要信息 if important_data.get("responseData"): print(f"获取主要信息：成功") elif important_data["headerInfos"]["code"] == "X201": print(f"获取主要信息：失败 {important_data['headerInfos']['reason']}") if not auto_login(): return important_data = telecom.qry_important_data() # 简化主要信息 try: summary = telecom.to_summary(important_data["responseData"]["data"]) except Exception as e: print( f"简化主要信息出错，提 Issue 请提供以下信息（隐私打码）：nn{json.dumps(important_data['responseData']['data'], ensure_ascii=False)}nn{e}" ) return if summary: print(f"简化主要信息：{summary}") account_config["summary"] = summary # 获取流量包明细 flux_package_str = "" user_flux_package = telecom.user_flux_package() if user_flux_package: print("获取流量包明细：成功") packages = user_flux_package["responseData"]["data"]["productOFFRatable"][ "ratableResourcePackages" ] for package in packages: package_icon = ( "🇨🇳" if "国内" in package["title"] else "📺" if "专用" in package["title"] else "🌎" ) flux_package_str += f"n{package_icon}{package['title']}n" for product in package["productInfos"]: if product["infiniteTitle"]: # 无限流量 flux_package_str += f"""🔹[{product['title']}]{product['infiniteTitle']}{product['infiniteValue']}{product['infiniteUnit']}/无限n""" else: flux_package_str += f"""🔹[{product['title']}]{product['leftTitle']}{product['leftHighlight']}{product['rightCommon']}n""" # 流量字符串 common_str = ( f"{telecom.convert_flow(summary['commonUse'], 'GB', 2)} / {telecom.convert_flow(summary['commonTotal'], 'GB', 2)} GB 🟢" if summary["flowOver"] == 0 else f"-{telecom.convert_flow(summary['flowOver'], 'GB', 2)} / {telecom.convert_flow(summary['commonTotal'], 'GB', 2)} GB 🔴" ) special_str = ( f"{telecom.convert_flow(summary['specialUse'], 'GB', 2)} / {telecom.convert_flow(summary['specialTotal'], 'GB', 2)} GB" if summary["specialTotal"] > 0 else "" ) # 基本信息 notify_str = f""" 📱 手机：{summary['phonenum']} 💰 余额：{round(summary['balance'] / 100, 2)} 📞 通话：{summary['voiceUsage']}{f" / {summary['voiceTotal']}" if summary['voiceTotal'] > 0 else ''} min 🌐 总流量 - 通用：{common_str}{f'{chr(10)} - 专用：{special_str}' if special_str else ''}""" # 流量包明细 if os.environ.get("TELECOM_FLUX_PACKAGE", "true").lower() != "false": notify_str += f"nn【流量包明细】nn{flux_package_str.strip()}" notify_str += f"nn查询时间：{summary['createTime']}" add_notify(notify_str.strip()) # 更新账号配置 CONFIG_DATA.setdefault("accounts", {})[phonenum] = account_config update_config() def main(): global CONFIG_DATA start_time = datetime.now() print(f"===============程序开始===============") print(f"⏰ 执行时间: {start_time.strftime('%Y-%m-%d %H:%M:%S')}") print() # 读取配置 if os.path.exists(CONFIG_PATH): print(f"⚙️ 正从 {CONFIG_PATH} 文件中读取配置") with open(CONFIG_PATH, "r", encoding="utf-8") as file: CONFIG_DATA = json.load(file) else: CONFIG_DATA = {} # 获取所有账号 accounts = [] # 从环境变量获取账号 if "TELECOM_USER" in os.environ: TELECOM_USERS = os.environ.get("TELECOM_USER").split("&") for TELECOM_USER in TELECOM_USERS: phonenum, password = ( TELECOM_USER[:11], TELECOM_USER[11:], ) accounts.append((phonenum, password)) # 从配置文件获取账号(兼容旧配置) if old_user := CONFIG_DATA.get("user"): if phonenum := old_user.get("phonenum"): password = old_user.get("password", "") accounts.append((phonenum, password)) if not accounts: exit("未找到任何账号配置，程序退出") print(f"共发现 {len(accounts)} 个账号需要处理") # 处理每个账号 for i, (phonenum, password) in enumerate(accounts, 1): print(f"n===== 开始处理账号 {i}/{len(accounts)}: {phonenum} =====") process_account(phonenum, password) # 通知 if NOTIFYS: notify_body = "nn".join(NOTIFYS) print(f"===============推送通知===============") send_notify(f"【电信套餐用量监控】共{len(accounts)}个账号", notify_body) print() def update_config(): # 更新配置 with open(CONFIG_PATH, "w", encoding="utf-8") as file: json.dump(CONFIG_DATA, file, ensure_ascii=False, indent=2) if __name__ == "__main__": main()

使用方法

1. 拉库订阅，ql repo https://github.com/Cp0204/ChinaTelecomMonitor.git "telecom_monitor" "" "telecom_class|notify"

2. 在脚本管理中，新增脚本，如neww.py，粘贴上述代码

3. 在环境变量中，新增环境变量TELECOM_USER，每个TELECOM_USER一个手机号密码，多个账号就多个TELECOM_USER，前11位手机号，后面的是密码

   环境变量示例：

   TELECOM_USER 18912345678password 手机号密码直接拼接，密码为6位数字，会自动截取

   TELECOM_FLUX_PACKAGE true (默认) 推送流量包明细，false 则只推送基本信息

4. 在定时任务中，添加任务，执行task neww.py

5. 推送效果如下

作者声明本文无利益相关，欢迎值友理性交流，和谐讨论～