project/spides/spiders/taobao/snapshot_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 MySQLPoolOnline
from pipelines.drug_pipelines import DrugPipeline
from spiders.taobao.snapshot_taobao_login import (TaobaoAutoCrawl)
from area_info.city_name_to_id import get_city
from oss_upload.oss_upload import AliyunOSSUploader
from urllib.parse import quote, unquote


logger = get_spider_logger("taobao")
from urllib.parse import quote



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



def build_taobao_search_url(keyword: str, page: int = 1) -> str:
    """
    构建淘宝搜索URL

    参数:
        keyword: 搜索关键词 (例如: "999 玉屏风口服液 10支")
        page: 页码，从1开始

    返回:
        完整的淘宝搜索URL字符串
    """
    # 对关键词进行URL编码（空格转为%20）
    encoded_keyword = quote(keyword, safe='').replace(' ', '%20')

    # 固定参数
    fixed_params = {
        "_input_charset": "utf-8",
        "commend": "all",
        "ie": "utf8",
        "preLoadOrigin": "https://www.taobao.com",
        "search_type": "item",
        "source": "suggest",
        "sourceId": "tb.index",
        "spm": "a21bo.jianhua/a.search_history.d1",
        "ssid": "s5-e",
        "tab": "all",
        "suggest_query": "",
    }

    # 动态参数
    dynamic_params = {
        "q": encoded_keyword,
        "page": str(page),
    }

    # 合并参数
    all_params = {**fixed_params, **dynamic_params}

    # 构建查询字符串并返回完整URL
    query_string = "&".join([f"{k}={v}" for k, v in all_params.items()])
    return f"https://s.taobao.com/search?{query_string}"

def extract_item_data(item_element):
    """
    从商品元素中提取数据
    """
    result = {
        "item_id": "",
        "title": "",
        "price": "",
        "realSales": "",
        "shopInfo": {"title": ""},
        "procity": "",
        "auctionURL": ""
    }

    # 1. 提取 item_id - 从 a 标签的 id 属性
    a_elem = item_element.ele('xpath=.//a[contains(@id, "item_id_")]')
    if a_elem:
        item_id_full = a_elem.attr('id')
        if item_id_full:
            result["item_id"] = item_id_full.replace("item_id_", "")

    # 2. 提取 title - 从 div 的 title 属性
    title_elem = item_element.ele('xpath=.//div[contains(@class, "title--")]')
    if title_elem:
        title = title_elem.attr('title')
        if not title:
            # 如果没有 title 属性，取文本内容
            title = title_elem.text
        result["title"] = title

    # 3. 提取 price - 整数部分 + 小数部分
    price_int = item_element.ele('xpath=.//div[contains(@class, "priceInt--")]')
    price_float = item_element.ele('xpath=.//div[contains(@class, "priceFloat--")]')
    if price_int and price_float:
        result["price"] = f"{price_int.text}.{price_float.text.replace('.', '')}"

    # 4. 提取 realSales (销量)
    sales_elem = item_element.ele('xpath=.//span[contains(@class, "realSales--")]')
    if sales_elem:
        result["realSales"] = sales_elem.text

    # 5. 提取 shopInfo.title (店铺名称)
    shop_elem = item_element.ele('xpath=.//span[contains(@class, "shopNameText--")]')
    if shop_elem:
        result["shopInfo"]["title"] = shop_elem.text

    # 6. 提取 procity (发货地)
    procity_elem = item_element.ele('xpath=.//div[contains(@class, "procity--")]/span')
    if procity_elem:
        result["procity"] = procity_elem.text

    # 7. 提取 auctionURL
    if a_elem:
        href = a_elem.attr('href')
        if href:
            result["auctionURL"] = href
        elif result["item_id"]:
            result["auctionURL"] = f"https://item.taobao.com/item.htm?id={result['item_id']}"

    return result


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/"
)





class TaobaoCrawl:
    def __init__(self, drug_dict=None):
        self.cookies = None
        self.db = MySQLPoolOnline()
        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
        if self.task_dict:
            self.get_product_data()
        self.is_no_product = 0
        self.driver=''
        self.ossuploader = AliyunOSSUploader()

    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)

    @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)

    @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(1, SEARCH_MAX_PAGE + 1):
            logger.info(f"正在爬取关键词：{keyword}，{page}页数据")

            # input_box = self.driver.ele('xpath=//*[@id="q"]')
            # input_box.input(keyword)
            # time.sleep(1)
            # button = self.driver.ele('xpath=//*[@id="J_TSearchForm"]/div[2]/button')
            # button.click()
            # time.sleep(2)
            base_url = build_taobao_search_url(keyword)
            tab = self.driver.latest_tab

            tab.listen.start('https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/')  # 开始监听，指定获取包含该文本的数据包
            if page==1:
                tab.get(base_url)
            else:
                next_btn = tab.ele('xpath=//*[@id="search-content-leftWrap"]//button[contains(@aria-label, "下一页")]')
                if not next_btn.click():
                    break

            for _ in range(5):
                response = tab.listen.wait()  # 等待并获取一个数据包
                if len(response.url)>3000:
                    res = response.response.raw_body
                    break
                else:
                    continue
            time.sleep(1.5)
            try:
                json_data = self._parse_jsonp_body(res)
                item_array = json_data.get("data", {}).get("itemsArray", [])
            except Exception as e:
                logger.warning(
                    "解析数据异常，%s 账号可能退出登录，尝试重新登录: %s",
                    self.account_name,
                    e,
                )
                break
            s2 = 0
            if not item_array:
                logger.warning("关键词 %s 第 %s 页未获取到商品数据", keyword, page)
                return
            elems =  tab.eles('xpath=//*[@id="content_items_wrapper"]/div//div/a/span/../../../..')
            title_eles = tab.eles('xpath=//*[@id="content_items_wrapper"]/div//div/a/span')
            for s,raw in enumerate(item_array):
                try:
                    item_id = raw.get("item_id", "")

                    if not item_id:
                        continue
                    while raw.get('shopInfo').get('title') != title_eles[s2].text:
                        s2 += 1
                    item_title = self.get_html_content(raw.get("title") or "")

                    if self.brand not in item_title:
                        self.is_no_product += 1
                        continue
                    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")
                    upload_key = hashlib.md5(item_url.encode("utf-8")).hexdigest()
                    for i in range(3):
                        try:
                            jpg_bytes = elems[s2].get_screenshot(as_bytes="jpg")
                            snapshot_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(upload_key))
                            break
                        except Exception as e:
                            snapshot_url=''
                            print(e)
                    s2+=1
                    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": 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)
                except:
                    continue


            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

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


    def update_cookie(self):
        taobao_auto = TaobaoAutoCrawl(self.account_name, self.ip, self.product)
        self.driver=taobao_auto.run()
        if not self.driver :
            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
        )
        if 1:
            if not self.update_cookie():
                return False


        logger.info("获取到账号: %s, ip: %s", self.account_name, self.ip)
        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)
        try:
            self.get_search()
            self.driver.quit()
        except Exception as e:
            print('数据获取错误',e)
        logger.info(
            "任务id:%s, 任务状态已更新, 产品名称:%s, 爬取数据:%s条",
            self.task_id,
            self.product,
            self.pipeline.crawl_count,
        )
        return self.pipeline.crawl_count, self.success