[Python+MySQL+PHP] 赛氪比赛管理器

项目背景

赛氪（Saikr）是一家专注于大学生竞赛活动的平台，致力于为全国大学生提供高含金量的竞赛信息、报名和成绩查询服务。
平台涵盖英语竞赛、数学竞赛、编程挑战赛、知识竞赛等多种赛事，知名竞赛如全国大学生英语竞赛、蓝桥杯、互联网+大学生创新创业大赛等均可在赛氪上报名和查询成绩。

说白了，赛氪是一个有一堆比赛的平台，应该涵盖了我们在大学期间能接触到的至少70%的校级以上比赛。

当然，里面的水赛也特别多……

我们学院参加各类竞赛并获奖可以申报专项奖学金，虽然金额不多，但是至少可以回本。参加的比赛必须满足：除优秀奖之外，总获奖率≤50%。

赛氪是一个很好的比赛信息来源之一。

截止24年11月3日笔者撰写本文，赛氪现在共登记有51032个比赛，除去不能报名的，至少也有上百个比赛，再除去不合申报要求的比赛，恐怕更少了。

需求拆解

标记看过的比赛。
屏蔽水赛或不满足我要求的比赛。
标记报名了的比赛，提醒我参赛。

方案选型

数据爬取: 使用 Python 和 requests 库爬取赛氪网的比赛信息。
数据存储: 采用 MySQL 数据库存储爬取到的比赛信息。
后端处理: 使用 PHP 作为后端语言进行数据处理与接口管理。
前端展示: 使用 H5 和 Bootstrap 框架构建前端用户界面。
数据更新: 青龙面板设置每日定时任务，自动更新和同步比赛信息。

效果展示

登录界面

管理界面

查看比赛界面

爬虫运行界面

源码展示

在这里，我先开源MySQL版的赛氪比赛抓取器。剩下的内容整理优化后再开源到gitee或者github.

Python和数据库部分

使用前，请建立一个MYSQL数据库，并输入以下建表指令：

表1：`contests`

（如果只想让Python爬虫跑起来，建立这个表就行）

