from __future__ import annotations import re from typing import Any, Dict, List, Tuple, TypedDict, Union from langchain_core.documents import Document from langchain_text_splitters.base import Language from langchain_text_splitters.character import RecursiveCharacterTextSplitter
class LineType(TypedDict):
"""Line type as typed dict.""" metadata: Dict[str, str] content: str
class HeaderType(TypedDict):
"""Header type as typed dict.""" level: int name: str data: str
class IdentifyHeaders:
"""Compute data for identifying header text."""
def __init__(self,pdf_doc: str,page = None,body_limit: float = 10):
"""Read all text and make a dictionary of fontsizes.
Args:
body_limit: consider text with larger font size as some header
"""
mydoc = fitz.open(pdf_doc)
fontsizes = {}
pages = range(mydoc.page_count)
for pno in pages:
page = mydoc.load_page(pno)
blocks = page.get_text("dict", flags=fitz.TEXTFLAGS_TEXT)["blocks"]
for span in [ # look at all non-empty horizontal spans
s
for b in blocks
for l in b["lines"]
for s in l["spans"]
if not is_white(s["text"])
]:
fontsz = round(span["size"])
count = fontsizes.get(fontsz, 0) + len(span["text"].strip())
fontsizes[fontsz] = count
mydoc.close()
self.header_id = {}
temp = sorted([(k, v) for k, v in fontsizes.items()],key=lambda i: i[1],reverse=True,)
b_limit = temp[0][0]
sizes = sorted([f for f in fontsizes.keys() if f > b_limit],reverse=True,)[:8]
for i, size in enumerate(sizes):
self.header_id[size] = "#" * (i + 1) + " "
def get_header_id(self, span: dict, page=None) -> str:
"""Return appropriate markdown header prefix.
Given a text span from a "dict"/"rawdict" extraction, determine the
markdown header prefix string of 0 to n concatenated '#' characters.
"""
fontsize = round(span["size"]) # compute fontsize
hdr_id = self.header_id.get(fontsize, "")
return hdr_id
def aggregate_lines_to_chunks(lines: List[LineType]) -> List[Document]:
"""Combine lines with common metadata into chunks
Args:
lines: Line of text / associated header metadata
"""
def split_text(text: str,headers_split: List[Tuple[str, str]]) -> List[Document]:
"""Split markdown file
Args:
text: Markdown file"""
lines = text.split("\n")
lines_with_metadata: List[LineType] = []
current_content: List[str] = []
current_metadata: Dict[str, str] = {}
current_metadata['type'] = 'text'
header_stack: List[HeaderType] = []
initial_metadata: Dict[str, str] = {}
in_code_block = False
opening_fence = ""
for line in lines:
stripped_line = line.strip()
stripped_line = "".join(filter(str.isprintable, stripped_line))
if stripped_line == :
continue
current_header_level = 0
if stripped_line.startswith("-"):
continue
elif stripped_line.startswith("```") or stripped_line.startswith("84.185.107.48"):
initial_metadata['type'] = 'code'
in_code_block = True
opening_fence = "```"
elif stripped_line.startswith("|"):
initial_metadata['type'] = 'table'
elif not in_code_block:
initial_metadata['type'] = 'text'
if in_code_block:
if stripped_line.startswith(opening_fence):
in_code_block = False
opening_fence = ""
for sep, name in headers_split: #if header create index
if stripped_line.startswith(sep) and (len(stripped_line) == len(sep) or stripped_line[len(sep)] == " "):
current_header_level = sep.count("#")
while (header_stack and header_stack[-1]["level"] >= current_header_level):
popped_header = header_stack.pop()
if popped_header["name"] in initial_metadata:
initial_metadata.pop(popped_header["name"])
header: HeaderType = {"level": current_header_level,"name": name,"data": stripped_line[len(sep):].strip()}
header_stack.append(header)
initial_metadata[name] = header["data"]
if current_metadata['type'] != initial_metadata['type']:
lines_with_metadata.append({"content":"\n".join(current_content),"metadata":current_metadata.copy()})
current_content.clear()
current_metadata = initial_metadata.copy()
if current_header_level == 0:
current_content.append(stripped_line)
else:
lines_with_metadata.append({"content":"\n".join(current_content),"metadata":current_metadata.copy()})
current_content.clear()
lines_with_metadata.append({"content":"\n".join(current_content),"metadata":current_metadata.copy()})
#lines_with_metadata.append({"content":"\n".join(current_content),"metadata":current_metadata})
aggregated_chunks = [x for x in lines_with_metadata if x['content'] != ]
# aggregated_chunks: List[LineType] = []
# for line in lines_with_metadata:
# if (aggregated_chunks and aggregated_chunks[-1]["metadata"] == line["metadata"]):
# aggregated_chunks[-1]["content"] += " \n" + line["content"]
# elif (aggregated_chunks
# and aggregated_chunks[-1]["metadata"] != line["metadata"]
# and len(aggregated_chunks[-1]["metadata"]) < len(line["metadata"])
# and aggregated_chunks[-1]["content"].split("\n")[-1][0] == "#"
# and False
# ):
# aggregated_chunks[-1]["content"] += " \n" + line["content"]
# aggregated_chunks[-1]["metadata"] = line["metadata"]
# else:
# aggregated_chunks.append(line)
return [
Document(page_content=chunk["content"], metadata=chunk["metadata"])
for chunk in aggregated_chunks
]