spider_work/jihuanshe_spider/jhs_sold_spider.py

# -*- 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))