import time import json import inspect import requests import uuid import random import urllib3 import asyncio import aiohttp import logging from datetime import datetime, timedelta # from book_data_buiilder import troov_dublin_visas_book_data_builder # 禁止显示 urllib3 的 SSL 警告信息 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) logger = logging.getLogger(__name__) TROOV_EMBASSY = { # 都柏林 'TroovFr_Dublin_Visas': { 'code': 'ambassade-de-france-en-irlande', 'name': 'Visas', 'teamId': '621540d353069dec25bd0045', 'zoneId': '624317926863643fe83c8548', 'chooseService': 'Visas', 'submitCountry': 'IE', 'submitCity': 'DUB', 'travelCountry': 'FR', 'visaCategory': 'Visas', 'website': 'https://consulat.gouv.fr/en/ambassade-de-france-en-irlande/appointment?name=Visas' }, # 日本 - 测试环境 'TroovFr_Tokyo_Visa': { 'code': 'ambassade-de-france-a-tokyo', 'name': 'Visa', 'teamId': '6238b4dfb1e5a274ff4c0f09', 'zoneId': '6242f7463f8ce81dec596054', 'chooseService': 'Visa', 'submitCountry': 'JP', 'submitCity': 'TYO', 'travelCountry': 'FR', 'visaCategory': 'Visa', 'website': 'https://consulat.gouv.fr/en/ambassade-de-france-a-tokyo/appointment?name=Visa' }, } class SessionExpiredError(Exception): """会话过期异常""" def __init__(self, message="SESSION EXPIRED"): super().__init__(message) def check_responsed_session_expired(resp_text: str): """检查响应内容是否包含会话过期标识，如有则抛出异常""" if "SESSION_EXPIRED" in resp_text or "SESSION_NOT_FOUND" in resp_text: raise SessionExpiredError("SESSION EXPIRED OR NOT FOUND") def is_session_remaining_life_zero(session_dic, max_lifetime_minutes=5): def should_expire(probability=0.6): """以给定概率返回 True（表示失效），否则返回 False""" return random.random() < probability now = datetime.utcnow() now_hour, now_minute = now.hour, now.minute session_time = datetime.strptime(session_dic['session_create_at'], "%Y-%m-%dT%H:%M:%S.%fZ") session_hour, session_minute = session_time.hour, session_time.minute # 🕒 计算 session 已存活时长（分钟） elapsed_minutes = (now - session_time).total_seconds() / 60 # ✅ 新增逻辑：session 超过 n 分钟就强制过期 if elapsed_minutes >= max_lifetime_minutes: return True # 🧠 原逻辑保持不变 if (now_minute >= 45 and session_hour == now_hour and 45 <= session_minute < 60) or \ (now_minute < 5 and session_hour == now_hour) or \ (now_minute >= 5 and now_minute < 45): return False expired = should_expire() return expired async def troov_handshake(embassy, proxy): url = "https://51.254.177.49/api/handshake" headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'origin': 'https://consulat.gouv.fr', 'referer': embassy['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-web': 'fr.gouv.consulat' } session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) try: async with session.head(url, headers=headers, proxy=proxy) as response: if response.status == 200: return ( response.headers.get('X-Gouv-App-Id'), response.headers.get('X-Gouv-Handshake') ) return None finally: await session.close() async def troov_create_session(proxy, captcha, embassy=TROOV_EMBASSY['TroovFr_Dublin_Visas']): handshake_ret = await troov_handshake(embassy, proxy) if not handshake_ret: logger.error(f'troov_handshake failed') return None x_gouv_app_id, x_gouv_handshake = handshake_ret reservation_session_ret = await troov_make_reservation_session(embassy, proxy, captcha, x_gouv_app_id, x_gouv_handshake) if not reservation_session_ret: logger.error(f'troov_make_reservation_session failed') return None x_gouv_handshake2, session_create_at, session_id = reservation_session_ret session_dic = { 'embassy': embassy, 'x_gouv_app_id':x_gouv_app_id, 'x-csrf-token': x_gouv_handshake2, 'session_create_at': session_create_at, 'session_id': session_id } status = await troov_update_dynamic_steps(proxy, session_dic) if status: return session_dic logger.error(f'troov_update_dynamic_steps failed') return None async def troov_refresh_session(proxy, session_dic): handshake_ret = await troov_handshake(proxy, session_dic['center']) if not handshake_ret: return None x_gouv_app_id, x_gouv_handshake = handshake_ret session_dic['x_gouv_app_id'] = x_gouv_app_id session_dic['x-csrf-token'] = x_gouv_handshake reservation_session_ret = await troov_get_reservation_session(proxy, session_dic) if not reservation_session_ret: return None return session_dic async def troov_make_reservation_session(embassy, proxy, capcha_str, handshake_gouv_appid, handshake_gouv_handshake): url = f"https://51.254.177.49/api/team/{embassy['teamId']}/reservations-session" payload = json.dumps({ "standaloneServiceName": embassy["name"], "sessionId": None, "captcha": capcha_str }) headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'content-type': 'application/json', 'origin': f'https://consulat.gouv.fr', 'referer': embassy["website"], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-csrf-token': handshake_gouv_handshake, 'x-gouv-app-id': handshake_gouv_appid, 'x-gouv-web': 'fr.gouv.consulat' } session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) try: async with session.post(url, headers=headers, data=payload, proxy=proxy) as response: response_text = await response.text() if response.status == 200: response_dic = json.loads(response_text) return response.headers['X-Gouv-Handshake'], response_dic['created_at'], response_dic['_id'] logger.error(f'troov_make_reservation_session {response.status}, {response_text}') return None finally: await session.close() async def troov_get_reservation_session(proxy, session_dic): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations-session" headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'origin': 'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat', } params = { 'sessionId': session_dic['session_id'], 'standaloneServiceName': session_dic['embassy']['name'] } session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) try: async with session.get(url, params=params, headers=headers, proxy=proxy) as response: if response.status == 200: return await response.json() else: rtext = await response.text() check_responsed_session_expired(rtext) return None finally: await session.close() async def troov_update_dynamic_steps(proxy, session_dic): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations-session/{session_dic['session_id']}/update-dynamic-steps" payload = json.dumps({ "key": "slotsSteps", "steps": [ { "stepType": "slotsStep", "name": "Visas", "numberOfSlots": 1, "dynamicStepIndex": 0, "zone_id": session_dic['embassy']['zoneId'], "value": { "lastSelectedDate": "", "label": session_dic['embassy']['name'], "accessibleCalendar": False, "hasSwitchedCalendar": False, "slots": {} } } ] }) headers = { 'accept': 'application/json, text/plain, */*', 'content-type': 'application/json', 'origin': f'https://consulat.gouv.fr', 'referer': session_dic["embassy"]["website"], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat', } session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) try: async with session.post(url, headers=headers, data=payload, proxy=proxy) as response: if response.status == 200: return True else: rtext = await response.text() check_responsed_session_expired(rtext) logger.error(f'troov_make_reservation_session {response.status}, {rtext}') return False finally: await session.close() async def troov_update_step_value(proxy, session_dic, slot_datetime_list: list): slots_data = {} for slot_datetime in slot_datetime_list: slot_date = slot_datetime['date'] slot_time = slot_datetime['time'] if slot_date not in slots_data: slots_data[slot_date] = [] _ = { "time": slot_time, "rate": "0.00", "capacity": 1, "numberOfApplicants": 1, "date": slot_date } slots_data[slot_date].append(_) url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations-session/{session_dic['session_id']}/update-step-value" payload = json.dumps({ "key": "slotsStep", "value": { "lastSelectedDate": list(slots_data.keys())[-1], "label": session_dic['embassy']['name'], "accessibleCalendar": False, "hasSwitchedCalendar": False, "slots": slots_data }, "stepIndex": 2, "dynamicStepIndex": 0 }) headers = { 'accept': 'application/json, text/plain, */*', 'content-type': 'application/json', 'origin': 'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat', } timeout = aiohttp.ClientTimeout(total=10) # 为每个请求创建全新的连接，不使用连接池 connector = aiohttp.TCPConnector(ssl=False, limit=1, force_close=True) async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session: async with session.post(url, headers=headers, data=payload, proxy=proxy) as response: rtext = await response.text() logger.info(f'update-step-value retcode={response.status}') if response.status == 200: logger.info(f'update-step-value retcode={response.status}, text={rtext}') return slot_datetime_list if response.status == 404: check_responsed_session_expired(rtext) response_dic = json.loads(rtext) taken_slots = response_dic.get('message', {}).get('takenSlots', []) taken_slot_datetimes = [] for ts in taken_slots: sdt = { "date": ts['slotDate'].split('T')[0], "time": ts['slotDate'].split('T')[1] } taken_slot_datetimes.append(sdt) return [x for x in slot_datetime_list if x not in taken_slot_datetimes] return None async def troov_get_exclude_days(proxy, session_dic): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations/exclude-days" payload = json.dumps({ "session": { session_dic['embassy']['zoneId']: True }, "sessionId": session_dic['session_id'] }) headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'content-type': 'application/json', 'origin': f'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat' } session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) try: async with session.post(url, headers=headers, data=payload, proxy=proxy) as response: if response.status == 200: return await response.json() else: rtext = await response.text() check_responsed_session_expired(rtext) return None finally: await session.close() async def troov_get_interval(proxy, session_dic): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations/get-interval?serviceId={session_dic['embassy']['zoneId']}" headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'origin': f'https://consulat.gouv.fr', 'referer': session_dic["embassy"]["website"], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat' } async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) as session: async with session.get(url, headers=headers, proxy=proxy) as response: if response.status == 200: return await response.json() else: rtext = await response.text() check_responsed_session_expired(rtext) return None async def troov_get_available_times(proxy, session_dic, date, places=1, capacity=2): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations/availability?name=Visas&date={date}&places={places}&matching=&maxCapacity={capacity}&sessionId={session_dic['session_id']}" headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'origin': f'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat' } async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) as session: async with session.get(url, headers=headers, proxy=proxy) as response: if response.status == 200: return await response.json() else: rtext = await response.text() check_responsed_session_expired(rtext) return None # async def troov_book(proxy, session_dic, date, slot, uinfo, captcha): # url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations/family" # book_body = troov_dublin_visas_book_data_builder(session_dic['session_id'], date, slot, uinfo, captcha) # payload = json.dumps(book_body) # headers = { # 'accept': 'application/json, text/plain, */*', # 'content-type': 'application/json', # 'origin': f'https://consulat.gouv.fr', # 'referer': session_dic['embassy']['website'], # 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', # 'x-csrf-token': session_dic['x-csrf-token'], # 'x-gouv-app-id': session_dic['x_gouv_app_id'], # 'x-gouv-web': 'fr.gouv.consulat', # } # session = aiohttp.ClientSession(connector=aiohttp.TCPConnector(ssl=False)) # try: # async with session.post(url, headers=headers, data=payload, proxy=proxy) as response: # if response.status == 200: # resp_text = await response.text() # if "qrcode" in resp_text: # return resp_text # else: # rtext = await response.text() # check_responsed_session_expired(rtext) # return None # finally: # await session.close() async def troov_get_open_days(proxy, session_dic): def get_dates_in_range(date_range): start_date = datetime.strptime(date_range["start"], "%Y-%m-%d") end_date = datetime.strptime(date_range["end"], "%Y-%m-%d") dates_list = [] current_date = start_date while current_date <= end_date: dates_list.append(current_date.strftime("%Y-%m-%d")) current_date += timedelta(days=1) return dates_list # 🚀 并发执行两个请求 interval, exclude_days = await asyncio.gather( troov_get_interval(proxy, session_dic), troov_get_exclude_days(proxy, session_dic) ) # logger.info(f'interval={interval}, exclude_days={exclude_days}') if interval and exclude_days: all_days = get_dates_in_range(interval) available_dates = [ day for day in all_days if day not in exclude_days and datetime.strptime(day, "%Y-%m-%d").weekday() not in (5, 6) ] return available_dates return None async def concurrency_full_coverage_lock(proxies, session_dic, slot_datetime_list, timeout=5): start_time = time.perf_counter() sem = asyncio.Semaphore(len(slot_datetime_list)) async def one_call(slot_datetime: dict): async with sem: t0 = time.perf_counter() try: result = await asyncio.wait_for( troov_update_step_value(random.choice(proxies), session_dic, [slot_datetime]), timeout=timeout ) locked = bool(result) status = "locked" if locked else "success" except asyncio.TimeoutError: locked = None status = "timeout" except SessionExpiredError: # ⚠️ 不吞掉这个异常，直接向上传递 raise except Exception as e: locked = None status = f"error: {repr(e)}" print(f'e={e}') finally: t1 = time.perf_counter() duration = t1 - t0 return { "slot_datetime": slot_datetime, "locked": locked, "duration": duration, "status": status } try: tasks = [one_call(slot_datetime) for slot_datetime in slot_datetime_list] results = await asyncio.gather(*tasks) except SessionExpiredError: # 🚨 如果任意任务抛出 SessionExpiredError，这里会捕获到 raise # 交由外部处理 end_time = time.perf_counter() total_time = end_time - start_time # 统计结果 locked_count = sum(1 for r in results if r["locked"] is True) success_count = sum(1 for r in results if r["locked"] is False) error_count = sum(1 for r in results if r["locked"] is None) avg_task_time = sum(r["duration"] for r in results) / len(results) if results else 0 locked_datetimes = [r['slot_datetime'] for r in results if r["locked"] is True] return { "total": len(results), "success": success_count, "locked": locked_count, "error": error_count, "slot_locked": locked_count > 0, "elapsed_time": total_time, "avg_task_time": avg_task_time, "session_dic": session_dic, "locked_datetimes": locked_datetimes } async def concurrency_lock(proxies, session_dic, slot_datetime_list, num_of_concurrency_size: int, timeout=5): start_time = time.perf_counter() sem = asyncio.Semaphore(num_of_concurrency_size) async def one_call(task_id: int): async with sem: t0 = time.perf_counter() try: result = await asyncio.wait_for( troov_update_step_value(random.choice(proxies), session_dic, slot_datetime_list), timeout=timeout ) locked = bool(result) status = "locked" if locked else "success" except asyncio.TimeoutError: locked = None status = "timeout" except SessionExpiredError: # ⚠️ 不吞掉这个异常，直接向上传递 raise except Exception as e: locked = None status = f"error: {repr(e)}" finally: t1 = time.perf_counter() duration = t1 - t0 return { "id": task_id, "locked": locked, "duration": duration, "status": status } try: tasks = [one_call(i) for i in range(num_of_concurrency_size)] results = await asyncio.gather(*tasks) except SessionExpiredError: # 🚨 如果任意任务抛出 SessionExpiredError，这里会捕获到 raise # 交由外部处理 end_time = time.perf_counter() total_time = end_time - start_time # 统计结果 locked_count = sum(1 for r in results if r["locked"] is True) success_count = sum(1 for r in results if r["locked"] is False) error_count = sum(1 for r in results if r["locked"] is None) avg_task_time = sum(r["duration"] for r in results) / len(results) if results else 0 return { "total": len(results), "success": success_count, "locked": locked_count, "error": error_count, "slot_locked": locked_count > 0, "elapsed_time": total_time, "avg_task_time": avg_task_time, "session_dic": session_dic } ########################################################!!!同步函数!!!######################################################## def troov_handshake_old(embassy, proxy): url = "https://51.254.177.49/api/handshake" headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'origin': 'https://consulat.gouv.fr', 'referer': embassy["website"], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-web': 'fr.gouv.consulat' } proxies = { "http": proxy, "https": proxy } response = requests.head(url, headers=headers, proxies=proxies, verify=False) if response.status_code == 200: return ( response.headers.get('X-Gouv-App-Id'), response.headers.get('X-Gouv-Handshake') ) return None def troov_create_session_old(proxy, captcha, embassy=TROOV_EMBASSY["TroovFr_Dublin_Visas"]): handshake_ret = troov_handshake_old(embassy, proxy) if not handshake_ret: return None x_gouv_app_id, x_gouv_handshake = handshake_ret reservation_session_ret = troov_make_reservation_session_old(embassy, proxy, captcha, x_gouv_app_id, x_gouv_handshake) if not reservation_session_ret: return None x_gouv_handshake2, session_create_at, session_id = reservation_session_ret session_dic = { 'embassy': embassy, 'x_gouv_app_id':x_gouv_app_id, 'x-csrf-token': x_gouv_handshake2, 'session_create_at': session_create_at, 'session_id': session_id } status = troov_update_dynamic_steps_old(proxy, session_dic) if status: return session_dic return None def troov_refresh_session_old(proxy, session_dic): handshake_ret = troov_handshake_old(session_dic['embassy'], proxy) if not handshake_ret: return None x_gouv_app_id, x_gouv_handshake = handshake_ret session_dic['x_gouv_app_id'] = x_gouv_app_id session_dic['x-csrf-token'] = x_gouv_handshake reservation_session_ret = troov_get_reservation_session_old(proxy, session_dic) if not reservation_session_ret: return None return session_dic def troov_make_reservation_session_old(embassy, proxy, capcha_str, handshake_gouv_appid, handshake_gouv_handshake): url = f"https://51.254.177.49/api/team/{embassy['teamId']}/reservations-session" payload = json.dumps({ "standaloneServiceName": embassy['name'], "sessionId": None, "captcha": capcha_str }) headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'content-type': 'application/json', 'origin': f'https://consulat.gouv.fr', 'referer': embassy['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-csrf-token': handshake_gouv_handshake, 'x-gouv-app-id': handshake_gouv_appid, 'x-gouv-web': 'fr.gouv.consulat' } proxies = { "http": proxy, "https": proxy } response = requests.post(url, headers=headers, data=payload, proxies=proxies, verify=False) if response.status_code == 200: response_dic = response.json() return response.headers['X-Gouv-Handshake'], response_dic['created_at'], response_dic['_id'] logger.error(f'troov_make_reservation_session_old {response.status_code}, {response.text}') return None def troov_update_dynamic_steps_old(proxy, session_dic): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations-session/{session_dic['session_id']}/update-dynamic-steps" payload = json.dumps({ "key": "slotsSteps", "steps": [ { "stepType": "slotsStep", "name": session_dic['embassy']['name'], "numberOfSlots": 1, "dynamicStepIndex": 0, "zone_id": session_dic['embassy']['zoneId'], "value": { "lastSelectedDate": "", "label": session_dic['embassy']['name'], "accessibleCalendar": False, "hasSwitchedCalendar": False, "slots": {} } } ] }) headers = { 'accept': 'application/json, text/plain, */*', 'content-type': 'application/json', 'origin': f'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat', } proxies = { "http": proxy, "https": proxy } response = requests.post(url, headers=headers, data=payload, proxies=proxies, verify=False) if response.status_code == 200: return True else: rtext = response.text check_responsed_session_expired(rtext) return False def troov_get_reservation_session_old(proxy, session_dic): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations-session" headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'origin': 'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat', } params = { 'sessionId': session_dic['session_id'], 'standaloneServiceName': session_dic['embassy']['name'] } proxies = { "http": proxy, "https": proxy } response = requests.get(url, params=params, headers=headers, proxies=proxies, verify=False) if response.status_code == 200: return response.json() else: rtext = response.text check_responsed_session_expired(rtext) return None def troov_get_interval_old(proxy, session_dic): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations/get-interval?serviceId={session_dic['embassy']['zoneId']}" headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'origin': f'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat' } proxies = { "http": proxy, "https": proxy } response = requests.get(url, headers=headers, proxies=proxies, verify=False) if response.status_code == 200: return response.json() else: rtext = response.text check_responsed_session_expired(rtext) return None def troov_get_exclude_days_old(proxy, session_dic): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations/exclude-days" payload = json.dumps({ "session": { session_dic['embassy']['zoneId']: True }, "sessionId": session_dic['session_id'] }) headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'content-type': 'application/json', 'origin': f'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat' } proxies = { "http": proxy, "https": proxy } response = requests.post(url, headers=headers, data=payload, proxies=proxies, verify=False) if response.status_code == 200: return response.json() else: rtext = response.text check_responsed_session_expired(rtext) return None def troov_get_open_days_old(proxy, session_dic): def get_dates_in_range(date_range): start_date = datetime.strptime(date_range["start"], "%Y-%m-%d") end_date = datetime.strptime(date_range["end"], "%Y-%m-%d") current_date = start_date dates_list = [] while current_date <= end_date: dates_list.append(current_date.strftime("%Y-%m-%d")) current_date += timedelta(days=1) return dates_list interval = troov_get_interval_old(proxy, session_dic) exclude_days = troov_get_exclude_days_old(proxy, session_dic) # exclude_days = ['2025-05-31'] if interval and exclude_days: all_days = get_dates_in_range(interval) available_dates = [ day for day in all_days if day not in exclude_days and datetime.strptime(day, "%Y-%m-%d").weekday() not in (5, 6) ] return available_dates return None def troov_get_available_times_old(proxy, session_dic, date, places=1, capacity=2): url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations/availability?name=Visas&date={date}&places={places}&matching=&maxCapacity={capacity}&sessionId={session_dic['session_id']}" headers = { 'accept': 'application/json, text/plain, */*', 'accept-language': 'zh-CN,zh;q=0.9,en;q=0.8', 'origin': f'https://consulat.gouv.fr', 'referer': session_dic['embassy']['website'], 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', 'x-gouv-app-id': session_dic['x_gouv_app_id'], 'x-gouv-web': 'fr.gouv.consulat' } proxies = { "http": proxy, "https": proxy } response = requests.get(url, headers=headers, proxies=proxies, verify=False) if response.status_code == 200: return response.json() else: rtext = response.text check_responsed_session_expired(rtext) return None # def troov_book_old(proxy, session_dic, date, slot, uinfo, captcha): # try: # url = f"https://51.254.177.49/api/team/{session_dic['embassy']['teamId']}/reservations/family" # book_body = troov_dublin_visas_book_data_builder(session_dic['session_id'], date, slot, uinfo, captcha) # payload = json.dumps(book_body) # headers = { # 'accept': 'application/json, text/plain, */*', # 'content-type': 'application/json', # 'origin': f'https://consulat.gouv.fr', # 'referer': session_dic['embassy']['website'], # 'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36', # 'x-csrf-token': session_dic['x-csrf-token'], # 'x-gouv-app-id': session_dic['x_gouv_app_id'], # 'x-gouv-web': 'fr.gouv.consulat', # } # proxy_config = { # "http": proxy, # "https": proxy # } # response = requests.post(url, headers=headers, data=payload, verify=False, proxies=proxy_config) # resp_text = response.text # logger.info(f'book code={response.status_code}, text={response.text}') # if response.status_code == 200: # if "qrcode" in resp_text: # return resp_text # else: # check_responsed_session_expired(resp_text) # except Exception as e: # logger.error(f"proxy={proxy}, session_dic={session_dic}, troov_book_old exception: {e}") # return None