import pymysql
import requests
import retrying
from lxml import etree
from queue import Queue
from feapder.network.user_agent import get
class AmazonShop:
def __init__(self):
# Database connection
self.db = pymysql.connect(host='localhost', user='root', password='root', db='py_spider')
self.cursor = self.db.cursor()
# Target API and Headers
self.amazon_api = 'https://www.amazon.cn/nav/ajax/hamburgerMainContent...'
self.headers = {
"Referer": "https://www.amazon.cn",
"User-Agent": "Mozilla/5.0...",
"X-Requested-With": "XMLHttpRequest",
}
# Queues for pipeline management
self.classify_url_queue = Queue() # Category URLs
self.detail_url_queue = Queue() # Product detail URLs
self.shop_info_queue = Queue() # Final product data
def create_table(self):
sql = """
create table if not exists amazon_shop(
id int primary key auto_increment,
price varchar(255) not null,
title varchar(255) not null,
goods_url varchar(255) not null
);
"""
self.cursor.execute(sql)
@retrying.retry(stop_max_attempt_number=3)
def get_shop_data(self, url):
# Dynamic User-Agent rotation
self.headers['User-Agent'] = get()
# To use proxies, uncomment:
# proxies = {'http': 'http://' + ip, 'https': 'https://' + ip}
# response = requests.get(url, headers=self.headers, proxies=proxies, timeout=2)
response = requests.get(url, headers=self.headers)
return response
def parse_shop_info(self):
while True:
goods_data = self.detail_url_queue.get()
goods_url = 'https://www.amazon.cn' + goods_data['detail_href']
try:
response = self.get_shop_data(goods_url)
tree = etree.HTML(response.text)
# Extract Title
title = tree.xpath('//div[@id="centerCol"]//h1/span/text()')[0] if tree.xpath('//div[@id="centerCol"]//h1/span/text()') else tree.xpath('//title/text()')[0]
# Extract Price
if tree.xpath('//div[@id="centerCol"]//div[@id="apex_desktop"]//span[@class="a-price-whole"]/text()'):
price = "¥" + tree.xpath('//div[@id="centerCol"]//div[@id="apex_desktop"]//span[@class="a-price-whole"]/text()')[0]
else:
price = '-'.join(tree.xpath('//td[@class="a-span12"]//span[@class="a-offscreen"]/text()'))
self.shop_info_queue.put((title.strip(), price, goods_url))
except Exception as e:
print('Error:', e)
finally:
self.detail_url_queue.task_done()
def save_shop_info(self):
while True:
info_list = []
for _ in range(30): # Batch insert 30 records
info = self.shop_info_queue.get()
info_list.append((0,) + info)
self.shop_info_queue.task_done()
sql = "insert into amazon_shop(id, price, title, goods_url) values (%s, %s, %s, %s);"
self.cursor.executemany(sql, info_list)