spider_work/baocui_spider/bc_new_daily_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2025/6/11 19:00
import inspect
import random
import time

import requests
import schedule
import user_agent
from loguru import logger
from mysql_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)
        sql_pool.update_one("update baocui_task set state = 1 where keyword = %s", (keyword,))


@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": ""
    }

    with requests.Session() as session:
        response = session.post(url, headers=headers, params=params, proxies=get_proxys(log), timeout=10)
    # response = requests.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)
    elif resp_json.get("code") == '0':
        msg = resp_json.get('msg')
        if '编号未收录' in msg:
            log.error(f"code == 0, {keyword} 获取失败, msg:{msg}")
            sql_pool.update_one("update baocui_task set state = 2 where keyword = %s", (keyword,))
        elif 'IP已被封禁' in msg:
            log.error(f"code == 0, {keyword} 获取失败, msg:{msg}")
            raise  Exception("IP已被封禁, 重新请求")
        else:
            log.error(f"code == 0, {keyword} 获取失败, msg:{msg}")
            sql_pool.update_one("update baocui_task set state = 3 where keyword = %s", (keyword,))
    else:
        log.error(f"code != 0 or 200, {keyword} 获取失败, msg:{resp_json.get('msg')}")
        sql_pool.update_one("update baocui_task set state = 3 where keyword = %s", (keyword,))


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), 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:
        while True:
            # keyword = 'E000011631'
            sql_code_list = sql_pool.select_all("select keyword from baocui_task where state = 0 limit 10000")
            sql_code_list = [i[0] for i in sql_code_list]
            # if not sql_code_list:
            #     log.debug(
            #         f"{inspect.currentframe().f_code.co_name} -> No len sql_code_list, Wait 1 hours again query.........")
            #     time.sleep(3600)
            #     sql_code_list = sql_pool.select_all("select keyword from baocui_task where state = 0 limit 10000")
            #     sql_code_list = [i[0] for i in sql_code_list]
            if not sql_code_list:
                log.debug(f"{inspect.currentframe().f_code.co_name} -> No len sql_code_list")
                break

            for keyword in sql_code_list:
                try:
                    get_bc_resp(log, keyword, sql_pool)
                except Exception as e:
                    log.error(f"Loop keyword:{keyword} 获取失败, error:{e}")
                    sql_pool.update_one("update baocui_task set state = 3 where keyword = %s", (keyword,))
                time.sleep(random.uniform(0.5, 1))
    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():
    """
    爬虫模块的启动文件
    """
    # 立即运行一次任务
    baocui_main(log=logger)

    # 设置定时任务
    schedule.every(30).days.at("00:01").do(baocui_main, log=logger)

    while True:
        schedule.run_pending()
        time.sleep(1)


if __name__ == '__main__':
    schedule_task()