update 6.13.1

jihuanshe_spider/YamlLoader.py

@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+#
+import os, re
+import yaml
+
+regex = re.compile(r'^\$\{(?P<ENV>[A-Z_\-]+\:)?(?P<VAL>[\w\.]+)\}$')
+
+class YamlConfig:
+    def __init__(self, config):
+        self.config = config
+
+    def get(self, key:str):
+        return YamlConfig(self.config.get(key))
+    
+    def getValueAsString(self, key: str):
+        try:
+            match = regex.match(self.config[key])
+            group = match.groupdict()
+            if group['ENV'] != None:
+                env = group['ENV'][:-1]
+                return os.getenv(env, group['VAL'])
+            return None
+        except:
+            return self.config[key]
+    
+    def getValueAsInt(self, key: str):
+        try:
+            match = regex.match(self.config[key])
+            group = match.groupdict()
+            if group['ENV'] != None:
+                env = group['ENV'][:-1]
+                return int(os.getenv(env, group['VAL']))
+            return 0
+        except:
+            return int(self.config[key])
+        
+    def getValueAsBool(self, key: str, env: str = None):
+        try:
+            match = regex.match(self.config[key])
+            group = match.groupdict()
+            if group['ENV'] != None:
+                env = group['ENV'][:-1]
+                return bool(os.getenv(env, group['VAL']))
+            return False
+        except:
+            return bool(self.config[key])
+    
+def readYaml(path:str = 'application.yml', profile:str = None) -> YamlConfig:
+    if os.path.exists(path):
+        with open(path) as fd:
+            conf = yaml.load(fd, Loader=yaml.FullLoader)
+
+    if profile != None:
+        result = path.split('.')
+        profiledYaml = f'{result[0]}-{profile}.{result[1]}'
+        if os.path.exists(profiledYaml):
+            with open(profiledYaml) as fd:
+                conf.update(yaml.load(fd, Loader=yaml.FullLoader))
+
+    return YamlConfig(conf)
+
+# res = readYaml()
+# mysqlConf = res.get('mysql')
+# print(mysqlConf)
+
+# print(res.getValueAsString("host"))
+# mysqlYaml = mysqlConf.getValueAsString("host")
+# print(mysqlYaml)
+# host = mysqlYaml.get("host").split(':')[-1][:-1]
+# port = mysqlYaml.get("port").split(':')[-1][:-1]
+# username = mysqlYaml.get("username").split(':')[-1][:-1]
+# password = mysqlYaml.get("password").split(':')[-1][:-1]
+# mysql_db = mysqlYaml.get("db").split(':')[-1][:-1]
+# print(host,port,username,password)

jihuanshe_spider/application.yml

@@ -0,0 +1,6 @@
+mysql:
+  host: ${MYSQL_HOST:100.64.0.25}
+  port: ${MYSQL_PROT:3306}
+  username: ${MYSQL_USERNAME:crawler}
+  password: ${MYSQL_PASSWORD:Pass2022}
+  db: ${MYSQL_DATABASE:crawler}

jihuanshe_spider/jhs_new_daily_spider.py

