amazon spider

作者: 是东东 | 来源:发表于2024-03-17 20:46 被阅读0次

经验总结：

一、亚马逊对于国外的IP友好，对于国内IP不友好
小规模使用香港VPN或者美国IP直接通过 request请求即可。robot几率大概是15%
如果使用国内IP robot几率就有90%

验证码识别解决：http://www.886it.cn/

二、如果你是大规模的用户这样，需要考虑以下几个方面：

高性能爬虫设计：

使用高效的异步网络库，如aiohttp或利用分布式爬虫系统，如feapder，并部署多个实例以提高并发能力。

IP解锁与代理管理：

国内代理池的构建和管理，可以自建或使用第三方服务。
实现自动切换IP的中间件，一旦检测到IP被封，自动更换代理。
定期验证代理池中IP的有效性，确保代理IP的质量。

数据解析与存储：

使用feapder框架进行数据抓取和解析，利用其强大的Xpath和CSS选择器。
数据存储可以考虑使用SQLite，MySQL、PostgreSQL或MongoDB。

时间与成本控制：

优化爬虫的抓取策略，比如合理安排爬取时间、避免重复爬取等。
成本控制方面，可以通过提高代理IP的复用率、减少无效请求、优化数据处理流程等方式来降低成本。

代码优化：

对爬虫代码进行性能分析和优化，减少不必要的计算和网络请求。
使用缓存策略，减少对同一URL的重复访问。

解决方案：

采用自建验证码识别系统，减少识别费用产生成本
采用chrome 动态采集+request 采集同时进行。chrome +webdriver 用于识别和解锁ip
这样就能实现数据的高效采集，一边解锁IP，一边上锁IP
这样IP被锁的概率由原先的80%可以降低到40%
解析页面数据
采集完后需要写入数据库中，这里有两种方式写入一种是同步写入，会有并发性能问题，二异步写入，性能提高700%
以下是异步写入数据库的方法
将采集好的数据，先保存为本地json，然后通过其他的进程异步上传，本进程只做采集。进过测试，改造成异步后性能提升700%

代码部分:

识别和解锁ip

is_robot = "/errors/validateCaptcha" in response.text
if (is_robot):
            # 机器人验证弹出 则标记被锁
            print(response.url + "mmmm" + proxy_about + '+++++++++' + "#机器人验证弹出================================")
            #读取验证码图片地址
            'div.a-box-inner div.a-text-center img'
            #执行IP解锁
            solution = get_amazon_code(code_img)

def get_amazon_code(captcha_url="https://images-na.ssl-images-amazon.com/captcha/ahkfsmoa/Captcha_lvghpcxkgh.jpg", captcha_type="amazon", key='B4EF4EV1O1',):
    # max req 20/qps
    payload = {"captcha_url": captcha_url, "key": key, "captcha_type": captcha_type, }
    headers = {
        "Accept": "*/*",
        "Accept-Language": "zh-CN,zh;q=0.9",
        "Cache-Control": "no-cache",
        "Connection": "keep-alive",
        "Content-Type": "application/x-www-form-urlencoded; charset=UTF-8",
        "Origin": "http://www.886it.cn",
        "Pragma": "no-cache",
        "Referer": "http://www.886it.cn/",
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
        "X-Requested-With": "XMLHttpRequest"
    }
    files = []
    response = requests.request("POST", "http://www.886it.cn/api/captcha/code", headers=headers, data=payload, files=files)
    return response.json()['code']

