spider_work/ags_spider/ags_pop_spider.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.12.10
# Date   : 2026/5/7 13:48
import inspect
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
"""
2026/5/7 网站改版  代码重构
"""
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


@retry(stop=stop_after_attempt(5), wait=wait_fixed(2), after=after_log)
def get_ags_pop_list(log, sql_pool):
    """
    获取卡组 数据报告 信息
    :param log: logger对象
    :param sql_pool: MySQLConnectionPool对象
    """
    crawl_date = time.strftime("%Y-%m-%d", time.localtime())
    headers = {
        "accept": "application/json",
        "user-agent": user_agent.generate_user_agent()
    }
    url = "https://app.agscard.com/api/v2/pop/categories"
    response = requests.get(url, headers=headers, timeout=10)
    # print(response.text)
    # print(response)
    response.raise_for_status()

    resp_json = response.json()
    resp_data = resp_json.get("data", [])
    if not resp_data:
        log.error("No data found in the response")
        return

    info_list = []
    for category_ in resp_data:
        category_id = category_.get("id")
        category_name = category_.get("name")
        category_link = f"https://agscard.com/pop/{category_id}"
        category_sets = category_.get("card_sets_count")
        category_cards = category_.get("card_products_count")
        category_graded = category_.get("total_population")

        data_dict = {
            'category_name': category_name,
            'category_link': category_link,
            'category_sets': category_sets,
            'category_cards': category_cards,
            'category_graded': category_graded,
            'crawl_date': crawl_date
        }
        # print(data_dict)
        info_list.append(data_dict)

    # 保存数据
    sql_pool.insert_many(table="ags_pop_record", data_list=info_list, ignore=True)


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def ags_pop_main(log):
    """
    主函数
    :param log: logger对象
    """
    log.info(
        f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................')

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=log)
    if not sql_pool.check_pool_health():
        log.error("数据库连接池异常")
        raise RuntimeError("数据库连接池异常")

    try:
        # 获取所有卡组中的卡牌信息
        get_ags_pop_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():
    """
    爬虫模块 定时任务 的启动文件
    """
    # 立即运行一次任务
    ags_pop_main(log=logger)

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

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


if __name__ == '__main__':
    schedule_task()
    # ags_pop_main(log=logger)