ann_md.py

ann_md.py — HTML 预处理与 Markdown 转换

解析流水线的第一道关卡：清洗噪音 DOM → 修复链接/图片路径 → 硬编码字段提取 → 表格转 Markdown → 渠道专项修复。六件事详解见 parsegpt 主篇。

代码

# -*- 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):
    """
    修复<a>链接：
    相对路径 → 绝对路径
    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):
    """
    修复<img>图片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()

项目分区导航：ann_img ⬅️ | 02-ann_md | ➡️ ann_model