commit
0194c93783
6 changed files with 208 additions and 0 deletions
@ -0,0 +1,32 @@
|
||||
# 使用Python官方镜像作为基础镜像 |
||||
FROM docker.1ms.run/library/python:3.11-slim |
||||
|
||||
# 设置工作目录 |
||||
WORKDIR /app |
||||
|
||||
# 设置环境变量 |
||||
# 默认检查间隔为30分钟(可被运行时环境变量覆盖) |
||||
ENV CHECK_INTERVAL_MINUTES=30 |
||||
# 企业微信webhook地址(可被运行时环境变量覆盖) |
||||
ENV WECHAT_WEBHOOK_URL='https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=4226c76e-725b-4990-b926-05f16142e513' |
||||
# 设置Python日志输出环境变量,确保日志不缓冲 |
||||
ENV PYTHONUNBUFFERED=1 |
||||
ENV PYTHONIOENCODING=utf-8 |
||||
|
||||
# 安装依赖 |
||||
COPY requirements.txt . |
||||
RUN pip install --no-cache-dir -r requirements.txt |
||||
|
||||
# 复制脚本文件 |
||||
COPY firmware_checker.py . |
||||
COPY firmware_versions.txt . |
||||
|
||||
# 创建卷,以便可以持久化和修改versions.txt文件 |
||||
VOLUME ["/app"] |
||||
|
||||
# 添加健康检查(可选) |
||||
HEALTHCHECK --interval=60s --timeout=10s --start-period=5s --retries=3 \ |
||||
CMD python -c "import os; import sys; sys.exit(0 if os.path.exists('/app/firmware_checker.py') else 1)" || exit 1 |
||||
|
||||
# 设置容器启动时执行的命令 |
||||
CMD ["python", "firmware_checker.py"] |
||||
@ -0,0 +1,2 @@
|
||||
docker build -t registry.cn-beijing.aliyuncs.com/zyh5/firmware-checker . |
||||
docker push registry.cn-beijing.aliyuncs.com/zyh5/firmware-checker:latest |
||||
@ -0,0 +1,171 @@
|
||||
import requests |
||||
import os |
||||
import json |
||||
import re |
||||
import time |
||||
import sys |
||||
import logging |
||||
|
||||
# 配置日志系统 |
||||
logging.basicConfig( |
||||
level=logging.INFO, |
||||
format='%(asctime)s - %(message)s', |
||||
handlers=[ |
||||
logging.StreamHandler(sys.stdout) |
||||
] |
||||
) |
||||
logger = logging.getLogger(__name__) |
||||
|
||||
# 确保日志输出不缓冲 |
||||
for handler in logging.getLogger().handlers: |
||||
handler.flush = sys.stdout.flush |
||||
|
||||
def get_firmware_list(): |
||||
"""访问API获取固件列表""" |
||||
url = 'https://app4.i4.cn/getFirmwareList.xhtml?model=iPhone+13+Pro+Max&fs=2&seltype=1&ios=' |
||||
try: |
||||
response = requests.get(url, timeout=10) |
||||
response.raise_for_status() # 检查请求是否成功 |
||||
return response.json() |
||||
except Exception as e: |
||||
logger.error(f"获取固件列表失败: {e}") |
||||
return None |
||||
|
||||
def extract_versions_from_api(data): |
||||
"""从API返回的数据中提取版本号""" |
||||
versions = [] |
||||
if data and data.get('success') and 'list' in data: |
||||
for item in data['list']: |
||||
if 'version' in item: |
||||
# 提取主要版本号(如26.1,忽略beta等后缀) |
||||
version = item['version'] |
||||
main_version_match = re.match(r'(\d+\.\d+)', version) |
||||
if main_version_match: |
||||
main_version = main_version_match.group(1) |
||||
versions.append(main_version) |
||||
return versions |
||||
|
||||
def get_local_versions(): |
||||
"""直接读取firmware_versions.txt文件中的版本号""" |
||||
local_versions = [] |
||||
file_path = os.path.join(os.getcwd(), 'firmware_versions.txt') |
||||
|
||||
if not os.path.exists(file_path): |
||||
logger.warning(f"版本文件不存在: {file_path}") |
||||
return local_versions |
||||
|
||||
try: |
||||
with open(file_path, 'r', encoding='utf-8') as f: |
||||
for line in f: |
||||
line = line.strip() |
||||
if line: |
||||
# 提取行中的版本号格式 |
||||
version_match = re.match(r'(\d+\.\d+)', line) |
||||
if version_match: |
||||
main_version = version_match.group(1) |
||||
local_versions.append(main_version) |
||||
else: |
||||
# 如果整行不是版本号格式,也添加到列表中 |
||||
local_versions.append(line) |
||||
except Exception as e: |
||||
logger.error(f"读取版本文件失败: {e}") |
||||
|
||||
return local_versions |
||||
|
||||
def send_wechat_message(message): |
||||
"""通过企业微信webhook发送消息""" |
||||
# 从环境变量获取webhook_url,如果没有则使用默认值 |
||||
webhook_url = os.environ.get( |
||||
'WECHAT_WEBHOOK_URL', |
||||
'https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=4226c76e-725b-4990-b926-05f16142e513' |
||||
) |
||||
headers = {'Content-Type': 'application/json'} |
||||
|
||||
# 企业微信机器人消息格式 |
||||
payload = { |
||||
"msgtype": "text", |
||||
"text": { |
||||
"content": message |
||||
} |
||||
} |
||||
|
||||
try: |
||||
response = requests.post(webhook_url, headers=headers, data=json.dumps(payload), timeout=10) |
||||
response.raise_for_status() |
||||
result = response.json() |
||||
if result.get('errcode') != 0: |
||||
logger.error(f"发送企业微信消息失败: {result.get('errmsg')}") |
||||
except Exception as e: |
||||
logger.error(f"发送企业微信消息失败: {e}") |
||||
|
||||
def check_versions(): |
||||
"""检查并比较版本号""" |
||||
# 获取API返回的版本号 |
||||
api_data = get_firmware_list() |
||||
if not api_data: |
||||
return |
||||
|
||||
api_versions = extract_versions_from_api(api_data) |
||||
api_versions_set = set(api_versions) |
||||
|
||||
# 获取本地txt文件中的版本号 |
||||
local_versions = get_local_versions() |
||||
if not local_versions: |
||||
# 如果没有本地版本,创建一个示例txt文件 |
||||
create_example_txt() |
||||
return |
||||
|
||||
# 找出不在本地的版本号 |
||||
new_versions = [] |
||||
for api_version in api_versions_set: |
||||
if api_version not in local_versions: |
||||
new_versions.append(api_version) |
||||
|
||||
if new_versions: |
||||
logger.info(f"发现新固件版本: {', '.join(new_versions)}") |
||||
# 构建消息内容并发送到企业微信 |
||||
message = f"✅ 固件版本检查完成\n发现新的验证版本:\n" + "\n".join([f" - {v}" for v in new_versions]) |
||||
send_wechat_message(message) |
||||
else: |
||||
logger.info("所有固件版本已是最新") |
||||
|
||||
def create_example_txt(): |
||||
"""创建示例txt文件""" |
||||
example_path = os.path.join(os.getcwd(), 'firmware_versions.txt') |
||||
try: |
||||
with open(example_path, 'w', encoding='utf-8') as f: |
||||
f.write("26.1\n") |
||||
logger.info(f"已创建版本文件: {example_path}") |
||||
except Exception as e: |
||||
logger.error(f"创建版本文件失败: {e}") |
||||
|
||||
def main(): |
||||
"""主函数,支持定时执行""" |
||||
# 从环境变量读取执行间隔,默认30分钟 |
||||
interval_env = os.environ.get('CHECK_INTERVAL_MINUTES') |
||||
try: |
||||
interval_minutes = int(interval_env) if interval_env else 30 |
||||
except ValueError: |
||||
logger.warning(f"检查间隔参数无效,使用默认值30分钟") |
||||
interval_minutes = 30 |
||||
|
||||
interval_seconds = interval_minutes * 60 |
||||
logger.info(f"固件版本检查服务启动,间隔: {interval_minutes}分钟,当前时间: {time.strftime('%Y-%m-%d %H:%M:%S')}") |
||||
|
||||
# 首次立即执行一次 |
||||
check_versions() |
||||
|
||||
# 然后定时执行 |
||||
try: |
||||
while True: |
||||
logger.info(f"等待下次检查,{interval_minutes}分钟后...") |
||||
time.sleep(interval_seconds) |
||||
check_versions() |
||||
except KeyboardInterrupt: |
||||
logger.info("服务已停止") |
||||
except Exception as e: |
||||
logger.error(f"服务运行出错: {e}") |
||||
sys.exit(1) |
||||
|
||||
if __name__ == "__main__": |
||||
main() |
||||
Loading…
Reference in new issue