K-hermes
edf4dfcd52
- 基于OpenResty的RESTful API服务 - 支持Cookie/Token等数据类型的存储和管理 - 登录认证保护 - Web管理界面 - 数据过期管理
146 lines
3.7 KiB
Python
Executable File
146 lines
3.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
通用数据同步服务 - 过期检测脚本
|
|
每天检查所有数据的过期状态，发送微信通知
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
|
|
DATA_DIR = Path("/home/hermes/projects/data-sync-service/data")
|
|
SERVER_URL = "http://47.122.126.244:5001"
|
|
|
|
|
|
def parse_datetime(dt_str):
|
|
"""解析时间字符串"""
|
|
try:
|
|
return datetime.strptime(dt_str, "%Y-%m-%d %H:%M:%S")
|
|
except:
|
|
return None
|
|
|
|
|
|
def load_index():
|
|
"""加载索引文件"""
|
|
index_file = DATA_DIR / "index.json"
|
|
if not index_file.exists():
|
|
return {"tags": []}
|
|
|
|
with open(index_file, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
|
|
def load_data(filename):
|
|
"""加载数据文件"""
|
|
filepath = DATA_DIR / filename
|
|
if not filepath.exists():
|
|
return None
|
|
|
|
with open(filepath, 'r', encoding='utf-8') as f:
|
|
return json.load(f)
|
|
|
|
|
|
def format_config_block(tag, data_type):
|
|
"""格式化配置代码块"""
|
|
config = {
|
|
"serverUrl": SERVER_URL,
|
|
"password": "admin123123123",
|
|
"tag": tag,
|
|
"type": data_type
|
|
}
|
|
return json.dumps(config, indent=2, ensure_ascii=False)
|
|
|
|
|
|
def send_expired_notification(tag, data_type, expires_at, login_url):
|
|
"""发送过期通知"""
|
|
config_json = format_config_block(tag, data_type)
|
|
|
|
message = f"""❌ 数据已过期
|
|
|
|
Tag: {tag}
|
|
类型: {data_type}
|
|
过期时间: {expires_at}
|
|
|
|
请重新登录同步：
|
|
{login_url}
|
|
|
|
配置信息（点击复制）：
|
|
```json
|
|
{config_json}
|
|
```"""
|
|
|
|
print(message)
|
|
|
|
|
|
def send_expiring_soon_notification(tag, data_type, remaining_days, login_url):
|
|
"""发送即将过期通知"""
|
|
message = f"""⚠️ 数据即将过期
|
|
|
|
Tag: {tag}
|
|
类型: {data_type}
|
|
剩余时间: {remaining_days:.1f} 天
|
|
|
|
建议重新登录同步：
|
|
{login_url}"""
|
|
|
|
print(message)
|
|
|
|
|
|
def check_expiry():
|
|
"""检查所有数据的过期状态"""
|
|
index = load_index()
|
|
now = datetime.now()
|
|
|
|
expired_count = 0
|
|
expiring_soon_count = 0
|
|
|
|
for item in index.get("tags", []):
|
|
tag = item.get("tag")
|
|
data_type = item.get("type", "unknown")
|
|
expires_at_str = item.get("expires_at")
|
|
filename = item.get("file")
|
|
|
|
if not expires_at_str or not filename:
|
|
continue
|
|
|
|
# 解析过期时间
|
|
expires_at = parse_datetime(expires_at_str)
|
|
if not expires_at:
|
|
continue
|
|
|
|
# 加载数据获取登录链接
|
|
data = load_data(filename)
|
|
login_url = "http://47.122.126.244:5001"
|
|
if data and data.get("metadata"):
|
|
login_url = data["metadata"].get("login_url", login_url)
|
|
|
|
# 计算剩余时间
|
|
time_diff = expires_at - now
|
|
remaining_seconds = time_diff.total_seconds()
|
|
remaining_days = remaining_seconds / 86400
|
|
|
|
# 检查状态
|
|
if remaining_seconds < 0:
|
|
# 已过期
|
|
expired_count += 1
|
|
send_expired_notification(tag, data_type, expires_at_str, login_url)
|
|
elif remaining_days < 2:
|
|
# 即将过期（<2天）
|
|
expiring_soon_count += 1
|
|
send_expiring_soon_notification(tag, data_type, remaining_days, login_url)
|
|
|
|
# 发送汇总（仅当有问题时）
|
|
if expired_count > 0 or expiring_soon_count > 0:
|
|
summary = f"\n📊 检查完成\n\n"
|
|
if expired_count > 0:
|
|
summary += f"❌ 已过期: {expired_count} 个\n"
|
|
if expiring_soon_count > 0:
|
|
summary += f"⚠️ 即将过期: {expiring_soon_count} 个\n"
|
|
print(summary)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
check_expiry()
|