import datetime import json import re import time from http.cookies import SimpleCookie from models.contact_pojo import ContactPojo from curl_cffi import requests from db.mongo_manager import MONGO_STORE_MANAGER from models.ReserveResultPojo import ReserveResultPojo, PublishType from models.jsdata_pojo import JsDataPojo from models.result_pojo import RequestResult from queue_message.CookiesPublisher import CookiesPublisher from utils.address_ip import get_address_ip from workers.captcha_result_getter import CaptchaResultGetter def get_chrome_version_from_ua(ua): _result = re.findall("Chrome\/([0-9]*)", ua) print(_result) if len(_result) > 0: return _result[0] else: "128" class Sender: def __init__(self, cookie_str, cookiesPublisher: CookiesPublisher, received_dict, proxy_to_use, logger): self.store_type = "random" self.cookie = SimpleCookie() self.cookiesPublisher = cookiesPublisher # self.cookie_str = 'datadome=~pxdHFAvsQl2rvDrTzhPgCHxu~4TBcePTTE~Cy8Rgol6oMRc11gA02VRp0Z3uEDUszCjacubNu7vbfQCh27gz8RC10u_325pt_gsMmJh1ScGvOofVJiVAbEKvSEUjd82;policy=accepted;app.sig=PhjmDkq_dI49pADppDNKxpLe_G4;app=eyJmbGFzaCI6e30sImNhY2hlZmxhc2giOltdLCJjc3JmU2VjcmV0IjoiYnRodHNYU1lvdnl4RzVGakpGRDZsQ0JtIn0=;lang=fr;' self.cookie_str = cookie_str self.logger = logger self._csrf = None self.received_dict = received_dict self.proxy_to_use = proxy_to_use self.cookie.load(self.cookie_str) def publish_message_to_queue(self, contact: ContactPojo, status: PublishType, url: str, store_type: str, model: str = ""): # create the message if url == "https://rendezvousparis.hermes.com/client/welcome": return id = url.split("/")[-1] result = ReserveResultPojo(type=status, phone=contact.phone, message=status.value, url=url, firstName=contact.first_name, lastName=contact.last_name, email=contact.mail, passport=contact.passport, ccid=contact.ccid) result.ua = contact.ua result.current_ip = contact.current_ip if "oxylabs" in self.proxy_to_use["http"]: if "mob" in self.proxy_to_use["http"]: result.proxy = "oxylabs_mob" else: result.proxy = "oxylabs_res" elif "anyip" in self.proxy_to_use["http"]: result.proxy = "anyip" elif "iproyal" in self.proxy_to_use["http"]: result.proxy = "iproyal" else: result.proxy = "data_impulse" result.id = id result.store_type = store_type result.model = model result.created_at = time.strftime("%H:%M:%S", time.localtime()) collection_name = str(datetime.date.today()) MONGO_STORE_MANAGER.insert_reserve_result(collection_name=collection_name, reserve=result) def apply_redirect(self, response, old_headers, contact, js_data, selected_store, model=""): # /client/register/5XD2E2 _res_headers = response.headers _location = _res_headers['location'] _redirect_url = "https://rendezvousparis.hermes.com" + _location # get set_cookies from res_headers _cookies_to_set = _res_headers['set-cookie'] _cookies_to_set = _cookies_to_set.replace("httponly, ", "") _cookies_to_set = _cookies_to_set.replace("httponly", "") _new_headers = old_headers _cookie_loader = SimpleCookie() _cookie_loader.load(_cookies_to_set) _new_cookies = {k: v.value for k, v in _cookie_loader.items()} for key in _new_cookies: print(f"Key: {key}, Value: {_new_cookies[key]}") _new_headers[key] = _new_cookies[key] _redirect_response = requests.get(_redirect_url, verify=False, headers=_new_headers, proxies=self.proxy_to_use) self.logger.info("{}:{}".format(contact.mail, response.url)) contact.ua = js_data.ua contact.current_ip = get_address_ip(proxy_to_use=self.proxy_to_use) _appointment_url = _redirect_url self.publish_message_to_queue(contact, status=PublishType.SUCCESS, url=_appointment_url, store_type=selected_store, model=model) self.cookie.load(_cookies_to_set) new_cookies = {k: v.value for k, v in self.cookie.items()} new_coolies_str = "" for key in new_cookies: new_coolies_str = new_coolies_str + key + "=" + new_cookies[key] + ";" self.logger.info("will publish to queue {}".format(new_coolies_str)) # upload the cookie to queue self.received_dict['cookiesStr'] = new_coolies_str self.logger.info("body in json:{}".format(json.dumps(self.received_dict))) if self.cookiesPublisher is not None: self.cookiesPublisher.publish_body(json.dumps(self.received_dict)) # self.cookie_str = new_coolies_str self.logger.info("successful request, mail:{}".format(contact.mail)) self.send_jsdata_request(js_data, _appointment_url, new_coolies_str) # create the headers def send_request(self, url, js_data: JsDataPojo, contact: ContactPojo, model: str, csrf: str = None) -> RequestResult: if csrf is None: if self._csrf is None: _csrf = '8Bs2dBwb-nHONOzo9Tei2CcMZglEfsRqUz8E' else: _csrf = self._csrf else: self._csrf = csrf _csrf = csrf self.logger.info("model is:{}".format(model)) # the following code is used to add model to user agent # js_data.ua = "Mozilla/5.0 (Linux; Android 9; {}) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/99.0.4844.58 Mobile Safari/537.36".format( # model) _chrome_code = get_chrome_version_from_ua(js_data.ua) _chrome_version = None _chrome_ua_header = '"Not A;Brand";v="99", "Chromium";v="99", "Google Chrome";v="99"' _chrome_ua_version = '"Not A;Brand";v="99.0.0.0", "Chromium";v="99.0.4844.58", "Google Chrome";v="99.0.4844.58"' _chrome_version = "99.0.4844.58" if _chrome_code == '128': _chrome_version = "128.0.6613.147" _chrome_ua_header = '"Google Chrome";v="128", "Not=A?Brand";v="8", "Chromium";v="128"' _chrome_ua_version = '"Google Chrome";v="128.0.6613.147", "Not=A?Brand";v="8.0.0.0", "Chromium";v="128.0.6613.147"' elif _chrome_code == '137': _chrome_version = "137.0.7151.89" _chrome_ua_header = '"Google Chrome";v="137", "Not/A)Brand";v="24", "Chromium";v="137"' _chrome_ua_version = '"Google Chrome";v="137.0.7151.89", "Not=A?Brand";v="8.0.0.0", "Chromium";v="137.0.7151.89"' elif _chrome_code == '129': _chrome_version = "129.0.6668.81" _chrome_ua_header = '"Google Chrome";v="129", "Not=A?Brand";v="8", "Chromium";v="129"' _chrome_ua_version = '"Google Chrome";v="129.0.6668.81", "Not=A?Brand";v="8.0.0.0", "Chromium";v="129.0.6668.81"' elif _chrome_code == '116': _chrome_version = "116.0.5845.114" _chrome_ua_header = '"Google Chrome";v="116", "Not=A?Brand";v="8", "Chromium";v="116"' elif _chrome_code == '114': _chrome_version = "114.0.5735.53" _chrome_ua_header = '"Google Chrome";v="114", "Not=A?Brand";v="8", "Chromium";v="114"' self.logger.info("chrome_ua_header is:{}".format(_chrome_ua_header)) self.logger.info("_chrome_ua_version is:{}".format(_chrome_ua_version)) # 默认为空，也就是随机 _selected_store = '' if contact.store is not None and contact.store != "random": _selected_store = contact.store self.logger.info("store is:{}".format(_selected_store)) self.logger.info("cookie_str is:{}".format(self.cookie_str)) headers = {'Content-Type': 'application/x-www-form-urlencoded', 'User-Agent': js_data.ua, 'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7', 'Cookie': self.cookie_str, 'Referer': 'https://rendezvousparis.hermes.com/client/register', 'Sec-Fetch-Mode': 'navigate', 'sec-ch-ua-platform': '"Android"', 'sec-ch-ua-arch': '""', 'Sec-Fetch-Site': 'same-origin', 'sec-ch-ua-mobile': '?1', 'sec-ch-ua': _chrome_ua_header, 'sec-ch-ua-model': '"{}"'.format(model), 'Sec-Fetch-Dest': 'document', 'Accept-Language': 'fr-FR,fr;q=0.6'} if _chrome_ua_version is not None: headers['sec-ch-ua-full-version-list'] = _chrome_ua_version data = {'check': '', '_csrf': _csrf, 'prefer': _selected_store, 'surname': contact.last_name.capitalize(), 'name': contact.first_name.capitalize(), 'phone_country': "FR", 'phone_number': "0" + contact.phone, 'email': contact.mail, 'passport_id': contact.passport, 'processing': 'on', 'cgu': 'on'} self.logger.info(data) self.logger.info("try to request for mail:{}".format(contact.mail)) print("send request with cookie:{}".format(self.cookie_str)) try: proxy_to_use = self.proxy_to_use # print(proxy_to_use) response = requests.post(url=url, proxies=proxy_to_use, verify=False, headers=headers, data=data, timeout=15, allow_redirects=False, impersonate="chrome99_android") self.logger.info(response.status_code) if response.status_code == 302: # add to mongodb self.logger.info(response.text) self.apply_redirect(response=response, old_headers=headers, contact=contact, js_data=js_data, selected_store=_selected_store, model=model) # self.logger.info("{}:{}".format(contact.mail, response.url)) # contact.ua = js_data.ua # contact.current_ip = get_address_ip(proxy_to_use=proxy_to_use) # _appointment_url = response.url # self.publish_message_to_queue(contact, status=PublishType.SUCCESS, url=_appointment_url, # store_type=_selected_store) # cookies_to_set = response.headers['set-cookie'] # self.cookie.load(cookies_to_set) # new_cookies = {k: v.value for k, v in self.cookie.items()} # new_coolies_str = "" # for key in new_cookies: # new_coolies_str = new_coolies_str + key + "=" + new_cookies[key] + ";" # self.logger.info("will publish to queue {}".format(new_coolies_str)) # # upload the cookie to queue # self.received_dict['cookiesStr'] = new_coolies_str # self.logger.info("body in json:{}".format(json.dumps(self.received_dict))) # if self.cookiesPublisher is not None: # self.cookiesPublisher.publish_body(json.dumps(self.received_dict)) # # self.cookie_str = new_coolies_str # self.logger.info("successful request, mail:{}".format(contact.mail)) # self.send_jsdata_request(js_data, _appointment_url, new_coolies_str) return RequestResult.SUCCESS else: print(response.text) if response.status_code == 403: # try to load seesion # session = HTMLSession() self.logger.info("blocked by hermes, mail:{}".format(contact.mail)) return RequestResult.BLOCKED elif response.status_code == 502: self.logger.info("proxy error, mail:{}".format(contact.mail)) return RequestResult.PROXY_ERROR return RequestResult.BLOCKED except Exception as Error: print(Error) return RequestResult.PROXY_ERROR def send_jsdata_request(self, js_data, _appointment_url, new_coolies_str): captchaResultGetter = CaptchaResultGetter() self.logger.info("send jsdata after successful, url is {}".format(_appointment_url)) _new_cookies = captchaResultGetter.get_valid_ch_cookie(self.proxy_to_use, js_data, old_valid_cookie=self.cookie_str, referrer=_appointment_url) self.logger.info("new_cookies after jsdata is:{}".format(_new_cookies)) def test_simple_cookie_(): _test = "datadome=TslVDPtxU67iBZVrAnv0p2yqAz78czwc~ARuu1XFFuToPAarC8IXt~~Pj48Zrte63FEV0XX3uUClA39dL3ExEkPaeZgprOVYrqk7niMQwDy~WhssNi18VJ8xjwsilR8S; Max-Age=31536000; Domain=.hermes.com; Path=/; Secure; SameSite=None, lang=fr; Path=/; Expires=Sat, 04 Oct 2025 15:06:01 GMT; SameSite=Strict, app=eyJmbGFzaCI6e30sImNhY2hlZmxhc2giOltdLCJjc3JmU2VjcmV0IjoiQ0plMmF0MDFPaEs4LWl5TzFSZUkxQmotIiwiYXBwb2ludG1lbnRfY29kZSI6IkhYQUhNTSIsImJsb2NrX3JlZ2lzdHJhdGlvbiI6ZmFsc2UsInJfaWQiOjIzfQ==; path=/; expires=Mon, 04 Nov 2024 15:06:02 GMT; samesite=lax; secure; httponly, app.sig=8TYE8eWCkY-v6ivfTh2gSJVdjC0; path=/; expires=Mon, 04 Nov 2024 15:06:02 GMT; samesite=lax; secure; httponly" # _test = "datadome=TslVDPtxU67iBZVrAnv0p2yqAz78czwc~ARuu1XFFuToPAarC8IXt~~Pj48Zrte63FEV0XX3uUClA39dL3ExEkPaeZgprOVYrqk7niMQwDy~WhssNi18VJ8xjwsilR8S; Max-Age=31536000; Domain=.hermes.com; Path=/; Secure; SameSite=None, lang=fr; Path=/; Expires=Sat, 04 Oct 2025 15:06:01 GMT; SameSite=Strict, app=eyJmbGFzaCI6e30sImNhY2hlZmxhc2giOltdLCJjc3JmU2VjcmV0IjoiQ0plMmF0MDFPaEs4LWl5TzFSZUkxQmotIiwiYXBwb2ludG1lbnRfY29kZSI6IkhYQUhNTSIsImJsb2NrX3JlZ2lzdHJhdGlvbiI6ZmFsc2UsInJfaWQiOjIzfQ==; path=/; expires=Mon, 04 Nov 2024 15:06:02 GMT; samesite=lax; secure; app.sig=8TYE8eWCkY-v6ivfTh2gSJVdjC0; path=/; expires=Mon, 04 Nov 2024 15:06:02 GMT; samesite=lax; secure;" _test = _test.replace("httponly, ", "") _test = _test.replace("httponly", "") print(_test) _cookie_loader = SimpleCookie() _cookie_loader.load(_test) _new_cookies = {k: v.value for k, v in _cookie_loader.items()} for key in _new_cookies: print(f"Key: {key}, Value: {_new_cookies[key]}") if __name__ == '__main__': test_simple_cookie_() # test_data_dict = {"glvd": "Qualcomm", "glrd": "Adreno (TM) 642L", "hc": 8, "br_oh": 763, "br_ow": 393, "br_h": 763, # "br_w": 393, "rs_h": 873, "rs_w": 393, "rs_cd": 24, "ars_h": 873, "ars_w": 393, "plg": 0, # "eva": 33, "vnd": "Google Inc.", "plu": [], # "ua": "Mozilla/5.0 (Linux; Android 12; 2109119DG) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/114.0.0.0 Mobile Safari/537.36", # "dvm": 8, "acw": "probably", "pr": 2.75, "ts_mtp": 5, "tz": -120, # "emd": "k:ai,vi,vi,ao d:defau,937bf,b8054,defau g:922f3,90ce3,d7683,defau", # "cookiesStr": "app.sig=ns7DOjAptiuvVR0Ks9PxS0DA0qs;datadome=tYdldjfLz6ea35aFFgS2SYv5WErQlk5PpbzSL6GXwe_APfR7quyH_2xmGVx2DZaUWz0PdDJn0Gtj6gaxYmlCMuR9Yrr3WBtt38570LQzGCipiDrmu7Z0hMYgvHGrX_zy;app=eyJmbGFzaCI6e30sImNhY2hlZmxhc2giOltdLCJjc3JmU2VjcmV0IjoiUmFrWGFycUlUaDJwMTduSmlfVU1wMnZXIn0=;policy=accepted;lang=fr;", # "model": "2109119DG"} # # test_data_dict = json.loads(_test_data_json) # js = JsDataPojo(glrd=test_data_dict['glrd'], glvd=test_data_dict['glvd'], hc=test_data_dict['hc'], # ua=test_data_dict['ua'], br_oh=test_data_dict['br_oh'], br_ow=test_data_dict['br_ow'], # ars_h=test_data_dict['ars_h'], ars_w=test_data_dict['ars_w'], pr=test_data_dict['pr'], # plg=test_data_dict['plg'], br_h=test_data_dict['br_h'], br_w=test_data_dict['br_w'], # plu=test_data_dict['plu'], vnd=test_data_dict['vnd'], dvm=test_data_dict['dvm'], # ts_mtp=test_data_dict['ts_mtp'], eva=test_data_dict['eva'], rs_h=test_data_dict['rs_h'], # rs_w=test_data_dict['rs_w'], rs_cd=test_data_dict['rs_cd'], emd=test_data_dict['emd']) # _cookie_str = test_data_dict["cookiesStr"] # print("_cookie_str: {}".format(_cookie_str)) # _sender = Sender(cookie_str=_cookie_str, cookiesPublisher=None, # received_dict=test_data_dict, # proxy_to_use={}, logger=logging.getLogger("test")) # _contact = ContactPojo(first_name="John", last_name="Doe", mail="john.doe@example.com", # passport_number="E1234567890", phone_number="66666666") # _contact.store = "" # _host = "https://24ea-2a01-e0a-115-bab0-6107-7f97-948-16ff.ngrok-free.app" # _sender.send_request( # url=_host, js_data=js, # contact=_contact, model=test_data_dict["model"]) # get_chrome_version_from_ua( # "Mozilla/5.0 (Linux; Android 10; K) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/128.0.0.0 Mobile Safari/537.36")