project/spides/spiders/taobao/taobao_crawl.py

import hashlib
import json
import random
import re
import time
from decimal import Decimal, InvalidOperation
from curl_cffi import requests
from lxml import etree
from commons.Logger import get_spider_logger
from commons.conn_mysql import MySQLPoolOn2
from pipelines.drug_pipelines import DrugPipeline
from spiders.taobao.taobao_login import (TaobaoAutoCrawl)
from area_info.city_name_to_id import get_city
from urllib.parse import quote, unquote
logger = get_spider_logger("taobao")

MTOP_APP_KEY = "12574478"
MTOP_APP_ID = "34385"
SEARCH_MAX_PAGE = 20
REQUEST_RETRY_COUNT = 3
COOKIE_MAX_AGE_SEC = 3600

headers = {
    "accept": "*/*",
    "accept-language": "zh-CN,zh;q=0.9",
    "referer": "https://s.taobao.com/search?page=1&q=999%E6%84%9F%E5%86%92%E7%81%B5&spm=a21bo.jianhua%2Fa.201867-main.d4_first.42f72a89n1ITMs&tab=mall",
    "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
    "sec-ch-ua-mobile": "?0",
    "sec-ch-ua-platform": '"Windows"',
    "sec-fetch-dest": "script",
    "sec-fetch-mode": "no-cors",
    "sec-fetch-site": "same-site",
    "user-agent": (
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
        "(KHTML, like Gecko) Chrome/142.0.0.0 Safari/537.36"
    ),
}

MTOP_URL = (
    "https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/"
)


def get_product_name(item):
    """递归查找 item 中 propertyName='产品名称' 的 propertyValueName"""
    if isinstance(item, dict):
        if item.get('propertyName') == '产品名称':
            return item.get('propertyValueName', '')
        for v in item.values():
            result = get_product_name(v)
            if result:
                return result
    elif isinstance(item, list):
        for sub in item:
            result = get_product_name(sub)
            if result:
                return result
    return ''


