ann_img.py

ann_img.py — 图片处理与 OCR 组件

下载公告图片 → MD5/感知哈希去重 → 二维码识别（纯二维码/混合图/普通图）→ PaddleOCR 文字识别 → 多层黑名单过滤 → 上传华为云 OBS 替换 src。五层流程详解见 parsegpt 主篇。

代码

# -*- coding: utf-8 -*-
"""
图片处理工具模块
功能：下载网页图片 → 本地缓存 → 尺寸检测 → 二维码识别 → OCR文字识别 → 图片上传云存储 → 过滤垃圾图片
适用：爬虫后处理HTML中的图片，清洗、标准化、提取信息
"""

from PIL import Image
from utils import ner_logger
import os
import io
import re
import sys
sys.path.append('../')  # 添加上级目录到Python路径，解决模块导入问题
import json

# 工具类导入：日志、MD5、文件下载、后缀获取、文本MD5等
from utils import ner_logger, getMD5Bytes, download_file, get_file_extension, get_md5_clear_text, getMD5Str
# 图片工具：透明度检测、读取图片、图片均值哈希（去重）
from utils_img import check_transparency_ratio, read_image_object, average_hash
# OCR识别接口
from api.ocr_api import Paddle_OCR
# 华为云OBS上传接口
from api.hwcloud_api import upload_file_to_obs
# URL工具：重组URL（标准化URL）
from utils_html import recombine_url
# 二维码检测工具
from utils_img import detect_qr_code

# ====================== 全局初始化 ======================
# 初始化 OCR 解析器（全局单例，避免重复加载）
ocr_parser = Paddle_OCR()


def url_to_base64(image_url, _cache_dir):
    """
    根据图片URL下载图片，完成：缓存、尺寸获取、MD5、二维码检测、OCR识别
    :param image_url: 图片网络地址
    :param _cache_dir: 本地缓存目录
    :return: 宽度、高度、二维码标记、图片属性字典、缓存路径、文件名、响应头
    """
    # 1. 下载图片内容
    _ok, image_content, headers = download_file(image_url)
    if not _ok:
        ner_logger.info(f"图片下载失败 {image_url}")
        return 0, 0, 'N', {}, '', '', ''

    # 2. 读取为PIL图片对象
    _ok, image = read_image_object(image_url, image_content)
    if not _ok:
        ner_logger.info(f"图片读取失败 {image_url}")
        return 0, 0, 'N', {}, '', '', ''

    # 3. 获取图片宽高
    width, height = image.size

    # 4. 计算图片内容MD5（唯一标识）
    img_md5 = getMD5Bytes(image_content)

    # 5. 获取图片后缀（.jpg/.png/.gif等）
    _ext = get_file_ext(image_url, image)

    # 6. 生成本地缓存文件名（MD5命名，避免重复）
    _cache_filename = f'{img_md5}_pic{_ext}'
    _cachefile = os.path.join(_cache_dir, _cache_filename)

    # 7. 写入本地缓存文件
    with open(_cachefile, 'wb') as f:
        f.write(image_content)

    ner_logger.info(f"图片尺寸: {width}x{height}, {image_url}")

    # 8. RGBA格式转RGB（统一格式，避免后续处理报错）
    if image.mode == 'RGBA':
        image = image.convert('RGB')

    # ====================== 二维码检测 ======================
    _qrcode = 'N'  # N=无二维码 H=含二维码 Y=纯二维码
    _props = {}
    try:
        # 仅对大于64*64的图片检测二维码
        if width * height > 64 * 64:
            _qrcode, _props = detect_qr_code(_cachefile, width, height)
    except Exception as e:
        ner_logger.info(f"二维码检测失败 {image_url}: {e}")

    # ====================== 构建图片基础属性 ======================
    _props['img_width'] = width
    _props['img_height'] = height
    _props['img_md5'] = img_md5
    _props['img_similar_md5'] = average_hash(_cachefile)  # 均值哈希，用于相似图片去重
    _props['img_count'] = 1  # 图片出现次数（去重用）

    # ====================== OCR文字识别 ======================
    _props['img_ocr'] = ''
    ner_logger.info(f"二维码结果:{_qrcode} / {image_url}")

    # 非纯二维码才进行OCR（纯二维码无需识别文字）
    if _qrcode != 'Y':
        try:
            # 仅对大于64*64的图片做OCR
            if width * height > 64 * 64:
                ner_logger.info(f"开始OCR识别 {_cachefile}")
                _ocr_text = ocr_parser.ocr_txt_new(_cachefile, width, height)
                _props['img_ocr'] = _ocr_text
        except Exception as e:
            import traceback
            error_info = traceback.format_exc()
            ner_logger.info(f"OCR识别失败 {image_url}: {e}\n{error_info}")

        # 如果图片含二维码，从OCR结果中提取二维码链接
        if _qrcode == 'H' and has_qrcode_from_img(_props['img_ocr']):
            set_qrcode_from_img(_cachefile, _props)

    # 返回所有处理结果
    return width, height, _qrcode, _props, _cachefile, _cache_filename, headers


