coordinate/france_visa_registration_bot.py

import time
import json
import os
import re
import uuid
import socket
import shutil
import random
import requests
import argparse
import concurrent.futures
import base64
from urllib.parse import urlencode
from datetime import datetime, timedelta
from typing import Optional, Dict
from DrissionPage.common import Keys
from DrissionPage import ChromiumPage, ChromiumOptions

import configure
from utils.cloudflare_bypass_for_scraping import CloudflareBypasser
from toolkit.vs_cloud_api import VSCloudApi
from toolkit.proxy_tunnel import ProxyTunnel
from vs_types import NotFoundError, PermissionDeniedError, RateLimiteddError, SessionExpiredOrInvalidError, BizLogicError 
from utils.mouse import HumanMouse
from utils.keyboard import HumanKeyboard
from utils.scroll import HumanScroll
from utils.fingerprint_utils import FingerprintGenerator
from toolkit.captcha_breaker import recognize_captcha_with_qwen

    
def load_proxies(pool_name):
    """从 config/proxies.json 读取对应的代理池"""
    config_path = os.path.join(os.path.dirname(__file__), 'config', 'proxies.json')
    try:
        with open(config_path, 'r', encoding='utf-8') as f:
            data = json.load(f)
            proxies = data.get(pool_name, [])
            if not proxies:
                raise ValueError(f"代理池 '{pool_name}' 为空或不存在！")
            return proxies
    except Exception as e:
        print(f"读取代理配置文件失败: {e}")
        exit(1) 

