data-sync-service/scripts/check_expiry.py

#!/usr/bin/env python3
"""
过期检测脚本 - 定时检查数据过期状态并发送微信通知
"""

import json
from pathlib import Path
from datetime import datetime

DATA_DIR = Path('/home/hermes/projects/data-sync-service/data')
INDEX_FILE = DATA_DIR / 'index.json'


def load_index():
    """读取索引文件"""
    if not INDEX_FILE.exists():
        return {"tags": {}}
    
    with open(INDEX_FILE, 'r', encoding='utf-8') as f:
        return json.load(f)


def load_data_file(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 check_expiry():
    """检查所有数据的过期状态"""
    index = load_index()
    
    tags = index.get('tags', {})
    if not tags:
        print("📊 暂无数据")
        return
    
    expired_list = []
    warning_list = []
    valid_list = []
    
    for tag, info in tags.items():
        expires_at_str = info.get('expires_at', '')
        if not expires_at_str:
            continue
        
        try:
            expires_at = datetime.strptime(expires_at_str, '%Y-%m-%d %H:%M:%S')
            now = datetime.now()
            remaining = (expires_at - now).total_seconds()
            remaining_hours = remaining / 3600
            remaining_days = remaining_hours / 24
            
            # 读取完整数据（获取 login_url）
            file_data = load_data_file(info['file'])
            login_url = ''
            if file_data and 'metadata' in file_data:
                login_url = file_data['metadata'].get('login_url', '')
            
            item = {
                'tag': tag,
                'type': info.get('type', 'unknown'),
                'expires_at': expires_at_str,
                'remaining_days': remaining_days,
                'login_url': login_url
            }
            
            if remaining < 0:
                expired_list.append(item)
            elif remaining < 2 * 24 * 3600:  # 少于2天
                warning_list.append(item)
            else:
                valid_list.append(item)
        
        except Exception as e:
            print(f"⚠️ 解析时间失败: {tag} - {e}")
    
    # 发送通知
    has_notification = False
    
    # 1. 紧急通知（已过期）
    for item in expired_list:
        has_notification = True
        send_expired_notification(item)
        print("")
    
    # 2. 提醒通知（即将过期）
    for item in warning_list:
        has_notification = True
        send_warning_notification(item)
        print("")
    
    # 3. 汇总统计
    if not has_notification:
        print("✅ 所有数据状态正常")
        print(f"\n有效数据：{len(valid_list)} 个")
        for item in valid_list:
            print(f"  - {item['tag']} (剩余 {item['remaining_days']:.1f} 天)")
    else:
        print(f"📊 数据状态统计")
        print(f"  ❌ 已过期: {len(expired_list)} 个")
        print(f"  ⚠️ 即将过期: {len(warning_list)} 个")
        print(f"  ✅ 正常: {len(valid_list)} 个")


def send_expired_notification(item):
    """发送过期通知"""
    tag = item['tag']
    data_type = item['type']
    expires_at = item['expires_at']
    login_url = item['login_url']
    
    # 生成配置 JSON
    config = {
        "serverUrl": "http://47.122.126.244:5001",
        "password": "admin123123123",
        "tag": tag,
        "type": data_type
    }
    config_json = json.dumps(config, indent=2, ensure_ascii=False)
    
    message = f"""❌ 数据已过期

Tag: {tag}
类型: {data_type}
过期时间: {expires_at}

请重新登录同步：
{login_url if login_url else '(无登录链接)'}

配置信息（点击复制）：
```json
{config_json}
```"""
    
    print(message)


def send_warning_notification(item):
    """发送提醒通知"""
    tag = item['tag']
    data_type = item['type']
    expires_at = item['expires_at']
    remaining_days = item['remaining_days']
    login_url = item['login_url']
    
    message = f"""⚠️ 数据即将过期

Tag: {tag}
类型: {data_type}
剩余时间: {remaining_days:.1f} 天
过期时间: {expires_at}

建议重新登录同步：
{login_url if login_url else '(无登录链接)'}"""
    
    print(message)


if __name__ == '__main__':
    check_expiry()