272 lines
16 KiB
Python
272 lines
16 KiB
Python
import datetime
|
|
import json
|
|
import logging
|
|
import random
|
|
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):
|
|
# 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"
|
|
else:
|
|
result.proxy = "data_impulse"
|
|
result.id = id
|
|
result.store_type = store_type
|
|
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):
|
|
# /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)
|
|
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))
|
|
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
|
|
if _chrome_code == '128':
|
|
_chrome_version = "128.0.6613.147"
|
|
elif _chrome_code == '129':
|
|
_chrome_version = "129.0.6668.54"
|
|
elif _chrome_code == '116':
|
|
_chrome_version = "116.0.5845.114"
|
|
elif _chrome_code == '114':
|
|
_chrome_version = "114.0.5735.53"
|
|
elif _chrome_code == '99':
|
|
_chrome_version = "99.0.4844.58"
|
|
# 99: " Not A;Brand";v="99", "Chromium";v="99", "Google Chrome";v="99"
|
|
# " Not A;Brand";v="99.0.0.0", "Chromium";v="99.0.4844.58", "Google Chrome";v="99.0.4844.58"
|
|
_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_ua_header = 'Not A;Brand";v="99", "Chromium";v="99", "Google Chrome";v="99"'
|
|
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_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': "+33 " + 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)
|
|
# 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")
|