【关键词列表页采集】—【排位解析】
asin_info = {}
asin_info[“title”] = title #商品标记
asin_info[“asin”] = dataAsin #商品asin
asin_info[“price”] = price #价格
asin_info[“ratting”] = ratting #打分
asin_info[“ratting_num”] = parseNumber(ratting_num) #打分数
asin_info[“goods_type”] = goods_type #广告还是自然排名
asin_info[“img_url”] = pic #首图
asin_info[“coupon”] = coupon #优惠券
asin_info[“deal”] = deal #促销
asin_info[“prime”] = prime #会员折扣
asin_info[“promotion”] = promotion
asin_info[“bs”] = bs
asin_info[“ac”] = ac
asin_info[“sales_num”] = sale_num #销量2024年更新
asin_info[“uuid”] = uuid #页面唯一标识

    def parseAdsLocation(self, adsItem, Respons, country_code, asins, pageNo):
        adsRankCount = 0  # 广告总排名
        natureRankCount = 0  # 自然总排名
        BestadsElement = False
        BestnatureElement = False

        # 判断是否存在数据
        searchResultElements = Respons.xpath('//span[@data-component-type="s-search-results"]')
        if (not searchResultElements):
            return [], [], 0

        # 获取所有商品信息
        asinElements = searchResultElements[0].xpath(
            '//div[@class="s-main-slot s-result-list s-search-results sg-row"]/div[@data-uuid!=""]')
        if (len(asinElements) == 0):
            print('没有广告数据====================================')
            return [], [], 0

        asd_asin_arr = []
        for asd_asin in asins.split(","):
            if (len(asd_asin) > 2): asd_asin_arr.append(asd_asin)

        Elements = []
        key_asins = []
        otherPlace = []
        for asinElement in asinElements:

            Element = adsItem.copy()

            # 品牌广告
            asinElement_text = asinElement.extract()
            if ("s-result-item s-widget s-widget-spacing-large AdHolder s-flex-full-width" in asinElement_text):
                for asd_asin in asd_asin_arr:
                    if (asd_asin in asinElement_text):
                        otherPlace.append('品牌广告')
                        break
                continue
            # 视频广告
            if ("sbv-video-single-product" in asinElement_text):
                for asd_asin in asd_asin_arr:
                    if (asd_asin in asinElement_text):
                        otherPlace.append('视频广告')
                        break
                continue
            if ("a-size-medium-plus a-color-base" in asinElement_text and "Amazon’s Choice" in asinElement_text):
                for asd_asin in asd_asin_arr:
                    if (asd_asin in asinElement_text):
                        otherPlace.append('AC推荐')
                        break
                continue
            if ("MAIN-SHOPPING_ADVISER" in asinElement_text and "HIGHLY RECOMMENDED" in asinElement_text):
                for asd_asin in asd_asin_arr:
                    if (asd_asin in asinElement_text):
                        otherPlace.append('ER推荐')
                        break
                continue

            if ("MAIN-SHOPPING_ADVISER" in asinElement_text and "Climate Pledge Friendly" in asinElement_text):
                print("CLIMATE PLEDGE FRIENDLY")
                for asd_asin in asd_asin_arr:
                    if (asd_asin in asinElement_text):
                        otherPlace.append('气候友好保证')
                        break
                continue
            if (
                    "a-size-medium-plus a-color-base" in asinElement_text and "Top rated from our brands" in asinElement_text):
                for asd_asin in asd_asin_arr:
                    if (asd_asin in asinElement_text):
                        otherPlace.append('TR推荐')
                        break
                continue
                # 高度推荐 HIGHLY RATED
            if ("a-size-medium-plus a-color-base" in asinElement_text and "Highly rated" in asinElement_text):
                print("hahahhahahaha**************", asd_asin_arr)
                for asd_asin in asd_asin_arr:
                    if (asd_asin in asinElement_text):
                        otherPlace.append('HR推荐')
                        break
                continue

                # print( Element['ads_rank']," Element['ads_rank']====================")
            sponsoredElements1 = asinElement.css('span[class="s-label-popover-default"]')
            sponsoredElements2 = asinElement.css('span[class="puis-label-popover-default"]')
            sponsoredElementsmx = asinElement.css('span[class="puis-label-popover-hover"]')
            sponsoredElements = False
            if (len(sponsoredElements1) > 0 or len(sponsoredElements2) > 0 or len(sponsoredElementsmx) > 0):
                sponsoredElements = True
            dataAsin = asinElement.css('div::attr(data-asin)').extract_first()
            if (len(str(dataAsin)) < 5): continue

            # 含有Amazon's Choice的商品
            # B0748W31L5-amazons-choice
            amazonsChoice = asinElement.xpath('//*[@id="' + dataAsin + '-amazons-choice"' + ']')
            # 含有best-seller的商品
            bestSeller = asinElement.xpath('//*[@id="' + dataAsin + '-best-seller"' + ']')

            if (amazonsChoice):
                ac = 1
            else:
                ac = 0

            if (bestSeller):
                bs = 1
            else:
                bs = 0

            # 当前产品的信息
            if (sponsoredElements):
                goods_type = "A"
            else:
                goods_type = "N"

            ratting = self.parseRatting(asinElement, country_code)
            ratting_num = asinElement.css('span[class="a-size-base s-underline-text"]::text').extract_first()
            price = self.parsePrice(asinElement, country_code)
            pic = asinElement.css('img[class="s-image"]::attr(src)').extract_first()
            title = self.parseTitle(asinElement, country_code)
            uuid = asinElement.css('div::attr(data-uuid)').extract_first()
            sale_num = asinElement.css(
                'div[class="a-row a-size-base"] span[class="a-size-base a-color-secondary"]::text').extract_first()

            coupon = asinElement.css(
                'span[class="a-size-base s-highlighted-text-padding aok-inline-block s-coupon-highlight-color"]::text').extract_first()
            if (coupon == None):
                coupon = ""
            if ("save" not in coupon.lower()):
                coupon = ""

            deal = ""
            deal = asinElement.css('span[class="a-badge-label-inner a-text-ellipsis"] span::text').extract_first()
            if (deal == None):
                deal = ""
            if ("deal" not in deal.lower()):
                deal = ""

            promotion = ""
            prime = ""
            secondary = asinElement.css('div[class="a-row a-size-base a-color-secondary"] span::text').extract_first()
            if (secondary == None):
                secondary = ""

            if ("prime" in secondary.lower()):
                prime = secondary.replace("join ", "").replace("to buy this item ", "")
            if ("promotion" in secondary.lower()):
                promotion = secondary
                # print(Respons.url)
            # print("ratting_num:",ratting_num)
            asin_info = {}
            asin_info["title"] = title  #商品标记
            asin_info["asin"] = dataAsin  #商品asin
            asin_info["price"] = price    #价格
            asin_info["ratting"] = ratting    #打分
            asin_info["ratting_num"] = parseNumber(ratting_num)  #打分数
            asin_info["goods_type"] = goods_type  #广告还是自然排名
            asin_info["img_url"] = pic   #首图
            asin_info["coupon"] = coupon   #优惠券
            asin_info["deal"] = deal   #促销
            asin_info["prime"] = prime  #会员折扣
            asin_info["promotion"] = promotion   
            asin_info["bs"] = bs   
            asin_info["ac"] = ac
            asin_info["sales_num"] = sale_num   #销量2024年更新
            asin_info["uuid"] = uuid  #页面唯一标识

            if (sponsoredElements):
                # 广告排名
                adsRankCount = adsRankCount + 1
                # 当前页广告数

                if (dataAsin in asins):
                    # print(adsRankCount,"广告排名 in ****************")
                    # 记录广告排名与页码以及页码排名
                    Element['goods_type'] = "A"
                    Element['ads_page_rank'] = adsRankCount
                    Element['ads_page_no'] = pageNo
                    Element['ads_asin_info'] = json.dumps(asin_info)
                    if (BestadsElement == False):
                        BestadsElement = Element

            else:
                # 自然排名
                natureRankCount = natureRankCount + 1
                # 当前页自然排名
                if (dataAsin in asins):
                    # print(adsRankCount,  "自然排名 in****************")
                    Element['goods_type'] = "N"
                    Element['nature_page_rank'] = natureRankCount
                    Element['nature_page_no'] = pageNo
                    Element['nature_asin_info'] = json.dumps(asin_info)
                    if (BestnatureElement == False):
                        BestnatureElement = Element

            key_asins.append(asin_info)

        all_count=0
        if (BestnatureElement != False):
            Elements.append(BestnatureElement)

        if (BestadsElement != False):
            Elements.append(BestadsElement)
            # 检查是否底部有广告
        if (len(Elements) == 0):
            Elements.append(adsItem)

        Elements_new = []
        for Element in Elements:
            Element['search_num'] = all_count
            Element['ads_page_num'] = adsRankCount
            Element['nature_page_num'] = natureRankCount
            Element["other_place"] = otherPlace
            Elements_new.append(Element)

        return Elements_new, key_asins, adsRankCount

