panleicim/appointment_request
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

redirection after send request

Browse Source
This commit is contained in:
PAN Lei
2024-10-04 17:52:33 +02:00
parent fd5bcf6096
commit 6976e0f484
1 changed files with 141 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files
workers/sender.py
+141 -31
View File
@@ -1,5 +1,6 @@
import datetime
import json
import logging
import random
import re
import time
@@ -13,6 +14,7 @@ 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):
@@ -63,11 +65,47 @@ class Sender:
collection_name = str(datetime.date.today())
MONGO_STORE_MANAGER.insert_reserve_result(collection_name=collection_name, reserve=result)
def send_request_for_list(self, contact_list: list):
pass
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]
def get_csrf(self):
pass
_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:
@@ -80,10 +118,23 @@ class Sender:
self._csrf = csrf
_csrf = csrf
self.logger.info("model is:{}".format(model))
_chrome_version = get_chrome_version_from_ua(js_data.ua)
_chrome_ua_header = '"Chromium";v="{}", "Not;A=Brand";v="24", "Google Chrome";v="{}"'.format(_chrome_version,
_chrome_version)
_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"
_chrome_ua_version = '"Chromium";v="{}", "Not;A=Brand";v="24.0.0.0", "Google Chrome";v="{}"'.format(
_chrome_version, _chrome_version)
_chrome_ua_header = '"Chromium";v="{}", "Not;A=Brand";v="24", "Google Chrome";v="{}"'.format(_chrome_code,
_chrome_code)
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":
@@ -92,16 +143,20 @@ class Sender:
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',
'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': model,
'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,
@@ -113,29 +168,34 @@ class Sender:
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)
timeout=15, allow_redirects=False)
self.logger.info(response.status_code)
if response.status_code == 200:
if response.status_code == 302:
# add to mongodb
self.logger.info(response.text)
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)
self.publish_message_to_queue(contact, status=PublishType.SUCCESS, url=response.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)))
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.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)
@@ -152,7 +212,57 @@ class Sender:
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__':
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")
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")