首先,您应该从亚马逊基础页面获取令牌anti-csrftoken-a2z:
-
使用特定的用户代理向www.amazon.com 发出请求:Mozilla ...
-
通过 XPATH 选择器获取 JSON 数据:
//span[@id='nav-global-location-data-modal-action']/@data-a-modal
来自此选择器的 JSON 示例:
{
"width": 375,
"closeButton": "false",
"popoverLabel": "Choose your location",
"ajaxHeaders": {
"anti-csrftoken-a2z": "ajaxHeaders >> anti-csrftoken-a2z"
},
"name": "glow-modal",
"url": "/gp/glow/get-address-selections.html?deviceType=desktop&pageType=Gateway&storeContext=NoStoreName&actionSource=desktop-modal",
"footer": "<span class=\"a-declarative\" data-action=\"a-popover-close\" data-a-popover-close=\"{}\"><span class=\"a-button a-button-primary\"><span class=\"a-button-inner\"><button name=\"glowDoneButton\" class=\"a-button-text\" type=\"button\">Done</button></span></span></span>",
"header": "Choose your location"
}
- 为下一个请求制作标头:
headers = {
"anti-csrftoken-a2z": `gMDCYRgjYFVWvjfmU70/qMURqYh7kAko11WlenYAAAAMAAAAAGGokFZyYXcAAAAA`,
"user-agent": "Mozila ..."
}
-
向链接发出请求:https://www.amazon.com/gp/glow/get-address-selections.html?deviceType=desktop&pageType=Gateway&storeContext=NoStoreName&actionSource=desktop-modal
带有来自第 2 步的标头和来自第 1 步的响应 cookie。
-
从响应中提取CSRF_TOKEN:
正则表达式:'CSRF_TOKEN : "(.+?)"'
-
为下一个请求制作标头:
headers = {
"anti-csrftoken-a2z": "CSRF token from step 4",
"user-agent": "Mozila ..."
}
- 向
https://www.amazon.com/gp/delivery/ajax/address-change.html 发出 POST 请求
使用表单数据:
{
"locationType": "LOCATION_INPUT",
"zipCode": "zip-code",
"storeContext": "generic",
"deviceType": "web",
"pageType": "Gateway",
"actionSource": "glow",
"almBrandId": "undefined",
}
带有来自第 5 步的标头和来自第 3 步的响应 cookie。
如果所有文件都应该得到这样的响应:
{
'isValidAddress': 1,
'isTransitOutOfAis': 0,
'address': {'locationType': 'LOCATION_INPUT', 'district': None,
'zipCode': '30322', 'addressId': None, 'isDefaultShippingAddress': 'false', 'obfuscatedId': None, 'isAccountAddress': 'false', 'state': 'GA',
'countryCode': 'US', 'addressLabel': None,
'city': 'ATLANTA', 'addressLine1': None}, 'sembuUpdated': 1
}
- 保存第 6 步中的响应 cookie 并将它们用于进一步的请求
包含所有逻辑的 Python 脚本:
import json
import requests
from parsel import Selector
AMAZON_US_URL = "https://www.amazon.com/"
AMAZON_ADDRESS_CHANGE_URL = (
"https://www.amazon.com/gp/delivery/ajax/address-change.html"
)
AMAZON_CSRF_TOKEN_URL = (
"https://www.amazon.com/gp/glow/get-address-selections.html?deviceType=desktop"
"&pageType=Gateway&storeContext=NoStoreName&actionSource=desktop-modal"
)
DEFAULT_USER_AGENT = (
"Mozilla/5.0 ..."
)
DEFAULT_REQUEST_HEADERS = {"Accept-Language": "en", "User-Agent": DEFAULT_USER_AGENT}
def get_amazon_content(start_url: str, cookies: dict = None) -> tuple:
response = requests.get(
url=start_url, headers=DEFAULT_REQUEST_HEADERS, cookies=cookies
)
response.raise_for_status()
return Selector(text=response.text), response.cookies
def get_ajax_token(content: Selector):
data = content.xpath(
"//span[@id='nav-global-location-data-modal-action']/@data-a-modal"
).get()
if not data:
raise ValueError("Invalid page content")
json_data = json.loads(data)
return json_data["ajaxHeaders"]["anti-csrftoken-a2z"]
def get_session_id(content: Selector):
session_id = content.re_first(r'session: \{id: "(.+?)"')
if not session_id:
raise ValueError("Session id not found")
return session_id
def get_token(content: Selector):
csrf_token = content.re_first(r'CSRF_TOKEN : "(.+?)"')
if not csrf_token:
raise ValueError("CSRF token not found")
return csrf_token
def send_change_location_request(zip_code: str, headers: dict, cookies: dict):
response = requests.post(
url=AMAZON_ADDRESS_CHANGE_URL,
data={
"locationType": "LOCATION_INPUT",
"zipCode": zip_code,
"storeContext": "generic",
"deviceType": "web",
"pageType": "Gateway",
"actionSource": "glow",
"almBrandId": "undefined",
},
headers=headers,
cookies=cookies,
)
assert response.json()["isValidAddress"], "Invalid change response"
return response.cookies
def get_session_cookies(zip_code: str):
response = requests.get(url=AMAZON_US_URL, headers=DEFAULT_REQUEST_HEADERS)
content = Selector(text=response.text)
headers = {
"anti-csrftoken-a2z": get_ajax_token(content=content),
"user-agent": DEFAULT_USER_AGENT,
}
response = requests.get(
url=AMAZON_CSRF_TOKEN_URL, headers=headers, cookies=response.cookies
)
content = Selector(text=response.text)
headers = {
"anti-csrftoken-a2z": get_token(content=content),
"user-agent": DEFAULT_USER_AGENT,
}
send_change_location_request(
zip_code=zip_code, headers=headers, cookies=dict(response.cookies)
)
# Verify that location changed correctly.
response = requests.get(
url=AMAZON_US_URL, headers=DEFAULT_REQUEST_HEADERS, cookies=response.cookies
)
content = Selector(text=response.text)
location_label = content.css("span#glow-ingress-line2::text").get().strip()
assert zip_code in location_label
if __name__ == "__main__":
get_session_cookies(zip_code="30322")
另外,使用 Scrapy 框架的类似逻辑:
from http.cookies import SimpleCookie
from scrapy import FormRequest, Request, Spider
from scrapy.http import HtmlResponse
class AmazonSessionSpider(Spider):
"""
Amazon spider for extracting location cookies.
"""
name = "amazon.com:location-session"
address_change_endpoint = "/gp/delivery/ajax/address-change.html"
csrf_token_endpoint = (
"/gp/glow/get-address-selections.html?deviceType=desktop"
"&pageType=Gateway&storeContext=NoStoreName&actionSource=desktop-modal"
)
countries_base_urls = {
"US": "https://www.amazon.com",
"GB": "https://www.amazon.co.uk",
"DE": "https://www.amazon.de",
"ES": "https://www.amazon.es",
}
default_headers = {
"sec-fetch-site": "none",
"sec-fetch-dest": "document",
"accept-language": "ru-RU,ru;q=0.9",
"connection": "close",
}
def __init__(self, country: str, zip_code: str, *args, **kwargs):
super().__init__(*args, **kwargs)
self.country = country
self.zip_code = zip_code
def start_requests(self):
"""
Make start request to main Amazon country page.
"""
request = Request(
url=self.countries_base_urls[self.country],
headers=self.default_headers,
callback=self.parse_ajax_token,
)
yield request
def parse_ajax_token(self, response: HtmlResponse):
"""
Parse ajax token from response.
"""
yield response.request.replace(
url=self.countries_base_urls[self.country] + self.csrf_token_endpoint,
headers={
"anti-csrftoken-a2z": self.get_ajax_token(response=response),
**self.default_headers,
},
callback=self.parse_csrf_token,
)
def parse_csrf_token(self, response: HtmlResponse):
"""
Parse CSRF token from response and make request to change Amazon location.
"""
yield FormRequest(
method="POST",
url=self.countries_base_urls[self.country] + self.address_change_endpoint,
formdata={
"locationType": "LOCATION_INPUT",
"zipCode": self.zip_code,
"storeContext": "generic",
"deviceType": "web",
"pageType": "Gateway",
"actionSource": "glow",
"almBrandId": "undefined",
},
headers={
"anti-csrftoken-a2z": self.get_csrf_token(response=response),
**self.default_headers,
},
callback=self.parse_session_cookies,
)
def parse_session_cookies(self, response: HtmlResponse) -> dict:
"""
Return cookies dict if location changed successfully.
"""
json_data = response.json()
if not json_data.get("isValidAddress"):
return {}
return self.extract_response_cookies(response=response)
@staticmethod
def get_ajax_token(response: HtmlResponse) -> str:
"""
Extract ajax token from response.
"""
data = response.xpath("//input[@id='glowValidationToken']/@value").get()
if not data:
raise ValueError("Invalid page content")
return data
@staticmethod
def get_csrf_token(response: HtmlResponse) -> str:
"""
Extract CSRF token from response.
"""
csrf_token = response.css("script").re_first(r'CSRF_TOKEN : "(.+?)"')
if not csrf_token:
raise ValueError("CSRF token not found")
return csrf_token
@staticmethod
def extract_response_cookies(response: HtmlResponse) -> dict:
"""
Extract cookies from response object
and return it in valid format.
"""
cookies = {}
cookie_headers = response.headers.getlist("Set-Cookie", [])
for cookie_str in cookie_headers:
cookie = SimpleCookie()
cookie.load(cookie_str.decode("utf-8"))
for key, raw_value in cookie.items():
cookies[key] = raw_value.value
return cookies
外壳命令:
scrapy crawl amazon.com:location-session -a country=US -a zip_code=30332