# -*- coding: utf-8 -*- # Author : Charley # Python : 3.12.10 # Date : 2026/5/12 15:43 import time import inspect import schedule import requests import user_agent from loguru import logger from parsel import Selector from datetime import datetime from mysql_pool import MySQLConnectionPool from tenacity import retry, stop_after_attempt, wait_fixed logger.remove() logger.add("./logs/{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00", format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}", level="DEBUG", retention="7 day") def after_log(retry_state): """ retry 回调 :param retry_state: RetryCallState 对象 """ # 检查 args 是否存在且不为空 if retry_state.args and len(retry_state.args) > 0: log = retry_state.args[0] # 获取传入的 logger else: log = logger # 使用全局 logger if retry_state.outcome.failed: log.warning( f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times") else: log.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded") @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_proxys(log): """ 获取代理 :return: 代理 """ tunnel = "x371.kdltps.com:15818" kdl_username = "t13753103189895" kdl_password = "o0yefv6z" try: proxies = { "http": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel}, "https": "http://%(user)s:%(pwd)s@%(proxy)s/" % {"user": kdl_username, "pwd": kdl_password, "proxy": tunnel} } return proxies except Exception as e: log.error(f"Error getting proxy: {e}") raise e def conversion_date(log, date_str): """ 日期格式转换 将 "May 1, 2026" 格式转换为 "YYYY-MM-DD" :param log: logger对象 :param date_str: 日期字符串，如 "May 1, 2026" :return: 格式化后的日期字符串 "YYYY-MM-DD" """ if not date_str or not isinstance(date_str, str): return None try: date_str = date_str.strip() date_object = datetime.strptime(date_str, "%B %d, %Y") return date_object.strftime("%Y-%m-%d") except ValueError: try: date_object = datetime.strptime(date_str, "%b %d, %Y") return date_object.strftime("%Y-%m-%d") except ValueError: log.error(f"日期格式转换失败: {date_str}") return None @retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log) def get_single_page(log, page, sql_pool): """ 获取单页数据 :param log: logger对象 :param page: 页码 :param sql_pool: MySQL连接池 :return: 该页数据条数 """ log.info(f"正在爬取第 {page} 页数据............") headers = { "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7", "referer": "https://myslabs.com/browse/archive/", "user-agent": user_agent.generate_user_agent() } url = "https://myslabs.com/browse/archive/" params = { # "page": "2" "page": f"{page}" } # response = requests.get(url, headers=headers, cookies=cookies, params=params) response = requests.get(url, headers=headers, params=params, timeout=22) response.raise_for_status() selector = Selector(response.text) tag_div_list = selector.xpath('//div[@class="slabs_cont padded-primary-container"]/div') info_list = [] for tag_div in tag_div_list: title = tag_div.xpath('./a/div/text()').get() price = tag_div.xpath('.//div[@class="item-price"]/text()').get() price = price.replace('$', '').strip() if price else None release_date = tag_div.xpath('.//small/text()').get() # May 1, 2026 release_date = conversion_date(log, release_date) front_img = tag_div.xpath( './/div[@class="slab_item_img"]//img/@src | .//div[@class="slab_item_img"]//img/@data-src').get() # print(front_img) front_img = front_img.split('?')[0] if front_img and '?' in front_img else front_img back_img = front_img.replace("1.png", "2.png") if front_img else None # src="https://cdn.myslabs.com/myslabs-prod/media/3c9dc5fb-7178-46d3-8581-245fa315ada4/2026/05/02/BXIJYTK_1777730735_1.png" # src="https://cdn.myslabs.com/myslabs-prod/media/3c9dc5fb-7178-46d3-8581-245fa315ada4/2026/05/02/BXIJYTK_1777730735_2.png" # src="https://cdn.myslabs.com/myslabs-prod/media/3c9dc5fb-7178-46d3-8581-245fa315ada4/2026/05/02/BXIJYTK_1777730735_1.png?width=360&height=610&quality=85" detail_url = tag_div.xpath('./a/@href').get() # detail_url = "https://myslabs.com" + detail_url if detail_url else None detail_id = detail_url.split("/")[-2] if detail_url else 1 data_dict = { "title": title.strip() if title else None, "price": price.replace(',', '') if price else None, "release_date": release_date, "front_img": front_img, "back_img": back_img, "detail_id": int(detail_id) if detail_id else 1 } # print(data_dict) info_list.append(data_dict) # 保存数据 sql_pool.insert_many(table="slabs_record", data_list=info_list, ignore=True) return len(tag_div_list) def get_sold_list(log, sql_pool): page = 1 max_page = 10 while page <= max_page: try: single_count = get_single_page(log, page, sql_pool) if single_count < 72: log.warning(f"第 {page} 页数据不足 72 条，结束爬取") break except Exception as e: log.error(f"爬取第 {page} 页数据时出错: {e}") page += 1 @retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log) def ms_main(log): """ 主函数 :param log: logger对象 """ log.info( f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................') # 配置 MySQL 连接池 sql_pool = MySQLConnectionPool(log=log) if not sql_pool: log.error("MySQL数据库连接失败") raise Exception("MySQL数据库连接失败") try: get_sold_list(log, sql_pool) except Exception as e: log.error(f'{inspect.currentframe().f_code.co_name} error: {e}') finally: log.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............') def schedule_task(): """ 设置定时任务 """ # ms_main(log=logger) schedule.every().day.at("05:00").do(ms_main, log=logger) while True: schedule.run_pending() time.sleep(1) if __name__ == '__main__': # ms_main(log=logger) schedule_task()