You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

如何结构化Python类以选择对应解析器且避免重复请求S3资源

优化方案

你的思路方向是对的,核心要解决的就是「S3请求和解析器实例化强绑定」的问题,调整结构后完全可以做到仅发起1次S3请求。

核心改造点

  • 拆分资源拉取和解析逻辑:把S3请求、生成BeautifulSoup的逻辑从Parser基类的初始化方法中剥离,支持直接传入已生成的soup对象初始化解析器
  • 把detect_parser改成类方法:不需要实例化解析器就能完成匹配判断,遍历检测阶段完全不会触发S3请求
  • 新增解析器工厂统一调度:负责拉取资源、匹配解析器、返回解析结果

改造后代码示例

1. 基类 parser.py 改造

from gzip import decompress
from bs4 import BeautifulSoup
import requests

class Parser:
    # 新增可选参数soup,外部传入已生成的soup就不会再发请求
    def __init__(self, page_id, soup=None):
        self.landing_page_endpoint = f"https://my_org.org/{page_id}"
        self.parser_name = None
        # 只有没传soup的时候才主动拉取
        self.soup = soup if soup else self.get_soup()

    def get_html(self):
        r = requests.get(self.landing_page_endpoint)
        html = decompress(r.content)
        return html

    def get_soup(self):
        soup = BeautifulSoup(self.get_html(), "html.parser")
        return soup

    def parse(self):
        """Core method that returns authors and associated affiliations."""
        raise NotImplementedError("子类必须实现parse方法")

    # 类方法,子类各自实现匹配逻辑,不需要实例化就能调用
    @classmethod
    def detect_parser(cls, soup):
        """根据soup的头部信息判断是否是当前解析器的适配类型,返回布尔值"""
        raise NotImplementedError("子类必须实现detect_parser方法")

2. 子类 parsers/gregory.py 改造

from parser import Parser

class Gregory(Parser):
    def __init__(self, doi, soup=None):
        super().__init__(doi, soup)
        self.parser_name = "gregory"

    @classmethod
    def detect_parser(cls, soup):
        # 示例:根据页面头部的meta信息判断是否匹配,可根据实际逻辑修改
        meta_tag = soup.find("meta", {"name": "source"})
        return meta_tag and meta_tag.get("content") == "gregory"

    def parse(self):
        my_parsed_info = 'asdf'
        return my_parsed_info

3. 新增解析器工厂 parser_factory.py

# 导入所有已实现的解析器子类
from parsers.gregory import Gregory
# 后续新增解析器只需要在这里加进列表即可
PARSER_CLASSES = [Gregory]

def get_parser_result(page_id):
    # 仅发起1次S3请求,生成soup
    from parser import Parser
    base_parser = Parser(page_id)
    target_soup = base_parser.soup
    # 遍历所有解析器类匹配,完全不需要实例化,不会触发额外请求
    for parser_cls in PARSER_CLASSES:
        if parser_cls.detect_parser(target_soup):
            # 匹配到后传入已有的soup实例化解析器,不会再发请求
            parser_instance = parser_cls(page_id, soup=target_soup)
            return parser_instance.parse()
    # 没有匹配到解析器的兜底逻辑
    raise ValueError(f"没有找到适配page_id={page_id}的解析器")

4. 调用层 views.py 改造

from flask import jsonify, request
from parser_factory import get_parser_result

page_id = request.args.get('page_id')
try:
    result = get_parser_result(page_id)
    return jsonify(result)
except ValueError as e:
    return jsonify({"error": str(e)}), 400

额外优化说明

后续新增解析器只需要两步:

  1. 继承Parser基类实现detect_parser和parse方法
  2. 把新的解析器类加到PARSER_CLASSES列表里即可,不需要修改调度逻辑

内容的提问来源于stack exchange,提问作者Casey

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.10.02 10:54:03