阿里巴巴供应商爬虫
起因
学了爬虫入门之后，打算找一个有难度的网站来实践，一开始打算找淘宝或者天猫（业界老大）来实践，但后续发现网上已经有很多这方面的项目，于是瞄上了阿里的国际网站阿里巴巴。开始一切顺利，没发现什么难度，后面发现供应商的联系方式需要登录，于是以其为目标开始写爬虫。
网站结构
1.阿里巴巴有个供应商分类的页面
https://www.alibaba.com/companies
2.点进去之后有个二级分类
3.再点进去就可以找到这个分类下的所有供应商
4.再进去就可以找供应商的联系方式了
5.这时要获取联系方式就需要登录
爬虫思路
1.先从供应商分类页面开始爬取，获取到所有的供应商联系方式的页面（即上面网站结构的第4步）存入数据库
2.编写另一个爬虫从数据库拿到联系方式页面，再用selenium登录获取联系方式（这一步是取巧了，但也导致爬取速度直线下降）
代码结构
#scrapy的代码结构
alibaba
-spider
--company.py #获取供应商联系方式页面
--contact.py #获取联系方式的页面
-items
-middlewares.py
-pipelines.py
-setting.pycompany.py
class CompanySpider(scrapy.Spider):
name = 'company'
allowed_domains = ['www.alibaba.com']
start_urls = ['https://www.alibaba.com/companies']
# 爬取大的分类信息
def parse(self, response):
categorys = response.css(".ui-box.ui-box-normal.ui-box-wrap.clearfix .g-cate-list dl dt a")
for cat in categorys:
name = cat.css("::text").extract_first()
url = cat.css("::attr(href)").extract_first()
yield scrapy.Request(url, callback=self.parseCategory, meta={'cat': name})
pass
# 爬取详细分类信息
def parseCategory(self, response):
cat = response.meta['cat']
ccats = response.css("#category-main-box .g-float-left>ul>li>a")
for ccat in ccats:
name = ccat.css("::text").extract_first()
url = ccat.css("::attr(href)").extract_first()
ccat_url = urljoin(self.start_urls[0], url)
yield scrapy.Request(ccat_url, callback=self.parseCompany, meta={'ccat': name, 'cat': cat})
pass
def parseCompany(self, response):
cat = response.meta['cat']
ccat = response.meta['ccat']
# 爬取下一页
next_url = response.css(".next::attr(href)").extract_first()
if (next_url):
yield scrapy.Request(urljoin(response.url, next_url), callback=self.parseCompany, meta={'cat': cat, 'ccat': ccat})
# 爬取公司信息
items = response.css(".item-main")
for item in items:
company = CompanyItem()
title = item.css(".item-title .title.ellipsis a")
company["name"] = title.css("::text").extract_first().strip()
url = title.css("::attr(href)").extract_first()
url = url.rpartition("/")[0]
company["url"] = url
attrs = item.css(".content.util-clearfix .right .attr")
for attr in attrs:
name = attr.css(".name::text").extract_first()
value = attr.css(".value")
if name == 'Main Products:':
mainProducts = value.css("::text").extract_first()
company["mainProducts"] = mainProducts
elif name == "Country/Region:":
region = value.css(".ellipsis.search::text").extract_first()
company["region"] = region
elif name == "Total Revenue:":
totalRevenue = value.css(".ellipsis.search::text").extract_first()
company["totalRevenue"] = totalRevenue
elif name == "Top 3 Markets:":
top3 = value.css(".ellipsis.search::text").extract()
company["top3Markets"] = ",".join(top3)
company["ccat"] = ccat
company["cat"] = cat
yield company
passcontact.py
class ContactSpider(scrapy.Spider):
name = 'contact'
allowed_domains = ['www.alibaba.com']
start_urls = []
count = 1000
def __init__(self):
# selenium
# options = Options()
# prefs = {"profile.managed_default_content_settings.images": 2}
# options.add_experimental_option("prefs", prefs)
# options.headless = True
# driver_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))+"\chromedriver.exe"
# print(driver_path)
# 创建Firefox浏览器的一个Options实例对象
profile = webdriver.FirefoxProfile()
# 禁用CSS加载
profile.set_preference('permissions.default.stylesheet', 2)
# 禁用images加载
profile.set_preference('permissions.default.image', 2)
# 禁用flash插件
fireFoxOptions = webdriver.FirefoxOptions()
fireFoxOptions.headless = True
profile.set_preference('dom.ipc.plugins.enabled.libflashplayer.so', False)
self.browser = webdriver.Firefox(executable_path=settings.FIREFOX_WEBDRIVER_PATH,
firefox_profile=profile, firefox_options=fireFoxOptions)
# mysql
self.client = pymysql.connect(
host=settings.MYSQL_HOST,
db=settings.MYSQL_DBNAME,
user=settings.MYSQL_USER,
passwd=settings.MYSQL_PASSWD,
charset='utf8'
)
self.cur = self.client.cursor()
# startUrls
sql = "select id,url from disct_company where id <= {} and id > {}".format(settings.END, settings.START)
try:
self.cur.execute(sql)
results = self.cur.fetchall()
for url in results:
id = url[0]
contact_url = url[1] # type: str
if not (contact_url.startswith("http:") or contact_url.startswith("https:")):
contact_url = "https:" + contact_url
if not (contact_url.endswith("/")):
contact_url += "/"
contact_url += ("contactinfo.html?%d" % (id))
self.start_urls.append(contact_url)
except Exception as error:
logging.error("contact spider init err:{}".format(error))
# 关闭资源
dispatcher.connect(receiver=self.spiderCloseHandler, signal=signals.spider_closed)
def spiderCloseHandler(self, spider):
self.browser.quit()
self.cur.close()
self.client.close()
def parse(self, response):
id = response.meta.get("id")
item = ContactItem()
item['id'] = id
name = response.css(".contact-name::text").extract_first()
if not name:
name = response.css(".contact-info .name::text").extract_first()
if name:
name = name.strip()
item['name'] = name
department = response.css(".contact-department::text").extract_first()
job = response.css(".contact-job::text").extract_first()
if not (department and job):
dds = response.css(".contact-info .dl-horizontal dd::text")
dts = response.css(".contact-info .dl-horizontal dt::text")
if dds:
for i in range(len(dds)):
dd = dds[i].extract()
dt = dts[i].extract()
if dt == 'Department:':
department = dd
elif dt == 'Job Title:':
job = dd
item['department'] = department
item['job'] = job
info_table = response.css(".info-table tr")
if info_table:
for info in info_table:
th = info.css("th::text").extract_first()
td = info.css("td::text").extract_first()
if th == 'Telephone:':
item['telephone'] = td
elif th == 'Mobile Phone:':
item['mobilePhone'] = td
elif th == 'Fax:':
item['fax'] = td
elif th == 'Address:':
item['address'] = td
elif th == 'Zip:':
item['zip'] = td
elif th == 'Country/Region:':
item['country'] = td
elif th == 'Province/State:':
item['province'] = td
elif th == 'City:':
item['city'] = td
else:
item['telephone'] = response.css(
".contact-detail .sensitive-info .dl-horizontal dd[data-role=phone]::text").extract_first()
item['mobilePhone'] = response.css(
".contact-detail .sensitive-info .dl-horizontal dd[data-role=mobile]::text").extract_first()
item['fax'] = response.css(
".contact-detail .sensitive-info .dl-horizontal dd[data-role=fax]::text").extract_first()
dds = response.css(".contact-detail .public-info .dl-horizontal dd::text").extract()
dts = response.css(".contact-detail .public-info .dl-horizontal dt::text").extract()
for i in range(len(dts)):
dt = dts[i]
dd = dds[i]
if dt == 'Address:':
item['address'] = dd
elif dt == 'Zip:':
item['zip'] = dd
elif dt == 'Country/Region:':
item['country'] = dd
elif dt == 'Province/State:':
item['province'] = dd
elif dt == 'City:':
item['city'] = dd
yield itemseleniumDownloaderMiddleware
class SeleniumDownloaderMiddleware(object):
def process_request(self, request, spider):
if spider.name == 'contact':
try:
url = request.url
spider.browser.get(url)
time.sleep(1)
if self.is_element_exist(".sens-mask .icbu-link-default", spider.browser):
viewButton = spider.browser.find_element_by_css_selector(".sens-mask .icbu-link-default")
elif self.is_element_exist(".contact-detail-mask .view-contact-trigger", spider.browser):
viewButton = spider.browser.find_element_by_css_selector(".contact-detail-mask .view-contact-trigger")
if viewButton:
spider.browser.execute_script("arguments[0].click();", viewButton)
# spider.browser.execute_script("arguments[0].scrollIntoView();", viewButton)
time.sleep(2)
if self.is_element_exist(".sc-hd-prefix2-dialog-bd", spider.browser):
spider.browser.execute_script("arguments[0].click();", spider.browser.find_element_by_css_selector(".sc-hd-prefix2-tab-trigger"))
time.sleep(1)
spider.browser.switch_to.frame("alibaba-login-box")
spider.browser.find_element_by_id("fm-login-id").send_keys("your username")
time.sleep(1)
spider.browser.find_element_by_id("fm-login-password").send_keys("your password")
time.sleep(1)
# submit = spider.browser.find_element_by_id("fm-login-submit")
spider.browser.execute_script("document.getElementById('fm-login-submit').click();")
# spider.browser.execute_script("arguments[0].scrollIntoView();", submit)
# submit.click()
time.sleep(5)
spider.browser.switch_to.default_content()
except Exception as error:
logging.error("{} err:{}".format(url, error))
return HtmlResponse(spider.browser.current_url, body=spider.browser.page_source, encoding='utf-8',
request=request)
return None
def is_element_exist(self, css, browser):
s = browser.find_elements_by_css_selector(css_selector=css)
if len(s) == 0:
return False
elif len(s) == 1:
return True
else:
return False项目地址
项目已上传github：https://github.com/shuo123/alibabaSpider