main.py

main.py — 入口与调度中心

通过 -m 参数控制运行模式，主要流程如下：

特点： 支持断点续跑（-s 指定单文件）、代理（-p）、分页起始页（-t）、单独处理某公司（-c）。

代码

# -*- coding: utf-8 -*-
"""
学校/企业公告数据爬虫主程序
功能：使用Playwright自动化爬取学校、企业公告数据 → 本地处理 → 上传云端API
支持：定时循环、单文件处理、单企业处理、代理、重试、分页控制
"""

# 导入标准库：时间处理、系统操作、命令行参数解析
import time
import os
import argparse

# 导入Playwright自动化工具（用于模拟浏览器爬取网页）
from playwright.sync_api import sync_playwright

# 导入自定义爬虫模块
from spider_sch import SpiderSch       # 学校爬虫类
from spider_com import SpiderCom       # 企业爬虫类
from spider_data import DEFAULT_WX_SCHOOL, DEFAULT_PCOUNT  # 常量：微信特殊学校、默认处理条数
from spider_data import SpiderData     # 数据处理/上传类

# 导入工具模块：日志、浏览器工具
from utils import set_logger_debug, ner_logger
from utils_playwright import get_browser

# ====================== 命令行参数配置 ======================
# 创建命令行参数解析器，用于接收外部运行指令
parser = argparse.ArgumentParser(description='学校/企业公告数据爬取主程序')

# 运行模式（核心参数）
parser.add_argument('-m', '--method', default="p", type=str, help='启动方式：p=爬学校,cp=爬企业,ann=处理公告等')
# 配置文件编号
parser.add_argument('-f', '--file', default="99", type=str, help='使用的配置文件编号，对应不同爬取任务')
# 版本说明
parser.add_argument('-v', '--version', default="", type=str, help='版本说明')
# 云端API环境（dev/prod）
parser.add_argument('-d', '--dist', default="dev", type=str, help='云端API接口环境：dev=测试,prod=正式')
# 指定只处理单个文件（用于断点续爬/修复）
parser.add_argument('-s', '--processfile', default="", type=str, help='指定需要处理的单个文件ID')
# 指定只处理单个公司
parser.add_argument('-c', '--company', default="", type=str, help='指定单独处理的公司名称/标识')
# 代理设置
parser.add_argument('-p', '--proxy', default="", type=str, help='浏览器代理地址，为空则不使用')
# 重试开关
parser.add_argument('-r', '--retry', default="1", type=str, help='是否允许重试：1=允许，0=不允许')
# 翻页起始页码
parser.add_argument('-t', '--pagestart', default="2", type=str, help='翻页开始页数，默认从第2页开始')

# 解析命令行传入的所有参数
args = parser.parse_args()

# ====================== 核心爬取函数 ======================
def clawler_main(s, _stat={}):
    """
    爬虫主入口：启动浏览器 → 遍历学校/企业节点 → 执行爬取
    :param s: 爬虫实例（SpiderSch / SpiderCom）
    :param _stat: 运行状态字典（记录配置、进度、过滤条件）
    """
    # 从爬虫实例中获取浏览器可执行文件路径
    executable_path = s.get_browser_path()

    # 启动Playwright上下文（管理浏览器生命周期）
    with sync_playwright() as p:
        # 创建浏览器实例（自动处理代理、无头模式、防检测）
        browser = get_browser(p, executable_path, args.proxy)
        # 将浏览器对象绑定到爬虫实例，供内部方法使用
        s.browser = browser

        # 新建浏览器标签页
        page = browser.new_page()

        # ===================== 爬取核心逻辑 =====================
        # 获取所有需要爬取的学校/企业节点（配置文件中定义）
        nodes = s.get_nodes()
        # 获取已爬取进度（用于断点续爬）
        process = s.get_progress()

        # 遍历所有节点（key=学校/企业ID，node=详情配置）
        for _key, _node in nodes.items():
            # 过滤：跳过微信专用特殊学校（单独处理）
            if _key == DEFAULT_WX_SCHOOL:
                continue

            # 过滤：如果指定了单独处理公司，则只爬该公司
            if 'company' in _stat and _stat['company'] != _key:
                continue

            # 遍历该节点下的所有学校/企业信息
            for _sch_info in _node:
                # 如果该节点在进度记录中 → 开始执行爬取
                if _key in process:
                    print(f"开始爬取 {_key}")
                    # 调用实例内部的run方法执行具体爬取逻辑
                    s.run(page, _key, _sch_info, _stat)

        # 爬取完成，等待10秒后关闭浏览器
        time.sleep(10)
        browser.close()

