--- title: "01-func_call" created: 2026-04-02 tags: - 项目 aliases: - func_call --- # func_call.py ### `func_call.py` — 运行时调度器 `importlib` 动态加载指定的 `gen_` 模块执行解析;结果为空自动触发 [[02-func_gen_bygpt|func_gen_bygpt]] 让大模型重新生成。完整流程见 [[00-auto_gen|auto_gen 主篇]]。 ## 代码 ```python # -*- 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') ``` --- **项目分区导航**:[[00-auto_gen|auto_gen]] ⬅️ | 01-func_call | ➡️ [[02-func_gen_bygpt|func_gen_bygpt]]