spider_work/baocui_spider/bc_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/2/27 11:57
import inspect
import requests
import user_agent
from loguru import logger
from mysq_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed

"""
D+9位数，从D000000001-D000200000
E+9位数，从E000000001-E000040000
C+9位数，从C000000001-C000600000
"""

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_data(sql_pool, info):
    """
    保存数据
    :param sql_pool: MySQL连接池对象
    :param info: 要保存的数据
    """
    sql = """
        INSERT INTO baocui_record (keyword, picurl, dt_pics, vr_pics, vr_date, title, name, year, score, card_type, card_num, card_score, card_centering, card_edges, card_corners, card_surface)
        VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
        """
    sql_pool.insert_one(sql, info)


def parse_data(resp_json: dict, keyword, sql_pool):
    """
    解析数据
    :param resp_json: response.json()的返回值 -> dict
    :param keyword: 查询的keyword
    :param sql_pool: MySQL连接池对象
    """
    info_dict = resp_json.get("info")
    if info_dict:
        picurl = info_dict.get("picurl")  # 正面图
        dt_pics = info_dict.get("dt_pics")  # 反面图
        vr_pics = info_dict.get("vr_pics")  # VR图
        vr_date = info_dict.get("vr_date")

        title = info_dict.get("details", {}).get("title")

        details_grade = info_dict.get("details", {}).get("grade", [])
        info_dict = {
            'name': None,
            'year': None,
            'score': None,
            'card_type': None,
            'card_num': None,
            'card_score': None,
            'card_centering': None,
            'card_edges': None,
            'card_corners': None,
            'card_surface': None

        }
        for grade in details_grade:
            grade_name = grade.get("title")
            grade_value = grade.get("val")
            # print(f"{grade_name}:{grade_value}")
            if grade_name == "名称":
                info_dict['name'] = grade_value
            elif grade_name == "年份":
                info_dict['year'] = grade_value
            elif grade_name == "分数":
                info_dict['score'] = grade_value
            elif grade_name == "卡种":
                info_dict['card_type'] = grade_value
            elif grade_name == "编号":
                info_dict['card_num'] = grade_value
            elif grade_name == "分值":
                info_dict['card_score'] = grade_value
            elif grade_name == "CENTERING":
                info_dict['card_centering'] = grade_value
            elif grade_name == "EDGES":
                info_dict['card_edges'] = grade_value
            elif grade_name == "CORNERS":
                info_dict['card_corners'] = grade_value
            elif grade_name == "SURFACE":
                info_dict['card_surface'] = grade_value

        info = (keyword, picurl, dt_pics, vr_pics, vr_date, title, info_dict.get('name'), info_dict.get('year'),
                info_dict.get('score'), info_dict.get('card_type'), info_dict.get('card_num'),
                info_dict.get('card_score'), info_dict.get('card_centering'), info_dict.get('card_edges'),
                info_dict.get('card_corners'), info_dict.get('card_surface'))
        # print(info)
        save_data(sql_pool, info)


@retry(stop=stop_after_attempt(5), wait=wait_fixed(1), after=after_log)
def get_bc_resp(log, keyword, sql_pool):
    """
    获取响应
    :param log: logger对象
    :param keyword: 查询的keyword
    :param sql_pool: MySQL连接池对象
    """
    headers = {
        "accept": "application/json",
        "accept-language": "en,zh-CN;q=0.9,zh;q=0.8",
        # "referer": "https://www.baocuicoin.com/",
        "user-agent": user_agent.generate_user_agent()
    }

    url = "https://web-api.baocuicoin.com/Search/index"
    params = {
        "keyword": keyword,
        # "code": "404"
        "code": ""
    }
    try:
        with requests.Session() as session:
            response = session.post(url, headers=headers, params=params, proxies=get_proxys(log), timeout=10)
        # print(response.text)
        # print(response.json())
        response.raise_for_status()
        resp_json = response.json()
        if not resp_json:
            log.error(f"Not resp_json, {keyword} 获取失败, msg:{resp_json.get('msg')}")
            return

        if resp_json.get("code") == '200':
            parse_data(resp_json, keyword, sql_pool)
        else:
            log.error(f"code != 200, {keyword} 获取失败, msg:{resp_json.get('msg')}")
    except Exception as e:
        log.warning(f"{inspect.currentframe().f_code.co_name} error: {e}")


def generate_keywords():
    """
    生成要查询的 keywords
    """
    start = 1
    # for prefix, end in [('C', 600001), ('D', 200001), ('E', 40001)]:
    for prefix, end in [('D', 200001), ('E', 40001)]:
        if prefix == 'D':
            start = 150503
        for i in range(start, end):
            yield f"{prefix}{i:09d}"


@retry(stop=stop_after_attempt(50), wait=wait_fixed(1800), after=after_log)
def baocui_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:
        # keyword = 'E000011631'
        for keyword in generate_keywords():
            try:
                get_bc_resp(log, keyword, sql_pool)
            except Exception as e:
                log.error(f"Loop keyword:{keyword} 获取失败, error:{e}")
    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__':
    baocui_main(logger)