@@ -0,0 +1,324 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# Date   : 2025/3/19 18:55
+import time
+import inspect
+import requests
+import schedule
+import user_agent
+from loguru import logger
+from datetime import datetime
+from mysq_pool import MySQLConnectionPool
+from tenacity import retry, stop_after_attempt, wait_fixed
+
+"""
+com.jihuanshe
+"""
+
+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 save_product_data(sql_pool, product_info: tuple):
+    """
+    保存 product 数据
+    :param sql_pool: MySQL连接池对象
+    :param product_info: 要保存的数据
+    """
+    sql = """
+        INSERT INTO jhs_product_record (seller_user_id, seller_username, product_id, app_id, nonce_str, signature, auction_product_name, auction_product_images, game_key, language_text, authenticator_name, grading, starting_price, max_bid_price, status, auction_product_start_time, auction_product_end_time)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"""
+    sql_pool.insert_one(sql, product_info)
+
+
+def save_biddings_data(sql_pool, biddings_list: list):
+    """
+    保存 biddings 数据
+    :param sql_pool: MySQL连接池对象
+    :param biddings_list: 要保存的数据 -> list
+    """
+    sql = """
+        INSERT INTO jhs_biddings_record (product_id, username, bid_price, bid_status, created_at)
+        VALUES (%s, %s, %s, %s, %s)"""
+    sql_pool.insert_all(sql, biddings_list)
+
+
+def save_shop_data(sql_pool, shop_info: tuple):
+    """
+    保存 product 数据
+    :param sql_pool: MySQL连接池对象
+    :param shop_info: 要保存的数据
+    """
+    sql = """
+        INSERT INTO jhs_shop_record (seller_user_id, seller_username, follower_count, success_order_user_count, seller_credit_rank_image_url)
+        VALUES (%s, %s, %s, %s, %s)"""
+    sql_pool.insert_one(sql, shop_info)
+
+
+def parse_data(log, resp_json, sql_product_id, sql_pool, sql_shop_id_list):
+    """
+    解析数据, 将解析后的数据, 分成三个表存储    product
+                                        biddings
+                                        shop
+    :param log: logger对象
+    :param resp_json: resp_json
+    :param sql_product_id: sql_product_id
+    :param sql_pool: MySQL连接池对象
+    :param sql_shop_id_list: sql_shop_id_list
+    """
+    sql_id = sql_product_id[0]
+    pid = sql_product_id[1]
+
+    status = resp_json.get('status')
+    if status not in ['ongoing', 'pending']:
+        """
+        获取 shop 信息
+        """
+        seller_user_id = resp_json.get('seller_user_id')
+        seller_username = resp_json.get('seller_username')
+
+        # print(seller_user_id)
+        # # 查询商家id在不在数据库中
+        # sql_exists_flag = """SELECT EXISTS (SELECT 1 FROM jhs_shop_record WHERE seller_user_id = %s) AS exists_flag"""
+        # exists_flag = sql_pool.select_one(sql_exists_flag, (seller_user_id,))
+        # exists_flag = exists_flag[0]
+        # if exists_flag == 1:
+
+        if str(seller_user_id) in sql_shop_id_list:
+            log.info(
+                f"----------------- The seller_user_id {seller_user_id} is already in the database, Not need save -----------------")
+        else:
+            follower_count = resp_json.get('follower_count')
+            success_order_user_count = resp_json.get('success_order_user_count')
+            seller_credit_rank_image_url = resp_json.get('seller_credit_rank_image_url')
+            shop_info = (
+                seller_user_id, seller_username, follower_count, success_order_user_count, seller_credit_rank_image_url
+            )
+            # print(shop_info)
+
+            try:
+                save_shop_data(sql_pool, shop_info)
+                sql_shop_id_list.append(seller_user_id)
+            except Exception as e:
+                if "Duplicate entry" in str(e):
+                    logger.warning(f"存在重复的 seller_user_id{seller_user_id}，跳过插入....")
+                else:
+                    logger.error(f"保存数据 seller_user_id 时出错: {str(e)}")
+
+            # sql_pool.update_one("update jhs_task set task_state = 1 where id = %s", (sql_id,))
+
+        """
+        获取 product 信息
+        """
+        app_id = resp_json.get('appId')
+        nonce_str = resp_json.get('nonceStr')
+        signature = resp_json.get('signature')
+
+        # auction_product_id = resp_json.get('auction_product_id')
+        auction_product_name = resp_json.get('auction_product_name')
+        auction_product_images_list = resp_json.get('auction_product_images', [])
+        if auction_product_images_list:
+            auction_product_images = ','.join(auction_product_images_list)
+        else:
+            auction_product_images = None
+
+        game_key = resp_json.get('game_key')
+        language_text = resp_json.get('language_text')
+        authenticator_name = resp_json.get('authenticator_name')
+        grading = resp_json.get('grading')
+        starting_price = resp_json.get('starting_price')
+        max_bid_price = resp_json.get('max_bid_price')
+
+        auction_product_start_timestamp = resp_json.get('auction_product_start_timestamp')
+        auction_product_start_time = datetime.fromtimestamp(auction_product_start_timestamp).strftime(
+            '%Y-%m-%d %H:%M:%S') if auction_product_start_timestamp else None
+        auction_product_end_timestamp = resp_json.get('auction_product_end_timestamp')
+        auction_product_end_time = datetime.fromtimestamp(auction_product_end_timestamp).strftime(
+            '%Y-%m-%d %H:%M:%S') if auction_product_end_timestamp else None
+
+        product_info = (
+            seller_user_id, seller_username, pid, app_id, nonce_str, signature, auction_product_name,
+            auction_product_images, game_key, language_text, authenticator_name, grading, starting_price, max_bid_price,
+            status, auction_product_start_time, auction_product_end_time
+        )
+        # print(product_info)
+        try:
+            save_product_data(sql_pool, product_info)
+            sql_pool.update_one("update jhs_task set task_state = 1 where id = %s", (sql_id,))
+        except Exception as e:
+            if "Duplicate entry" in str(e):
+                logger.warning(f"存在重复的 pid{pid}，跳过插入....")
+            else:
+                logger.error(f"保存数据时出错: {str(e)}")
+            return
+
+        """
+        获取 biddings 信息
+        """
+        biddings = resp_json.get('biddings', [])
+        # print(biddings)
+
+        # 创建一个字典来存储每个用户的最高出价记录
+        highest_bids = {}
+
+        for record in biddings:
+            username = record['username']
+            bid_price = float(record['bid_price'])  # 将出价转换为浮点数以便比较
+
+            # 如果用户不在字典中，或者当前出价高于已存储的最高出价，则更新记录
+            if username not in highest_bids or bid_price > float(highest_bids[username]['bid_price']):
+                highest_bids[username] = record
+
+        bids_list = list(highest_bids.values())
+        biddings_list = [
+            (pid, record['username'], record['bid_price'], record['bid_status'], record['created_at'])
+            for record in bids_list
+        ]
+        # print(biddings_list)
+        if biddings_list:
+            save_biddings_data(sql_pool, biddings_list)
+        else:
+            log.info(f"................ No biddings found for product {pid}, Not need save ................")
+
+    else:
+        log.info(f"................ The product {pid} is ongoing or pending, Not need parse ................")
+        sql_pool.update_one("update jhs_task set task_state = 4 where id = %s", (sql_id,))
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def get_resp(log, sql_product_id, sql_pool, sql_shop_id_list):
+    """
+    获取 response 响应
+    :param log: logger对象
+    :param sql_product_id: sql_product_id
+    :param sql_pool: MySQL连接池对象
+    :param sql_shop_id_list: sql_shop_id_list
+    """
+    sql_id = sql_product_id[0]
+    pid = sql_product_id[1]
+    headers = {
+        "accept": "application/json, text/plain, */*",
+        "referer": "https://www.jihuanshe.com/",
+        "user-agent": user_agent.generate_user_agent()
+    }
+    url = "https://api.jihuanshe.com/api/market/share/auction-product"
+    params = {
+        "auction_product_id": pid,
+        "url": f"https://www.jihuanshe.com/app/auction?auctionProductId={pid}"
+    }
+    response = requests.get(url, headers=headers, params=params, timeout=5, proxies=get_proxys(log))
+    # print(response.json())
+    # print(response)
+    resp_json = response.json()
+    if resp_json:
+        if resp_json.get("code") == 440:
+            log.debug(f"< 抱歉，该竞价商品已被移除, pid:{pid} >")
+            sql_pool.update_one("update jhs_task set task_state = 2 where id = %s", (sql_id,))
+        else:
+            try:
+                parse_data(log, resp_json, sql_product_id, sql_pool, sql_shop_id_list)
+            except Exception as e:
+                log.error(f"get_resp call parse_data() sql_product_id:{sql_product_id} 获取失败, error:{e}")
+
+
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def jhs_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:
+        # max_sql_id = sql_pool.select_one("select max(product_id) from jhs_product_record")
+        # if max_sql_id:
+        #     max_sql_id = max_sql_id[0]
+        # else:
+        #     max_sql_id = 422342
+        # log.debug(f"当前数据库中最大的 product_id max_sql_id:{max_sql_id}")
+
+        # 从数据库获取需要爬取的 product_id  从最大的id max_sql_id 开始
+        product_id_list = sql_pool.select_all(
+            "SELECT id, product_id FROM jhs_task WHERE task_state != 1 AND id > 420107 LIMIT 10000")
+            # "SELECT id, product_id FROM jhs_task WHERE task_state IN (0, 2) AND id > 420107 LIMIT 10000")
+        # "SELECT id, product_id FROM jhs_task WHERE task_state IN (0, 2) AND id > %s LIMIT 6000", (max_sql_id,))
+        # "SELECT id, product_id FROM jhs_task WHERE task_state IN (0, 2) AND id < 376575 ORDER BY id DESC")
+        product_id_list = [keyword for keyword in product_id_list]
+
+        sql_shop_id_list = sql_pool.select_all("SELECT seller_user_id FROM jhs_shop_record")
+        sql_shop_id_list = [keyword[0] for keyword in sql_shop_id_list]
+        for sql_product_id in product_id_list:
+            try:
+                get_resp(log, sql_product_id, sql_pool, sql_shop_id_list)
+            except Exception as e:
+                log.error(f"Loop sql_product_id:{sql_product_id} 获取失败, error:{e}")
+                sql_pool.update_one("update jhs_task set task_state = 3 where id = %s", (sql_product_id[0],))
+    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():
+    """
+    设置定时任务
+    """
+    jhs_main(log=logger)
+
+    schedule.every().day.at("01:31").do(jhs_main, log=logger)
+    while True:
+        schedule.run_pending()
+        time.sleep(1)
+
+
+if __name__ == '__main__':
+    schedule_task()
+    # get_resp(logger, (438807, 438807), MySQLConnectionPool(log=logger),[])

