---
title: "02-spider_com.py"
created: 2026-04-03
tags:
- 项目
aliases:
- spider_com.py
---
# spider_com.py
### `spider_com.py` — 大公司官网爬虫
核心类 `SpiderCom`，结构与 `SpiderSch` 高度相似，但针对企业招聘页有额外扩展：
- 支持**无限滚动页面**：通过 `scrollHeight` 判断是否到底，自动滚动加载
- 支持**"加载更多"按钮**：可通过元素点击或自定义函数两种方式触发
- 支持**分页翻页**：`page_count` 和 `page_func_name` 控制自动翻页
- 支持三种特殊数据获取模式：
- `api`：调用百度数据处理 API（`auto_api_proc`）
- `on_response`：监听网络响应（`on_response_proc`）
- 默认：常规 Playwright DOM 爬取
- 详情页使用**新 Tab 开启**（`browser.new_page()`），而学校爬虫复用同一 Tab
- 对已爬取文件会更新修改时间戳，用于判断数据新鲜度
## 代码
```python
# -*- coding: utf-8 -*-
"""
企业招聘爬虫核心类：SpiderCom
功能：自动化爬取企业招聘/职位信息
支持：断点续爬、动态滚动加载、点击加载更多、分页翻页、API接口爬取、请求监听爬取
特点：智能获取真实链接、文件去重、异常状态码处理、跨平台运行、防反爬机制
"""
import os, hashlib, json, time
import configparser
import glob
import requests
import re
from collections import defaultdict
# 企业专用函数调用模块
from auto_gen_com.func_call import call_func, execute_page_action
# 工具类：日志、URL解析、时间判断、文本处理、随机数
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
from utils_resume import remove_announcement_word
# Playwright工具：点击获取链接、iframe提取、页面跳转识别
from utils_playwright import click_by_text_and_get_url, get_iframe_urls, get_redirect_url
# 扩展爬取方式：API专用爬取、网络请求监听爬取
from auto_api.baidu_data_proc_api import api_proc as auto_api_proc
from auto_on_response.main_proc import on_response_proc
# ===================== 全局常量配置 =====================
# 企业爬虫进度文件（断点续爬）
TMP_PROGRESS_FILE = "data/progress_com.txt"
# 首页加载停顿时间（防反爬）
FIRST_PAUSE_TIME = 10
# 页面超时时间（毫秒）：60秒
PAGE_TIMEOUT = 60000
# 企业爬虫默认函数包路径
DEFAULT_FUNC_PACKAGE = "auto_gen_com.gen"
# 配置文件节点名称
DEFAULT_COMMON = "Common"
DEFAULT_TEMPLATE = "Template"
DEFAULT_COM = "Company"
class SpiderCom():
"""企业职位/招聘信息爬虫核心类"""
def __init__(self, _file="99"):
"""
初始化企业爬虫
:param _file: 配置文件编号，对应 setting_com_xx.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_com_{_file}.ini", encoding="utf-8")
# 爬取进度列表（断点续爬）
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_com_{self.file}.txt"
def print_all_com(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_com_{i}.ini"
if os.path.exists(_confilefile):
config.read(_confilefile, encoding="utf-8")
# 解析企业节点
for _key in config.options(DEFAULT_COM):
if _key.startswith("com_"):
_svalue = config.get(DEFAULT_COM, _key)
_value = json.loads(_svalue)
for _com_info in _value:
com = _com_info.get("com_name")
com_webname = _com_info.get("com_webname")
printdict[com].append(com_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 get_browser_path(self):
"""根据操作系统（Windows/Mac）获取浏览器路径"""
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 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:
_msg = "\n".join(_remain)
ner_logger.info(f'剩余进度 {len(_remain) / len(self.get_nodes()):.0%}，继续爬取\n{_msg}')
return _remain
elif len(_finish) > 0:
_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):
"""
从配置中读取所有企业节点
格式：com_00001、com_00002...
"""
nodes = {}
for i in range(1, 100000):
num = str(i).zfill(5) # 转为5位数字
_key = f"com_{num}"
if self.config.has_option(DEFAULT_COM, _key):
_svalue = self.config.get(DEFAULT_COM, _key)
_value = json.loads(_svalue)
# 从模板补充配置字段
self.supplement_node_info(_value)
nodes[_key] = _value
return nodes
def get_other_type(self):
"""获取微信等外部页面样式配置"""
return self.config.get(DEFAULT_COMMON, "weixin_style")
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 supplement_node_info(self, _node):
"""
从模板配置中补充缺失字段
如选择器、函数名、点击方式、翻页配置等
"""
for _com_info in _node:
_template = _com_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 _com_info:
_com_info[_key] = _va
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. 支持正则匹配class
"""
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
# 正则匹配class（兜底方案）
selector1_re = f"{selector1}_re"
ner_logger.info(f"尝试使用正则 {sch_info.get(selector1_re)}")
if sch_info.get(selector1_re):
regex_pattern = sch_info.get(selector1_re)
div_elements = page.query_selector_all('div')
for element in div_elements:
class_name = element.get_attribute('class') or ''
if re.search(regex_pattern, class_name):
return True, f"div.{class_name}"
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都视为成功
if status in [200, 412]:
page.wait_for_load_state('load', timeout=PAGE_TIMEOUT)
try:
page.wait_for_load_state('networkidle', timeout=30000)
except Exception as e:
ner_logger.info(f"等待网络空闲出错: {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, func_name="table_func_name"):
"""执行页面前置任务（如关闭弹窗、点击加载）"""
func_package = DEFAULT_FUNC_PACKAGE
table_func_name = sch_info.get(func_name)
if table_func_name:
package_func_name = f"{func_package}.{table_func_name}"
ner_logger.info(f"执行前置任务 {package_func_name}")
return execute_page_action(package_func_name, page)
return False
def api_proc(self, page, _key, com_info, _stat):
"""API专用爬取模式（直接调用接口）"""
sch_name = com_info.get("com_name")
sch_webname = com_info.get("com_webname")
print(f"API模式爬取：{sch_name} - {sch_webname}")
# 预先打开页面
pre_open_url = com_info.get("pre_open_url")
self.open_with_url(page, pre_open_url)
# 遍历URL执行API爬取
urls = com_info.get("urls")
for i, k in enumerate(urls):
url = urls.get(k)
auto_api_proc(self, _key, com_info, k, url, _stat)
time.sleep(FIRST_PAUSE_TIME * 30)
self.write_process_file(_key)
def on_resp_proc(self, page, _key, com_info, _stat):
"""请求监听爬取模式（监听网络请求获取数据）"""
sch_name = com_info.get("com_name")
sch_webname = com_info.get("com_webname")
print(f"请求监听模式爬取：{sch_name} - {sch_webname}")
pre_open_url = com_info.get("pre_open_url")
self.open_with_url(page, pre_open_url)
urls = com_info.get("urls")
for i, k in enumerate(urls):
url = urls.get(k)
on_response_proc(self, page, _key, com_info, k, url, _stat)
time.sleep(FIRST_PAUSE_TIME * 30)
self.write_process_file(_key)
def run(self, page, _key, com_info, _stat):
"""
单个企业爬虫主入口
支持：普通爬取、API爬取、请求监听爬取
"""
sch_name = com_info.get("com_name")
sch_webname = com_info.get("com_webname")
data_proc_type = com_info.get("data_proc_type", "")
# 特殊爬取模式：API / 请求监听
if data_proc_type == "api":
return self.api_proc(page, _key, com_info, _stat)
elif data_proc_type == "on_response":
return self.on_resp_proc(page, _key, com_info, _stat)
# 普通爬取模式
print(f"普通模式爬取企业：{sch_name} - {sch_webname}")
urls = com_info.get("urls")
for i, k in enumerate(urls):
url = urls.get(k)
# 预先打开首页（降低反爬）
pre_open_url = com_info.get("pre_open_url")
if pre_open_url:
self.open_with_url(page, pre_open_url)
time.sleep(FIRST_PAUSE_TIME)
# 打开列表页
self.open_with_url(page, url)
# 爬取列表数据
self.get_page_data(page, _key, com_info, url, k)
# ===================== 分页翻页逻辑 =====================
_page_count = 0
_page_start = 2
# 全量爬取：1000页
if "page_func_name" in com_info and 'method' in _stat and _stat['method'] == "cp_full":
_page_count = 1000
# 普通爬取：3页
elif "page_count" in com_info and com_info['page_count'] == 'Y':
_page_count = 3
# 命令行指定起始页
if "page_start" in _stat:
_page_start = _stat['page_start']
# 执行翻页
if _page_count > 1:
func_name = com_info.get("page_func_name")
for i in range(2, _page_count):
time.sleep(2)
_ok = self.pre_page_run(page, com_info, "page_func_name")
# 跳过起始页之前的页面
if _ok and i < _page_start:
ner_logger.info(f"分页 {i} 跳过")
time.sleep(3)
elif _ok:
ner_logger.info(f"分页 {i} 执行成功")
time.sleep(FIRST_PAUSE_TIME)
_purl = url + f"&p={i}"
self.get_page_data(page, _key, com_info, _purl, k)
else:
ner_logger.info(f"分页 {i} 执行失败")
break
time.sleep(FIRST_PAUSE_TIME)
# 写入进度
self.write_process_file(_key)
def get_page_data(self, page, _key, sch_info, url, k):
"""
爬取列表页数据
流程：打开 → 滚动加载 → 点击加载更多 → 解析 → 保存 → 遍历详情
"""
ner_logger.info(f"开始爬取列表：{_key} / {url}")
time.sleep(5)
# 获取列表选择器
_ok, table_selector = self.get_selector_text(page, sch_info, "table_selector", "table_selectors")
if not _ok:
ner_logger.error(f"列表无匹配元素，需人工处理：{table_selector}")
return [False]
# ===================== 智能滚动加载（动态页面必备） =====================
last_height = page.evaluate("document.body.scrollHeight")
scroll_count = 0
max_scrolls = 9999
while scroll_count < max_scrolls:
page.evaluate("window.scrollTo(0, document.body.scrollHeight)")
time.sleep(2)
new_height = page.evaluate("document.body.scrollHeight")
if new_height == last_height:
ner_logger.info(f"滚动完成，共{scroll_count + 1}次")
break
last_height = new_height
scroll_count += 1
ner_logger.info(f"第{scroll_count}次滚动")
# ===================== 点击加载更多 =====================
def click_page_run(p, si, fn="click_load_more_func_name"):
fp = DEFAULT_FUNC_PACKAGE
f_name = si.get(fn)
if f_name:
return execute_page_action(f"{fp}.{f_name}", p)
return False
# 配置开启才执行
click_load_more = sch_info.get("click_load_more", "")
if click_load_more and click_load_more.upper() == "Y":
max_clicks = int(sch_info.get("max_load_more_clicks", 1))
click_count = 0
method = sch_info.get("load_more_method", "function")
if method == "element":
selector = sch_info.get("load_more_selector", "")
while click_count < max_clicks:
try:
btn = page.query_selector(selector)
if btn and btn.is_visible():
btn.click()
click_count += 1
time.sleep(5)
else:
break
except:
break
else:
while click_count < max_clicks:
if click_page_run(page, sch_info):
click_count += 1
time.sleep(5)
else:
break
ner_logger.info(f"共点击加载更多 {click_count} 次")
# 提取列表HTML
tableObj = page.locator(table_selector)
html_content = tableObj.inner_html()
# 保存列表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")
with open(tmp_file, "w", encoding="utf-8") as f:
f.write(f"{html_content}
")
# 调用解析函数生成JSON
func_name = sch_info.get("func_name")
pkg_func = f"{DEFAULT_FUNC_PACKAGE}.{func_name}"
tmp_json = f"{key_tmp_dir}/index_{_hash}.json"
call_func(pkg_func, html_content, tmp_json)
time.sleep(FIRST_PAUSE_TIME)
# 遍历详情页
with open(tmp_json, "r", encoding="utf-8") as f:
data = json.load(f)
for item in data:
self.get_page_detail_data(page, _key, url, k, key_tmp_dir, sch_info, item)
time.sleep(get_random_number())
return [True]
def get_page_detail_data(self, page, _key, url, _k, key_tmp_dir, sch_info, _item):
"""爬取详情页（核心）：智能获取链接、去重、保存HTML+JSON"""
# 获取职位信息
_link = _item.get("link")
area = _item.get("hd_loc")
_click_text = sch_info.get("click_text")
_click_type = sch_info.get("click_type")
_max_parent_level = sch_info.get("max_parent_level")
# 处理JS链接
if _link and _link.startswith("javascript:"):
_link = ""
# 无链接 → 通过点击标题获取
if not _link or _click_text == 'Y':
_text = _item.get("announcement_name")
hd_loc = _item.get("hd_loc", "")
combined_text = _text + hd_loc
_hash = hashlib.md5(combined_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") as f:
_link = f.read()
# 点击获取
if not _link:
new_url, _ = click_by_text_and_get_url(
page, url, _text, _click_type, area, _max_parent_level, url
)
if new_url:
_link = new_url
with open(tmp_file, "w") as f:
f.write(_link)
if not _link:
return False
# 拼接真实URL
domain = sch_info.get("json_domain")
_fullurl = self.get_full_url(domain, _link)
# 前端路由#不处理跳转
if "#" not in _fullurl:
_final_link = get_final_url(_fullurl)
if _fullurl != _final_link:
_fullurl = _final_link
# 文件去重
_hash = hashlib.md5(_fullurl.encode()).hexdigest()
tmp_file = os.path.join(key_tmp_dir, f"detail_{_hash}.html")
tmp_json = os.path.join(key_tmp_dir, f"detail_{_hash}.json")
# 已存在 → 仅更新时间
if os.path.exists(tmp_file) and os.path.exists(tmp_json):
t = time.time()
os.utime(tmp_file, (t, t))
os.utime(tmp_json, (t, t))
return True
# 获取详情内容
_ok, content = self.get_page_detail_content(page, sch_info, domain, _fullurl)
if not _ok:
return False
# 保存文件
with open(tmp_file, "w", encoding="utf-8") as f:
f.write(content)
# 保存JSON元数据
_item['full_url'] = _fullurl
_item['last_url'] = page.url
_item['file_path'] = tmp_file
_item['parent_url'] = url
_item['channel'] = _key
_item['job_type'] = _k.split("_")[0]
with open(tmp_json, "w", encoding="utf-8") as f:
json.dump(_item, f, ensure_ascii=False)
time.sleep(get_random_number())
return True
def get_page_detail_content(self, page, sch_info, domain, _fullurl, _redirect=True):
"""
获取详情页正文
新开标签页 → 处理弹窗 → 提取内容 → 关闭页面
"""
# 新开页面爬取详情（不污染列表页）
new_page = self.browser.new_page()
response = new_page.goto(_fullurl, wait_until="networkidle", timeout=3200000)
new_page.wait_for_timeout(1500)
# 自动关闭常见弹窗
for btn in ["Accept", "同意", "Continue", "OK"]:
try:
new_page.locator(f"button:has-text('{btn}')").click(timeout=1500)
except:
pass
time.sleep(get_random_number() * 2)
# 状态码判断
if not response:
new_page.close()
return False, ""
status = response.status
if status != 200:
ner_logger.info(f"页面异常 {status}：{_fullurl}")
new_page.close()
return False, ""
# 处理页面跳转
if 'redirect_url' in sch_info and _redirect:
_ok, jump_url = get_redirect_url(new_page)
if _ok:
return self.get_page_detail_content(new_page, sch_info, domain, jump_url, False)
# 获取正文
full_html = new_page.content()
_ok, selector = self.get_selector_text(new_page, sch_info, "detail_selector", "detail_selectors")
if _ok:
obj = new_page.locator(selector)
if obj.count() == 1:
content = f"{obj.inner_html()}
"
new_page.close()
return True, content
new_page.close()
return True, full_html
```
---
**项目分区导航**：[[01-main.py|main.py]] ⬅️ | 02-spider_com.py | ➡️ [[03-spider_data.py|spider_data.py]]