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"[ALERT] {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())