jihuanshe_spider/jhs_sold_spider.py

@@ -0,0 +1,308 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# Date   : 2025/3/17 10:37
+import inspect
+import requests
+import user_agent
+from loguru import logger
+from datetime import datetime
+from mysq_pool import MySQLConnectionPool
+from tenacity import retry, stop_after_attempt, wait_fixed
+
+"""
+com.jihuanshe
+"""
+
+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 save_product_data(sql_pool, product_info: tuple):
+    """
+    保存 product 数据
+    :param sql_pool: MySQL连接池对象
+    :param product_info: 要保存的数据
+    """
+    sql = """
+        INSERT INTO jhs_product_record (seller_user_id, seller_username, product_id, app_id, nonce_str, signature, auction_product_name, auction_product_images, game_key, language_text, authenticator_name, grading, starting_price, max_bid_price, status, auction_product_start_time, auction_product_end_time)
+        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)"""
+    sql_pool.insert_one(sql, product_info)
+
+
+def save_biddings_data(sql_pool, biddings_list: list):
+    """
+    保存 biddings 数据
+    :param sql_pool: MySQL连接池对象
+    :param biddings_list: 要保存的数据 -> list
+    """
+    sql = """
+        INSERT INTO jhs_biddings_record (product_id, username, bid_price, bid_status, created_at)
+        VALUES (%s, %s, %s, %s, %s)"""
+    sql_pool.insert_all(sql, biddings_list)
+
+
+def save_shop_data(sql_pool, shop_info: tuple):
+    """
+    保存 product 数据
+    :param sql_pool: MySQL连接池对象
+    :param shop_info: 要保存的数据
+    """
+    sql = """
+        INSERT INTO jhs_shop_record (seller_user_id, seller_username, follower_count, success_order_user_count, seller_credit_rank_image_url)
+        VALUES (%s, %s, %s, %s, %s)"""
+    sql_pool.insert_one(sql, shop_info)
+
+
+def parse_data(log, resp_json, sql_product_id, sql_pool, sql_shop_id_list):
+    """
+    解析数据, 将解析后的数据, 分成三个表存储    product
+                                        biddings
+                                        shop
+    :param log: logger对象
+    :param resp_json: resp_json
+    :param sql_product_id: sql_product_id
+    :param sql_pool: MySQL连接池对象
+    :param sql_shop_id_list: sql_shop_id_list
+    """
+    sql_id = sql_product_id[0]
+    pid = sql_product_id[1]
+
+    status = resp_json.get('status')
+    if status not in ['ongoing', 'pending']:
+        """
+        获取 shop 信息
+        """
+        seller_user_id = resp_json.get('seller_user_id')
+        # print(type(seller_user_id))
+        seller_username = resp_json.get('seller_username')
+
+        # print(seller_user_id)
+        # # 查询商家id在不在数据库中
+        # sql_exists_flag = """SELECT EXISTS (SELECT 1 FROM jhs_shop_record WHERE seller_user_id = %s) AS exists_flag"""
+        # exists_flag = sql_pool.select_one(sql_exists_flag, (seller_user_id,))
+        # exists_flag = exists_flag[0]
+        # if exists_flag == 1:
+
+        if str(seller_user_id) in sql_shop_id_list:
+            log.info(
+                f"----------------- The seller_user_id {seller_user_id} is already in the database, Not need save -----------------")
+        else:
+            follower_count = resp_json.get('follower_count')
+            success_order_user_count = resp_json.get('success_order_user_count')
+            seller_credit_rank_image_url = resp_json.get('seller_credit_rank_image_url')
+            shop_info = (
+                seller_user_id, seller_username, follower_count, success_order_user_count, seller_credit_rank_image_url
+            )
+            # print(shop_info)
+            try:
+                save_shop_data(sql_pool, shop_info)
+                sql_shop_id_list.append(seller_user_id)
+            except Exception as e:
+                if "Duplicate entry" in str(e):
+                    logger.warning(f"存在重复的 seller_user_id{seller_user_id}，跳过插入....")
+                else:
+                    logger.error(f"保存数据 seller_user_id 时出错: {str(e)}")
+
+        """
+        获取 product 信息
+        """
+        app_id = resp_json.get('appId')
+        nonce_str = resp_json.get('nonceStr')
+        signature = resp_json.get('signature')
+
+        # auction_product_id = resp_json.get('auction_product_id')
+        auction_product_name = resp_json.get('auction_product_name')
+        auction_product_images_list = resp_json.get('auction_product_images', [])
+        if auction_product_images_list:
+            auction_product_images = ','.join(auction_product_images_list)
+        else:
+            auction_product_images = None
+
+        game_key = resp_json.get('game_key')
+        language_text = resp_json.get('language_text')
+        authenticator_name = resp_json.get('authenticator_name')
+        grading = resp_json.get('grading')
+        starting_price = resp_json.get('starting_price')
+        max_bid_price = resp_json.get('max_bid_price')
+
+        auction_product_start_timestamp = resp_json.get('auction_product_start_timestamp')
+        auction_product_start_time = datetime.fromtimestamp(auction_product_start_timestamp).strftime(
+            '%Y-%m-%d %H:%M:%S') if auction_product_start_timestamp else None
+        auction_product_end_timestamp = resp_json.get('auction_product_end_timestamp')
+        auction_product_end_time = datetime.fromtimestamp(auction_product_end_timestamp).strftime(
+            '%Y-%m-%d %H:%M:%S') if auction_product_end_timestamp else None
+
+        product_info = (
+            seller_user_id, seller_username, pid, app_id, nonce_str, signature, auction_product_name,
+            auction_product_images, game_key, language_text, authenticator_name, grading, starting_price, max_bid_price,
+            status, auction_product_start_time, auction_product_end_time
+        )
+        # print(product_info)
+        try:
+            save_product_data(sql_pool, product_info)
+            sql_pool.update_one("update jhs_task set task_state = 1 where id = %s", (sql_id,))
+        except Exception as e:
+            if "Duplicate entry" in str(e):
+                logger.warning(f"存在重复的 pid{pid}，跳过插入....")
+            else:
+                logger.error(f"保存数据时出错: {str(e)}")
+            return
+
+        """
+        获取 biddings 信息
+        """
+        biddings = resp_json.get('biddings', [])
+        # print(biddings)
+
+        # 创建一个字典来存储每个用户的最高出价记录
+        highest_bids = {}
+
+        for record in biddings:
+            username = record['username']
+            bid_price = float(record['bid_price'])  # 将出价转换为浮点数以便比较
+
+            # 如果用户不在字典中，或者当前出价高于已存储的最高出价，则更新记录
+            if username not in highest_bids or bid_price > float(highest_bids[username]['bid_price']):
+                highest_bids[username] = record
+
+        bids_list = list(highest_bids.values())
+        biddings_list = [
+            (pid, record['username'], record['bid_price'], record['bid_status'], record['created_at'])
+            for record in bids_list
+        ]
+        # print(biddings_list)
+        if biddings_list:
+            save_biddings_data(sql_pool, biddings_list)
+        else:
+            log.info(f"................ No biddings found for product {pid}, Not need save ................")
+
+    else:
+        log.info(f"................ The product {pid} is ongoing or pending, Not need parse ................")
+        sql_pool.update_one("update jhs_task set task_state = 4 where id = %s", (sql_id,))
+
+
+@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
+def get_resp(log, sql_product_id, sql_pool, sql_shop_id_list):
+    """
+    获取 response 响应
+    :param log: logger对象
+    :param sql_product_id: sql_product_id
+    :param sql_pool: MySQL连接池对象
+    :param sql_shop_id_list: sql_shop_id_list
+    """
+    sql_id = sql_product_id[0]
+    pid = sql_product_id[1]
+    headers = {
+        "accept": "application/json, text/plain, */*",
+        "referer": "https://www.jihuanshe.com/",
+        "user-agent": user_agent.generate_user_agent()
+    }
+    url = "https://api.jihuanshe.com/api/market/share/auction-product"
+    params = {
+        "auction_product_id": pid,
+        "url": f"https://www.jihuanshe.com/app/auction?auctionProductId={pid}"
+    }
+    response = requests.get(url, headers=headers, params=params, timeout=5, proxies=get_proxys(log))
+    # print(response.json())
+    # print(response)
+    resp_json = response.json()
+    if resp_json:
+        if resp_json.get("code") == 440:
+            log.debug(f"< 抱歉，该竞价商品已被移除, pid:{pid} >")
+            sql_pool.update_one("update jhs_task set task_state = 2 where id = %s", (sql_id,))
+        else:
+            try:
+                parse_data(log, resp_json, sql_product_id, sql_pool, sql_shop_id_list)
+                # sql_pool.update_one("update jhs_task set task_state = 1 where id = %s", (sql_id,))
+            except Exception as e:
+                log.error(f"get_resp call parse_data() sql_product_id:{sql_product_id} 获取失败, error:{e}")
+
+
+@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
+def jhs_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:
+        # while True:
+        # product_id_list = sql_pool.select_all(
+        #     "SELECT id, product_id FROM jhs_task WHERE task_state = 0 AND id <= 420107 ORDER BY product_id DESC")
+
+        # 补充了17511条
+        product_id_list = sql_pool.select_all(
+            "SELECT jt.id ,jt.product_id FROM jhs_task jt LEFT JOIN jhs_product_record jpr ON jt.product_id = jpr.product_id WHERE jt.task_state = 1 AND jpr.product_id IS NULL")
+        product_id_list = [keyword for keyword in product_id_list]
+
+        sql_shop_id_list = sql_pool.select_all("SELECT seller_user_id FROM jhs_shop_record")
+        sql_shop_id_list = [keyword[0] for keyword in sql_shop_id_list]
+        # if not product_id_list:
+        #     log.info(f'没有待采集商品,等待十分钟再查询采集任务............')
+        #     time.sleep(600)
+        # continue
+
+        for sql_product_id in product_id_list:
+            try:
+                get_resp(log, sql_product_id, sql_pool, sql_shop_id_list)
+            except Exception as e:
+                log.error(f"Loop sql_product_id:{sql_product_id} 获取失败, error:{e}")
+                sql_pool.update_one("update jhs_task set task_state = 3 where id = %s", (sql_product_id[0],))
+    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} 运行结束,等待下一轮的采集任务............')
+
+
+if __name__ == '__main__':
+    jhs_main(logger)
+    # get_resp(logger, (1, 7), MySQLConnectionPool(log=logger))

