--- title: "01-ann_img" created: 2026-04-02 tags: - 项目 aliases: - ann_img --- # ann_img.py ### `ann_img.py` — 图片处理与 OCR 组件 下载公告图片 → MD5/感知哈希去重 → 二维码识别（纯二维码/混合图/普通图）→ PaddleOCR 文字识别 → 多层黑名单过滤 → 上传华为云 OBS 替换 `src`。五层流程详解见 [[00-parsegpt|parsegpt 主篇]]。 ## 代码 ```python # -*- 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中所有标签 流程：下载 → 检测 → 过滤 → 上传云存储 → 替换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 ``` --- **项目分区导航**：[[00-parsegpt|parsegpt]] ⬅️ | 01-ann_img | ➡️ [[02-ann_md|ann_md]]