入库效果

这里是亚马逊amazon 解析页面评论的python demo

def parseReview_element(self, element, countryCode):

        # review唯一标识
        reviewId = element.css('div::attr(id)').extract_first()
        if (reviewId == None):
            # 不是本国的review，跳过
            return None
        element_head = element.css("#customer_review-" + reviewId).extract_first()
        if (element_head == None):
            # 不是本国的review，跳过
            return None

        # 亚马逊用户id与名称

        amazonUserId = self.user(element)

        amazonUserName = str(element.css(".a-profile-name::text").extract_first())[0: 90]

        # 星级与title
        star = self.star(element)
        title = self.title(element)
        # review date
        reviewDateElement = element.css('span[data-hook*="review-date"]').extract_first()

        reviewDate = ReviewDateParseHelper().reviewDate(html_clear(reviewDateElement))
        if (reviewDate == ReviewDateParseHelper().LocalDateTimeUtils.parse("2099-12-31",
                                                                           ReviewDateParseHelper().LocalDateTimeUtils.YYYY_MM_DD)):
            logging.debug("解析评论日期出错*********************************")
            logging.debug(element.extract())
            logging.debug("解析评论日期出错*********************************")
        # Verified_Purchase 购买认证
        verifiedElement = element.css('span[data-hook*="avp-badge"]').extract_first()
        verified = 1 if verifiedElement else 0

        # 点赞数
        helpful_num = element.css('span[data-hook*="helpful-vote-statement"]::text').extract_first()
        if (helpful_num):
            helpful_num = "".join(list(filter(str.isdigit, helpful_num)))
            helpful_num = int(helpful_num) if (helpful_num != "") else 1
        else:
            helpful_num = 0

        # *review 内容
        content = html_clear(element.css('span[data-hook*="review-body"]').extract_first())
        originReviewDate = html_clear(reviewDateElement)
        user_info_ext = ",".join(element.css('span[class*="c7yTopDownDashedStrike"]::text').extract())

        # 解析media

        videos = []
        has_video = element.css("div[class*='cr-video-desktop']").extract_first()
        if (has_video):
            video = {}
            video['videoUrl'] = element.css("input[value$='.mp4']::attr(value)").extract_first()
            video['videoSlateImgUrl'] = element.css(
                "div[class*='cr-video-desktop']::attr(data-thumbnail-url)").extract_first()
            videos.append(video)

        pics = []
        has_pics = element.css("div[class*=review-image-tile-section] span a img::attr(src)").extract()
        if (has_pics):
            for pic in has_pics:
                pics.append(pic)

        if(len(pics)==0):
            has_pics2 = element.css("div[class*=cr-lightbox-image-thumbnails] img::attr(src)").extract()
            if (has_pics2):
                for pic in has_pics2:
                    pics.append(pic)

        if (len(videos) > 0 or len(pics) > 0):
            media = {}
            media['videos'] = videos
            media['reviewImageUrls'] = list(set(pics))
            media = json.dumps(media, ensure_ascii=False)
        else:
            media = ""
        print("media:==============",media)
        content = content.replace('The media could not be loaded.', " ").strip().replace("\n", " ").replace("\r", " ")

        Review = {}
        Review['country_code'] = countryCode
        Review['review_id'] = reviewId
        Review['amazon_user_id'] = amazonUserId
        Review['amazon_user_name'] = amazonUserName
        Review['verified'] = verified
        Review['star'] = star
        Review['title'] = title
        Review['content'] = content
        Review['helpful_num'] = helpful_num  # helpful_num
        Review['review_date'] = reviewDate
        Review['user_info_ext'] = user_info_ext
        Review['origin_review_date'] = originReviewDate
        Review['media'] = media
        Review['create_time'] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time())))
        Review['update_time'] = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime(int(time.time())))
        return Review

