func_call.py

func_call.py — 运行时调度器

importlib 动态加载指定的 gen_ 模块执行解析；结果为空自动触发 func_gen_bygpt 让大模型重新生成。完整流程见 auto_gen 主篇。

代码

# -*- coding: utf-8 -*-
"""
动态加载 & 执行自动生成的解析函数
功能：
1. 动态导入大模型生成的 Python 解析代码
2. 执行 HTML 表格解析，输出 JSON
3. 自动校验结果，为空则重新生成解析函数
4. 提供爬虫动作统一调用入口
"""

import importlib
import sys
sys.path.append('../')
import json

from auto_gen.func_gen_bygpt import gen_func_bygpt
from utils import ner_logger

# ====================== 加载模块并执行解析函数 ======================
def load_and_execute(module_name, func_name, html_content, tmp_file):
    """
    动态加载 Python 模块，并执行指定解析函数
    :param module_name: 模块名（自动生成的代码文件）
    :param func_name: 函数名 → 固定 extract_table_from_html
    :param html_content: 要解析的 HTML 原文
    :param tmp_file: 输出 JSON 路径
    :return: 解析是否成功
    """
    try:
        # 动态导入模块
        module = importlib.import_module(module_name)
        # 获取函数
        func = getattr(module, func_name)
        # 执行解析
        func(html_content, tmp_file)

    except Exception as e:
        # 执行失败 → 写入空列表
        with open(tmp_file, 'w', encoding='utf-8') as f:
            f.write("[]")
        ner_logger.error(f"执行解析html文件失败：{tmp_file} {e}！")

    # 检查解析结果
    return check_result(module_name, html_content, tmp_file)

# ====================== 解析结果校验：空则重新生成代码 ======================
def check_result(module_name, html_content, tmp_file):
    """
    校验解析后的 JSON 是否有效
    无效 → 触发重新生成解析函数
    """
    try:
        # 读取解析结果
        with open(tmp_file, 'r', encoding='utf-8') as f:
            result = f.read()
            result_json = json.loads(result)

            # 结果为空 → 重新生成代码
            if len(result_json) == 0:
                ner_logger.warning(f"解析结果为空，重新生成解析函数：{module_name}")
                gen_func_bygpt(module_name, html_content)

            return len(result_json) > 0

    except Exception as e:
        ner_logger.error(f"解析JSON失败：{tmp_file}，需人工处理！错误：{e}")
        return False

# ====================== 外部调用入口：执行解析 ======================
def call_func(func_name, html_content, tmp_fname):
    """
    外部统一调用入口
    :param func_name: 模块名
    :param html_content: HTML 内容
    :param tmp_fname: 输出 JSON 文件
    :return: 解析成功/失败
    """
    tmp_file = tmp_fname
    # 加载并执行自动生成的 extract_table_from_html 函数
    _ok = load_and_execute(func_name, 'extract_table_from_html', html_content, tmp_file)
    return _ok

# ====================== 动态执行爬虫页面动作 ======================
def execute_page_action(module_name, page):
    """
    执行爬虫页面处理函数
    :param module_name: 爬虫模块名
    :param page: 页面参数
    """
    try:
        module = importlib.import_module(module_name)
        func = getattr(module, "crawl_page")
        func(page)
    except Exception as e:
        ner_logger.error(f"执行动作错误：{module_name} {e}！")

# ====================== 动态执行首页爬虫，返回 URL 列表 ======================
def execute_index_action(module_name, page, sch_info):
    """
    执行首页爬虫，返回抓取的 URL 列表
    :param module_name: 爬虫模块
    :param page: 页码
    :param sch_info: 学校配置
    :return: url 列表
    """
    urls = []
    try:
        module = importlib.import_module(module_name)
        func = getattr(module, "crawl_page")
        urls = func(page, sch_info)
    except Exception as e:
        ner_logger.error(f"执行首页爬虫错误：{module_name} {e}！")
    return urls


# 使用示例
# load_and_execute('gen_00001', 'extract_table_from_html', html_content, 'auto_gen/tmp/result.json')

项目分区导航：auto_gen ⬅️ | 01-func_call | ➡️ func_gen_bygpt