---
title: "04-spider_sch.py"
created: 2026-04-03
tags:
- 项目
aliases:
- spider_sch.py
---
# spider_sch.py
### `spider_sch.py` — 学校官网爬虫
核心类 `SpiderSch`，负责从高校招聘页爬取数据。
**流程：**
1. 读取 INI 配置文件（`setting_sch_*.ini`），加载每所学校的URL、CSS选择器、解析函数等参数
2. `run()` → 逐个学校打开首页，调用 `get_page_data()` 爬取列表页
3. `get_page_data()` → 用 Playwright 渲染页面，提取列表 HTML，调用自动生成的解析函数（`auto_gen.gen.*`）解析成 JSON
4. `get_page_detail_data()` → 逐条访问详情页，处理各种边界情况（无链接时点击、相对路径、外部跳转、iframe、微信链接等）
5. 详情页内容和元数据分别保存为 `.html` 和 `.json` 文件
**亮点：**
- 支持进度持久化（`progress_*.txt`），程序中断后从断点继续
- 标题过滤：用正则黑名单文件剔除无关公告
- 日期过滤：只爬取180天内的公告
- 多种异常兜底：selector找不到时有备选 selector、iframe 穿透、微信链接特殊处理
## 代码
```python
# -*- coding: utf-8 -*-
"""
学校爬虫核心类：SpiderSch
功能：根据配置文件自动爬取学校招聘/公告信息
流程：打开列表页 → 解析标题/时间/链接 → 进入详情页 → 提取内容 → 保存HTML/JSON
支持：断点续爬、标题过滤、iframe解析、跳转链接、微信文章、BS4提取、防重复爬取
"""
import os, hashlib, json, time
import configparser
import glob
import requests
import re
from collections import defaultdict
# 自动生成的函数调用工具
from auto_gen.func_call import call_func, execute_page_action, execute_index_action
# 工具类：日志、URL处理、文本校验、时间判断、随机数、HTML清洗等
from utils import ner_logger, get_long_url_domain
from utils_date import is_near_month
from utils import get_final_url, check_contact, check_url_type, get_random_number
from utils import is_wechat_url
from utils_html import clean_html, find_wx_url
from utils_html import get_directory_from_url
from utils_resume import remove_announcement_word
from utils_playwright import click_by_text_and_get_url, get_iframe_urls, get_redirect_url
from utils_bs4 import get_node_text
# ===================== 全局常量配置 =====================
# 临时爬取进度文件（断点续爬用）
TMP_PROGRESS_FILE = "data/progress.txt"
# 首页加载停顿时间（防止访问过快）
FIRST_PAUSE_TIME = 10
# 页面超时时间（毫秒）：60秒
PAGE_TIMEOUT = 60000
# 默认函数执行包路径
DEFAULT_FUNC_PACKAGE = "auto_gen.gen"
# 配置文件节点名称
DEFAULT_COMMON = "Common"
DEFAULT_TEMPLATE = "Template"
DEFAULT_SCH = "School"
class SpiderSch():
"""学校招聘/公告爬虫核心类"""
def __init__(self, _file="99"):
"""
初始化爬虫
:param _file: 配置文件编号，如 setting_sch_10.ini
"""
# 浏览器实例（外部传入）
self.browser = None
# 配置文件编号
self.file = _file
# 配置读取器
self.config = configparser.ConfigParser()
# 按优先级加载配置文件（后面覆盖前面）
self.config.read("data/setting_default.ini", encoding="utf-8") # 默认配置
self.config.read("data/setting_template.ini", encoding="utf-8") # 模板配置
self.config.read(f"data/setting_sch_{_file}.ini", encoding="utf-8") # 学校专属配置
# 标题必须包含的关键词（|分隔）
self.title_includes = self.config.get(DEFAULT_COMMON, "title_include").split("|")
# 标题排除关键词（从黑名单文件读取）
with open("data/black_wx_exclude_title.txt", encoding="utf-8") as f:
_wxlist = f.read().splitlines()
self.title_excludes = list(set(_wxlist)) # 去重
# 爬取进度列表（断点续爬）
self.progress_list = []
if os.path.exists(self.get_progress_file()):
with open(self.get_progress_file(), "r", encoding="utf-8") as f:
for line in f.readlines():
if line.strip():
self.progress_list = [line.strip()]
# 初始化临时文件保存目录
TMP_DIR = self.get_savepath()
ner_logger.info(f"临时目录：{TMP_DIR}")
def get_progress_file(self):
"""获取当前配置对应的进度文件路径"""
return f"data/progress_{self.file}.txt"
def print_all_sch(self):
"""
打印所有已配置的学校名称（用于调试/核对）
遍历0-99号配置文件，汇总所有学校并按出现次数展示
"""
config = configparser.ConfigParser()
config.read("data/setting_default.ini", encoding="utf-8")
printdict = defaultdict(list)
# 加载所有配置文件
for i in range(0, 100):
_confilefile = f"data/setting_sch_{i}.ini"
if os.path.exists(_confilefile):
config.read(_confilefile, encoding="utf-8")
# 解析所有学校节点
for _key in config.options(DEFAULT_SCH):
if _key.startswith("sch_"):
_svalue = config.get(DEFAULT_SCH, _key)
_value = json.loads(_svalue)
for _sch_info in _value:
sch = _sch_info.get("sch_name")
sch_webname = _sch_info.get("sch_webname")
printdict[sch].append(sch_webname)
# 打印只出现一次的学校
for _key in printdict.keys():
if len(printdict[_key]) == 1:
print(f'{_key} : {" ".join(printdict[_key])}')
print("-" * 20)
# 打印出现多次的学校
for _key in printdict.keys():
if len(printdict[_key]) > 1:
print(f'{_key} : {" ".join(printdict[_key])}')
def is_title_include(self, title):
"""
标题过滤：判断是否需要爬取
逻辑：排除黑名单关键词 → 允许爬取
"""
_title = remove_announcement_word(title.strip())
for _blkre in self.title_excludes:
if _blkre.strip() == "":
continue
pattern = re.compile(_blkre)
if re.findall(pattern, _title):
ner_logger.info(f"招聘公告的标题不符合要求，被过滤掉 {title} : {_blkre}")
return False
return True
def get_other_type(self):
"""获取微信等外部页面样式配置"""
return self.config.get(DEFAULT_COMMON, "weixin_style")
def is_external_link(self, domain, _fullurl):
"""
判断是否是外部链接
:return: True=外部链接 False=本站链接
"""
# 同域名 → 内部
if _fullurl.startswith(domain):
return False
# 微信链接 → 视为内部
if 'mp.weixin.qq.com' in _fullurl:
return False
return True
def get_browser_path(self):
"""根据操作系统获取浏览器路径"""
tmp_dir = ""
if os.name == "nt": # Windows
tmp_dir = self.config.get(DEFAULT_COMMON, "browser_path_win")
if os.name == "posix": # Mac/Linux
tmp_dir = self.config.get(DEFAULT_COMMON, "browser_path_mac")
if not os.path.exists(tmp_dir):
os.makedirs(tmp_dir)
return tmp_dir
def get_savepath(self, _tmp="/data/tmp"):
"""获取数据保存根目录"""
if os.name == "nt":
return self.config.get(DEFAULT_COMMON, "savepath_win") + _tmp
if os.name == "posix":
return self.config.get(DEFAULT_COMMON, "savepath_mac") + _tmp
return ""
def get_md_exe(self):
"""获取Markdown转换工具路径（Windows/Mac）"""
if os.name == "nt":
return self.config.get(DEFAULT_COMMON, "md_path_win")
if os.name == "posix":
return self.config.get(DEFAULT_COMMON, "md_path_mac")
return ""
def get_html_md_exe(self):
"""获取HTML转MD工具路径"""
if os.name == "nt":
return self.config.get(DEFAULT_COMMON, "html_md_path_win")
if os.name == "posix":
return self.config.get(DEFAULT_COMMON, "html_md_path_mac")
return ""
def write_process_file(self, line):
"""
写入进度文件（断点续爬）
每次爬完一个学校就写入，下次从这里继续
"""
with open(self.get_progress_file(), "w", encoding="utf-8") as f:
f.write(line)
f.write("\n")
if not line in self.progress_list:
self.progress_list = [line]
def get_progress(self):
"""
获取爬取进度（断点续爬核心）
:return: 待爬取学校列表
"""
_finish = []
_remain = []
_find = False
for _key, _node in self.get_nodes().items():
if _key in self.progress_list:
_finish.append(_key)
_find = True
elif _find:
_remain.append(_key)
else:
_finish.append(_key)
# 输出进度日志
if len(_remain) > 0 and (not 'sch_88888' in _remain or len(_remain) > 1):
_msg = "\n".join(_remain)
ner_logger.info(f'剩余进度 {len(_remain)/len(self.get_nodes()):.0%}，人工启动重新跑\n{_msg}')
return _remain
elif len(_finish) > 0 and (not 'sch_88888' in _finish or len(_finish) > 1):
_msg = "\n".join(_finish)
ner_logger.info(f'重新进度 {len(_finish)/len(self.get_nodes()):.0%}，人工启动重新跑\n{_msg}')
return _finish
return _finish + _remain
def get_nodes(self):
"""
从配置中读取所有学校节点
格式：sch_00001、sch_00002...
"""
nodes = {}
for i in range(1, 100000):
num = str(i).zfill(5) # 转为5位数字
_key = f"sch_{num}"
if self.config.has_option(DEFAULT_SCH, _key):
_svalue = self.config.get(DEFAULT_SCH, _key)
_value = json.loads(_svalue)
# 补充模板字段
self.supplement_node_info(_value)
nodes[_key] = _value
return nodes
def supplement_node_info(self, _node):
"""
从模板配置中补充缺失的字段
如选择器、函数名、点击方式等
"""
for _sch_info in _node:
_template = _sch_info.get("template")
if _template:
_tv = self.config.get(DEFAULT_TEMPLATE, _template)
_tvjson = json.loads(_tv)
for _key, _va in _tvjson.items():
if not _key in _sch_info:
_sch_info[_key] = _va
def get_full_url(self, domain, _link):
"""拼接完整URL（处理相对路径）"""
if _link.startswith("http"):
return _link
if _link.startswith("/"):
return f'{domain}{_link}'
return f'{domain}/{_link}'
def get_key_dir(self, _key):
"""为每个学校创建独立临时目录"""
TMP_DIR = self.get_savepath()
key_tmp_dir = f"{TMP_DIR}/{_key}"
if not os.path.exists(key_tmp_dir):
os.makedirs(key_tmp_dir)
return key_tmp_dir
def get_selector_text(self, page, sch_info, selector1, selector2, style3=""):
"""
智能获取页面选择器（容错机制）
1. 先尝试主选择器 selector1
2. 失败则尝试备选列表 selector2
3. 最后尝试通用样式 style3
"""
html = page.content()
table_selector = sch_info.get(selector1)
style_element = page.query_selector(table_selector)
if not style_element:
table_selectors = sch_info.get(selector2)
if table_selectors:
table_selectors = table_selectors + style3
elif style3:
table_selectors = style3
if table_selectors:
for _selector in table_selectors.split("|"):
if not _selector.strip():
continue
ner_logger.info(f"尝试使用 {_selector}")
style_element = page.query_selector(_selector)
if style_element:
table_selector = _selector
break
if style_element:
ner_logger.info(f"找到元素 {table_selector}")
return True, table_selector
ner_logger.info(f"没有找到元素在页面内容里面:\n{html}")
ner_logger.info(f"没有找到元素 {table_selector}")
return False, table_selector
def open_with_url(self, page, url, refer=""):
"""
安全打开页面
支持：防盗链Refer、状态码判断、加载等待、异常捕获
"""
try:
if refer:
page.set_extra_http_headers({"Referer": refer})
response = page.goto(url, timeout=PAGE_TIMEOUT)
if response:
status = response.status
# 200/412都视为成功（部分学校特殊返回412）
if status in [20, 412]:
try:
page.wait_for_load_state('load', timeout=PAGE_TIMEOUT)
page.wait_for_load_state('networkidle', timeout=120000)
except Exception as e:
ner_logger.info(f"尝试打开url时出错 networkidle: {e}")
time.sleep(3)
return True
else:
ner_logger.debug(f"页面状态码错误 {status}")
elif page.url == url:
ner_logger.debug(f"无返回值，默认打开成功：{url}")
time.sleep(3)
return True
except Exception as e:
ner_logger.debug(f"操作超时 {e}")
ner_logger.debug(f"页面存在问题 {url}")
return False
def pre_page_run(self, page, sch_info):
"""
页面前置任务：进入列表后执行自定义JS操作
如点击加载更多、关闭弹窗等
"""
func_package = DEFAULT_FUNC_PACKAGE
table_func_name = sch_info.get("table_func_name")
if table_func_name:
package_func_name = f"{func_package}.{table_func_name}"
execute_page_action(package_func_name, page)
ner_logger.info(f"执行前置任务 {package_func_name}")
def get_index_list(self, page, sch_info):
"""
获取列表页URL
支持：固定配置 / 动态函数获取
"""
if 'index_url_func' in sch_info:
_ok = self.open_with_url(page, sch_info.get("urls").get('k1'))
if not _ok:
return []
func_name = sch_info.get("index_url_func")
func_package = DEFAULT_FUNC_PACKAGE
package_func_name = f"{func_package}.{func_name}"
urls = execute_index_action(package_func_name, page, sch_info)
ner_logger.info(f"执行获取index_url_func\n{urls}")
return urls
else:
return sch_info.get("urls")
def run(self, page, _key, sch_info, _stat={}):
"""
单个学校爬虫入口
遍历所有列表页 → 爬取数据 → 记录进度
"""
_ret_list = []
sch_name = sch_info.get("sch_name")
sch_webname = sch_info.get("sch_webname")
print(f"爬取学校 {sch_name} - {sch_webname}")
# 获取所有列表URL
urls = self.get_index_list(page, sch_info)
for i, k in enumerate(urls):
url = urls.get(k)
# 预先打开域名首页（降低反爬概率）
pre_open_url = sch_info.get("pre_open_url")
if pre_open_url:
_ok = self.open_with_url(page, pre_open_url)
if not _ok:
ner_logger.info(f"预先打开页面失败 {pre_open_url}")
return False
time.sleep(FIRST_PAUSE_TIME)
else:
domain = get_long_url_domain(url)
if domain:
_ok = self.open_with_url(page, domain[0])
if not _ok:
return False
time.sleep(FIRST_PAUSE_TIME)
# 爬取当前列表页数据
_list = self.get_page_data(page, _key, sch_info, url)
if False in _list:
_ret_list.append(False)
else:
_ret_list.append(True)
time.sleep(FIRST_PAUSE_TIME)
# 写入进度：该学校爬取完成
self.write_process_file(_key)
def get_page_data(self, page, _key, sch_info, url):
"""
爬取列表页数据
打开 → 提取列表 → 调用解析函数 → 遍历详情
"""
_ret_list = []
ner_logger.info(f"开始爬取链接：{_key} / {url}")
# 打开列表页
_ok = self.open_with_url(page, url)
if not _ok:
ner_logger.info(f"页面打开失败，跳过 {url}")
_ret_list.append(False)
return _ret_list
# 获取列表选择器
_ok, table_selector = self.get_selector_text(page, sch_info, "table_selector", "table_selectors")
if not _ok:
ner_logger.error(f"列表页面无匹配元素，需人工处理：{table_selector}")
_ret_list.append(False)
return _ret_list
# 执行前置操作
self.pre_page_run(page, sch_info)
# 定位列表元素
tableObj = page.locator(table_selector)
if tableObj.count() > 1:
# 取第一个可见元素
for i in range(tableObj.count()):
_tableObj = tableObj.nth(i)
if _tableObj.is_visible():
tableObj = _tableObj
break
if tableObj.count() > 1:
tableObj = tableObj.nth(0)
# 保存列表HTML
key_tmp_dir = self.get_key_dir(_key)
_hash = hashlib.md5(url.encode("utf-8")).hexdigest()
tmp_file = os.path.join(key_tmp_dir, f"index_{_hash}.html")
_context_outtext = tableObj.inner_html()
with open(tmp_file, "w", encoding="utf-8") as f:
f.write(f"{_context_outtext}
")
# 调用解析函数生成JSON
func_package = DEFAULT_FUNC_PACKAGE
func_name = sch_info.get("func_name")
package_func_name = f"{func_package}.{func_name}"
tmp_fname = f'{key_tmp_dir}/index_{_hash}.json'
_ok = call_func(package_func_name, _context_outtext, tmp_fname)
if not _ok:
ner_logger.info(f"列表解析失败，跳过 {url}")
_ret_list.append(False)
return _ret_list
time.sleep(FIRST_PAUSE_TIME)
# 遍历每条公告，进入详情页
with open(tmp_fname, "r", encoding="utf-8") as f:
_data = json.load(f)
for _item in _data:
_ok = self.get_page_detail_data(page, _key, url, key_tmp_dir, sch_info, _item)
time.sleep(get_random_number())
_ret_list.append(_ok)
return _ret_list
def get_page_detail_data(self, page, _key, url, key_tmp_dir, sch_info, _item):
"""
爬取详情页数据（核心）
逻辑：
1. 时间过滤（180天内）
2. 标题过滤
3. 智能获取真实链接（点击/相对路径/JS）
4. 去重爬取
5. 保存HTML+JSON
"""
# 时间过滤：只爬取近180天
_publish_time = _item.get("publish_time")
_item_title = _item.get("announcement_name")
if _publish_time and not is_near_month(_publish_time):
ner_logger.info(f"过期跳过 {_publish_time} | {_item_title}")
return True
# 标题黑名单过滤
if not self.is_title_include(_item_title):
return True
# 获取链接
_link = _item.get("link")
_click_text = sch_info.get("click_text")
_click_type = sch_info.get("click_type")
# 处理JS链接
if _link and _link.startswith("javascript:"):
_link = ""
# 处理相对路径 ./
elif _link and _link.startswith("./"):
_url_path = get_directory_from_url(url)
_link = _link.replace("./", _url_path)
# 无链接 → 通过点击标题获取
if not _link or _click_text == 'Y':
_text = _item_title
_hash = hashlib.md5(_text.encode("utf-8")).hexdigest()
tmp_file = os.path.join(key_tmp_dir, f"detail_{_hash}.url")
# 从缓存读取
if os.path.exists(tmp_file):
with open(tmp_file, "r", encoding="utf-8") as f:
_link = f.read()
ner_logger.info(f"从缓存读取链接 {_link}")
# 缓存无则模拟点击
if not _link or _click_text == 'Y':
try:
page.goto(url)
except Exception as e:
return False
time.sleep(get_random_number())
self.pre_page_run(page, sch_info)
# 点击标题获取真实URL
new_url, content = click_by_text_and_get_url(page, url, _text, _click_type)
if new_url:
_link = new_url
with open(tmp_file, "w", encoding="utf-8") as f:
f.write(_link)
ner_logger.info(f"点击获取链接 {_link}")
if not _link:
return False
# 拼接完整URL
domain = sch_info.get("json_domain")
_fullurl = self.get_full_url(domain, _link)
_final_link = get_final_url(_fullurl)
if _fullurl != _final_link:
_fullurl = _final_link
# 文件去重：已存在则跳过
_hash = hashlib.md5(_fullurl.encode("utf-8")).hexdigest()
tmp_file = os.path.join(key_tmp_dir, f"detail_{_hash}.html")
tmp_json_file = os.path.join(key_tmp_dir, f"detail_{_hash}.json")
if os.path.exists(tmp_file) and os.path.exists(tmp_json_file):
print(f"已存在，跳过 {_fullurl}")
return True
# 获取详情内容
_ok, _context_outtext, _context_full_outtext = self.get_page_detail_content(
page, sch_info, domain, _fullurl, refer=url
)
if not _ok or not _context_outtext:
return False
# 最终真实URL
_last_url = page.url
# 内容过短 → 尝试查找微信文章
if len(_context_outtext) < 400 or 'search_wx_file' in sch_info:
_search_wx = sch_info.get("search_wx_file", "N")
_ok, wx_url = find_wx_url(_context_outtext, _search_wx)
if _ok:
_last_url = wx_url
# 保存文件
with open(tmp_file, "w", encoding="utf-8") as f:
f.write(_context_outtext)
with open(tmp_file + ".full", "w", encoding="utf-8") as f:
f.write(_context_full_outtext)
# 保存JSON元数据
_item['full_url'] = _fullurl
_item['last_url'] = _last_url
_item['file_path'] = tmp_file
_item['contact'] = check_contact(_context_outtext)
_item['parent_url'] = url
_item['channel'] = _key
_item['type_url'] = check_url_type(_fullurl, _last_url)
with open(tmp_json_file, "w", encoding="utf-8") as f:
f.write(json.dumps(_item, ensure_ascii=False))
time.sleep(get_random_number())
return True
def get_page_detail_content(self, page, sch_info, domain, _fullurl, refer="", noiframe=True):
"""
获取详情页正文内容
支持：
1. 普通页面选择器
2. iframe嵌套
3. 微信文章
4. BS4解析
5. 302跳转
"""
_context_outtext = ""
_context_full_outtext = ""
# 打开页面
_ok = self.open_with_url(page, _fullurl, refer)
if not _ok:
return False, "", ""
# 微信文章直接返回全文
if is_wechat_url(_fullurl):
return True, page.content(), page.content()
# 使用BS4解析
if sch_info.get('use_bs4'):
html = get_node_text(_fullurl, sch_info['use_bs4'])
return True, html, html
# 处理页面跳转
if 'redirect_url' in sch_info:
_ok, jump_url = get_redirect_url(page)
if _ok and noiframe:
return self.get_page_detail_content(page, sch_info, domain, jump_url, refer, False)
# 清洗全文HTML
_context_full_outtext = clean_html(page.content())
# 获取正文选择器
_style3 = self.get_other_type()
_ok, detail_selector = self.get_selector_text(
page, sch_info, "detail_selector", "detail_selectors", _style3
)
if _ok:
detailObj = page.locator(detail_selector)
if detailObj.count() != 1:
return False, "", _context_full_outtext
html = detailObj.inner_html()
return True, f"{html}
", _context_full_outtext
# 处理iframe
iframe_selector = sch_info.get("detail_iframe", "")
iurls = get_iframe_urls(page, iframe_selector)
if noiframe and iurls:
return self.get_page_detail_content(page, sch_info, domain, iurls[0], refer, False)
# 外部链接直接返回全文
if self.is_external_link(domain, _fullurl):
return True, _context_full_outtext, _context_full_outtext
ner_logger.error(f"详情页无匹配元素，需人工处理：{detail_selector}")
return False, "", _context_full_outtext
```
---
**项目分区导航**：[[03-spider_data.py|spider_data.py]] ⬅️ | 04-spider_sch.py | ➡️ [[00-parsegpt|parsegpt]]