spider_data.py
spider_data.py — 数据处理与上传
核心类 SpiderData，消费爬虫产出的 JSON+HTML 文件，驱动后续处理流水线。
process_announcement_data() 是核心方法，根据 proc_type 参数执行不同操作：
| proc_type | 操作 |
|---|---|
ann |
HTML → Markdown → 大模型解析公告结构化数据 |
wx |
微信文章特殊处理（含图片复杂度检测）后同上 |
cjob |
解析公司职位数据（parse_cjob） |
up_api |
将模型结果上传到云端（学校公告） |
up_api_cjob |
将模型结果上传到云端（公司职位） |
其他功能：
- 黑名单系统：基于 URL、图片 MD5、文本内容 MD5 过滤垃圾数据
- 微信文章图片复杂度检测：GIF、小尺寸无文字图片过多时跳过，避免大模型处理失败
- 手工微信文章支持：
pre_wx_article()将本地 HTML 文件整理成统一数据格式 - 用
.ok/.err/.expired后缀文件做状态标记，避免重复处理
代码
# -*- coding: utf-8 -*-
import os
import glob
import json
import shutil
import time
from utils import ner_logger,is_wechat_url,getMD5Str
from utils_html import get_weixin_info,get_weixin_hand_url,clean_weixin_html
from utils_date import check_file_modification_time,check_file_modification_time_old
from utils_playwright import get_wx_url_content
from parsegpt.ann_md import html2md_with_fix,md_to_html
from parsegpt.ann_model import parse_announcement
from parsegpt.cjob_model import parse_cjob
from api.quanzhi_api import upload_cloud,upload_cloud_job
from spider_sch import DEFAULT_COMMON
# 默认手工维护微信文章的特殊学校代码（固定ID）
DEFAULT_WX_SCHOOL = "sch_88888"
# 默认一次处理的数据条数（防止一次性处理过多）
DEFAULT_PCOUNT = 3
# 数据处理核心类：负责爬虫结果清洗、MD转HTML、大模型解析、上传云端
class SpiderData():
"""数据处理中心：清洗爬虫数据 → 转换格式 → 大模型抽取 → 上传API"""
def __init__(self,_spider_sch):
"""
初始化数据处理器
:param _spider_sch: 爬虫主类实例（传递配置、工具方法）
"""
self.spider_sch = _spider_sch
# ===================== 加载图片/文本黑名单（过滤垃圾内容） =====================
# 图片URL黑名单
with open("data/black_img_urls.txt",encoding="utf-8") as f:
self.black_img_urls_list = f.read().splitlines()
# 图片MD5黑名单
with open("data/black_img_md5.txt",encoding="utf-8") as f:
self.black_img_md5_list = f.read().splitlines()
# 图片内容MD5黑名单
with open("data/black_img_md5_content.txt",encoding="utf-8") as f:
self.black_img_md5_content_list = f.read().splitlines()
# 文本黑名单（暂时空）
self.black_text_list = []
# 加载完整URL黑名单并计算MD5加入黑名单
with open("data/black_img_urls.full.txt",encoding="utf-8") as f:
for line in f.read().splitlines():
_md5 = getMD5Str(line.strip())
self.black_img_urls_list.append(_md5)
# 去重
self.black_img_urls_list = list(set(self.black_img_urls_list))
# 打印黑名单加载数量
ner_logger.info(f"读取黑名单数量,url:{len(self.black_img_urls_list)} ,md5: {len(self.black_img_md5_list)},text : {len(self.black_text_list)}")
# ===================== 黑名单检查工具 =====================
def check_url_in_blacklist(self,_md5):
"""检查图片MD5是否在黑名单中"""
if _md5 in self.black_img_urls_list:
return True
if _md5 in self.black_img_md5_list:
return True
return False
def check_md5_txt_in_blacklist(self,_txt_md5):
"""检查文本内容MD5是否在黑名单"""
if _txt_md5 in self.black_img_md5_content_list:
return True
return False
# ===================== 路径/配置获取工具 =====================
def get_wx_path(self):
"""获取跨平台微信文章本地存放路径"""
if os.name == "nt":
return self.spider_sch.config.get(DEFAULT_COMMON,"wxpath_win")
if os.name == "posix":
return self.spider_sch.config.get(DEFAULT_COMMON,"wxpath_mac")
return ""
def get_wx_url(self,_data):
"""
从数据中智能提取微信文章URL
优先：type_url=wxwz → 其次last_url
"""
if 'type_url' in _data and _data['type_url'] == "wxwz" and is_wechat_url(_data['full_url']):
return _data['full_url']
elif "last_url" in _data and is_wechat_url(_data['last_url']):
return _data['last_url']
else:
return ""
# ===================== 外部命令调用工具 =====================
def proc_wechaturl_md(self,_url,_dir,_outfile):
"""调用外部exe将微信文章转成MD文件"""
_exe = self.spider_sch.get_md_exe()
_cmd = f"{_exe} -image=url --dir {_dir} --output {_outfile} \"{_url}\" "
ner_logger.info(f"开始执行微信文章转换md {_cmd}")
os.system(_cmd)
def proc_html_md(self,_inputfile,_outfile):
"""调用外部exe将HTML转成MD文件"""
_exe = self.spider_sch.get_html_md_exe()
_cmd = f"{_exe} --output-overwrite --plugin-table --exclude-selector=\".ad\" --input \"{_inputfile}\" --output \"{_outfile}\""
ner_logger.info(f"开始执行html转换md {_cmd}")
os.system(_cmd)
# ===================== 核心：处理公告/招聘数据 =====================
def process_announcement_data(self,_key,sch_info,_stat,proc_type = "ann"):
"""
统一数据处理入口
:param _key: 机构ID（sch_xx / com_xx）
:param sch_info: 机构配置信息
:param _stat: 全局状态（计数、去重、配置）
:param proc_type: 处理类型 wx=微信 / ann=公告 / cjob=企业职位 / up_api=上传
"""
_keydir = self.spider_sch.get_key_dir(_key)
_ar_dir = self.spider_sch.get_savepath(f"/data/ardata/{_key}")
_cache_dir = self.spider_sch.get_savepath(f"/data/cache")
# 创建输出目录
if not os.path.exists(_ar_dir):
os.makedirs(_ar_dir)
if not os.path.exists(_cache_dir):
os.makedirs(_cache_dir)
ner_logger.info(f"_keydir:{_keydir},{proc_type}")
# 遍历该机构下所有JSON数据文件
for _file in glob.glob(f"{_keydir}/*.json"):
_filename = os.path.basename(_file)
# 跳过列表页index文件，只处理详情detail_文件
if _filename.startswith("index"):
continue
if not (_filename.startswith("detail_") and _filename.endswith(".json")):
continue
# ===================== 打开JSON & 基础过滤 =====================
with open(_file,"r",encoding="utf-8") as f:
_hfile = _file.replace(".json",".html")
# 公告/职位必须有HTML文件
if proc_type in ["ann","cjob"] and not os.path.exists(_hfile) :
continue
# 标记：本次处理过的文件，避免重复
_tag_info = f"{_filename}_{proc_type}"
# 10分钟内新文件 → 跳过（防止爬虫还在写入）
if check_file_modification_time(_file):
ner_logger.info(f"文件10s内生成，跳过 {_file}")
continue
# 10天前旧文件 → 跳过
if check_file_modification_time_old(_file):
if not _tag_info in _stat['all_proc_list']:
_stat['all_proc_list'].append(_tag_info)
continue
# ===================== 定义各类输出文件路径 =====================
_ar_file = os.path.join(_ar_dir,_filename) # 归档JSON
_md_file = _ar_file.replace(".json",".md") # MD文件
_fix_file = _ar_file.replace(".json",".html") # 清洗后HTML
_model_file = _ar_file.replace(".json",".model.json") # 大模型结果
_up_api_file = _model_file.replace(".json",f".json.{_stat['dist']}.ok") # 上传成功标记
_up_api_err_file = _model_file.replace(".json",f".json.{_stat['dist']}.err") # 上传失败
_expired_file = _ar_file.replace(".json",".json.expired") # 过期不处理
_job_ann_file = _ar_file.replace(".json",".json.job") # 职位标记
# 已过期 → 跳过
if os.path.exists(_expired_file):
continue
# 上传模式必须有模型文件
if proc_type in ["up_api","up_api_cjob"] and not os.path.exists(_model_file):
continue
# 加载原始数据
_data = json.load(f)
_title = _data['announcement_name']
# 标题不符合关键词 → 跳过
if proc_type in ["wx",'ann'] and not self.spider_sch.is_title_include(_title):
ner_logger.info(f"标题在关键词排除之列，{_file} / {_title}不需要爬取，跳过")
continue
# 获取微信URL
wx_url = self.get_wx_url(_data)
# 调试：只处理指定前缀文件
if 'pfile' in _stat and not _filename.startswith(_stat['pfile']):
continue
# 调试：强制重新处理
if 'pfile' in _stat and _filename.startswith(_stat['pfile']):
if os.path.exists(_model_file) and proc_type in ["wx",'ann',"cjob"]:
os.remove(_model_file)
if os.path.exists(_up_api_file) and proc_type in ["up_api","up_api_cjob"]:
os.remove(_up_api_file)
# 已生成模型 → 跳过
if os.path.exists(_model_file) and proc_type in ["wx","ann",'cjob']:
continue
# 已上传 → 跳过
if os.path.exists(_up_api_file) and proc_type in ["up_api","up_api_cjob"]:
continue
# 本次已处理 → 跳过
if _tag_info in _stat['all_proc_list'] and not 'pfile' in _stat:
_stat['all_proc_list'].append(_tag_info)
continue
time.sleep(0.5)
# ===================== 分类型处理 =====================
# 1. 处理微信文章
if (proc_type == "wx" and wx_url) or (proc_type == "ann" and wx_url):
_ok,_wx_file = self.pre_wx_article_html(_data,_cache_dir,wx_url)
if _ok:
_ok,_hfile,_mdfile = self.process_wechat_data(sch_info,_file,_ar_dir,_filename,_data,_cache_dir,wx_url,_wx_file)
if _ok:
_ok = self.process_gonggao_data(sch_info,_filename,_data,_cache_dir,_md_file,_fix_file,_model_file,_hfile,proc_type,_expired_file,_stat)
if _ok == "ok":
_stat['total'] = _stat.get('total',0) +1
_stat['p_count'] +=1
if _stat['total'] >= DEFAULT_PCOUNT:
ner_logger.info(f"共处理{_stat['p_count']}条wx数据完成，退出")
return
# 2. 处理普通公告
elif proc_type == "ann" and not wx_url:
_ok = self.process_gonggao_data(sch_info,_filename,_data,_cache_dir,_md_file,_fix_file,_model_file,_hfile,proc_type,_expired_file,_stat)
if _ok == "ok":
_stat['total'] = _stat.get('total',0)+1
_stat['p_count'] +=1
if _stat['total'] >= DEFAULT_PCOUNT:
ner_logger.info(f"共处理{_stat['p_count']}条ann数据完成，退出")
return
# 3. 处理企业职位
elif proc_type == "cjob":
ner_logger.info(f"开始处理到公司职位文件:{_data['announcement_name']} - {_hfile}")
_ok,_msg = parse_cjob(self,_model_file,_data,sch_info,_expired_file,_hfile,_stat)
if _ok == "ok":
_stat['total'] = _stat.get('total',0)+1
_stat['p_count'] +=1
if _stat['total'] >= DEFAULT_PCOUNT:
ner_logger.info(f"共处理{_stat['p_count']}条cjob数据完成，退出")
return
# 4. 上传公告到API
elif proc_type == "up_api" and os.path.exists(_model_file) and not os.path.exists(_job_ann_file):
if os.path.exists(_up_api_file):
continue
ner_logger.info(f"开始处理数据上传云端 - {_hfile} \n {_model_file}")
_ok,_msg,_code= upload_cloud(_model_file,_stat['dist'])
if _ok and _code == "200":
with open(_up_api_file,"w",encoding="utf-8") as f:
f.write(_msg)
time.sleep(2)
elif _ok and _code != "200":
with open(_expired_file,"w",encoding="utf-8") as f:
f.write(_msg)
else:
with open(_up_api_err_file,"w",encoding="utf-8") as f:
f.write(_msg)
# 5. 上传职位到API
elif (proc_type == "up_api_cjob" or (proc_type == "up_api" and os.path.exists(_job_ann_file))) and os.path.exists(_model_file):
if os.path.exists(_up_api_file):
continue
ner_logger.info(f"开始处理职位数据上传云端 - {_hfile} \n {_model_file}")
_ok,_msg,_code= upload_cloud_job(_model_file,_stat['dist'],_stat['retry'])
if _ok:
with open(_up_api_file,"w",encoding="utf-8") as f:
f.write(_msg)
time.sleep(2)
else:
with open(_up_api_err_file,"w",encoding="utf-8") as f:
f.write(_msg)
# ===================== 公告处理：HTML → MD → 大模型抽取 =====================
def process_gonggao_data(self,sch_info,_filename,_data,_cache_dir,_md_file,_fix_file,_model_file,_hfile,proc_type,_expired_file,_stat):
"""处理学校/企业公告：清洗HTML → 转MD → 大模型抽取结构化数据"""
ner_logger.info(f"分析文件{_data['announcement_name']} - {_md_file}")
with open(_hfile,"r",encoding="utf-8") as f:
_html = f.read()
# HTML 转 MD + 清洗
_ok,_info,_full_text = html2md_with_fix(self,_data,_fix_file,_md_file,_html,sch_info,_cache_dir,_hfile)
if _ok and len(_info) > 2 and 'props' in _info:
# 微信文章额外检查图片复杂度
if proc_type == "wx":
_ok = self.check_wx_file(_data,_hfile,_model_file)
if not _ok:
with open(_expired_file,'a',encoding='utf-8') as fw:
fw.write(f"不能处理这个微信公告(283)\n")
return ""
# 大模型抽取结构化字段
(_ok,msg) = parse_announcement(_md_file,_fix_file,_model_file,_info,_full_text,proc_type,sch_info,_expired_file,_stat)
if _ok:
ner_logger.info(f"大模型处理文件成功:{_data['announcement_name']} - {_md_file}")
return "ok"
else:
with open(_expired_file,'a',encoding='utf-8') as fw:
fw.write(f"{msg}\n")
else:
with open(_expired_file,'a',encoding='utf-8') as fw:
fw.write(f"不能处理这个微信公告(299） html2md_with_fix \n")
return ""
# ===================== 微信文章专用处理 =====================
def process_wechat_data(self,sch_info,_file,_ar_dir,_filename,_data,_cache_dir,wx_url,_wx_file):
"""微信文章专用流程：转MD → 转回HTML → 供后续清洗"""
ner_logger.info(f"处理微信数据{_data['announcement_name']} - {wx_url} \n {_file}")
_md_filename = _filename.replace(".json",".md")
_md_cache_dir = f'{_cache_dir}_md'
if not os.path.exists(_md_cache_dir):
os.makedirs(_md_cache_dir)
_md_file = os.path.join(_md_cache_dir,_md_filename)
# 重新生成
if os.path.exists(_md_file):
os.remove(_md_file)
# 调用工具转MD
self.proc_wechaturl_md(_wx_file,_md_cache_dir,_md_filename)
if not os.path.exists(_md_file):
return "","",""
# MD内容过短 → 无效
with open(_md_file,"r",encoding="utf-8") as f:
_md_text = f.read()
if len(_md_text) < 50:
return "","",""
# MD 转 HTML
_html_file = _md_file.replace(".md",".html")
_ok = md_to_html(_md_file,_html_file)
if _ok :
return "ok",_html_file,_md_file
return "","",""
def pre_wx_article_html(self,_data,_cache_dir,wx_url):
"""预先下载微信文章到本地，清洗HTML，提取信息"""
_wx_cache_dir = f'{_cache_dir}_wx'
if not os.path.exists(_wx_cache_dir):
os.makedirs(_wx_cache_dir)
_md5_url = getMD5Str(wx_url)
_wx_file = os.path.join(_wx_cache_dir,f"{_md5_url}.html")
_wx_file_0 = os.path.join(_wx_cache_dir,f"{_md5_url}.html.0")
_wx_file_config = os.path.join(_wx_cache_dir,f"{_md5_url}.html.config")
# 未缓存 → 下载
if not os.path.exists(_wx_file) or not os.path.exists(_wx_file_config):
executable_path = self.spider_sch.get_browser_path()
_ok,_content,image_lists = get_wx_url_content(executable_path,wx_url)
if _ok:
with open(_wx_file,"w",encoding="utf-8") as f:
f.write(_content)
with open(_wx_file_config,"w",encoding="utf-8") as f:
json.dump(image_lists,f,ensure_ascii=False,indent=4)
else:
return False,""
# 备份原始文件
if not os.path.exists(_wx_file_0):
shutil.copy(_wx_file,_wx_file_0)
# 提取微信文章信息（作者、时间等）
get_weixin_info(_wx_file,_data)
# 清洗微信HTML垃圾
clean_weixin_html(_wx_file,_wx_file_0)
return True,_wx_file
# ===================== 手工微信文章导入 =====================
def pre_wx_article(self):
"""导入本地手工整理好的微信HTML文章，生成标准JSON结构"""
_wx_path = self.get_wx_path()
ner_logger.info(f"wx文章路径{_wx_path}")
# 固定导入到默认微信机构ID
_keydir = self.spider_sch.get_key_dir(DEFAULT_WX_SCHOOL)
if not os.path.exists(_keydir):
os.makedirs(_keydir)
# 遍历所有HTML
for _file in glob.glob(f"{_wx_path}/**/*.html",recursive=True):
_file_err_file = _file.replace(".html",".html.err")
_filename = os.path.basename(_file)
# 提取标题、链接、日期
title,link, wxdate = get_weixin_hand_url(_file)
_file_json = f'detail_{getMD5Str(link)}.json'
_file_file = f'{_keydir}/{_file_json}'
# 标记不可处理 → 删除并跳过
if os.path.exists(_file_err_file):
if os.path.exists(_file_file):
os.remove(_file_file)
continue
# 已处理 → 跳过
if os.path.exists(_file_file):
continue
# 构造标准数据结构
_data = {
"announcement_name":title,
"publish_time":wxdate,
"link":link,
"full_url":link,
"last_url":"",
"parent_url":"http://mp.weixin.qq.com",
"upload": "",
"contact": "无",
"type_url":"wxwz",
"channel": DEFAULT_WX_SCHOOL,
"wx_name":"",
"wx_title":title,
"wx_public_time":wxdate,
"wx_source_file":_file,
}
# 写入标准JSON
with open(_file_file,"w",encoding="utf-8") as f:
f.write(json.dumps(_data,ensure_ascii=False,indent=4))
# ===================== 微信文章质量检查（过滤复杂图片） =====================
def check_wx_file(self,_data,_hfile,_model_file):
"""
检查微信文章是否适合自动处理
规则：小图/动态图过多 → 判定复杂，不处理
"""
count = 0
_props = _data['props']
img_urls = _props.get('img_urls',{})
wx_source_file = _hfile
if 'wx_source_file' in _data:
wx_source_file = _data['wx_source_file']
_ok_file = wx_source_file.replace(".html",".html.ok")
_err_file = wx_source_file.replace(".html",".html.err")
# 图片复杂度评分
for _url,_dict in img_urls.items():
img_ocr = _dict['img_ocr']
img_width = _dict['img_width']
img_height = _dict['img_height']
if 'full_qr' in _dict and _dict['full_qr'] == 'Y':
continue
# 极小图
if img_width * img_height < 100 * 100 and len(img_ocr) < 5:
count += 2
# 中小图
elif img_width * img_height < 500 * 500 and len(img_ocr) < 5:
count += 1
# GIF动图
elif _url.endswith(".gif"):
count += 2
# 分数过高 → 不处理
if count > 10:
with open(_err_file,"w",encoding="utf-8") as f:
f.write("不好处理")
return False
# 标记可处理
with open(_ok_file,"w",encoding="utf-8") as f:
f.write(f"可以处理{_model_file}")
return True
项目分区导航：spider_com.py ⬅️ | 03-spider_data.py | ➡️ spider_sch.py
💬 评论