func_call.py

func_call.py — 运行时调度器（大公司版）

与 auto_gen 的 func_call 同源，面向大公司招聘页面的解析调度：动态加载解析函数 → 执行 → 结果为空触发大模型重新生成。

代码

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

from auto_gen_com.func_gen_bygpt import gen_func_bygpt
from utils import ner_logger

# 动态加载模块并执行解析函数，异常时写入空数组，最后校验结果
def load_and_execute(module_name, func_name, html_content, tmp_file):
    # 动态导入指定模块
    module = importlib.import_module(module_name)
    # 获取模块中指定名称的函数
    func = getattr(module, func_name)
    # 执行解析函数
    try:
        func(html_content, tmp_file)
    except Exception as e:
        # 执行失败则写入空JSON列表，保证文件格式合法
        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):
    # 读取解析输出文件
    with open(tmp_file, 'r', encoding='utf-8') as f:
        result = f.read()
        result_json = []
        try:
            # 转换为JSON对象
            result_json = json.loads(result)
            # 解析结果为空，触发大模型重新生成解析函数
            if len(result_json) == 0:
                gen_func_bygpt(module_name, html_content)
        except:
            ner_logger.error(f"执行解析html文件失败：{tmp_file},人工处理！")
        
        # 有数据返回True，无数据返回False
        if len(result_json) == 0:
            return False
        return True

# 外部调用入口：加载自动生成的解析函数并执行
def call_func(func_name, html_content, tmp_fname):
    tmp_file = tmp_fname
    # 执行动态加载和解析
    _ok = load_and_execute(func_name, 'extract_table_from_html', html_content, tmp_file)
    return _ok

# 动态执行页面爬虫函数，返回执行结果
def execute_page_action(module_name, page):
    # 动态导入爬虫模块
    module = importlib.import_module(module_name)
    # 获取页面爬取函数
    func = getattr(module, "crawl_page")
    # 执行并返回结果
    try:
        return func(page)
    except Exception as e:
        ner_logger.error(f"执行动作错误：{module_name}{e}！")
    return False

# 动态执行首页爬虫，返回抓取的URL列表
def execute_index_action(module_name, page, sch_info):
    urls = []
    # 动态导入爬虫模块
    module = importlib.import_module(module_name)
    # 获取首页爬取函数
    func = getattr(module, "crawl_page")
    # 执行爬虫获取链接列表
    try:
        urls = func(page, sch_info)
    except Exception as e:
        ner_logger.error(f"执行动作错误：{module_name}  {e}！")
    return urls

# 使用示例
# 假设有一个名为 example.py 的文件，其中包含一个名为 example_function 的函数
# load_and_execute('gen_00001', 'extract_table_from_html',"",'a.txt')

项目分区导航：click_50001 ⬅️ | 02-func_call | ➡️ func_gen_bygpt