class FranceVisaRegistrator:
    def __init__(self, france_visa_url, proxy_config: Optional[Dict]=None, capsolver_key: Optional[str]=None, user_inputs: Optional[Dict]=None):
        self.proxy_config = proxy_config
        self.capsolver_key = capsolver_key
        self.user_inputs = user_inputs
        # 隔离的用户数据目录
        self.instance_id = uuid.uuid4().hex[:8]
        self.france_visa_url = france_visa_url
        # self.instance_id = '18d389e9'
        self.workspace = os.path.abspath(os.path.join("data/temp_browser_data", f"reg_session_{self.instance_id}"))
        self.page = None
        self.mouse = None
        self.keyboard = None
        
        # 持有隧道实例
        self.tunnel = None

    def _log(self, msg):
        now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        print(f"[{now}][TLS-Reg-{self.instance_id}] {msg}")

    def _get_free_port(self):
        """获取可用端口，防止 DrissionPage 解析日志报错"""
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(('', 0))
            return s.getsockname()[1]
        
    def save_screenshot(self, name_prefix):
        try:
            timestamp = int(time.time())
            filename = f"{self.instance_id}_{name_prefix}_{timestamp}.jpg"
            save_path = os.path.join("data", filename)
            os.makedirs("data", exist_ok=True)
            self.page.get_screenshot(path=save_path, full_page=False)
            self._log(f"Screenshot saved to {save_path}")
        except Exception as e:
            self._log(f"Failed to save screenshot: {e}")

    def init_browser(self):
        """初始化独立、配置好代理的浏览器环境"""
        self._log("Initializing browser...")
        co = ChromiumOptions()
        
        # 1. 端口与路径隔离
        port = self._get_free_port()
        co.set_local_port(port)
        co.set_user_data_path(self.workspace)
        
        chrome_path = configure.CHROME_PATH
        if not chrome_path:
            chrome_path = os.getenv("CHROME_BIN")
        if chrome_path and os.path.exists(chrome_path):
            co.set_paths(browser_path=chrome_path)
        
        # 2. 代理配置 (支持账号密码)
        if self.proxy_config and self.proxy_config.get("ip"):
            p = self.proxy_config
            if p.get("username") and p.get("password"):
                self.tunnel = ProxyTunnel(p['ip'], p['port'], p['username'], p['password'])
                local_proxy = self.tunnel.start()
                self._log(f"Tunnel started at {local_proxy}")
                co.set_argument(f'--proxy-server={local_proxy}')
            else:
                proxy_str = f"{p.get('proto', 'http')}://{p['ip']}:{p['port']}"
                co.set_argument(f'--proxy-server={proxy_str}')
        else:
            self._log("[WARN] No proxy configured!")

        fingerprint_gen = FingerprintGenerator()
        specific_fp = fingerprint_gen.generate(self.instance_id)
        self._log(f'browser fingerprint={specific_fp}')
        # 3. 反爬及稳定性配置
        co.headless(False)
        co.set_argument('--no-sandbox')
        co.set_argument('--lang=en-us')
        co.set_argument('--accept-lang=en-us')
        # co.set_argument('--disable-gpu')
        co.set_argument('--disable-dev-shm-usage')
        co.set_argument('--window-size=1920,1080')
        co.set_argument('--disable-blink-features=AutomationControlled')
        co.set_argument(f"--fingerprint={specific_fp.get('seed')}")
        co.set_argument(f"--fingerprint-platform={specific_fp.get('platform')}")
        co.set_argument(f"--fingerprint-brand={specific_fp.get('brand')}")
        self.page = ChromiumPage(co)
        self.page.get(self.france_visa_url)
        time.sleep(5)
        cf_bypasser = CloudflareBypasser(self.page, log=True)
        cf_bypasser.bypass(max_retry=8)
        time.sleep(3)
        cf_bypasser.handle_waiting_room()
        
        self._log("正在初始化拟人化工具...")
        self.mouse = HumanMouse(self.page, debug=True)
        self.keyboard = HumanKeyboard(self.page)
        self._log("随机化鼠标开始位置...")
        viewport_width = self.page.rect.viewport_size[0]
        viewport_height = self.page.rect.viewport_size[1]
        init_x = random.randint(10, viewport_width - 10)
        init_y = random.randint(10, viewport_height - 10)
        self.mouse.move(init_x, init_y)
        
    def register(self):
        username = self.user_inputs.get('username')
        first_name = self.user_inputs.get('first_name')
        last_name = self.user_inputs.get('last_name')
        password = f'Visafly@1234'
        self.page.wait.ele_deleted('tag:h2@@text():Log in to France-Visas', timeout=5)
        self.page.ele('tag:button@@text():Create an account').click()
        time.sleep(5)
        
        self.page.ele('tag:input@@name=lastName').input(last_name)
        self.page.ele('tag:input@@name=firstName').input(first_name)
        self.page.ele('tag:input@@name=email').input(username)
        self.page.ele('tag:input@@name=emailVerif').input(username)
        self.page.ele('tag:input@@name=password').input(password)
        self.page.ele('tag:input@@name=password-confirm').input(password)
        self.page.ele('tag:select@@name=ddeLanguage').select('English')
        captcha = self.page.ele('#captchaComponent').ele('tag:img')
        src = captcha.attr("src")
        print(src)
        base64_data = src.split(",")[1]
        with open("captcha.png", "wb") as f:
            f.write(base64.b64decode(base64_data))
        result = recognize_captcha_with_qwen("captcha.png", "sk-893e895724c6403d81374e515ffaf427")
        print(f'captcha result={result}')
        self.page.ele('tag:input@@name=captchaFormulaireExtInput').input(result)
        self.page.ele("tag:button@@text():Create an account").click()
        
    def activate(self, sent_at=None):
        username = self.user_inputs.get('username')
        email_box = 'visafly666@gmail.com'
        sender = 'noreply at interieur.gouv.fr'
        recipient = username
        subject_keywords = 'Create your France-Visas account'
        body_keywords = ''
        
        if not sent_at:
            now_utc = datetime.utcnow()
            sent_at = now_utc.strftime("%Y-%m-%d %H:%M:%S")
        
        content_out = VSCloudApi.Instance().fetch_mail_content(
            email=email_box,
            sender=sender,
            recipient=recipient,
            subject_keywords=subject_keywords,
            body_keywords=body_keywords,
            sent_date=sent_at,
            expiry=600
        )
        self._log(f'activate email content={content_out}')
        match = re.search(r'https://\S+', content_out)
        activate_link = match.group(0) if match else None
        self.page.get(activate_link)
        time.sleep(3)
    
    def make_account_useful(self):
        
        def fill_date_field(page, selector, date_str):
            if not date_str:
                return
                
            ele = page.ele(selector)
            ele.scroll.to_see(center=True)
            
            js_detect_format = """
                const parts = new Intl.DateTimeFormat().formatToParts(new Date(2023, 11, 31));
                let format = [];
                for (let part of parts) {
                    if (part.type === 'year') format.push('Y');
                    if (part.type === 'month') format.push('M');
                    if (part.type === 'day') format.push('D');
                }
                return format;
            """
            date_format = page.run_js(js_detect_format)
            
            year, month, day = date_str.split('-')
            date_dict = {
                'Y': year,
                'M': month.zfill(2),
                'D': day.zfill(2)
            }
            ele.click()
            time.sleep(0.1)
            page.actions.type(Keys.LEFT * 3)
            time.sleep(0.1)
            for i, char in enumerate(date_format):
                val = date_dict[char]
                page.actions.type(val)
                time.sleep(0.1)
                if char == 'Y':
                    if i < 2:
                        page.actions.type(Keys.RIGHT)
                        time.sleep(0.1)
                else:
                    pass
        
        passport_no = self.user_inputs.get('passport_no')
        passport_issue_date = self.user_inputs.get('passport_issue_date')
        passport_expiry_date = self.user_inputs.get('passport_expiry_date')
        nationality = self.user_inputs.get('nationality')
        passport_issue_from = self.user_inputs.get('passport_issue_from')
        self.page.ele('#formHeader:navigationLanguage_input').select('English')
        time.sleep(3)
        self.page.ele('#formAccueilUsager:ajouterGroupe').click()
        time.sleep(5)
        self.page.ele('#formStep1:visas-selected-nationality_input').select(nationality)
        self.page.ele('#formStep1:Visas-selected-deposit-country_input').select('Ireland')
        self.page.ele('#formStep1:Visas-selected-stayDuration_input').select('Short-stay (≤ 90 days)')
        self.page.ele('#formStep1:Visas-selected-destination_input').select('France')
        self.page.ele('#formStep1:Visas-selected-deposit-town_input').select('Dublin')
        self.page.ele('#formStep1:Visas-selected-authority_input').select(passport_issue_from)
        self.page.ele('#formStep1:Visas-dde-travel-document_input').select('Ordinary passport')
        self.page.ele('#formStep1:Visas-dde-travel-document-number').input(passport_no)
        fill_date_field(self.page, '#formStep1:Visas-dde-release_date_real_input', passport_issue_date)
        fill_date_field(self.page, '#formStep1:Visas-dde-expiration_date_input', passport_expiry_date)
        self.page.ele('#formStep1:Visas-selected-purposeCategory_input').select('Tourism')
        self.page.ele('#formStep1:Visas-selected-purpose_input').select('Tourism / Private visit')
        self.page.ele('#formStep1:btnVerifier').click()
        time.sleep(3)
        self.page.ele('#formStep1:btnSuivant').click()
        time.sleep(3)
        self.page.ele('#formStep1:btnValiderModal').click()
        time.sleep(3)
        self.page.ele('.iconeDDEIdPanel').click()
        time.sleep(0.5)
        self.page.ele('text():My applications').click()
        time.sleep(3)
        html_content = self.page.html
        match = re.search(r'FRA1[A-Z0-9]+', html_content)
        if not match:
            raise BizLogicError(message='FRA1 not found')
        fra_number = match.group(0)
        print(fra_number)
        return fra_number
    
    def cleanup(self):
        """清理浏览器进程和缓存文件夹"""
        self._log("Cleaning up resources...")
        if self.page:
            try: self.page.quit()
            except: pass
        if os.path.exists(self.workspace):
            time.sleep(1) # 等待文件锁释放
            shutil.rmtree(self.workspace, ignore_errors=True)
            

def main():
    france_visa_url = 'https://application-form.france-visas.gouv.fr/fv-fo-dde/'
    proxy_config = {
        'ip': '127.0.0.1',
        'port': 7890,
        'username': '',
        'password': ''
    }
    capsolver_key = ''
    user_inputs = {
        "username": "ManaliAshokGaikwad26@gmail-app.com",
        "first_name": "Manali Ashok",
        "last_name": "Gaikwad",
        "nationality": "Indian",
        "passport_issue_from": "India",
        "passport_no": "Z4413123",
        "passport_issue_date": "2018-01-15",
        "passport_expiry_date": "2028-01-14",
    }
    bot = FranceVisaRegistrator(
        france_visa_url, 
        proxy_config=proxy_config, 
        capsolver_key=capsolver_key, 
        user_inputs=user_inputs
    )
    bot.init_browser()
    now_utc = datetime.utcnow()
    sent_at = now_utc.strftime("%Y-%m-%d %H:%M:%S")
    bot.register() 
    bot.activate(sent_at=sent_at)
    bot.make_account_useful()

if __name__ == "__main__":
    main()