如何为AI代理收集签证预约数据

pip install git+https://github.com/capsolver-ai/capsolver-core.git
pip install aiohttp schedule python-telegram-bot

export CAPSOLVER_API_KEY="your-capsolver-api-key"

VISA_PORTALS = {
    "us_visa_cgifederal": {
        "name": "US Visa (CGI Federal)",
        "url": "https://www.usvisascheduling.com",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "site_key": "6Lc...",  # 通过浏览器检查识别
        "trigger": "login_page_and_date_selection",
        "check_interval_minutes": 10
    },
    "vfs_global_schengen": {
        "name": "VFS Global (Schengen)",
        "url": "https://visa.vfsglobal.com",
        "captcha_type": "AntiTurnstileTaskProxyLess",
        "site_key": "0x4A...",
        "trigger": "appointment_search",
        "check_interval_minutes": 15
    },
    "uk_visa_tls": {
        "name": "UK Visa (TLS Contact)",
        "url": "https://visas-immigration.service.gov.uk",
        "captcha_type": "ReCaptchaV2TaskProxyLess",
        "site_key": "6Le...",
        "trigger": "every_page_load",
        "check_interval_minutes": 10
    }
}

import asyncio
import aiohttp
import time
from capsolver_core import create_capsolver, CaptchaType, CaptchaInfo

class VisaAppointmentMonitor:
    """监控签证预约可用性并处理CAPTCHA。"""
    
    def __init__(self, api_key: str):
        self.cap = create_capsolver(api_key=api_key)
        self.session = None
        self.last_check = {}
    
    async def solve_portal_captcha(self, portal_config: dict) -> str:
        """在签证门户上解决CAPTCHA并返回令牌。"""
        captcha_type_map = {
            "ReCaptchaV2TaskProxyLess": CaptchaType.RECAPTCHA_V2,
            "AntiTurnstileTaskProxyLess": CaptchaType.CLOUDFLARE,
            "ReCaptchaV3TaskProxyLess": CaptchaType.RECAPTCHA_V3
        }
        
        info = CaptchaInfo(
            type=captcha_type_map[portal_config["captcha_type"]],
            website_url=portal_config["url"],
            website_key=portal_config["site_key"]
        )
        
        solution = await self.cap.solve(info)
        return solution.token
    
    async def check_availability(self, portal_key: str) -> dict:
        """在特定门户上检查预约可用性。"""
        portal = VISA_PORTALS[portal_key]
        
        try:
            # 解决CAPTCHA以访问门户
            token = await self.solve_portal_captcha(portal)
            
            # 使用令牌进行身份验证并检查时段
            # （实现因门户而异）
            availability = await self._fetch_slots(portal, token)
            
            self.last_check[portal_key] = {
                "timestamp": time.time(),
                "slots_found": len(availability),
                "status": "success"
            }
            
            return {
                "portal": portal["name"],
                "available_slots": availability,
                "checked_at": time.strftime("%Y-%m-%d %H:%M:%S")
            }
            
        except Exception as e:
            self.last_check[portal_key] = {
                "timestamp": time.time(),
                "status": "error",
                "error": str(e)
            }
            return {"portal": portal["name"], "error": str(e)}
    
    async def _fetch_slots(self, portal: dict, captcha_token: str) -> list:
        """使用解决的令牌获取可用预约时段。"""
        if not self.session:
            self.session = aiohttp.ClientSession()
        
        # 提交CAPTCHA令牌并检索预约数据
        headers = {"Content-Type": "application/x-www-form-urlencoded"}
        data = {
            "g-recaptcha-response": captcha_token,
            # 每个门户特有的表单字段
        }
        
        async with self.session.post(
            f"{portal['url']}/appointment/check",
            data=data, headers=headers
        ) as resp:
            result = await resp.json()
            return result.get("available_dates", [])
    
    async def close(self):
        if self.session:
            await self.session.close()
        await self.cap.aclose()

import asyncio
from datetime import datetime

class VisaSlotAlertSystem:
    """持续监控并即时警报。"""
    
    def __init__(self, monitor: VisaAppointmentMonitor):
        self.monitor = monitor
        self.alert_callbacks = []
    
    def add_alert(self, callback):
        self.alert_callbacks.append(callback)
    
    async def run_monitoring_loop(self, portals: list, interval_minutes: int = 10):
        """为指定的门户运行持续监控。"""
        print(f"开始{len(portals)}个门户的签证监控...")
        
        while True:
            for portal_key in portals:
                result = await self.monitor.check_availability(portal_key)
                
                if result.get("available_slots"):
                    # 发现时段 - 触发警报
                    await self._send_alerts(result)
                    print(f"[警报] 在{result['portal']}发现时段！")
                else:
                    print(f"[{datetime.now().strftime('%H:%M')}] "
                          f"{result.get('portal', portal_key)}: 无时段")
                
                # 门户检查之间的简短延迟
                await asyncio.sleep(5)
            
            # 等待下一次循环
            await asyncio.sleep(interval_minutes * 60)
    
    async def _send_alerts(self, result):
        for callback in self.alert_callbacks:
            await callback(result)

# Telegram通知示例
async def telegram_alert(result):
    """当发现时段时发送Telegram通知。"""
    message = (
        f"🎫 签证预约可用！\n"
        f"门户: {result['portal']}\n"
        f"时段: {len(result['available_slots'])}\n"
        f"日期: {', '.join(result['available_slots'][:5])}\n"
        f"检查时间: {result['checked_at']}"
    )
    # 通过Telegram Bot API发送
    print(message)

# 运行系统
async def main():
    monitor = VisaAppointmentMonitor(api_key="YOUR_CAPSOLVER_API_KEY")
    alert_system = VisaSlotAlertSystem(monitor)
    alert_system.add_alert(telegram_alert)
    
    await alert_system.run_monitoring_loop(
        portals=["us_visa_cgifederal", "vfs_global_schengen"],
        interval_minutes=10
    )

asyncio.run(main())