CREATE TABLE `contests` (
  `contest_id` INT(11) NOT NULL AUTO_INCREMENT,
  `contest_url` VARCHAR(255) NOT NULL,
  `contest_name` VARCHAR(255) DEFAULT NULL,
  `org_real_name` VARCHAR(255) DEFAULT NULL,
  `org_avatar` VARCHAR(255) DEFAULT NULL,
  `org_univs_name` VARCHAR(255) DEFAULT NULL,
  `contest_start_time` DATETIME DEFAULT NULL,
  `contest_end_time` DATETIME DEFAULT NULL,
  `regist_start_time` DATETIME DEFAULT NULL,
  `regist_end_time` DATETIME DEFAULT NULL,
  `is_valid` TINYINT(1) DEFAULT NULL,
  `is_contest_status` TINYINT(1) DEFAULT NULL,
  `look_count` INT(11) DEFAULT 0,
  `focus_num` INT(11) DEFAULT 0,
  `content` TEXT DEFAULT NULL,
  `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  `updated_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  PRIMARY KEY (`contest_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8_general_ci;

表2：`contest_tags`

CREATE TABLE `contest_tags` (
  `tag_id` INT(11) NOT NULL AUTO_INCREMENT,
  `user_id` INT(11) NOT NULL,
  `contest_id` INT(11) NOT NULL,
  `tag_type` ENUM('like', 'block') NOT NULL,
  `is_registered` TINYINT(1) DEFAULT 0,
  `next_contest_time` DATETIME DEFAULT NULL,
  `created_at` TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
  PRIMARY KEY (`tag_id`),
  KEY `user_id` (`user_id`),
  KEY `tag_type` (`tag_type`),
  KEY `contest_id` (`contest_id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8_general_ci;

第一个表contests存储竞赛相关的信息，第二个表contest_tags用于用户与竞赛之间的标签关系，支持多种类型标签（如点赞和屏蔽）。

Python代码：

import requests
import mysql.connector
from mysql.connector import errorcode
from datetime import datetime

# 配置数据库连接
config = {
    'user': '数据库用户名',
    'password': '数据库密码',
    'host': '数据库IP地址',  # 通常为localhost
    'database': '数据库名',
    'raise_on_warnings': True
}

# 连接到数据库
try:
    cnx = mysql.connector.connect(**config)
    cursor = cnx.cursor()
except mysql.connector.Error as err:
    if err.errno == errorcode.ER_ACCESS_DENIED_ERROR:
        print("用户名或密码错误")
    elif err.errno == errorcode.ER_BAD_DB_ERROR:
        print("数据库不存在")
    else:
        print(err)
    exit(1)

# 按报名时间降序排列的比赛列表，相当于是先抓取最新的，如果要更改抓取的条数只需要更改limit参数为别的数值，具体对应https://new.saikr.com/contests页面
LIST_API_URL = "https://apiv4buffer.saikr.com/api/pc/contest/lists?page=1&limit=150&univs_id=&class_id=&level=0&sort=0"

def convert_unix_to_datetime(timestamp):
    return datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')

def fetch_contest_list():
    try:
        response = requests.get(LIST_API_URL)
        response.raise_for_status()
        data = response.json()
        if data['code'] == 200:
            return data['data']['list']
        else:
            print(f"列表API返回错误代码: {data['code']}, 信息: {data['msg']}")
            return []
    except requests.RequestException as e:
        print(f"请求列表API失败: {e}")
        return []

    #抓取详细的比赛信息，甚至比赛内容都在里面，所以拿到这个内容之后甚至可以自己复刻一个赛氪了
def fetch_contest_info(contest_url_suffix):
    info_api_url = f"https://apiv4buffer.saikr.com/api/pc/contest/info?contest_url={contest_url_suffix}&isp="
    try:
        response = requests.get(info_api_url)
        response.raise_for_status()
        data = response.json()
        if data['code'] == 200:
            return data['data']
        else:
            print(f"信息API返回错误代码: {data['code']}, 信息: {data['msg']}")
            return None
    except requests.RequestException as e:
        print(f"请求信息API失败: {e}")
        return None

    #解析比赛内容
def insert_or_update_contest(contest_summary, contest_info):
    contest_id = contest_summary['contest_id']
    contest_url_suffix = contest_summary['contest_url'].replace("vse/", "")
    complete_contest_url = f"https://new.saikr.com/vse/{contest_url_suffix}"

    # Convert UNIX timestamps to datetime strings
    regist_start_time = convert_unix_to_datetime(contest_summary['regist_start_time'])
    regist_end_time = convert_unix_to_datetime(contest_summary['regist_end_time'])
    contest_start_time = convert_unix_to_datetime(contest_summary['contest_start_time'])
    contest_end_time = convert_unix_to_datetime(contest_summary['contest_end_time'])

    is_valid = 1 if contest_summary.get('is_exam', 0) == 0 else 0
    is_contest_status = 1 if contest_summary.get('is_contest_status', 0) == 1 else 0
    
    # 从 contest_info 中获取 look_count 和 focus_num
    look_count = contest_info.get('look_count', 0)
    focus_num = contest_info.get('focus_num', 0)
    
    content = contest_info.get('content', '')

    insert_query = """
    INSERT INTO contests (contest_id, contest_url, contest_name, org_real_name, org_avatar, org_univs_name,
                          contest_start_time, contest_end_time, regist_start_time, regist_end_time,
                          is_valid, is_contest_status, look_count, focus_num, content)
    VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
    ON DUPLICATE KEY UPDATE
        contest_url=VALUES(contest_url),
        contest_name=VALUES(contest_name),
        org_real_name=VALUES(org_real_name),
        org_avatar=VALUES(org_avatar),
        org_univs_name=VALUES(org_univs_name),
        contest_start_time=VALUES(contest_start_time),
        contest_end_time=VALUES(contest_end_time),
        regist_start_time=VALUES(regist_start_time),
        regist_end_time=VALUES(regist_end_time),
        is_valid=VALUES(is_valid),
        is_contest_status=VALUES(is_contest_status),
        look_count=VALUES(look_count),
        focus_num=VALUES(focus_num),
        content=VALUES(content)
    """
    contest_data = (
        contest_id,
        complete_contest_url,
        contest_summary['contest_name'],
        contest_info.get('org_real_name', ''),  # 从 contest_info 获取
        contest_info.get('org_avatar', ''),
        contest_info.get('org_univs_name', ''),  # 从 contest_info 获取
        contest_start_time,
        contest_end_time,
        regist_start_time,
        regist_end_time,
        is_valid,
        is_contest_status,
        look_count,
        focus_num,
        content
    )
    try:
        cursor.execute(insert_query, contest_data)
        cnx.commit()
        print(f"已插入或更新比赛ID {contest_id}")
    except mysql.connector.Error as err:
        print(f"插入或更新比赛ID {contest_id} 失败: {err}")

def main():
    contest_list = fetch_contest_list()
    if not contest_list:
        print("没有获取到任何比赛数据。")
        return
    
    for contest_summary in contest_list:
        contest_id = contest_summary['contest_id']
        contest_url_suffix = contest_summary['contest_url'].replace("vse/", "")
        contest_info = fetch_contest_info(contest_url_suffix)
        if contest_info:
            insert_or_update_contest(contest_summary, contest_info)
        else:
            print(f"跳过比赛ID {contest_id} 由于无法获取详细信息。")

    # 关闭数据库连接
    cursor.close()
    cnx.close()
    print("数据抓取完成。")

if __name__ == "__main__":
    main()

手机上阅读

最后一次更新于2024-11-03

Soulxyzの日常

不要让时代的悲哀成为你的悲哀