class TaobaoCrawl:
    def __init__(self, drug_dict=None,scheduler = None):
        self.cookies = None
        self.db = MySQLPoolOn2()
        self.pipeline = DrugPipeline("taobao")
        self.session = None
        self.proxies = None
        self.account_name = None
        self.ip = None
        self.cookie_stamp = None
        self.platform = 1
        self.task_dict = drug_dict or {}
        self.collect_task_id = None
        self.success = True
        self.scheduler = scheduler 
        if self.task_dict:
            self.get_product_data()
        self.is_no_product = 0

    def get_product_data(self):
        self.task_id = self.task_dict["id"]
        self.company_id = self.task_dict["company_id"]
        self.product = self.task_dict["product_name"]
        self.product_desc = self.task_dict.get("product_specs", "")
        self.brand = self.task_dict.get("product_brand", "")
        self.product_keyword = self.task_dict.get("product_keyword", "")
        self.collect_task_id = self.task_dict.get("collect_task_id", "")
        self.sampling_cycle = self.task_dict.get("sampling_cycle", "")
        self.sampling_start_time = self.task_dict.get("sampling_start_time", "")
        self.sampling_end_time = self.task_dict.get("sampling_end_time", "")
        self.collect_equipment_id = self.task_dict.get("collect_equipment_id", "")
        self.account_id = self.task_dict.get("collect_equipment_account_id", "")
        self.collect_region_id = self.task_dict.get("collect_region_id", "")
        self.collect_round = self.task_dict.get("collect_round", 1)
        self.start_page = self.task_dict.get(self.task_dict.get("current_page")+1, 1)
        self.report_data = {'task_id':self.task_id,'platform':self.platform,'username':self.scheduler.username}

    def _error_report(self,data):
            self.report_data.update(data)
            self.scheduler.stop()
            self.scheduler.post_report(self.report_data)

    def _success_report(self,data):
            self.report_data.update(data)
            self.scheduler.post_report(self.report_data)

    @staticmethod
    def _normalize_url(url):
        if not url:
            return ""
        url = str(url)
        if url.startswith("//"):
            return "https:" + url
        return url

    @staticmethod
    def _extract_shop_id(shop_url):
        if not shop_url:
            return ""
        shop_re = re.search(r"appUid=(\w+)", shop_url)
        if shop_re:
            return shop_re.group(1)
        return hashlib.md5(shop_url.encode("utf-8")).hexdigest()

    @staticmethod
    def _sql_literal(value):
        """避免拼接账号名时单引号打断 SQL（非完整防注入，仅兜底）。"""
        if value is None:
            return ""
        return str(value).replace("'", "''").replace("\\", "\\\\")

    def init_session(self):
        self.session = requests.Session(impersonate="chrome124")
        self.session.cookies.update(self.cookies or {})
        self.session.headers.update(headers)
        if self.proxies:
            self.session.proxies.update(self.proxies)

    def init_session_lose(self,res,keyword,page):
        if '哎哟喂,被挤爆啦' in res.text:
            if not self.update_cookie():
                return False

            res = self._request_search_page(keyword, page)
            if '哎哟喂,被挤爆啦' in res.text:
                return False

        return True


    @staticmethod
    def _is_transport_error(err):
        msg = str(err or "")
        return ("curl: (16)" in msg) or ("Failed to perform" in msg)

    def get_token(self, t, app_key, data_str):
        _m_h5_tk = (self.cookies or {}).get("_m_h5_tk", "")
        token = _m_h5_tk.split("_")[0] if _m_h5_tk else ""
        text = f"{token}&{t}&{app_key}&{data_str}"
        return hashlib.md5(text.encode()).hexdigest()

    def get_html_content(self, res_html):
        if not res_html:
            return ""
        ele_html = etree.HTML(res_html)
        if ele_html is None:
            return str(res_html)
        text_list = ele_html.xpath(".//text()")
        return "".join(text_list)

    def _build_search_payload(self, keyword, page, page_size=50):
        return {
            "appId": MTOP_APP_ID,
            "params": {
                "device": "HMA-AL00",
                "isBeta": "false",
                "grayHair": "false",
                "from": "nt_history",
                "brand": "HUAWEI",
                "info": "wifi",
                "index": "4",
                "rainbow": "",
                "schemaType": "auction",
                "elderHome": "false",
                "isEnterSrpSearch": "true",
                "newSearch": "false",
                "network": "wifi",
                "subtype": "",
                "hasPreposeFilter": "false",
                "prepositionVersion": "v2",
                "client_os": "Android",
                "gpsEnabled": "false",
                "searchDoorFrom": "srp",
                "debug_rerankNewOpenCard": "false",
                "homePageVersion": "v7",
                "searchElderHomeOpen": "false",
                "search_action": "initiative",
                "sugg": "_4_1",
                "sversion": "13.6",
                "style": "list",
                "ttid": "600000@taobao_pc_10.7.0",
                "needTabs": "true",
                "areaCode": "CN",
                "vm": "nw",
                "countryNum": "156",
                "m": "pc",
                "page": page,
                "n": 48,
                "q": keyword,
                "qSource": "url",
                "pageSource": "",
                "channelSrp": "",
                "tab": "all",
                "pageSize": str(page_size),
                "sourceS": "2",
                "ntoffset": "0",
                "filterTag": "",
                "service": "",
                "prop": "",
                "loc": "",
                "categoryp": "",
                "screenResolution": "1920x1080",
                "viewResolution": "1092x4722",
                "userAgent": headers["user-agent"],
                "couponUnikey": "",
                "subTabId": "",
                "np": "",
                "clientType": "h5",
                "isNewDomainAb": "false",
                "forceOldDomain": "false",
            },
        }

    def _request_search_page(self, keyword, page):
        t = str(int(time.time() * 1000))
        data = self._build_search_payload(keyword, page)
        data_str = json.dumps(data, separators=(",", ":"))
        sign = self.get_token(t, MTOP_APP_KEY, data_str)
        params = {
            "jsv": "2.7.4",
            "appKey": MTOP_APP_KEY,
            "t": t,
            "sign": sign,
            "api": "mtop.relationrecommend.wirelessrecommend.recommend",
            "v": "2.0",
            "timeout": "10000",
            "type": "jsonp",
            "dataType": "jsonp",
            "callback": "",
            "data": data_str,
        }
        return self.session.get(MTOP_URL, params=params, timeout=30)

    def _parse_jsonp_body(self, res_text):
        res_text = (res_text or "").strip()
        json_str = res_text
        m = re.match(r"^[^(]*\((.*)\)\s*;?\s*$", res_text, re.DOTALL)
        if m:
            json_str = m.group(1)
        return json.loads(json_str)

    def get_search(self):
        keyword = self.product
        if self.brand:
            keyword = (self.brand + " " + self.product).strip()

        if self.product_desc:
            keyword = (keyword + " " + self.product_desc).strip()

        for page in range(self.start_page,30):
            logger.info(f"正在爬取关键词：{keyword}，{page}页数据")
            for attempt in range(1, REQUEST_RETRY_COUNT + 1):
                try:
                    if not self.session:
                        self.init_session()
                    res = self._request_search_page(keyword, page)
                    if not self.init_session_lose(res,keyword,page):
                        #这里要警告，属于账号异常，账号即使刷新还是不恢复
                        self.success=False
                        self._error_report({'is_finished':0,'need_reassign':1,'current_page':self.start_page,'exception_type':2})
                        return

                    if res.status_code != 200:
                        logger.warning(
                            "请求失败 HTTP %s，第%s/%s 次重试",
                            res.status_code,
                            attempt,
                            REQUEST_RETRY_COUNT,
                        )
                        time.sleep(random.randint(3, 8))
                        continue
                except Exception as e:
                    if self._is_transport_error(e):
                        logger.warning(
                            "检测到网络传输异常(curl)，重建会话后重试: %s",
                            e,
                        )
                        self.init_session()
                    logger.warning(
                        "请求异常，第%s/%s 次重试: %s",
                        attempt,
                        REQUEST_RETRY_COUNT,
                        e,
                    )
                    # 指数退避，避免连续瞬时失败
                    time.sleep(min(3 * attempt, 10))
                    continue

                try:
                    json_data = self._parse_jsonp_body(res.text)
                    item_array = json_data.get("data", {}).get("itemsArray", [])
                    break
                except Exception as e:
                    logger.warning(
                        "解析数据异常，%s 账号可能退出登录，尝试重新登录: %s",
                        self.account_name,
                        e,
                    )
                    if self.update_cookie():
                        self.init_session()
                    else:
                        self._error_report({'is_finished':0,'need_reassign':1,'current_page':page,'exception_type':2})
                        return

            else:
                logger.warning("关键词 %s 第 %s 页连续重试失败", keyword, page)
                continue

            if not item_array:
                logger.warning("关键词 %s 第 %s 页未获取到商品数据", keyword, page)
                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page})
                return

            for raw in item_array:
                try:
                    item_id = raw.get("item_id", "")
                except:
                    break
                if not item_id:
                    continue

                item_title = self.get_html_content(raw.get("title") or "")
                try:
                    propertyValueName =get_product_name(raw)
                except:
                    propertyValueName = ''

                if self.brand not in item_title:
                    self.is_no_product += 1
                    continue
                if propertyValueName:
                    if self.product not in propertyValueName:
                        self.is_no_product += 1
                        continue
                else:
                    if self.product not in item_title:
                        self.is_no_product += 1
                        continue
                if "+" in item_title:
                    continue

                if self.product_desc:
                    if self.product_desc in item_title:
                        crawl_product_desc = self.product_desc
                    else:
                        crawl_product_desc = ""
                else:
                    crawl_product_desc = ""

                self.is_no_product = 0
                status = 1
                if self.product_keyword:
                    search_keyword_list = self.product_keyword.split(",")
                    for search_keyword in search_keyword_list:
                        if search_keyword.strip() not in item_title:
                            status = 0
                if status == 0:
                    continue

                item_price = raw.get("price")
                item_price_show = raw.get("priceShow", {}).get("price", 0)
                item_sales = raw.get("realSales") or ""

                sale_num = ""
                sales_m = re.search(r"(.*?)人付款", item_sales)
                if sales_m:
                    sale_num = sales_m.group(1)

                item_url = self._normalize_url(raw.get("auctionURL"))
                match = re.search(r'skuId=([^&]*)', unquote(item_url))

                if match:
                    if match.group(1)!='null':
                        item_id = item_id + '_' + match.group(1)



                shop_name = raw.get("shopInfo", {}).get("title", "")

                area_str = (raw.get("procity", "") or "").strip()
                city_id, province_id, city, province = get_city(area_str)

                shop_url = self._normalize_url(
                    raw.get("shopInfo", {}).get("url", "")
                )

                structured_list = raw.get("structuredUSPInfo",{})
                for structured in structured_list:
                    if structured.get("propertyName","") == "规格":
                        crawl_product_desc = structured.get("propertyValueName","")

                pic_path = raw.get("pic_path", "")
                raw_price = item_price_show
                if raw_price in (None, ""):
                    price = Decimal("0.00")
                else:
                    try:
                        price = Decimal(str(raw_price)).quantize(Decimal("0.00"))
                    except (InvalidOperation, ValueError):
                        price = Decimal("0.00")

                scrape_date = time.strftime("%Y-%m-%d")
                update_time = time.strftime("%Y-%m-%d %H:%M:%S")
                snapshot_url = self._normalize_url(pic_path) if pic_path else ""

                # 字段与 yaofangwang_crawl 对齐；键顺序须与 commons.sql_data.RETRIEVE_SCRAPE_INSERT_COLUMNS 一致
                product = {
                    "platform": self.platform,
                    "item_id": item_id,
                    "enterprise_id": self.company_id,
                    "product_name": item_title,
                    "spec": crawl_product_desc,
                    "one_price": "",
                    "detail_url": item_url,
                    "shop_name": shop_name,
                    "anonymous_store_name": "",
                    "shop_url": shop_url,
                    "city_name": "",
                    "city_id": "",
                    "province_name": "",
                    "province_id": "",
                    "shipment_city_name": city,
                    "shipment_city_id": city_id,
                    "shipment_province_name": province,
                    "shipment_province_id": province_id,
                    "area_info": area_str,
                    "factory_name": "",
                    "scrape_date": scrape_date,
                    "price": price,
                    "sales": sale_num,
                    "stock_count": "",
                    "snapshot_url": "",
                    "approval_num": "",
                    "produced_time": "",
                    "deadline": "",
                    "update_time": update_time,
                    "insert_time": update_time,
                    "number": 1,
                    "product_brand": self.brand or "",
                    "collect_task_id": self.collect_task_id,
                    "task_id": self.task_id,
                    "search_name": self.product,
                    "company_name": "",
                    "collect_config_info": json.dumps(
                        {
                            "sampling_cycle": self.sampling_cycle,
                            "sampling_start_time": self.sampling_start_time,
                            "sampling_end_time": self.sampling_end_time,
                        }
                    ),
                    "account_id": self.account_id,
                    "collect_region_id": self.collect_region_id,
                    "collect_round": self.collect_round,
                    "is_sold_out": 0

                }

                try:
                    self.pipeline.storge_data(product)
                    logger.info("%s", json.dumps(product, ensure_ascii=False, default=str))
                except Exception as e:
                    logger.exception("写入数据库失败: %s", e)

            logger.info(
                "关键词 %s 第 %s 页爬取完成",
                keyword,
                page,
            )

            total_page = (
                (json_data or {}).get("data", {}).get("mainInfo", {}).get("totalPage")
            )

            try:
                total_page_int = int(total_page) if total_page is not None else 50
            except (TypeError, ValueError):
                total_page_int = 50
            self.report_data.update({'total_pages':total_page_int})

            if self.scheduler.end ==True:
                logger.info('心跳失败')
                self._error_report({'is_finished':0,'need_reassign':0,'current_page':page})
                break

            if page >= total_page_int:
                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page})
                break
            if self.is_no_product > 100:
                self._success_report({'is_finished':1,'need_reassign':0,'current_page':page})
                break
            sleep_second = random.uniform(30, 60)
            logger.info("第 %s 页爬取完成，休息 %.1fs", page, sleep_second)
            self._success_report({'is_finished':0,'need_reassign':0,'current_page':page})
            time.sleep(sleep_second)

    def update_cookie(self):
        taobao_auto = TaobaoAutoCrawl(self.account_name, self.ip, self.product)
        if not taobao_auto.run():
            return False
        safe_name = self._sql_literal(self.account_name)
        sql_account = (
            f"select * from `retrieve_collect_equipment_account` where `username`='{safe_name}'"
        )
        account_list = self.db.select_data(sql_account)
        if not account_list:
            logger.error("账号 %s 未查询到 cookie 信息", self.account_name)
            return False
        cookie_str = account_list[0].get("cookie_str")
        if not cookie_str:
            logger.error("账号 %s cookie 为空", self.account_name)
            return False
        try:
            self.cookies = json.loads(cookie_str)
        except Exception as e:
            logger.error("账号 %s cookie 解析失败: %s", self.account_name, e)
            return False
        return True

    def get_account(self):
        sql_account = """
                      SELECT * FROM `retrieve_collect_equipment_account` WHERE `id` = %s and `status` = 0
                      """
        account_list = self.db.select_data(sql_account,self.account_id)
        if not account_list:
            return False

        account_dict = account_list[0]
        self.ip = account_dict.get("ip")
        cookie_str = account_dict.get("cookie_str")

        self.ip = account_dict.get("ip")
        self.account_name = account_dict.get("username")
        self.login_username = account_dict.get("phone", "")
        self.login_password = account_dict.get("password", "")
        self.cookie_stamp = account_dict.get("update_time")
        if self.ip:
            account_proxy = f"http://{self.ip}"
            self.proxies = {"http": account_proxy, "https": account_proxy}
        else:
            self.proxies = None

        need_refresh = (
                not cookie_str
                or int(time.time()) - int(self.cookie_stamp or 0) > COOKIE_MAX_AGE_SEC
        )
        print(account_dict)
        if need_refresh:
            if not self.update_cookie():
                return False
        else:
            try:
                self.cookies = json.loads(cookie_str)
            except Exception as e:
                logger.error("cookie 解析失败，尝试刷新: %s", e)
                if not self.update_cookie():
                    return False

        logger.info("获取到账号: %s, ip: %s", self.account_name, self.ip)
        self.init_session()
        return True

    def run(self):
        if not self.get_account():
            logger.info("==================当前无账号可用==================")
            self.success = False
            return self.pipeline.crawl_count, self.success

        logger.info("获取到账号:%s,代理ip:%s", self.account_name, self.ip)

        self.get_search()
        logger.info(
            "任务id:%s, 任务状态已更新, 产品名称:%s, 爬取数据:%s条",
            self.task_id,
            self.product,
            self.pipeline.crawl_count,
        )
        return self.pipeline.crawl_count, self.success