jihuanshe_spider/jhs_start_spider.py

@@ -0,0 +1,21 @@
+# -*- coding: utf-8 -*-
+# Author : Charley
+# Python : 3.10.8
+# Date   : 2025/3/20 13:14
+import subprocess
+import time
+
+
+def start_spider(script_name):
+    """启动指定的爬虫脚本"""
+    print(f"Starting {script_name}...")
+    # 使用subprocess.Popen启动一个新的进程，这样即使父进程退出，子进程也能继续运行
+    subprocess.Popen(["python", script_name])
+
+
+# 启动已经内置了定时任务的爬虫
+# start_spider("jhs_sold_spider.py")
+# time.sleep(1)
+start_spider("jhs_new_daily_spider.py")
+
+print("All spiders have been started. Press Ctrl+C to stop the main process.")

jihuanshe_spider/mysq_pool.py

@@ -0,0 +1,191 @@
+# -*- coding: utf-8 -*-
+# Author  : Charley
+# Python  : 3.8.10
+# Date: 2024-08-05 19:42
+import pymysql
+import YamlLoader
+from loguru import logger
+from retrying import retry
+from dbutils.pooled_db import PooledDB
+
+# 获取yaml配置
+yaml = YamlLoader.readYaml()
+mysqlYaml = yaml.get("mysql")
+sql_host = mysqlYaml.getValueAsString("host")
+sql_port = mysqlYaml.getValueAsInt("port")
+sql_user = mysqlYaml.getValueAsString("username")
+sql_password = mysqlYaml.getValueAsString("password")
+sql_db = mysqlYaml.getValueAsString("db")
+
+
+class MySQLConnectionPool:
+    """
+    MySQL连接池
+    """
+
+    def __init__(self, mincached=4, maxcached=5, maxconnections=10, log=None):
+        """
+        初始化连接池
+        :param mincached: 初始化时，链接池中至少创建的链接，0表示不创建
+        :param maxcached: 池中空闲连接的最大数目（0 或 None 表示池大小不受限制）
+        :param maxconnections: 允许的最大连接数（0 或 None 表示任意数量的连接）
+        """
+        # 使用 loguru 的 logger，如果传入了其他 logger，则使用传入的 logger
+        self.log = log or logger
+        self.pool = PooledDB(
+            creator=pymysql,
+            mincached=mincached,
+            maxcached=maxcached,
+            maxconnections=maxconnections,
+            blocking=True,  # 连接池中如果没有可用连接后，是否阻塞等待。True，等待；False，不等待然后报错
+            host=sql_host,
+            port=sql_port,
+            user=sql_user,
+            password=sql_password,
+            database=sql_db
+        )
+
+    @retry(stop_max_attempt_number=100, wait_fixed=600000)
+    def _get_connection(self):
+        """
+        获取连接
+        :return: 连接
+        """
+        try:
+            return self.pool.connection()
+        except Exception as e:
+            self.log.error(f"Failed to get connection from pool: {e}, wait 10 mins retry")
+            raise e
+
+    @staticmethod
+    def _close_connection(conn):
+        """
+        关闭连接
+        :param conn: 连接
+        """
+        if conn:
+            conn.close()
+
+    # @retry(stop_max_attempt_number=5, wait_fixed=1000)
+    def _execute(self, query, args=None, commit=False):
+        """
+        执行SQL
+        :param query: SQL语句
+        :param args: SQL参数
+        :param commit: 是否提交事务
+        :return: 查询结果
+        """
+        conn = None
+        cursor = None
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            cursor.execute(query, args)
+            if commit:
+                conn.commit()
+            self.log.debug(f"sql _execute , Query: {query}, Rows: {cursor.rowcount}")
+            return cursor
+        except Exception as e:
+            if conn and not commit:
+                conn.rollback()
+            self.log.error(f"Error executing query: {e}")
+            raise e
+        finally:
+            if cursor:
+                cursor.close()
+            self._close_connection(conn)
+
+    def select_one(self, query, args=None):
+        """
+        执行查询，返回单个结果
+        :param query: 查询语句
+        :param args: 查询参数
+        :return: 查询结果
+        """
+        cursor = self._execute(query, args)
+        return cursor.fetchone()
+
+    def select_all(self, query, args=None):
+        """
+        执行查询，返回所有结果
+        :param query: 查询语句
+        :param args: 查询参数
+        :return: 查询结果
+        """
+        cursor = self._execute(query, args)
+        return cursor.fetchall()
+
+    def insert_one(self, query, args):
+        """
+        执行单条插入语句
+        :param query: 插入语句
+        :param args: 插入参数
+        """
+        self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_one 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        return self._execute(query, args, commit=True)
+
+    def insert_all(self, query, args_list):
+        """
+        执行批量插入语句，如果失败则逐条插入
+        :param query: 插入语句
+        :param args_list: 插入参数列表
+        """
+        conn = None
+        cursor = None
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            cursor.executemany(query, args_list)
+            conn.commit()
+            self.log.debug(f"sql insert_all , SQL: {query}, Rows: {cursor.rowcount}")
+            self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data insert_all 入库中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        except Exception as e:
+            conn.rollback()
+            self.log.error(f"Batch insertion failed after 5 attempts. Trying single inserts. Error: {e}")
+            # 如果批量插入失败，则逐条插入
+            rowcount = 0
+            for args in args_list:
+                self.insert_one(query, args)
+                rowcount += 1
+            self.log.debug(f"Batch insertion failed. Inserted {rowcount} rows individually.")
+        finally:
+            cursor.close()
+            self._close_connection(conn)
+
+    def update_one(self, query, args):
+        """
+        执行单条更新语句
+        :param query: 更新语句
+        :param args: 更新参数
+        """
+        self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data update_one 更新中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        return self._execute(query, args, commit=True)
+
+    def update_all(self, query, args_list):
+        """
+        执行批量更新语句，如果失败则逐条更新
+        :param query: 更新语句
+        :param args_list: 更新参数列表
+        """
+        conn = None
+        cursor = None
+        try:
+            conn = self._get_connection()
+            cursor = conn.cursor()
+            cursor.executemany(query, args_list)
+            conn.commit()
+            self.log.debug(f"sql update_all , SQL: {query}, Rows: {cursor.rowcount}")
+            self.log.info('>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>data update_all 更新中>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
+        except Exception as e:
+            conn.rollback()
+            self.log.error(f"Error executing query: {e}")
+            # 如果批量更新失败，则逐条更新
+            rowcount = 0
+            for args in args_list:
+                self.update_one(query, args)
+                rowcount += 1
+            self.log.debug(f'Batch update failed. Updated {rowcount} rows individually.')
+
+        finally:
+            cursor.close()
+            self._close_connection(conn)

jihuanshe_spider/requirements.txt

@@ -0,0 +1,13 @@
+-i https://mirrors.aliyun.com/pypi/simple/
+airtest==1.3.5
+DBUtils==3.1.0
+loguru==0.7.3
+parsel==1.10.0
+pocoui==1.0.94
+PyMySQL==1.1.1
+PyYAML==6.0.2
+requests==2.32.3
+retrying==1.3.4
+schedule==1.2.2
+tenacity==9.0.0
+user_agent==0.1.10