snapshot_taobao_crawl2.py 29 KB

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