Monitor whether there are slot on the iOS TestFlight

testflight_monitor.py

import requests
import time
import re
import logging
import asyncio
from telegram.ext import Application
import argparse
from datetime import datetime

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler("testflight_monitor.log"),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

class TestFlightMonitor:
    def __init__(self, telegram_token, chat_id, check_interval=60):
        """
        Initializes the TestFlight monitor.
        
        Args:
            telegram_token (str): Telegram Bot API token.
            chat_id (str): Telegram chat ID to send notifications to.
            check_interval (int): Check interval in seconds.
        """
        self.telegram_token = telegram_token
        self.chat_id = chat_id
        self.check_interval = check_interval
        self.application = Application.builder().token(telegram_token).build()
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.114 Safari/537.36',
            'Accept-Language': 'en-US,en;q=0.9',
        }
        
    def check_testflight_link(self, testflight_url):
        """
        Checks if there are open slots for a TestFlight link.
        
        Args:
            testflight_url (str): TestFlight beta link.
            
        Returns:
            tuple: (bool, str) - Whether there are open slots and the app name.
        """
        try:
            response = requests.get(testflight_url, headers=self.headers)
            
            if response.status_code != 200:
                logger.error(f"请求失败: {response.status_code}")
                return False, None
            
            # Extract app name for notification
            app_name = "Unknown App"
            name_match = re.search(r'<title>Join the (.*) beta - TestFlight - Apple</title>', response.text)
            if name_match:
                app_name = name_match.group(1)
            
            # Check if the beta is full
            if "This beta is full" in response.text or "已满" in response.text:
                logger.info(f"Beta is full: {testflight_url}")
                return False, app_name
            
            # If "This beta is full" text is not detected, there might be open slots
            logger.info(f"可能有空位! 应用: {app_name}, 链接: {testflight_url}")
            return True, app_name
            
        except Exception as e:
            logger.error(f"检查TestFlight链接时出错: {e}")
            return False, None
    
    async def send_telegram_notification(self, testflight_url, app_name="未知应用"):
        """
        发送Telegram通知
        
        Args:
            testflight_url (str): 有空位的TestFlight链接
            app_name (str): 应用名称
        """
        try:
            message = f"🎉 TestFlight slot available!\n\n"
            message += f"📱 App: {app_name}\n"
            message += f"🔗 Link: {testflight_url}\n"
            message += f"⏰ 检测时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
            
            await self.application.bot.send_message(chat_id=self.chat_id, text=message)
            logger.info(f"已发送Telegram通知")
        except Exception as e:
            logger.error(f"发送Telegram通知时出错: {e}")
    
    async def monitor(self, testflight_urls):
        """
        持续监控TestFlight链接
        
        Args:
            testflight_urls (list): 要监控的TestFlight链接列表
        """
        if isinstance(testflight_urls, str):
            testflight_urls = [testflight_urls]
            
        logger.info(f"开始监控 {len(testflight_urls)} 个TestFlight链接，检查间隔: {self.check_interval}秒")
        
        # 异步启动应用但不阻塞
        await self.application.initialize()
        
        while True:
            for url in testflight_urls:
                has_space, app_name = self.check_testflight_link(url)
                if has_space:
                    await self.send_telegram_notification(url, app_name)
                    
            await asyncio.sleep(self.check_interval)
            
    async def shutdown(self):
        """关闭Telegram应用"""
        await self.application.shutdown()

async def main_async(args):
    monitor = TestFlightMonitor(
        telegram_token=args.token,
        chat_id=args.chat_id,
        check_interval=args.interval
    )
    
    try:
        await monitor.monitor(args.urls)
    except KeyboardInterrupt:
        logger.info("程序被用户中断")
        await monitor.shutdown()
    except Exception as e:
        logger.error(f"程序出错: {e}")
        await monitor.shutdown()

def main():
    parser = argparse.ArgumentParser(description='监控TestFlight链接是否有空位')
    parser.add_argument('--token', required=True, help='Telegram Bot API令牌')
    parser.add_argument('--chat_id', required=True, help='Telegram聊天ID')
    parser.add_argument('--urls', required=True, nargs='+', help='要监控的TestFlight链接')
    parser.add_argument('--interval', type=int, default=60, help='检查间隔(秒)，默认60秒')
    
    args = parser.parse_args()
    
    # 运行异步主函数
    asyncio.run(main_async(args))

if __name__ == "__main__":
    main()