亚马逊日期采集是一个非常繁琐的事情，涵盖全部站点，所有语言US FR DE IT ES SE AE 包括阿拉伯语等，特别是阿拉伯语，最让人吐血
如下是解析US UK CA MX站点日期的代码，大家可以参考然后取解析其他站点的数据。
日期解析只能一个站点一个站点的解析，没有其他技巧

def reviewDateUK(self, reviewDate):
        try:
            reviewDate = re.sub(".*on ", "",reviewDate)
            reviewDate = re.sub(",", "",reviewDate)
            dates = reviewDate.split(" ")
            reviewDate = dates[2] + "-" + self.LocalDateTimeUtils.transferMonth(dates[1]) + "-" + dates[0]

            parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_M_D)
            if (parse == None):
                parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_MM_DD)
            return parse
        except BaseException:
            return self.LocalDateTimeUtils.parse("2099-12-31", self.LocalDateTimeUtils.YYYY_MM_DD)

    """
     * US->on February 26, 2016
     """

    def reviewDateUS(self, reviewDate):
        try:
            reviewDate = re.sub( ".*on ", "",reviewDate)
            reviewDate = re.sub(",", "",reviewDate)
            dates = reviewDate.split(" ")
            reviewDate = dates[2] + "-" + self.LocalDateTimeUtils.transferMonth(dates[0]) + "-" + dates[1]
            parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_M_D)
            if (parse == None):
                parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_MM_DD)

            return parse
        except BaseException:
            return self.LocalDateTimeUtils.parse("2099-12-31", self.LocalDateTimeUtils.YYYY_MM_DD)

    """
     * AU->on 26 February, 2016
     """

    def reviewDateAU(self, reviewDate):
        try:
            reviewDate = re.sub(".*on ", "",reviewDate)
            reviewDate = re.sub(",", "",reviewDate)
            dates = reviewDate.split(" ")
            reviewDate = dates[2] + "-" + self.LocalDateTimeUtils.transferMonth(dates[1]) + "-" + dates[0]

            parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_M_D)
            if (parse == None):
                parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_MM_DD)
            return parse
        except BaseException:
            return self.LocalDateTimeUtils.parse("2099-12-31", self.LocalDateTimeUtils.YYYY_MM_DD)

    """
     * US->on February 26, 2016
     * CA->on March 7, 2016
     """

    def reviewDateCA(self, reviewDate):
        try:
            reviewDate = re.sub(".*on ", "",reviewDate)
            reviewDate = re.sub(",", "",reviewDate)
            dates = reviewDate.split(" ")
            reviewDate = dates[2] + "-" + self.LocalDateTimeUtils.transferMonth(dates[0]) + "-" + dates[1]

            parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_M_D)
            if (parse == None):
                parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_MM_DD)
            return parse
        except BaseException:
            return self.LocalDateTimeUtils.parse("2099-12-31", self.LocalDateTimeUtils.YYYY_MM_DD)

    """
     * US->on February 26, 2016
     * MX->en 8 de noviembre de 2016
     """

    def reviewDateMX(self, reviewDate):
        try:
            reviewDate = re.sub( ".*el ", "",reviewDate)
            dates = reviewDate.split(" de ")
            reviewDate = dates[2] + "-" + self.LocalDateTimeUtils.transferMonth(dates[1]) + "-" + dates[0]

            parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_M_D)
            if (parse == None):
                parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_MM_DD)
            return parse
        except BaseException:
            return self.LocalDateTimeUtils.parse("2099-12-31", self.LocalDateTimeUtils.YYYY_MM_DD)

    """
     *
     * @param reviewDate
     """

    def reviewDateJP(self, reviewDate):
        try:
            # 2021-3-18に本でレビュー済み
            reviewDate = reviewDate.replace("に本でレビュー済み", "")
            reviewDate = reviewDate.replace("に日本でレビュー済み", "")
            reviewDate = reviewDate.replace("评论于", "")
            reviewDate = reviewDate.replace("在日本 🇯🇵 发布", "")
            reviewDate = re.sub("年", "-",reviewDate)
            reviewDate = re.sub("月", "-",reviewDate)
            reviewDate = re.sub("日", "",reviewDate)
            reviewDate = reviewDate.strip()
            parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_M_D)
            if (parse == None):
                parse = self.LocalDateTimeUtils.parse(reviewDate, self.LocalDateTimeUtils.YYYY_MM_DD)
            return parse
        except BaseException as e:
            print("s",e,"s")
            return self.LocalDateTimeUtils.parse("2099-12-31", self.LocalDateTimeUtils.YYYY_MM_DD)

amazon spider

经验总结：

高性能爬虫设计：

解决方案：

代码部分:

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读