河小马 
Aliexpress 是中国最大的全球电子商务商店之一，也是一个流行的网络抓取目标。 Aliexpress 包含数百万种产品和产品评论，可用于市场分析、商业智能和直销。 在本教程中，我们将了解如何抓取 Aliexpress。我们将从通过抓取搜索系统来查找产品开始。然后我们将抓取找到的产品数据、定价和客户评论。 这将是一个相对简单的爬虫，只需几行 Python 代码。让我们开始吧！
为什么要抓取 Aliexpress？
抓取Aliexpress数据的原因有很多。首先，由于Aliexpress 是世界上最大的电子商务平台，因此它是商业智能或市场分析的主要目标。了解顶级产品及其在Aliexpress上的元信息可以在业务和市场分析中发挥巨大优势。 另一个常见用途是电子商务，主要是通过直销——本世纪最大的新兴市场之一是整理产品清单并直接转售，而不是管理仓库。在这种情况下，许多商店策展人会抓取Aliexpress 的产品，为他们的直销店生成精选产品列表。
项目设置
在本教程中，我们将使用带有两个包的 Python：
- httpx – HTTP 客户端库，可以让我们与 AliExpress.com 的服务器进行通信
- parsel – HTML 解析库，它将帮助我们解析从网络上抓取的 HTML 文件以获取酒店数据。
- loguru [可选] – 漂亮的日志库，可以帮助我们跟踪正在发生的事情。
这些包可以通过pip命令轻松安装：
$ pip install httpx parsel loguru
或者，您可以自由地换成httpx任何其他 HTTP 客户端库，例如requests，因为我们只需要基本的 HTTP 函数，这些函数在每个库中几乎都是可以互换的。至于，parsel另一个不错的选择是beautifulsoup包。
寻找Aliexpress产品
在 Aliexpress 上发现产品的方法有很多。 我们可以使用搜索系统来查找我们想要抓取的产品或探索许多产品类别。无论我们采用哪种方法，我们的关键目标都是相同的 – 抓取产品预览和分页。 让我们看一下在搜索或类别视图中使用的Aliexpress列表页面：
如果我们查看搜索页面或类别页面的页面源代码，我们可以看到所有产品预览都存储在一个 javascript 变量中，该变量window.runParams隐藏在<script>页面 HTML 源代码的标记中：
这是一种常见的 Web 开发模式，可以使用 javascript 实现动态数据管理。不过，这对我们来说是个好消息，因为我们可以使用正则表达式模式获取这些数据并像 Python 字典一样解析它！ 让我们编写我们的爬虫代码的第一部分——产品预览解析器，我们将使用它从类别或搜索结果页面中提取产品数据：
from parsel import Selector
import json
def extract_search(response) -> Dict:
"""extract json data from search page"""
# find script with result.pagectore data in it._it_t_=
script_with_data = sel.xpath('//script[contains(text(),"window.runParams")]')
# select page data from javascript variable in script tag using regex
data = json.loads(script_with_data.re(r'_init_data_\s*=\s*{\s*data:\s*({.+}) }')[0])
return data['data']['root']['fields']
def parse_search(response):
"""Parse search page response for product preview results"""
data = extract_search(response)
parsed = []
for result in data["mods"]["itemList"]["content"]:
parsed.append(
{
"id": result["productId"],
"url": f"https://www.aliexpress.com/item/{result['productId']}.html",
"type": result["productType"], # can be either natural or ad
"title": result["title"]["displayTitle"],
"price": result["prices"]["salePrice"]["minPrice"],
"currency": result["prices"]["salePrice"]["currencyCode"],
"trade": result.get("trade", {}).get("tradeDesc"), # trade line is not always present
"thumbnail": result["image"]["imgUrl"].lstrip("/"),
"store": {
"url": result["store"]["storeUrl"],
"name": result["store"]["storeName"],
"id": result["store"]["storeId"],
"ali_id": result["store"]["aliMemberId"],
},
}
)
return parsed让我们通过抓取单个 Aliexpress 列表页面（类别页面或搜索结果页面）来尝试我们的解析器：
运行代码和示例输出
if __name__ == "__main__":
# for example, this category is for android phones:
resp = httpx.get("https://www.aliexpress.com/category/5090301/cellphones.html", follow_redirects=True)
print(json.dumps(parse_search(resp), indent=2))
[
{
"id": "3256804075561256",
"url": "https://www.aliexpress.com/item/3256804075561256.html",
"type": "ad",
"title": "2G/3G Smartphones Original 512MB RAM/1G RAM 4GB ROM android mobile phones new cheap celulares FM unlocked 4.0inch cell",
"price": 21.99,
"currency": "USD",
"trade": "8 sold",
"thumbnail": "ae01.alicdn.com/kf/S1317aeee4a064fad8810a58959c3027dm/2G-3G-Smartphones-Original-512MB-RAM-1G-RAM-4GB-ROM-android-mobile-phones-new-cheap-celulares.jpg_220x220xz.jpg",
"store": {
"url": "www.aliexpress.com/store/1101690689",
"name": "New 123 Store",
"id": 1101690689,
"ali_id": 247497658
}
}
...
]有很多有用的信息，但我们将解析器限制在最基本的部分以保持简洁。接下来让我们把这个解析器用于实际的抓取。
抓取Aliexpress搜索
现在我们已经准备好产品预览解析器，我们需要一个抓取器循环来遍历搜索结果以收集所有可用结果——而不仅仅是第一页：
from loguru import logger as log
import httpx
async def scrape_search(query: str, session: httpx.AsyncClient, sort_type="default"):
"""Scrape all search results and return parsed search result data"""
query = query.replace(" ", "+")
async def scrape_search_page(page):
"""Scrape a single aliexpress search page and return all embedded JSON search data"""
log.info(f"scraping search query {query}:{page} sorted by {sort_type}")
resp = await session.get(
"https://www.aliexpress.com/wholesale?trafficChannel=main"
f"&d=y&CatId=0&SearchText={query}<ype=wholesale&SortType={sort_type}&page={page}"
)
return resp
# scrape first search page and find total result count
first_page = await scrape_search_page(query, session, 1)
first_page_data = extract_search(first_page)
page_size = first_page_data["pageInfo"]["pageSize"]
total_pages = int(math.ceil(first_page_data["pageInfo"]["totalResults"] / page_size))
if total_pages > 60:
log.warning(f"query has {total_pages}; lowering to max allowed 60 pages")
total_pages = 60
# scrape remaining pages concurrently
log.info(f'scraping search "{query}" of total {total_pages} sorted by {sort_type}')
other_pages = await asyncio.gather(*[scrape_search_page(page=i) for i in range(1, total_pages + 1)])
product_previews = []
for response in [first_page, *other_pages]:
product_previews.extend(parse_search(response))
return product_previews上面，我们定义了我们的scrape_search函数，我们使用一个常见的网络抓取习惯用法来进行已知长度的分页。 我们抓取第一页以提取总页数，并同时抓取其余页面。 现在，我们可以找到产品了，让我们来看看如何抓取产品数据、定价信息和评论！
抓取Aliexpress产品
要抓取 Aliexpress 产品，我们只需要一个产品数字 ID，我们已经在上一章中通过从 Aliexpress 搜索中抓取产品预览找到了它。例如，此手钻产品aliexpress.com/item/4000927436411.html的数字 ID 为4000927436411。 要解析产品数据，我们可以使用我们在搜索解析器中使用的相同技术——数据隐藏在window.runParams变量data键下的 HTML 文档中：
from parsel import Selector
def parse_product(response):
"""parse product HTML page for product data"""
sel = Selector(text=response.text)
# find the script tag containing our data:
script_with_data = sel.xpath('//script[contains(text(),"window.runParams")]')
# extract data using a regex pattern:
data = json.loads(script_with_data.re(r"data: ({.+?}),\n")[0])
product = {
"name": data["titleModule"]["subject"],
"total_orders": data["titleModule"]["formatTradeCount"],
"feedback": data["titleModule"]["feedbackRating"],
"variants": [],
}
# every product variant has it's own price and ID number (sku):
for sku in data["skuModule"]["skuPriceList"]:
product["variants"].append(
{
"name": sku["skuAttr"].split("#", 1)[1].split(";")[0],
"sku": sku["skuId"],
"available": sku["skuVal"]["availQuantity"],
"full_price": sku["skuVal"]["skuAmount"]["value"],
"discount_price": sku["skuVal"]["skuActivityAmount"]["value"],
"currency": sku["skuVal"]["skuAmount"]["currency"],
}
)
# data variable contains much more information - so feel free to explore it,
# but to keep things brief we focus on essentials in this article
return product
async def scrape_products(ids, session: httpx.AsyncClient):
"""scrape aliexpress products by id"""
log.info(f"scraping {len(ids)} products")
responses = await asyncio.gather(*[session.get(f"https://www.aliexpress.com/item/{id_}.html") for id_ in ids])
results = []
for response in responses:
results.append(parse_product(response))
return results在这里，我们定义了我们的产品抓取功能，它接收产品 ID、抓取 HTML 内容并提取每个产品的隐藏产品 JSON。如果我们为我们的钻孔产品运行它，我们应该看到一个格式良好的响应：
运行代码和示例输出
# Let's use browser like request headers for this scrape to reduce chance of being blocked or asked to solve a captcha
BASE_HEADERS = {
"user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/96.0.4664.110 Safari/537.36",
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8",
"accept-language": "en-US;en;q=0.9",
"accept-encoding": "gzip, deflate, br",
}
async def run():
async with httpx.AsyncClient(headers=BASE_HEADERS) as session:
print(json.dumps(await scrape_products(["4000927436411"], session), indent=2))
if __name__ == "__main__":
import asyncio
asyncio.run(run())
[
{
"name": "Mini Wireless Drill Electric Carving Pen Variable Speed USB Cordless Drill Rotary Tools Kit Engraver Pen for Grinding Polishing",
"total_orders": "3824",
"feedback": {
"averageStar": "4.8",
"averageStarRage": "96.4",
"display": true,
"evarageStar": "4.8",
"evarageStarRage": "96.4",
"fiveStarNum": 1724,
"fiveStarRate": "88",
"fourStarNum": 170,
"fourStarRate": "9",
"oneStarNum": 21,
"oneStarRate": "1",
"positiveRate": "87.6",
"threeStarNum": 45,
"threeStarRate": "2",
"totalValidNum": 1967,
"trialReviewNum": 0,
"twoStarNum": 7,
"twoStarRate": "0"
},
"variants": [
{
"name": "Red",
"sku": 10000011265318724,
"available": 1601,
"full_price": 16.24,
"discount_price": 12.99,
"currency": "USD"
},
...
}
]使用这种方法，我们丢弃的数据比我们在页面的可见 HTML 中看到的要多得多。我们获得了 SKU 编号、库存可用性、详细定价和评论评分元信息。我们只缺少评论本身，所以让我们看看如何检索评论数据。
抓取Aliexpress评论
Aliexpress 的产品评论需要对其后端 API 提出额外请求。如果我们启动 Network Inspector devtools（在主要浏览器中按 F12，然后选择“网络”选项卡），当我们单击下一个审核页面时，我们可以看到正在发出后台请求：
当我们点击第 2 页链接时，我们可以看到正在发出后台请求
让我们在我们的抓取工具中复制这个请求：
def parse_review_page(response):
"""parse single review page"""
sel = Selector(response.text)
parsed = []
for review_box in sel.css(".feedback-item"):
# to get star score we have to rely on styling where's 1 star == 20% width, e.g. 4 stars is 80%
stars = int(review_box.css(".star-view>span::attr(style)").re("width:(\d+)%")[0]) / 20
# to get options we must iterate through every options container
options = {}
for option in review_box.css("div.user-order-info>span"):
name = option.css("strong::text").get("").strip()
value = "".join(option.xpath("text()").getall()).strip()
options[name] = value
# parse remaining fields
parsed.append(
{
"country": review_box.css(".user-country>b::text").get("").strip(),
"text": review_box.xpath('.//dt[contains(@class,"buyer-feedback")]/span[1]/text()').get("").strip(),
"post_time": review_box.xpath('.//dt[contains(@class,"buyer-feedback")]/span[2]/text()').get("").strip(),
"stars": stars,
"order_info": options,
"user_name": review_box.css(".user-name>a::text").get(),
"user_url": review_box.css(".user-name>a::attr(href)").get(),
}
)
return parsed
async def scrape_product_reviews(seller_id: str, product_id: str, session: httpx.AsyncClient):
"""scrape all reviews of aliexpress product"""
async def scrape_page(page):
log.debug(f"scraping review page {page} of product {product_id}")
data = f"ownerMemberId={seller_id}&memberType=seller&productId={product_id}&companyId=&evaStarFilterValue=all+Stars&evaSortValue=sortlarest%40feedback&page={page}¤tPage={page-1}&startValidDate=&i18n=true&withPictures=false&withAdditionalFeedback=false&onlyFromMyCountry=false&version=&isOpened=true&translate=+Y+&jumpToTop=true&v=2"
resp = await session.post(
"https://feedback.aliexpress.com/display/productEvaluation.htm",
data=data,
headers={**session.headers, "Content-Type": "application/x-www-form-urlencoded"},
)
return resp
# scrape first page of reviews and find total count of review pages
first_page = await scrape_page(page=1)
sel = Selector(text=first_page.text)
total_reviews = sel.css("div.customer-reviews").re(r"\((\d+)\)")[0]
total_pages = int(math.ceil(int(total_reviews) / 10))
# then scrape remaining review pages concurrently
log.info(f"scraping reviews of product {product_id}, found {total_reviews} total reviews")
other_pages = await asyncio.gather(*[scrape_page(page) for page in range(1, total_pages + 1)])
reviews = []
for resp in [first_page, *other_pages]:
reviews.extend(parse_review_page(resp))
return reviews对于抓取评论，我们使用我们之前学到的相同的分页习惯用法——我们请求第一页，查找总数并同时检索其余页面。 此外，由于评论仅在 HTML 结构中可用，因此我们必须深入研究 HTML 解析。我们遍历每个评论框并提取核心详细信息，如星级、评论文本和标题等 – 所有这些都使用一些聪明的 XPath 和 CSS 选择器！ 现在，我们有了 Aliexpress 评论抓取工具，让我们试一试。为此，我们需要卖家 ID 和产品 ID，这是我们之前在产品数据抓取工具（字段sellerId和productId）中找到的
运行代码和示例输出
async def run():
async with httpx.AsyncClient(headers=BASE_HEADERS) as session:
print(json.dumps(await scrape_product_reviews("220712488", "4000714658687", session), indent=2))
if __name__ == "__main__":
asyncio.run(run())
[
{
"country": "BR",
"text": "As requested and",
"post_time": "31 May 2022 16:11",
"stars": 5.0,
"order_info": {
"Color:": "DKCD20FU-Li SET2",
"Ships From:": "China",
"Logistics:": "Seller's Shipping Method"
},
"user_name": "S***s",
"user_url": "feedback.aliexpress.com/display/detail.htm?ownerMemberId=XXXXXXXXX==&memberType=buyer"
},
...
]至此，我们已经涵盖了Aliexpress 的主要抓取目标——我们抓取搜索以查找产品，抓取产品页面以查找产品数据，抓取产品评论以收集反馈情报。
网络抓取 aliexpress.com 是否合法？
是的。Aliexpress 产品数据是公开的，我们不会提取任何个人或私人信息。以缓慢、尊重的速度抓取 aliexpress.com 属于道德抓取定义。