def has_qrcode_from_img(img_ocr):
    """
    判断OCR文本中是否包含二维码相关关键词（用于辅助判断二维码）
    暂时直接返回True，不做关键词过滤
    """
    # 原逻辑：包含'扫描','投递','应聘'等关键词则返回True
    return True


def get_file_ext(image_url, image):
    """
    获取图片文件后缀（优先从URL获取，失败则从图片格式获取）
    :param image_url: 图片URL
    :param image: PIL图片对象
    :return: 后缀名 .jpg/.png/.gif/.webp
    """
    # 从URL提取后缀
    _filesuffix = get_file_extension(image_url)
    if _filesuffix:
        return _filesuffix

    ner_logger.info(f"URL无法获取后缀，使用图片格式 {image_url}: {image.format}")

    # 从PIL图片格式判断
    if image.format == "WEBP":
        return ".webp"
    if image.format == "PNG":
        return ".png"
    if image.format == "JPEG":
        return ".jpg"
    if image.format == "GIF":
        return ".gif"

    # 默认返回png
    return '.png'


def get_img_real_size_config(_data):
    """
    加载图片真实显示尺寸配置文件（用于修正图片渲染大小）
    :param _data: 配置数据
    :return: 图片真实尺寸字典 {url: {rendered_width, rendered_height}}
    """
    if 'wx_code_file_config' in _data:
        _config_file = _data['wx_code_file_config']
        if os.path.exists(_config_file):
            with open(_config_file, 'r', encoding='utf-8') as f:
                _config = json.load(f)
                return _config
    return {}


def get_img_size(_rendered_img, _url):
    """
    从配置中获取图片实际渲染大小（修正网页显示尺寸）
    :param _rendered_img: 尺寸配置字典
    :param _url: 图片URL
    :return: 是否找到、宽度、高度
    """
    _combiled_url = recombine_url(_url)
    # 模糊匹配URL
    for kurl, size in _rendered_img.items():
        _combiled_kurl = recombine_url(kurl)
        if kurl in _url or _url in kurl or _combiled_kurl == _combiled_url:
            return True, int(float(size['rendered_width'])), int(float(size['rendered_height']))
    return False, 0, 0


