snapshot_taobao_crawl2.py 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765
  1. import hashlib
  2. import json
  3. import random
  4. import re
  5. import time
  6. from decimal import Decimal, InvalidOperation
  7. from curl_cffi import requests
  8. from lxml import etree
  9. from commons.Logger import get_spider_logger
  10. from commons.conn_mysql import MySQLPoolOn2
  11. from pipelines.drug_pipelines import DrugPipeline
  12. from spiders.taobao.snapshot_taobao_login import (TaobaoAutoCrawl)
  13. from area_info.city_name_to_id import get_city
  14. from oss_upload.oss_upload import AliyunOSSUploader
  15. from urllib.parse import quote, unquote
  16. from commons.config import (
  17. TB_DEVICE_ID,
  18. MTOP_APP_KEY,
  19. MTOP_APP_ID,
  20. TAOBAO_SNAP_SEARCH_MAX_PAGE,
  21. TAOBAO_REQUEST_RETRY_COUNT,
  22. TAOBAO_COOKIE_MAX_AGE_SEC,
  23. TAOBAO_SNAP_LISTEN_RETRY,
  24. UA_TAOBAO,
  25. TAOBAO_NO_MATCH_THRESHOLD,
  26. )
  27. logger = get_spider_logger("taobao")
  28. from urllib.parse import quote
  29. SEARCH_MAX_PAGE = TAOBAO_SNAP_SEARCH_MAX_PAGE
  30. REQUEST_RETRY_COUNT = TAOBAO_REQUEST_RETRY_COUNT
  31. COOKIE_MAX_AGE_SEC = TAOBAO_COOKIE_MAX_AGE_SEC
  32. def build_taobao_search_url(keyword: str, page: int = 1) -> str:
  33. """
  34. 构建淘宝搜索URL
  35. 参数:
  36. keyword: 搜索关键词 (例如: "999 玉屏风口服液 10支")
  37. page: 页码,从1开始
  38. 返回:
  39. 完整的淘宝搜索URL字符串
  40. """
  41. # 对关键词进行URL编码(空格转为%20)
  42. encoded_keyword = quote(keyword, safe='').replace(' ', '%20')
  43. # 固定参数
  44. fixed_params = {
  45. "_input_charset": "utf-8",
  46. "commend": "all",
  47. "ie": "utf8",
  48. "preLoadOrigin": "https://www.taobao.com",
  49. "search_type": "item",
  50. "source": "suggest",
  51. "sourceId": "tb.index",
  52. "spm": "a21bo.jianhua/a.search_history.d1",
  53. "ssid": "s5-e",
  54. "tab": "all",
  55. "suggest_query": "",
  56. }
  57. # 动态参数
  58. dynamic_params = {
  59. "q": encoded_keyword,
  60. "page": str(page),
  61. }
  62. # 合并参数
  63. all_params = {**fixed_params, **dynamic_params}
  64. # 构建查询字符串并返回完整URL
  65. query_string = "&".join([f"{k}={v}" for k, v in all_params.items()])
  66. return f"https://s.taobao.com/search?{query_string}"
  67. def extract_item_data(item_element):
  68. """
  69. 从商品元素中提取数据
  70. """
  71. result = {
  72. "item_id": "",
  73. "title": "",
  74. "price": "",
  75. "realSales": "",
  76. "shopInfo": {"title": ""},
  77. "procity": "",
  78. "auctionURL": ""
  79. }
  80. # 1. 提取 item_id - 从 a 标签的 id 属性
  81. a_elem = item_element.ele('xpath=.//a[contains(@id, "item_id_")]')
  82. if a_elem:
  83. item_id_full = a_elem.attr('id')
  84. if item_id_full:
  85. result["item_id"] = item_id_full.replace("item_id_", "")
  86. # 2. 提取 title - 从 div 的 title 属性
  87. title_elem = item_element.ele('xpath=.//div[contains(@class, "title--")]')
  88. if title_elem:
  89. title = title_elem.attr('title')
  90. if not title:
  91. # 如果没有 title 属性,取文本内容
  92. title = title_elem.text
  93. result["title"] = title
  94. # 3. 提取 price - 整数部分 + 小数部分
  95. price_int = item_element.ele('xpath=.//div[contains(@class, "priceInt--")]')
  96. price_float = item_element.ele('xpath=.//div[contains(@class, "priceFloat--")]')
  97. if price_int and price_float:
  98. result["price"] = f"{price_int.text}.{price_float.text.replace('.', '')}"
  99. # 4. 提取 realSales (销量)
  100. sales_elem = item_element.ele('xpath=.//span[contains(@class, "realSales--")]')
  101. if sales_elem:
  102. result["realSales"] = sales_elem.text
  103. # 5. 提取 shopInfo.title (店铺名称)
  104. shop_elem = item_element.ele('xpath=.//span[contains(@class, "shopNameText--")]')
  105. if shop_elem:
  106. result["shopInfo"]["title"] = shop_elem.text
  107. # 6. 提取 procity (发货地)
  108. procity_elem = item_element.ele('xpath=.//div[contains(@class, "procity--")]/span')
  109. if procity_elem:
  110. result["procity"] = procity_elem.text
  111. # 7. 提取 auctionURL
  112. if a_elem:
  113. href = a_elem.attr('href')
  114. if href:
  115. result["auctionURL"] = href
  116. elif result["item_id"]:
  117. result["auctionURL"] = f"https://item.taobao.com/item.htm?id={result['item_id']}"
  118. return result
  119. headers = {
  120. "accept": "*/*",
  121. "accept-language": "zh-CN,zh;q=0.9",
  122. "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",
  123. "sec-ch-ua": '"Not:A-Brand";v="99", "Google Chrome";v="145", "Chromium";v="145"',
  124. "sec-ch-ua-mobile": "?0",
  125. "sec-ch-ua-platform": '"Windows"',
  126. "sec-fetch-dest": "script",
  127. "sec-fetch-mode": "no-cors",
  128. "sec-fetch-site": "same-site",
  129. "user-agent": (
  130. UA_TAOBAO,
  131. ),
  132. }
  133. MTOP_URL = (
  134. "https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/"
  135. )
  136. class TaobaoCrawl:
  137. def __init__(self, drug_dict=None, scheduler=None, driver=None, cumulative_pages=0, cumulative_items=0, cumulative_stored=0, cumulative_skipped=0):
  138. self.cookies = None
  139. self.db = MySQLPoolOn2()
  140. self.pipeline = DrugPipeline("taobao")
  141. self.session = None
  142. self.proxies = None
  143. self.account_name = None
  144. self.ip = None
  145. self.cookie_stamp = None
  146. self.platform = 1
  147. self.task_dict = drug_dict or {}
  148. self.collect_task_id = None
  149. self.success = True
  150. self.scheduler = scheduler
  151. self.report_data = {}
  152. if self.task_dict:
  153. self.get_product_data()
  154. self.is_no_product = 0
  155. self.driver = driver
  156. self.ossuploader = AliyunOSSUploader()
  157. self.cumulative_pages = cumulative_pages
  158. self.cumulative_items = cumulative_items
  159. self.cumulative_stored = cumulative_stored
  160. self.cumulative_skipped = cumulative_skipped
  161. self.page_stored = 0
  162. def get_product_data(self):
  163. self.task_id = self.task_dict["id"]
  164. self.company_id = self.task_dict["company_id"]
  165. self.product = self.task_dict["product_name"]
  166. self.product_desc = self.task_dict.get("product_specs", "")
  167. self.brand = self.task_dict.get("product_brand", "")
  168. self.product_keyword = self.task_dict.get("product_keyword", "")
  169. self.collect_task_id = self.task_dict.get("collect_task_id", "")
  170. self.sampling_cycle = self.task_dict.get("sampling_cycle", "")
  171. self.sampling_start_time = self.task_dict.get("sampling_start_time", "")
  172. self.sampling_end_time = self.task_dict.get("sampling_end_time", "")
  173. self.collect_equipment_id = self.task_dict.get("collect_equipment_id", "")
  174. self.account_id = self.task_dict.get("collect_equipment_account_id", "101")
  175. self.collect_region_id = self.task_dict.get("collect_region_id", "")
  176. self.collect_round = self.task_dict.get("collect_round", 1)
  177. self.report_data = {'task_id': self.task_id, 'platform': self.platform, 'username': self.task_dict.get("username", TB_DEVICE_ID)}
  178. def _error_report(self, data):
  179. self.report_data.update(data)
  180. if self.scheduler:
  181. self.scheduler.stop()
  182. self.scheduler.post_report(self.report_data)
  183. def _success_report(self, data):
  184. self.report_data.update(data)
  185. if self.scheduler:
  186. self.scheduler.post_report(self.report_data)
  187. @staticmethod
  188. def _normalize_url(url):
  189. if not url:
  190. return ""
  191. url = str(url)
  192. if url.startswith("//"):
  193. return "https:" + url
  194. return url
  195. @staticmethod
  196. def _extract_shop_id(shop_url):
  197. if not shop_url:
  198. return ""
  199. shop_re = re.search(r"appUid=(\w+)", shop_url)
  200. if shop_re:
  201. return shop_re.group(1)
  202. return hashlib.md5(shop_url.encode("utf-8")).hexdigest()
  203. @staticmethod
  204. def _sql_literal(value):
  205. """避免拼接账号名时单引号打断 SQL(非完整防注入,仅兜底)。"""
  206. if value is None:
  207. return ""
  208. return str(value).replace("'", "''").replace("\\", "\\\\")
  209. def init_session(self):
  210. self.session = requests.Session(impersonate="chrome124")
  211. self.session.cookies.update(self.cookies or {})
  212. self.session.headers.update(headers)
  213. if self.proxies:
  214. self.session.proxies.update(self.proxies)
  215. @staticmethod
  216. def _is_transport_error(err):
  217. msg = str(err or "")
  218. return ("curl: (16)" in msg) or ("Failed to perform" in msg)
  219. def get_token(self, t, app_key, data_str):
  220. _m_h5_tk = (self.cookies or {}).get("_m_h5_tk", "")
  221. token = _m_h5_tk.split("_")[0] if _m_h5_tk else ""
  222. text = f"{token}&{t}&{app_key}&{data_str}"
  223. return hashlib.md5(text.encode()).hexdigest()
  224. def get_html_content(self, res_html):
  225. if not res_html:
  226. return ""
  227. ele_html = etree.HTML(res_html)
  228. if ele_html is None:
  229. return str(res_html)
  230. text_list = ele_html.xpath(".//text()")
  231. return "".join(text_list)
  232. def _build_search_payload(self, keyword, page, page_size=50):
  233. return {
  234. "appId": MTOP_APP_ID,
  235. "params": {
  236. "device": "HMA-AL00",
  237. "isBeta": "false",
  238. "grayHair": "false",
  239. "from": "nt_history",
  240. "brand": "HUAWEI",
  241. "info": "wifi",
  242. "index": "4",
  243. "rainbow": "",
  244. "schemaType": "auction",
  245. "elderHome": "false",
  246. "isEnterSrpSearch": "true",
  247. "newSearch": "false",
  248. "network": "wifi",
  249. "subtype": "",
  250. "hasPreposeFilter": "false",
  251. "prepositionVersion": "v2",
  252. "client_os": "Android",
  253. "gpsEnabled": "false",
  254. "searchDoorFrom": "srp",
  255. "debug_rerankNewOpenCard": "false",
  256. "homePageVersion": "v7",
  257. "searchElderHomeOpen": "false",
  258. "search_action": "initiative",
  259. "sugg": "_4_1",
  260. "sversion": "13.6",
  261. "style": "list",
  262. "ttid": "600000@taobao_pc_10.7.0",
  263. "needTabs": "true",
  264. "areaCode": "CN",
  265. "vm": "nw",
  266. "countryNum": "156",
  267. "m": "pc",
  268. "page": page,
  269. "n": 48,
  270. "q": keyword,
  271. "qSource": "url",
  272. "pageSource": "",
  273. "channelSrp": "",
  274. "tab": "all",
  275. "pageSize": str(page_size),
  276. "sourceS": "2",
  277. "ntoffset": "0",
  278. "filterTag": "",
  279. "service": "",
  280. "prop": "",
  281. "loc": "",
  282. "categoryp": "",
  283. "screenResolution": "1920x1080",
  284. "viewResolution": "1092x4722",
  285. "userAgent": headers["user-agent"],
  286. "couponUnikey": "",
  287. "subTabId": "",
  288. "np": "",
  289. "clientType": "h5",
  290. "isNewDomainAb": "false",
  291. "forceOldDomain": "false",
  292. },
  293. }
  294. def _request_search_page(self, keyword, page):
  295. t = str(int(time.time() * 1000))
  296. data = self._build_search_payload(keyword, page)
  297. data_str = json.dumps(data, separators=(",", ":"))
  298. sign = self.get_token(t, MTOP_APP_KEY, data_str)
  299. params = {
  300. "jsv": "2.7.4",
  301. "appKey": MTOP_APP_KEY,
  302. "t": t,
  303. "sign": sign,
  304. "api": "mtop.relationrecommend.wirelessrecommend.recommend",
  305. "v": "2.0",
  306. "timeout": "10000",
  307. "type": "jsonp",
  308. "dataType": "jsonp",
  309. "callback": "",
  310. "data": data_str,
  311. }
  312. return self.session.get(MTOP_URL, params=params, timeout=30)
  313. def _parse_jsonp_body(self, res_text):
  314. res_text = (res_text or "").strip()
  315. json_str = res_text
  316. m = re.match(r"^[^(]*\((.*)\)\s*;?\s*$", res_text, re.DOTALL)
  317. if m:
  318. json_str = m.group(1)
  319. return json.loads(json_str)
  320. def handle_bxb_slider(self, tab):
  321. """处理淘宝风控滑块"""
  322. try:
  323. for attempt in range(3):
  324. slider = None
  325. iframe = None
  326. for _ in range(3):
  327. iframe = tab.get_frame('#baxia-dialog-content', timeout=1)
  328. if iframe:
  329. slider = iframe.ele('#nc_1_n1z', timeout=1)
  330. if slider: break
  331. slider = tab.ele('#nc_1_n1z', timeout=1)
  332. if slider: break
  333. time.sleep(1)
  334. if not slider:
  335. return False
  336. logger.info(f"检测到淘宝滑块,第 {attempt + 1} 次尝试拖拽...")
  337. tab.actions.move_to(slider).wait(random.uniform(0.4, 0.8))
  338. tab.actions.hold(slider)
  339. # 生成拟人轨迹(缓动物理模型)
  340. distance = 300
  341. current = 0
  342. mid = distance * random.uniform(0.6, 0.8)
  343. t = 0.2
  344. v = 0
  345. while current < distance:
  346. a = random.uniform(2, 5) if current < mid else -random.uniform(1.5, 3)
  347. v0 = v
  348. v = v0 + a * t
  349. move = max(0.5, v0 * t + 0.5 * a * t * t)
  350. current += move
  351. y_offset = random.choice([0, 0, 0, -1, 1])
  352. duration = random.uniform(0.01, 0.03)
  353. tab.actions.move(move, y_offset, duration=duration)
  354. # 随机往回滑一点点,更像人的手抖
  355. tab.actions.move(-random.randint(1, 3), random.randint(-1, 1), duration=0.05)
  356. tab.actions.release()
  357. time.sleep(2)
  358. # 检查是否出现“验证失败”
  359. error_text = None
  360. if iframe:
  361. error_text = iframe.ele('text:验证失败', timeout=1)
  362. else:
  363. error_text = tab.ele('text:验证失败', timeout=1)
  364. if error_text:
  365. logger.warning("滑块验证失败被识破,点击刷新重试...")
  366. error_text.click()
  367. time.sleep(2)
  368. continue
  369. else:
  370. logger.info("滑块拖拽完成,大概率通过了验证")
  371. return True
  372. except Exception as e:
  373. logger.error(f"处理淘宝滑块异常: {e}")
  374. return False
  375. def get_search(self):
  376. keyword = self.product
  377. if self.brand:
  378. keyword = (self.brand + " " + self.product).strip()
  379. if self.product_desc:
  380. keyword = (keyword + " " + self.product_desc).strip()
  381. for page in range(1, SEARCH_MAX_PAGE + 1):
  382. logger.info(f"正在爬取关键词:{keyword},{page}页数据")
  383. self.page_stored = 0
  384. base_url = build_taobao_search_url(keyword)
  385. tab = self.driver.latest_tab
  386. tab.listen.start('https://h5api.m.taobao.com/h5/mtop.relationrecommend.wirelessrecommend.recommend/2.0/') # 开始监听,指定获取包含该文本的数据包
  387. if page==1:
  388. tab.get(base_url)
  389. self.handle_bxb_slider(tab)
  390. else:
  391. next_btn = tab.ele('xpath=//*[@id="search-content-leftWrap"]//button[@class="next-btn next-medium next-btn-normal next-pagination-item next-next"]')
  392. if not next_btn.click():
  393. break
  394. self.handle_bxb_slider(tab)
  395. res = None
  396. for _ in range(TAOBAO_SNAP_LISTEN_RETRY):
  397. response = tab.listen.wait(timeout=5) # 添加超时防止卡死
  398. if response:
  399. print(f"拦截到数据包 URL 长度: {len(response.url)}")
  400. temp_res = response.response.raw_body
  401. # 通过判断响应体内是否包含 itemsArray 关键字,来准确抓住真正的商品数据包
  402. is_target = False
  403. if isinstance(temp_res, bytes):
  404. is_target = b'"itemsArray"' in temp_res
  405. else:
  406. is_target = '"itemsArray"' in str(temp_res)
  407. if is_target:
  408. res = temp_res
  409. break
  410. else:
  411. print("该包不是主要搜索结果(无itemsArray),继续等待...")
  412. else:
  413. logger.warning("等待数据包超时,检查是否被滑块拦截...")
  414. if self.handle_bxb_slider(tab):
  415. logger.info("滑块处理完毕,继续等待数据包...")
  416. continue
  417. else:
  418. logger.warning("未检测到滑块且超时,退出等待。")
  419. break
  420. time.sleep(1.5)
  421. try:
  422. if not res:
  423. logger.warning("未获取到有效的搜索数据包,跳过当前页")
  424. break
  425. json_data = self._parse_jsonp_body(res)
  426. item_array = json_data.get("data", {}).get("itemsArray", [])
  427. except Exception as e:
  428. logger.warning(
  429. "解析数据异常,%s 账号可能退出登录,尝试重新登录: %s",
  430. self.account_name,
  431. e,
  432. )
  433. break
  434. s2 = 0
  435. if not item_array:
  436. logger.warning("关键词 %s 第 %s 页未获取到商品数据", keyword, page)
  437. return
  438. elems = tab.eles('xpath=//*[@id="content_items_wrapper"]/div//div/a/span/../../../..')
  439. title_eles = tab.eles('xpath=//*[@id="content_items_wrapper"]/div//div/a/span')
  440. for s,raw in enumerate(item_array):
  441. try:
  442. item_id = raw.get("item_id", "")
  443. if not item_id:
  444. continue
  445. while raw.get('shopInfo').get('title') != title_eles[s2].text:
  446. s2 += 1
  447. item_title = self.get_html_content(raw.get("title") or "")
  448. if self.brand not in item_title:
  449. self.is_no_product += 1
  450. continue
  451. if self.product not in item_title:
  452. self.is_no_product += 1
  453. continue
  454. if "+" in item_title:
  455. continue
  456. if self.product_desc:
  457. if self.product_desc in item_title:
  458. crawl_product_desc = self.product_desc
  459. else:
  460. crawl_product_desc = ""
  461. else:
  462. crawl_product_desc = ""
  463. self.is_no_product = 0
  464. status = 1
  465. if self.product_keyword:
  466. search_keyword_list = self.product_keyword.split(",")
  467. for search_keyword in search_keyword_list:
  468. if search_keyword.strip() not in item_title:
  469. status = 0
  470. if status == 0:
  471. continue
  472. item_price = raw.get("price")
  473. item_price_show = raw.get("priceShow", {}).get("price", 0)
  474. item_sales = raw.get("realSales") or ""
  475. sale_num = ""
  476. sales_m = re.search(r"(.*?)人付款", item_sales)
  477. if sales_m:
  478. sale_num = sales_m.group(1)
  479. item_url = self._normalize_url(raw.get("auctionURL"))
  480. match = re.search(r'skuId=([^&]*)', unquote(item_url))
  481. if match:
  482. if match.group(1) != 'null':
  483. item_id = item_id + '_' + match.group(1)
  484. shop_name = raw.get("shopInfo", {}).get("title", "")
  485. area_str = (raw.get("procity", "") or "").strip()
  486. city_id, province_id, city, province = get_city(area_str)
  487. shop_url = self._normalize_url(
  488. raw.get("shopInfo", {}).get("url", "")
  489. )
  490. structured_list = raw.get("structuredUSPInfo",{})
  491. for structured in structured_list:
  492. if structured.get("propertyName","") == "规格":
  493. crawl_product_desc = structured.get("propertyValueName","")
  494. pic_path = raw.get("pic_path", "")
  495. raw_price = item_price_show
  496. if raw_price in (None, ""):
  497. price = Decimal("0.00")
  498. else:
  499. try:
  500. price = Decimal(str(raw_price)).quantize(Decimal("0.00"))
  501. except (InvalidOperation, ValueError):
  502. price = Decimal("0.00")
  503. upload_key = hashlib.md5(item_url.encode("utf-8")).hexdigest()
  504. for i in range(3):
  505. try:
  506. jpg_bytes = elems[s2].get_screenshot(as_bytes="jpg")
  507. snapshot_url = self.ossuploader.upload_from_bytes(jpg_bytes, str(upload_key))
  508. break
  509. except Exception as e:
  510. snapshot_url=''
  511. print(e)
  512. s2+=1
  513. scrape_date = time.strftime("%Y-%m-%d")
  514. update_time = time.strftime("%Y-%m-%d %H:%M:%S")
  515. #snapshot_url = self._normalize_url(pic_path) if pic_path else ""
  516. # 字段与 yaofangwang_crawl 对齐;键顺序须与 commons.sql_data.RETRIEVE_SCRAPE_INSERT_COLUMNS 一致
  517. product = {
  518. "platform": self.platform,
  519. "item_id": item_id,
  520. "enterprise_id": self.company_id,
  521. "product_name": item_title,
  522. "spec": crawl_product_desc,
  523. "one_price": "",
  524. "detail_url": item_url,
  525. "shop_name": shop_name,
  526. "anonymous_store_name": "",
  527. "shop_url": shop_url,
  528. "city_name": "",
  529. "city_id": "",
  530. "province_name": "",
  531. "province_id": "",
  532. "shipment_city_name": city,
  533. "shipment_city_id": city_id,
  534. "shipment_province_name": province,
  535. "shipment_province_id": province_id,
  536. "area_info": area_str,
  537. "factory_name": "",
  538. "scrape_date": scrape_date,
  539. "price": price,
  540. "sales": sale_num,
  541. "stock_count": "",
  542. "snapshot_url": snapshot_url,
  543. "approval_num": "",
  544. "produced_time": "",
  545. "deadline": "",
  546. "update_time": update_time,
  547. "insert_time": update_time,
  548. "number": 1,
  549. "product_brand": self.brand or "",
  550. "collect_task_id": self.collect_task_id,
  551. "task_id": self.task_id,
  552. "search_name": self.product,
  553. "company_name": "",
  554. "collect_config_info": json.dumps(
  555. {
  556. "sampling_cycle": self.sampling_cycle,
  557. "sampling_start_time": self.sampling_start_time,
  558. "sampling_end_time": self.sampling_end_time,
  559. }
  560. ),
  561. "account_id": self.account_id,
  562. "collect_region_id": self.collect_region_id,
  563. "collect_round": self.collect_round,
  564. "is_sold_out": 0
  565. }
  566. try:
  567. affected_rows = self.pipeline.storge_data(product)
  568. if affected_rows and affected_rows > 0:
  569. self.page_stored += 1
  570. logger.info("%s", json.dumps(product, ensure_ascii=False, default=str))
  571. except Exception as e:
  572. logger.exception("写入库失败: %s", e)
  573. except:
  574. continue
  575. self.cumulative_pages += 1
  576. self.cumulative_items += len(item_array)
  577. page_skipped = len(item_array) - self.page_stored
  578. self.cumulative_stored += self.page_stored
  579. self.cumulative_skipped += page_skipped
  580. logger.info(
  581. "关键字 %s 第 %s 页获取完成, 本页获取量: %d, 本页已入库: %d, 本页未入库: %d | 账号测试总页数: %d, 账号测试总数据量: %d, 总已入库: %d, 总未入库: %d",
  582. keyword,
  583. page,
  584. len(item_array),
  585. self.page_stored,
  586. page_skipped,
  587. self.cumulative_pages,
  588. self.cumulative_items,
  589. self.cumulative_stored,
  590. self.cumulative_skipped
  591. )
  592. print(f"[{keyword}] 当前页数: {page}, 本页获取数据量: {len(item_array)}, 本页已入库: {self.page_stored}, 本页未入库: {page_skipped} | 账号测试总页数: {self.cumulative_pages}, 账号测试总数据量: {self.cumulative_items}, 总已入库: {self.cumulative_stored}, 总未入库: {self.cumulative_skipped}")
  593. total_page = (
  594. (json_data or {}).get("data", {}).get("mainInfo", {}).get("totalPage")
  595. )
  596. try:
  597. total_page_int = int(total_page) if total_page is not None else 50
  598. except (TypeError, ValueError):
  599. total_page_int = 50
  600. self.report_data.update({'total_pages': total_page_int})
  601. if self.scheduler and self.scheduler.end:
  602. logger.info('心跳失败')
  603. self._error_report({'is_finished': 0, 'need_reassign': 0, 'current_page': page})
  604. break
  605. if page >= total_page_int or page == SEARCH_MAX_PAGE:
  606. self._success_report({'is_finished': 1, 'need_reassign': 0, 'current_page': page})
  607. break
  608. if self.is_no_product > TAOBAO_NO_MATCH_THRESHOLD:
  609. self._success_report({'is_finished': 1, 'need_reassign': 0, 'current_page': page})
  610. break
  611. sleep_second = random.uniform(30, 60)
  612. logger.info("第 %s 页爬取完成,休息 %.1fs", page, sleep_second)
  613. self._success_report({'is_finished': 0, 'need_reassign': 0, 'current_page': page})
  614. time.sleep(sleep_second)
  615. def update_cookie(self):
  616. taobao_auto = TaobaoAutoCrawl(self.account_name, self.ip, self.product)
  617. self.driver=taobao_auto.run()
  618. if not self.driver :
  619. return False
  620. return True
  621. def get_account(self):
  622. sql_account = """
  623. SELECT * FROM `retrieve_collect_equipment_account` WHERE `id` = %s and `status` = 0
  624. """
  625. account_list = self.db.select_data(sql_account,self.account_id)
  626. if not account_list:
  627. return False
  628. account_dict = account_list[0]
  629. self.ip = account_dict.get("ip")
  630. cookie_str = account_dict.get("cookie_str")
  631. self.ip = account_dict.get("ip")
  632. self.account_name = account_dict.get("username")
  633. self.login_username = account_dict.get("phone", "")
  634. self.login_password = account_dict.get("password", "")
  635. self.cookie_stamp = account_dict.get("update_time")
  636. if self.ip:
  637. account_proxy = f"http://{self.ip}"
  638. self.proxies = {"http": account_proxy, "https": account_proxy}
  639. else:
  640. self.proxies = None
  641. need_refresh = (
  642. not cookie_str
  643. or int(time.time()) - int(self.cookie_stamp or 0) > COOKIE_MAX_AGE_SEC
  644. )
  645. if not self.driver:
  646. if not self.update_cookie():
  647. return False
  648. logger.info("获取到账号: %s, ip: %s", self.account_name, self.ip)
  649. return True
  650. def run(self):
  651. if not self.get_account():
  652. logger.info("==================当前无账号可用==================")
  653. self.success = False
  654. return self.pipeline.crawl_count, self.success, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped
  655. logger.info("获取到账号:%s,代理ip:%s", self.account_name, self.ip)
  656. try:
  657. self.get_search()
  658. # self.driver.quit()
  659. except Exception as e:
  660. print('数据获取错误',e)
  661. logger.info(
  662. "任务id:%s, 任务状态已更新, 产品名称:%s, 爬取数据:%s条",
  663. self.task_id,
  664. self.product,
  665. self.pipeline.crawl_count,
  666. )
  667. return self.pipeline.crawl_count, self.success, self.driver, self.cumulative_pages, self.cumulative_items, self.cumulative_stored, self.cumulative_skipped