--- title: "02-ann_md" created: 2026-04-02 tags: - 项目 aliases: - ann_md --- # ann_md.py ### `ann_md.py` — HTML 预处理与 Markdown 转换 解析流水线的第一道关卡：清洗噪音 DOM → 修复链接/图片路径 → 硬编码字段提取 → 表格转 Markdown → 渠道专项修复。六件事详解见 [[00-parsegpt|parsegpt 主篇]]。 ## 代码 ```python # -*- coding: utf-8 -*- """ HTML → Markdown 转换核心模块 功能：清洗HTML → 修复链接/图片 → 表格处理 → 转MD → 格式优化 → 提取纯文本 适用：爬虫页面结构化转换、招聘/学校公告内容标准化处理 """ import html2text from bs4 import BeautifulSoup import re import sys sys.path.append('../') import markdown import os import base64 # 图片处理工具：替换图片URL为云存储地址 from parsegpt.ann_img import replace_img_urls_with_base64 # 工具类：日志、MD5、去除括号字符 from utils import ner_logger, getMD5Str, remove_brackets # HTML/MD修复工具：黑名单清洗、格式修复、URL路径处理 from utils_html import fix_html_blacklist, fix_md, get_directory_from_url, get_directory_url # 华为云OBS上传接口 from api.hwcloud_api import upload_file_to_obs # 图片工具：保存Base64图片到本地 from utils_img import save_base64_img def md_to_html(md_file_path, html_file_path): """ Markdown文件 转换为 HTML文件 :param md_file_path: 输入MD文件路径 :param html_file_path: 输出HTML文件路径 :return: 转换成功/失败 """ try: with open(md_file_path, 'r', encoding='utf-8') as md_file: md_content = md_file.read() # markdown库转HTML html_content = markdown.markdown(md_content) with open(html_file_path, 'w', encoding='utf-8') as html_file: html_file.write(html_content) return True except FileNotFoundError: print(f"文件 {md_file_path} 未找到，请检查文件路径是否正确。") except Exception as e: print(f"转换过程中出现错误: {e}") return False def html2md_with_fix(spider_data, _data, _fix_file, _md_file, _html, sch_info, _cache_dir, _hfile): """ 【核心入口】HTML内容 → 清洗修复 → 转MD → 保存文件 完整流程：修复HTML → 表格处理 → 转MD → 格式优化 → 提取全文本 :return: 处理状态、配置数据、纯文本内容 """ _props = {} # 存储图片/链接等扩展属性 # 1. 清洗修复HTML（垃圾标签、链接、图片、黑名单过滤） _ok, _fix_html = fix_html(spider_data, _data, _html, sch_info, _props, _cache_dir, _hfile) if not _ok: return False, "", "" # 把图片/链接属性存入配置 _data['props'] = _props # 保存修复后的干净HTML with open(_fix_file, "w", encoding="utf-8") as f: f.write(_fix_html) # 2. 处理表格并转换为MD文本 _ok, md_text = html2md_table(spider_data, _fix_html, _fix_file, _data) if not _ok: return False, "", "" # 3. 修复MD格式 md_text = fix_md(md_text, _props) # 4. 特定学校（98534）专属格式优化 md_text = common_process_sch_98534(_data, md_text) # 保存最终MD文件 with open(_md_file, "w", encoding="utf-8") as f: f.write(md_text) # 5. 从修复后的HTML提取纯文本 _full_text = get_html_content(_fix_html, _props) return True, _data, _full_text def html2md_table(spider_data, _fix_html, _fix_file, _data): """ 智能表格处理 + HTML转MD 规则：无合并单元格 → 调用外部工具；有合并单元格 → 直接用html2text """ # 默认无表格 _data["has_table"] = "" soup = BeautifulSoup(_fix_html, 'html.parser') tables = soup.find_all('table') # 检测是否有合并单元格 / 空单元格 _has_colspan = False _has_table = "" if tables: for table in tables: rows = table.find_all('tr') # 行数>3 标记有表格 if len(rows) > 3: _has_table = "OK" # 检查每一格 for row in rows: cells = row.find_all(['td', 'th']) for cell in cells: # 有合并行/列 if 'rowspan' in cell.attrs or 'colspan' in cell.attrs: _has_colspan = True break # 单元格内容为空 if not cell.text.strip(): _has_colspan = True break if _has_colspan: break if _has_colspan: break # 无复杂表格 → 调用外部proc_html_md处理 if not _has_colspan: _htmlfile = _fix_file _htmlfile_md = f"{_htmlfile}.md" spider_data.proc_html_md(_htmlfile, _htmlfile_md) if os.path.exists(_htmlfile_md): with open(_htmlfile_md, "r", encoding="utf-8") as f: _full_text = f.read() # 清理临时文件 os.remove(_htmlfile_md) return True, _full_text else: # 有复杂表格，标记 _data["colspan"] = "OK" # 有表格，直接使用html2text转换 _data["has_table"] = _has_table return True, html2md(_fix_html) def common_process_sch_98534(_data, md_text): """ 【学校专属】sch_98534 招聘页面MD格式美化 功能：把岗位信息自动加粗、格式化、规整换行 """ if _data['channel'] in ['sch_98534']: try: # 匹配 ### 岗位信息 到 ### 岗位要求 之间的内容 block_re = re.compile(r'(### 岗位信息.*?)(?=### 岗位要求)', re.S) def _repl(m: re.Match) -> str: s = m.group(1) # 加粗字段名 s = re.sub(r'\n-\s+', '\n\n- **', s) s = re.sub(r'：\n\n', '：**\n\n', s) # 格式规整 s = re.sub(r'\*{3,}', '**', s) s = re.sub(r'\n{3,}', '\n\n', s) s = re.sub(r'(?m)^-\s***', '**', s) s = re.sub(r'(?<=**)\n\n([^**\n-].*)', r'\1', s) s = re.sub(r'**\n**', '**\n\n**', s) return s md_text = block_re.sub(_repl, md_text) ner_logger.info(f"针对mdfile的优化，指定学校98534{md_text}") except Exception as e: ner_logger.error(f"针对mdfile的优化，指定学校98534 Error:\n{e}") return md_text def html2md(htmltext): """ 基础HTML转MD（使用html2text库） 关闭软换行，避免格式错乱 """ text_maker = html2text.HTML2Text() text_maker.soft_break = '' # 关闭自动软换行 result = text_maker.handle(htmltext) result = re.sub(r'-\n', '-', result) # 修复列表换行错乱 return result def fix_html(spider_data, _data, htmltext, sch_info, _props, _cache_dir, _hfile): """ HTML清洗修复总入口 流程：解析 → 提取硬编码字段 → 清理垃圾 → 修复链接 → 修复图片 → 黑名单 → 图片上传云存储 """ _is_external_link = "" soup = BeautifulSoup(htmltext, 'html.parser') # 如果页面包含body，视为外部完整页面，不做深度处理 if soup.find('body'): _data['is_external_link'] = "OK" # 1. 提取配置中写死的标签内容（公司名、公告名等） find_hardcode_tag(soup, sch_info, _data, _hfile) # 2. 清理HTML文字垃圾（黑名单关键词） fix_html_blacklist(spider_data, soup, sch_info) # 3. 移除无用DIV/标签 fix_html_div(spider_data, soup, sch_info, _data) # 4. 修复a链接相对路径为绝对路径 fix_html_a(spider_data, soup, sch_info, _props, _data) # 5. 修复img图片src相对路径/base64 fix_html_img_src(spider_data, soup, sch_info, _data, _cache_dir) # 6. 移除黑名单图片 rm_html_blacklist(spider_data, soup, sch_info) # 7. 图片下载 → 检测 → 上传OBS → 替换src _ok = replace_img_urls_with_base64(spider_data, _data, soup, _props, _cache_dir) return _ok, soup.prettify() def find_hardcode_tag(soup, sch_info, _data, _hfile): """ 从配置中读取固定标签选择器，提取关键信息： 公司名、公告名、 tuning文本 """ _hardcode_tag = [] if 'detail_hd_company' in sch_info: _hardcode_tag = sch_info['detail_hd_company'].split('|') # 提取公司名称 for _tagstr in _hardcode_tag: _tag = _tagstr.split("#") company_tag = soup.find(_tag[0], class_=_tag[1]) if company_tag and len(company_tag.text.strip()) > 2: _data['hd_company'] = remove_brackets(company_tag.text.strip()) # 从完整页面文件提取公告、公司、优化文本 fix_full_html_extract(sch_info, _data, _hfile, 'detail_hd_ann_full', 'hd_ann') fix_full_html_extract(sch_info, _data, _hfile, 'detail_hd_company_full', 'hd_company') fix_full_html_extract(sch_info, _data, _hfile, 'detail_tuning_classes_full', 'tuning_content') def fix_full_html_extract(sch_info, _data, _hfile, _tag, _tk): """ 从完整页面.html.full文件中提取指定标签内容（用于无法从正文提取的场景） """ if _tag in sch_info: _full_file = f'{_hfile}.full' ner_logger.info(f"{_tag} {_full_file}") if os.path.exists(_full_file): with open(_full_file, "r", encoding="utf-8") as f: _full_text = f.read() soup_full = BeautifulSoup(_full_text, 'html.parser') _hardcode_ann = sch_info[f'{_tag}'].split('|') for _tagstr in _hardcode_ann: _tag = _tagstr.split("!") if "#" in _tag[1]: _tag1 = _tag[1].split("#") ann_tag = soup_full.find(_tag[0], class_=_tag1[0]) if ann_tag: _data[f'{_tk}'] = ann_tag.get(_tag1[1]) else: ann_tag = soup_full.find(_tag[0], class_=_tag[1]) if ann_tag and len(ann_tag.text.strip()) > 2: _data[f'{_tk}'] = remove_brackets(ann_tag.text.strip()) def rm_html_blacklist(spider_data, soup, sch_info): """ 移除图片黑名单：根据URL MD5判断，直接删除img标签 """ _domain_url = sch_info['json_domain'] for img_tag in soup.find_all('img'): _href = img_tag.get('src') if _href is None: continue _md5 = getMD5Str(_href) ner_logger.info(f"图片 {_href},{_md5}") if spider_data.check_url_in_blacklist(_md5): img_tag.decompose() ner_logger.info(f"图片在黑名单里面,已经移除掉 {_href}") def fix_html_div(spider_data, soup, sch_info, _data): """ 移除无用HTML标签： 按class、id、指定序号、自定义标签 移除垃圾模块 """ _all_div_class = sch_info.get('detail_rm_classes', '').split('|') _all_div_id = sch_info.get('detail_rm_ids', '').split('|') _rmlist = [] _index = [] # 移除指定class的div for div in soup.find_all('div'): _clist = div.get('class') if _clist: class_str = ".".join(_clist) for _class in _all_div_class: if '^' in _class: _sclass = _class.split('^') if class_str == _sclass[0] and int(_sclass[1]) == len(_index): _rmlist.append(div) if class_str == _sclass[0] and div not in _index: _index.append(div) elif class_str == _class: _rmlist.append(div) # 移除指定id for _id in _all_div_id: if div.get('id') == _id: _rmlist.append(div) # 移除其他自定义标签（p/span等） _all_oth_class = sch_info.get('detail_rm_oth_classes', '').split('|') for _oclass in _all_oth_class: if '!' in _oclass: _h, _c = _oclass.split('!') o_tags = soup.find_all(_h, class_=_c) for tag in o_tags: _rmlist.append(tag) # 提取tuning内容并移除标签 _all_tuning_content = [] _all_tuning_class = sch_info.get('detail_tuning_classes', '').split('|') for _oclass in _all_tuning_class: if '!' in _oclass: _h, _c = _oclass.split('!') ner_logger.info(f"需要优化的div,other 检测 {_h} {_c}") o_tags = soup.find_all(_h, class_=_c) or soup.find_all(_h, id=_c) for tag in o_tags: _all_tuning_content.append(tag.get_text(separator='\n', strip=True)) _rmlist.append(tag) _data['tuning_content'] = _all_tuning_content # 执行删除 for div in _rmlist: div.decompose() def fix_html_a(spider_data, soup, sch_info, _props, _data): """ 修复链接： 相对路径 → 绝对路径 javascript链接 → 直接删除 记录所有链接到props """ _props_hrefs = {} _domain_url = sch_info['json_domain'] for a in soup.find_all('a'): _href = a.get('href') if not _href: continue # 移除脚本链接 if _href.startswith('javascript'): a.decompose() continue # 绝对路径直接保留 if _href.startswith('http'): _props_hrefs[_href] = "" continue # 修复相对路径 / if _href.startswith('/'): _href = _domain_url + _href a['href'] = _href _props_hrefs[_href] = "" # 修复相对路径 ./ if _href.startswith('./'): httpdir = get_directory_from_url(_data['full_url']) _href = httpdir + _href[2:] a['href'] = _href _props_hrefs[_href] = "" _props['hrefs'] = _props_hrefs def fix_html_img_src(spider_data, soup, sch_info, _data, _cache_dir): """ 修复图片src： 相对路径 → 绝对路径 base64 → 保存本地 → 上传OBS """ _domain_url = sch_info['json_domain'] _pre_http = "https:" if not _domain_url.startswith('http:') else "http:" for img_tag in soup.find_all('img'): _href = img_tag.get('src') if not _href: continue if _href.startswith('http'): continue # 修复 //domain.com if _href.startswith('//'): _href = _pre_http + _href # 修复 /path elif _href.startswith('/'): _href = _domain_url + _href # 修复 ../path elif _href.startswith('../'): _lasturl = _data.get('last_url', _data['full_url']) httpdir = get_directory_url(_lasturl) _href = httpdir + _href[3:] # 修复普通相对路径 elif re.match(r'^[a-zA-Z0-9]{1,30}/', _href): _href = f"{_domain_url}/{_href}" # 过滤 file:// elif _href.startswith("file://"): _href = "" # 处理 base64 图片 elif "data:image/png;base64" in _href: _md5 = getMD5Str(_href) _cache_file = f"{_cache_dir}/{_md5}.png" if save_base64_img(_cache_file, _href): _href = upload_obs(f"{_md5}.png", _cache_file, {'Content-Type': 'image/png'}) else: _href = "" elif "data:image/jpeg;base64" in _href: _md5 = getMD5Str(_href) _cache_file = f"{_cache_dir}/{_md5}.jpeg" if save_base64_img(_cache_file, _href): _href = upload_obs(f"{_md5}.jpeg", _cache_file, {'Content-Type': 'image/jpeg'}) else: _href = "" img_tag['src'] = _href def upload_obs(_cache_filename, _cachefile, headers): """ 上传文件到华为云OBS """ _ok, _url = upload_file_to_obs(_cache_filename, _cachefile, headers) return _url if _ok else _cachefile def get_html_content(htmltext, _props): """ 提取HTML纯文本： 把图片OCR文字插入到图片位置，再提取全文 """ soup = BeautifulSoup(htmltext, 'html.parser') # 把图片OCR文字插入到img标签后 for img_tag in soup.find_all('img'): _href = img_tag.get('src') _ohref = img_tag.get('q_url') target_url = _ohref or _href if target_url and 'img_urls' in _props: for _url, imgmap in _props['img_urls'].items(): if _url == target_url and 'img_ocr' in imgmap: img_tag.insert_after(imgmap['img_ocr']) # 返回干净文本 return soup.get_text() ``` --- **项目分区导航**：[[01-ann_img|ann_img]] ⬅️ | 02-ann_md | ➡️ [[03-ann_model|ann_model]]