def replace_img_urls_with_base64(spider_data, _data, soup, _props, _cache_dir):
    """
    核心函数：处理HTML中所有<img>标签
    流程：下载 → 检测 → 过滤 → 上传云存储 → 替换src → 清洗垃圾图片
    :param spider_data: 爬虫数据对象（黑名单校验）
    :param _data: 配置数据
    :param soup: BeautifulSoup对象
    :param _props: 输出属性字典
    :param _cache_dir: 缓存目录
    :return: 处理是否成功
    """
    _props_imgs = {}  # 存储所有图片信息
    _rendered_imgs = get_img_real_size_config(_data)  # 图片真实显示尺寸
    _multipics = {}  # 记录无文字图片（用于去重删除）

    # 遍历HTML中所有img标签
    for img_tag in soup.find_all('img'):
        img_url = img_tag.get('src')
        # 只处理http/https开头的有效图片
        if img_url and img_url.startswith(('http://', 'https://')):
            ner_logger.info(f"处理图片: {img_url}")

            # 下载并解析图片
            width, height, _qrcode, _iprops, _cachefile, _cache_filename, headers = url_to_base64(img_url, _cache_dir)

            # 下载失败 → 直接删除标签
            if width * height == 0:
                img_tag.decompose()
                continue

            # 原始真实尺寸
            intrinsic_width = width
            intrinsic_height = height

            # 使用配置中的渲染尺寸（如果有）
            _ok, rendered_width, rendered_height = get_img_size(_rendered_imgs, img_url)
            if _ok:
                width = rendered_width
                height = rendered_height
                if width * height != intrinsic_width * intrinsic_height:
                    ner_logger.info(f"使用渲染尺寸: {width}x{height}, 真实尺寸{intrinsic_width}x{intrinsic_height},{img_url}")

            # ====================== 图片黑名单过滤 ======================
            # 1. 内容MD5黑名单
            if spider_data.check_url_in_blacklist(_iprops['img_md5']):
                img_tag.decompose()
                ner_logger.info(f"图片MD5在黑名单，已移除 {img_url}")
                continue

            # 2. 相似哈希MD5黑名单
            if spider_data.check_url_in_blacklist(_iprops['img_similar_md5']):
                img_tag.decompose()
                ner_logger.info(f"图片相似哈希在黑名单，已移除 {img_url},{_iprops['img_similar_md5']}")
                continue

            # 3. OCR文字内容黑名单
            if len(_iprops['img_ocr']) > 30:
                _md5txt = get_md5_clear_text(_iprops['img_ocr'])
                _mdtv = getMD5Str(_md5txt)
                ner_logger.info(f"图片OCR文本MD5 {_mdtv}")
                if spider_data.check_md5_txt_in_blacklist(_mdtv):
                    img_tag.decompose()
                    ner_logger.info(f"图片OCR文本在黑名单，已移除 {img_url}")
                    continue

            # ====================== 尺寸/质量过滤 ======================
            # 过小图片 + 无文字 → 删除
            if _qrcode == 'N' and len(_iprops['img_ocr']) < 5 and (
                    width * height < 120 * 120 or (width < 40 and height < 500) or (width < 500 and height < 40)):
                img_tag.decompose()
                ner_logger.info(f"图片过小且无文字，已移除 {img_url}")
                continue

            # 学校就业指导中心垃圾图过滤
            if _qrcode in ['H'] and width * height < 300 * 700 and probe_school_blackimg(_iprops['img_ocr']):
                img_tag.decompose()
                ner_logger.info(f"学校垃圾广告图，已移除 {img_url}")
                continue

            # 高透明 + 无文字 → 删除
            if _qrcode in ['N'] and len(_iprops['img_ocr']) < 5 and check_transparency_ratio(_cachefile):
                img_tag.decompose()
                ner_logger.info(f"图片透明度过高，已移除 {img_url}")
                continue

            # 记录无文字图片（后续去重）
            if _qrcode == 'N' and len(_iprops['img_ocr']) < 20:
                _multipics[img_tag] = img_url

            # ====================== 给img标签添加自定义属性 ======================
            img_tag['q_width'] = width
            img_tag['q_height'] = height
            img_tag['q_qrcode'] = _qrcode
            img_tag['q_url'] = img_url

            # ====================== 上传图片到华为云OBS ======================
            _ok, _url = upload_file_to_obs(_cache_filename, _cachefile, headers)
            if _ok:
                # 替换src为云存储地址
                img_tag['src'] = f"{_url}"
                # 纯二维码添加尺寸缩放参数
                if _qrcode == 'Y' and width > 100:
                    img_tag['src'] = f"{_url}?x-image-process=image/resize,m_fixed,w_{width},h_{height}"
                # 记录云存储地址
                _iprops['qz_img_url'] = _url
                _iprops['rendered_width'] = width
                _iprops['rendered_height'] = height
            else:
                ner_logger.info(f"OBS上传失败，使用原图 {img_url}")
                return False

            # ====================== 图片去重计数 ======================
            if img_url in _props_imgs:
                _props_imgs[img_url]['img_count'] += 1
            else:
                # 相似图片合并计数
                for _k, _v in _props_imgs.items():
                    if _v['img_similar_md5'] == _iprops['img_similar_md5']:
                        _v['img_count'] += 1
                        _iprops['img_count'] += 1
                        break
                _props_imgs[img_url] = _iprops

            # 标记超长图片
            if width > 500 and height > 3000:
                _data["is_large_image"] = 'OK'

    # ====================== 移除重复无意义图片 ======================
    for img_tag, img_url in _multipics.items():
        if img_url in _props_imgs and _props_imgs[img_url]['img_count'] > 1:
            img_tag.decompose()
            ner_logger.info(f"无文字重复图片，已移除 {img_url}")

    # 保存所有图片信息
    _props["img_urls"] = _props_imgs
    return True


