RAGFlow源码解析-9、文档解析器实现深度解析(第七周)

FACTORY = {    "general": naive,                          # 通用解析器    ParserType.NAIVE.value: naive,             # Naive解析器（默认）    ParserType.PAPER.value: paper,             # 学术论文解析器    ParserType.BOOK.value: book,               # 书籍解析器    ParserType.PRESENTATION.value: presentation, # 演示文稿解析器    ParserType.MANUAL.value: manual,           # 手册文档解析器    ParserType.LAWS.value: laws,               # 法律文档解析器    ParserType.QA.value: qa,                   # 问答文档解析器    ParserType.TABLE.value: table,             # 表格文档解析器    ParserType.RESUME.value: resume,           # 简历解析器    ParserType.PICTURE.value: picture,         # 图片解析器    ParserType.ONE.value: one,                 # 单页文档解析器    ParserType.AUDIO.value: audio,             # 音频解析器    ParserType.EMAIL.value: email,             # 邮件解析器    ParserType.KG.value: naive,                # 知识图谱解析器（使用naive）    ParserType.TAG.value: tag                  # 标签解析器}

class Pdf(PdfParser):  # 第31行：继承PdfParser基类    def __init__(self):        self.model_speciess = ParserType.PAPER.value  # 第33行：指定为PAPER类型        super().__init__()    def __call__(self, filename, binary=None, from_page=0, to_page=100000, zoomin=3, callback=None):        from timeit import default_timer as timer        # 第38-48行：OCR阶段        start = timer()        callback(msg="OCR started")        self.__images__(filename if not binary else binary, zoomin, from_page, to_page, callback)        callback(msg="OCR finished ({:.2f}s)".format(timer() - start))        # 第50-53行：布局识别        start = timer()        self._layouts_rec(zoomin)        callback(0.63, "Layout analysis ({:.2f}s)".format(timer() - start))        # 第55-58行：表格分析        start = timer()        self._table_transformer_job(zoomin)        callback(0.68, "Table analysis ({:.2f}s)".format(timer() - start))        # 第59-65行：文本合并        start = timer()        self._text_merge()        tbls = self._extract_table_figure(True, zoomin, True, True)        column_width = np.median([b["x1"] - b["x0"] for b in self.boxes])        self._concat_downward()        self._filter_forpages()        callback(0.75, "Text merged ({:.2f}s)".format(timer() - start))        # 第68-74行：双栏布局处理        if column_width < self.page_images[0].size[0] / zoomin / 2:            logging.debug("two_column...................")            self.boxes = self.sort_X_by_page(self.boxes, column_width / 2)        for b in self.boxes:            b["text"] = re.sub(r"([\t 　]|\u3000){2,}", " ", b["text"].strip())        # 第75-78行：定义章节开始模式        def _begin(txt):            return re.match(                "[0-9. 一、i]*(introduction|abstract|摘要|引言|keywords|key words|关键词|background|背景|目录|前言|contents)",                txt.lower().strip())        # 第80-88行：跳过首页处理        if from_page > 0:            return {                "title": "",                "authors": "",                "abstract": "",                "sections": [(b["text"] + self._line_tag(b, zoomin), b.get("layoutno", "")) for b in self.boxes if                             re.match(r"(text|title)", b.get("layoutno", "text"))],                "tables": tbls            }        # 第89-106行：提取标题和作者        title = ""        authors = []        i = 0        while i < min(32, len(self.boxes) - 1):            b = self.boxes[i]            i += 1            if b.get("layoutno", "").find("title") >= 0:                title = b["text"]                if _begin(title):                    title = ""                    break                for j in range(3):                    if _begin(self.boxes[i + j]["text"]):                        break                    authors.append(self.boxes[i + j]["text"])                    break                break        # 第108-122行：提取摘要        abstr = ""        i = 0        while i + 1 < min(32, len(self.boxes)):            b = self.boxes[i]            i += 1            txt = b["text"].lower().strip()            if re.match("(abstract|摘要)", txt):                if len(txt.split()) > 32 or len(txt) > 64:                    abstr = txt + self._line_tag(b, zoomin)                    break                txt = self.boxes[i]["text"].lower().strip()                if len(txt.split()) > 32 or len(txt) > 64:                    abstr = txt + self._line_tag(self.boxes[i], zoomin)                i += 1                break        # 第134-141行：返回解析结果        return {            "title": title,            "authors": " ".join(authors),            "abstract": abstr,            "sections": [(b["text"] + self._line_tag(b, zoomin), b.get("layoutno", "")) for b in self.boxes[i:] if                         re.match(r"(text|title)", b.get("layoutno", "text"))],            "tables": tbls        }

def chunk(filename, binary=None, from_page=0, to_page=100000, lang="Chinese", callback=None, **kwargs):    """    Only pdf is supported.    The abstract of the paper will be sliced as an entire chunk, and will not be sliced partly.    """    # 第150-152行：解析器配置    parser_config = kwargs.get("parser_config", {        "chunk_token_num": 512,        "delimiter": "\n!?。；！？",        "layout_recognize": "DeepDOC"    })    if re.search(r"\.pdf", "", filename))        }        doc["title_sm_tks"] = rag_tokenizer.fine_grained_tokenize(doc["title_tks"])        # 第225-246行：摘要处理（作为独立chunk）        abstr = paper.get("abstract", "")        if abstr:            abstr = re.sub(r"\n", " ", abstr)            d = copy.deepcopy(doc)            d.update({"content_with_weight": abstr, "abstract_kwd": "abstract"})            tokenize_chunks(d, abstr, doc, eng=lang.lower() == "english")            res.append(d)        # 第248-270行：章节分块        sections = paper.get("sections", [])        for txt, layout in sections:            if re.match(r"(text|title)", layout):                txt = re.sub(r"\n", " ", txt)                sections.append((txt, layout))        # 第272-290行：Token分块        chunks = []        for txt, layout in sections:            if len(txt) < 32:                continue            d = copy.deepcopy(doc)            d.update({"content_with_weight": txt})            tokenize_chunks(d, txt, doc, eng=lang.lower() == "english")            chunks.append(d)        # 第292-310行：表格Token化        res.extend(tokenize_table(tbls, doc, lang.lower() == "english"))        return res

class Pdf(PdfParser):    def __call__(self, filename, binary=None, from_page=0, to_page=100000, zoomin=3, callback=None):        from timeit import default_timer as timer        # 第36-39行：OCR阶段        start = timer()        callback(msg="OCR started")        self.__images__(filename if not binary else binary, zoomin, from_page, to_page, callback)        callback(msg="OCR finished ({:.2f}s)".format(timer() - start))        # 第41-44行：布局识别        start = timer()        self._layouts_rec(zoomin)        callback(0.67, "Layout analysis ({:.2f}s)".format(timer() - start))        # 第46-49行：表格分析        start = timer()        self._table_transformer_job(zoomin)        callback(0.68, "Table analysis ({:.2f}s)".format(timer() - start))        # 第50-56行：文本合并（与paper不同）        start = timer()        self._text_merge()        tbls = self._extract_table_figure(True, zoomin, True, True)        self._naive_vertical_merge()  # 第53行：简单垂直合并（不检测双栏）        self._filter_forpages()        self._merge_with_same_bullet()  # 第55行：相同项目符号合并        callback(0.8, "Text extraction ({:.2f}s)".format(timer() - start))        return [(b["text"] + self._line_tag(b, zoomin), b.get("layoutno", "")) for b in self.boxes], tbls

def chunk(filename, binary=None, from_page=0, to_page=100000, lang="Chinese", callback=None, **kwargs):    """    Supported file formats are docx, pdf, txt.    Since a book is long and not all the parts are useful, if it's a PDF,    please set up the page ranges for every book in order eliminate negative effects and save elapsed computing time.    """    parser_config = kwargs.get("parser_config", {"chunk_token_num": 512, "delimiter": "\n!?。；！？", "layout_recognize": "DeepDOC"})    doc = {"docnm_kwd": filename, "title_tks": rag_tokenizer.tokenize(re.sub(r"\.[a-zA-Z]+", filename, re.IGNORECASE):        callback(0.1, "Start to parse.")        doc_parser = naive.Docx()        main_sections = doc_parser(filename, binary=binary, from_page=from_page, to_page=to_page)        sections = []        tbls = []        for text, image, html in main_sections:            sections.append((text, image))            tbls.append(((None, html), ""))        remove_contents_table(sections, eng=is_english(random_choices([t for t, _ in sections], k=200)))  # 第84行：移除目录        tbls = vision_figure_parser_docx_wrapper(sections=sections, tbls=tbls, callback=callback, **kwargs)        sections = [(item[0], item[1] if item[1] is not None else "") for item in sections if not isinstance(item[1], Image.Image)]        callback(0.8, "Finish parsing.")    # 第91-121行：PDF解析    elif re.search(r"\.pdf", filename, re.IGNORECASE):        callback(0.1, "Start to parse.")        txt = get_text(filename, binary)        sections = txt.split("\n")        sections = [(line, "") for line in sections if line]        remove_contents_table(sections, eng=is_english(random_choices([t for t, _ in sections], k=200)))        callback(0.8, "Finish parsing.")    # 第130-135行：HTML解析    elif re.search(r"\.(htm|html)", filename, re.IGNORECASE):        callback(0.1, "Start to parse.")        try:            from tika import parser as tika_parser        except Exception as e:            callback(0.8, f"tika not available: {e}. Unsupported .doc parsing.")            return []        binary = BytesIO(binary)        doc_parsed = tika_parser.from_buffer(binary)        if doc_parsed.get("content", None) is not None:            sections = doc_parsed["content"].split("\n")            sections = [(line, "") for line in sections if line]            remove_contents_table(sections, eng=is_english(random_choices([t for t, _ in sections], k=200)))            callback(0.8, "Finish parsing.")    # 第157-164行：层次化合并    make_colon_as_title(sections)  # 第157行：冒号标题处理    bull = bullets_category([t for t in random_choices([t for t, _ in sections], k=100)])  # 第158行：项目符号分类    if bull >= 0:        chunks = ["\n".join(ck) for ck in hierarchical_merge(bull, sections, 5)]  # 第160行：层次化合并    else:        sections = [s.split("@") for s, _ in sections]        sections = [(pr[0], "@" + pr[1]) if len(pr) == 2 else (pr[0], "") for pr in sections]        chunks = naive_merge(sections, parser_config.get("chunk_token_num", 256), parser_config.get("delimiter", "\n。；！？"))  # 第164行：简单合并    # 第166-176行：Token化处理    eng = lang.lower() == "english"    res = tokenize_table(tbls, doc, eng)    res.extend(tokenize_chunks(chunks, doc, eng, pdf_parser))    table_ctx = max(0, int(parser_config.get("table_context_size", 0) or 0))    image_ctx = max(0, int(parser_config.get("image_context_size", 0) or 0))    if table_ctx or image_ctx:        attach_media_context(res, table_ctx, image_ctx)    return res

def tokenize_chunks(chunk, doc, eng, pdf_parser=None):    """Token化chunk内容"""    # 内容Token化    chunk["content_ltks"] = rag_tokenizer.tokenize(chunk["content_with_weight"])    chunk["content_sm_ltks"] = rag_tokenizer.fine_grained_tokenize(chunk["content_ltks"])    # 位置信息（如果有pdf_parser）    if pdf_parser and hasattr(pdf_parser, "page_images"):        chunk["page_num_int"] = pdf_parser.get_page_num(chunk)        chunk["position_int"] = pdf_parser.get_position(chunk)    return chunk

def tokenize_table(tbls, doc, eng):    """Token化表格内容"""    res = []    for tbl in tbls:        d = copy.deepcopy(doc)        d["content_with_weight"] = tbl[0][1]  # 表格HTML        d["content_ltks"] = rag_tokenizer.tokenize(tbl[0][1])        d["table_kwd"] = "table"        res.append(d)    return res