# ====================== 数据处理函数 ======================
def process_main_wx(s, d, _stat={}):
    """
    处理微信渠道数据（单独处理）
    :param s: 学校爬虫实例
    :param d: 数据处理实例
    :param _stat: 状态参数
    """
    nodes = s.get_nodes()
    for _key, _node in nodes.items():
        # 跳过微信特殊学校
        if _key == DEFAULT_WX_SCHOOL:
            continue
        # 遍历并处理公告数据（渠道标记为wx）
        for _sch_info in _node:
            d.process_announcement_data(_key, _sch_info, _stat, "wx")

def process_main_announcement(s, d, _stat={}, proc_type="ann", _uapi="up_api"):
    """
    处理爬取到的公告数据（清洗、解析、入库）
    :param proc_type: 处理类型 ann=普通公告 cjob=企业职位
    :param _uapi: 上传API类型
    """
    nodes = s.get_nodes()
    for _key, _node in nodes.items():
        if _key == DEFAULT_WX_SCHOOL:
            continue

        for _sch_info in _node:
            # 执行数据解析处理
            d.process_announcement_data(_key, _sch_info, _stat, proc_type)

            # 达到默认批量处理条数 → 自动触发上传
            if 'total' in _stat and _stat['total'] >= DEFAULT_PCOUNT:
                ner_logger.info(f"{_stat['total']}条数据已处理完成")
                # 调用上传接口
                d.process_announcement_data(_key, _sch_info, _stat, _uapi)
                ner_logger.info(f"{_stat['total']}条数据已上传完成")
                # 重置计数器
                _stat['total'] = 0

def process_main_wxhand(s, d, _stat={}):
    """
    处理【微信手工录入】的文章数据（专用渠道）
    """
    # 预先初始化微信文章数据
    d.pre_wx_article()
    nodes = s.get_nodes()

    for _key, _node in nodes.items():
        # 只处理微信特殊学校
        if _key != DEFAULT_WX_SCHOOL:
            continue

        ner_logger.info(f"开始处理特殊学校数据：{_key}")
        for _sch_info in _node:
            d.process_announcement_data(_key, _sch_info, _stat, "wx")

def process_main_upapi(s, d, _stat={}, _uapi="up_api"):
    """
    统一上传数据到云端API
    """
    nodes = s.get_nodes()
    for _key, _node in nodes.items():
        if _key == DEFAULT_WX_SCHOOL:
            continue

        for _sch_info in _node:
            d.process_announcement_data(_key, _sch_info, _stat, _uapi)

def process_main_wxhand_api(s, d, _stat={}):
    """
    上传微信手工数据到云端API
    """
    nodes = s.get_nodes()
    for _key, _node in nodes.items():
        # 只处理微信特殊学校
        if _key != DEFAULT_WX_SCHOOL:
            continue

        ner_logger.info(f"开始处理特殊学校数据上传：{_key} {DEFAULT_WX_SCHOOL}")
        for _sch_info in _node:
            d.process_announcement_data(_key, _sch_info, _stat, "up_api")