def intit_blacklist():
    """
    初始化学校垃圾图片关键词黑名单（从文件加载）
    :return: 关键词列表
    """
    _blacklist = []
    with open("data/black_img_text.txt", encoding="utf-8") as f:
        for _line in f.read().splitlines():
            _line = _line.strip()
            if _line:
                _blacklist.append(_line.strip())
    return _blacklist


# 全局加载垃圾图片关键词
sch_blacklist = intit_blacklist()


def probe_school_blackimg(_text):
    """
    检测图片文字是否为学校垃圾广告（就业中心、推广类）
    :param _text: OCR文字
    :return: True=垃圾图 False=正常图
    """
    if len(_text) > 50:
        return False
    # 正则匹配关键词
    for _blkre in sch_blacklist:
        pattern = re.compile(_blkre)
        if re.findall(pattern, _text):
            return True
    return False


def set_qrcode_from_img(_cache_file, _props):
    """
    从图片中截取二维码区域 → 生成新图片 → 上传OBS → 记录二维码链接
    :param _cache_file: 原图路径
    :param _props: 属性字典
    """
    _qk = None
    _qr = None
    _all_qr = {}

    # 遍历属性中以http开头的二维码坐标
    for _k, _v in _props.items():
        if _k.startswith('http') and len(_props[_k]) > 0 and len(_props[_k].split(',')) == 4:
            try:
                ner_logger.info(f"截取二维码区域 {_cache_file}")
                # 解析坐标：左、上、宽、高
                left, top, width, height = _props[_k].split(',')
                left, top, width, height = int(float(left)), int(float(top)), int(float(width)), int(float(height))

                # 打开原图并裁剪
                image = Image.open(_cache_file)
                cropped_image = image.crop((left, top, left + width, top + height))

                # 保存到内存
                byte_arr = io.BytesIO()
                cropped_image.save(byte_arr, format='PNG')

                # 临时保存
                _filename = os.path.splitext(_cache_file)
                _nfilename = _filename[0] + '_qr.png'
                with open(_nfilename, 'wb') as f:
                    f.write(byte_arr.getvalue())

                # 上传OBS
                _md5 = getMD5Bytes(byte_arr.getvalue())
                _qrfilename = f'{_md5}_qr.png'
                _ok, url = upload_file_to_obs(_qrfilename, _nfilename)

                if _ok:
                    if not _qr:
                        _qk = _k
                        _qr = url
                    _all_qr[_k] = url
                    os.remove(_nfilename)  # 删除临时文件

            except Exception as e:
                ner_logger.info(f"二维码截取失败: {e}")

    # 保存最终二维码信息
    if _qk:
        _props['inside_qr_link'] = _qk
        _props['inside_qr_pic'] = _qr
        _props['all_qr_pics'] = _all_qr

项目分区导航：parsegpt ⬅️ | 01-ann_img | ➡️ ann_md