# ====================== 主调度函数（根据-m参数执行对应流程） ======================
def run_periodically(s, cs, d):
    """
    周期运行调度器：根据命令行参数 -m 执行对应任务
    :param s: 学校爬虫实例
    :param cs: 企业爬虫实例
    :param d: 数据处理实例
    """
    # 初始化状态字典：存储运行配置、过滤条件、计数器
    _stat = {}

    # ===================== 读取命令行参数到状态字典 =====================
    # 单文件处理
    if args.processfile != "":
        _stat['pfile'] = args.processfile
    # 单企业处理
    if args.company != "":
        _stat['company'] = args.company
    # 起始页码
    if args.pagestart != "":
        _stat['page_start'] = int(args.pagestart)
    else:
        _stat['page_start'] = 2
    # 云端环境
    _stat['dist'] = args.dist
    # 是否重试
    _stat['retry'] = args.retry
    # 已处理公告计数
    _stat['p_count'] = 0
    # 已处理列表（去重）
    _stat['all_proc_list'] = []

    # 打印模式：仅输出所有学校列表，不爬取
    if args.method in ["o"]:
        s.print_all_sch()

    # ===================== 企业爬虫流程 =====================
    # 1. 爬取企业原始数据
    if args.method in ["cp", 'cp_full']:
        _stat['method'] = args.method
        clawler_main(cs, _stat)
        time.sleep(1 * 6)
        ner_logger.info("第一步：爬取企业数据完成，休息后继续处理")

    # 2. 处理企业职位数据
    if args.method in ["cjob"]:
        process_main_announcement(cs, d, _stat, "cjob", "up_api_cjob")
        time.sleep(1 * 6)
        ner_logger.info("第二步：处理职位数据完成")

    # 3. 上传企业数据到云
    if args.method in ["cjob_api"]:
        process_main_upapi(cs, d, _stat, "up_api_cjob")
        time.sleep(1 * 6)
        ner_logger.info("第四步：企业数据上传完成")

    # 循环执行企业爬取+处理+上传
    if args.method in ["all_job"]:
        ner_logger.info("启动企业公告循环处理任务")
        _loop_times = 0
        while True:
            _all_total = 0
            _stat['total'] = 0

            # 处理数据
            process_main_announcement(cs, d, _stat, "cjob", "up_api_cjob")
            _all_total += _stat['total']
            time.sleep(6)

            # 上传数据
            _stat['total'] = 0
            process_main_upapi(cs, d, _stat, "up_api_cjob")
            _all_total += _stat['total']
            time.sleep(6)

            # 长期无数据则退出循环
            if _all_total == 0 and _loop_times > 200:
                ner_logger.info("无新数据，结束循环")
                break
            _loop_times += 1

    # ===================== 学校爬虫流程 =====================
    # 1. 爬取学校数据
    if args.method in ["p"]:
        clawler_main(s, _stat)
        time.sleep(6)
        ner_logger.info("第一步：学校数据爬取完成")

    # 循环爬取学校（每8小时一次）
    if args.method in ["all_p"]:
        ner_logger.info("启动学校循环爬取任务（8小时/次）")
        clawler_main(s, _stat)
        time.sleep(60 * 60 * 8)
        ner_logger.info("循环爬取完成，休息8小时")

    # ===================== 学校数据处理流程 =====================
    # 2. 处理学校公告
    if args.method in ["ann"]:
        process_main_announcement(s, d, _stat)
        time.sleep(6)
        ner_logger.info("第二步：公告处理完成")

    # 3. 处理学校微信数据
    if args.method in ["ann_wx"]:
        process_main_wx(s, d, _stat)
        time.sleep(6)
        ner_logger.info("第三步：微信数据处理完成")

    # 4. 上传学校数据到云
    if args.method in ["ann_api"]:
        process_main_upapi(s, d, _stat)
        time.sleep(6)
        ner_logger.info("第四步：数据上传完成")

    # 循环处理学校公告+微信+上传
    if args.method in ["all_ann"]:
        ner_logger.info("启动学校公告循环处理任务")
        _loop_times = 0
        while True:
            _loop_times += 1
            _all_total = 0
            _stat['total'] = 0

            # 上传
            process_main_upapi(s, d, _stat)
            _all_total += _stat['total']
            time.sleep(6)

            # 处理公告
            _stat['total'] = 0
            process_main_announcement(s, d, _stat)
            _all_total += _stat['total']
            time.sleep(6)

            # 处理微信
            _stat['total'] = 0
            process_main_wx(s, d, _stat)
            _all_total += _stat['total']
            time.sleep(6)

            # 无数据则退出
            if _all_total == 0 and _loop_times > 20:
                ner_logger.info("无新数据，结束循环")
                break

    # ===================== 微信手工数据流程 =====================
    # 处理微信手工数据
    if args.method in ["wxhand"]:
        process_main_wxhand(s, d, _stat)
        ner_logger.info("第一步：微信手工数据处理完成")

    # 上传微信手工数据
    if args.method in ["wxhand_api"]:
        process_main_wxhand_api(s, d, _stat)
        ner_logger.info("第二步：微信手工数据上传完成")

    # 循环处理微信手工数据
    if args.method in ["all_wxhand"]:
        ner_logger.info("启动微信循环处理任务")
        while True:
            _stat['total'] = 0
            process_main_wxhand_api(s, d, _stat)
            time.sleep(6)

            _stat['total'] = 0
            process_main_wxhand(s, d, _stat)
            time.sleep(6)

            _stat['total'] = 0
            process_main_wxhand_api(s, d, _stat)
            time.sleep(6)

# ====================== 程序入口 ======================
if __name__ == '__main__':
    # 初始化日志系统
    set_logger_debug(args.file)

    # 初始化三大核心实例
    s = SpiderSch(args.file)       # 学校爬虫
    cs = SpiderCom(args.file)      # 企业爬虫
    d = SpiderData(s)              # 数据处理/上传

    # 如果输入版本参数 → 打印帮助说明
    if args.version != '':
        print("===== 使用说明 =====")
        print("-m / --method : 运行模式")
        print("  p      = 爬取学校数据")
        print("  cp     = 爬取企业数据")
        print("  ann    = 处理学校公告")
        print("  cjob   = 处理企业职位")
        print("  ann_api= 上传学校数据")
        print("  wxhand = 处理微信手工数据")
        print("-f / --file   : 配置文件编号")
        exit()

    # 启动调度器，开始执行任务
    run_periodically(s, cs, d)

项目分区导航：全链路流程梳理 ⬅️ | 01-main.py | ➡️ spider_com.py