Akjava/open_Deep-Research-DuckDuckGo-Groq
2
1# This is copied from Magentic-one's great repo: https://github.com/microsoft/autogen/blob/v0.4.4/python/packages/autogen-magentic-one/src/autogen_magentic_one/markdown_browser/mdconvert.py2# Thanks to Microsoft researchers for open-sourcing this!3# type: ignore4import base645import copy6import html7import json8import mimetypes9import os10import re11import shutil12import subprocess13import sys14import tempfile15import traceback16from typing import Any, Dict, List, Optional, Union17from urllib.parse import parse_qs, quote, unquote, urlparse, urlunparse18 19import mammoth20import markdownify21import pandas as pd22import pdfminer23import pdfminer.high_level24import pptx25 26# File-format detection27import puremagic28import pydub29import requests30import speech_recognition as sr31from bs4 import BeautifulSoup32from youtube_transcript_api import YouTubeTranscriptApi33from youtube_transcript_api.formatters import SRTFormatter34 35 36class _CustomMarkdownify(markdownify.MarkdownConverter):37 """38 A custom version of markdownify's MarkdownConverter. Changes include:39 40 - Altering the default heading style to use '#', '##', etc.41 - Removing javascript hyperlinks.42 - Truncating images with large data:uri sources.43 - Ensuring URIs are properly escaped, and do not conflict with Markdown syntax44 """45 46 def __init__(self, **options: Any):47 options["heading_style"] = options.get("heading_style", markdownify.ATX)48 # Explicitly cast options to the expected type if necessary49 super().__init__(**options)50 51 def convert_hn(self, n: int, el: Any, text: str, convert_as_inline: bool) -> str:52 """Same as usual, but be sure to start with a new line"""53 if not convert_as_inline:54 if not re.search(r"^\n", text):55 return "\n" + super().convert_hn(n, el, text, convert_as_inline) # type: ignore56 57 return super().convert_hn(n, el, text, convert_as_inline) # type: ignore58 59 def convert_a(self, el: Any, text: str, convert_as_inline: bool):60 """Same as usual converter, but removes Javascript links and escapes URIs."""61 prefix, suffix, text = markdownify.chomp(text) # type: ignore62 if not text:63 return ""64 href = el.get("href")65 title = el.get("title")66 67 # Escape URIs and skip non-http or file schemes68 if href:69 try:70 parsed_url = urlparse(href) # type: ignore71 if parsed_url.scheme and parsed_url.scheme.lower() not in ["http", "https", "file"]: # type: ignore72 return "%s%s%s" % (prefix, text, suffix)73 href = urlunparse(parsed_url._replace(path=quote(unquote(parsed_url.path)))) # type: ignore74 except ValueError: # It's not clear if this ever gets thrown75 return "%s%s%s" % (prefix, text, suffix)76 77 # For the replacement see #29: text nodes underscores are escaped78 if (79 self.options["autolinks"]80 and text.replace(r"\_", "_") == href81 and not title82 and not self.options["default_title"]83 ):84 # Shortcut syntax85 return "<%s>" % href86 if self.options["default_title"] and not title:87 title = href88 title_part = ' "%s"' % title.replace('"', r"\"") if title else ""89 return "%s[%s](%s%s)%s" % (prefix, text, href, title_part, suffix) if href else text90 91 def convert_img(self, el: Any, text: str, convert_as_inline: bool) -> str:92 """Same as usual converter, but removes data URIs"""93 94 alt = el.attrs.get("alt", None) or ""95 src = el.attrs.get("src", None) or ""96 title = el.attrs.get("title", None) or ""97 title_part = ' "%s"' % title.replace('"', r"\"") if title else ""98 if convert_as_inline and el.parent.name not in self.options["keep_inline_images_in"]:99 return alt100 101 # Remove dataURIs102 if src.startswith("data:"):103 src = src.split(",")[0] + "..."104 105 # TODO add options106 src="#"107 return "" % (alt, src, title_part)108 109 def convert_soup(self, soup: Any) -> str:110 return super().convert_soup(soup) # type: ignore111 112 113class DocumentConverterResult:114 """The result of converting a document to text."""115 116 def __init__(self, title: Union[str, None] = None, text_content: str = ""):117 self.title: Union[str, None] = title118 self.text_content: str = text_content119 120 121class DocumentConverter:122 """Abstract superclass of all DocumentConverters."""123 124 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:125 raise NotImplementedError()126 127 128class PlainTextConverter(DocumentConverter):129 """Anything with content type text/plain"""130 131 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:132 # Guess the content type from any file extension that might be around133 content_type, _ = mimetypes.guess_type("__placeholder" + kwargs.get("file_extension", ""))134 135 # Only accept text files136 if content_type is None:137 return None138 # elif "text/" not in content_type.lower():139 # return None140 141 text_content = ""142 with open(local_path, "rt", encoding="utf-8") as fh:143 text_content = fh.read()144 return DocumentConverterResult(145 title=None,146 text_content=text_content,147 )148 149 150class HtmlConverter(DocumentConverter):151 """Anything with content type text/html"""152 153 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:154 # Bail if not html155 extension = kwargs.get("file_extension", "")156 if extension.lower() not in [".html", ".htm"]:157 return None158 159 result = None160 with open(local_path, "rt", encoding="utf-8") as fh:161 result = self._convert(fh.read())162 163 return result164 165 def _convert(self, html_content: str) -> Union[None, DocumentConverterResult]:166 """Helper function that converts and HTML string."""167 168 # Parse the string169 soup = BeautifulSoup(html_content, "html.parser")170 171 # Remove javascript and style blocks172 for script in soup(["script", "style"]):173 script.extract()174 175 # Print only the main content176 body_elm = soup.find("body")177 webpage_text = ""178 if body_elm:179 webpage_text = _CustomMarkdownify().convert_soup(body_elm)180 else:181 webpage_text = _CustomMarkdownify().convert_soup(soup)182 183 assert isinstance(webpage_text, str)184 185 return DocumentConverterResult(186 title=None if soup.title is None else soup.title.string, text_content=webpage_text187 )188 189 190class WikipediaConverter(DocumentConverter):191 """Handle Wikipedia pages separately, focusing only on the main document content."""192 193 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:194 # Bail if not Wikipedia195 extension = kwargs.get("file_extension", "")196 if extension.lower() not in [".html", ".htm"]:197 return None198 url = kwargs.get("url", "")199 if not re.search(r"^https?:\/\/[a-zA-Z]{2,3}\.wikipedia.org\/", url):200 return None201 202 # Parse the file203 soup = None204 with open(local_path, "rt", encoding="utf-8") as fh:205 soup = BeautifulSoup(fh.read(), "html.parser")206 207 # Remove javascript and style blocks208 for script in soup(["script", "style"]):209 script.extract()210 211 # Print only the main content212 body_elm = soup.find("div", {"id": "mw-content-text"})213 title_elm = soup.find("span", {"class": "mw-page-title-main"})214 215 webpage_text = ""216 main_title = None if soup.title is None else soup.title.string217 218 if body_elm:219 # What's the title220 if title_elm and len(title_elm) > 0:221 main_title = title_elm.string # type: ignore222 assert isinstance(main_title, str)223 224 # Convert the page225 webpage_text = f"# {main_title}\n\n" + _CustomMarkdownify().convert_soup(body_elm)226 else:227 webpage_text = _CustomMarkdownify().convert_soup(soup)228 229 return DocumentConverterResult(230 title=main_title,231 text_content=webpage_text,232 )233 234 235class YouTubeConverter(DocumentConverter):236 """Handle YouTube specially, focusing on the video title, description, and transcript."""237 238 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:239 # Bail if not YouTube240 extension = kwargs.get("file_extension", "")241 if extension.lower() not in [".html", ".htm"]:242 return None243 url = kwargs.get("url", "")244 if not url.startswith("https://www.youtube.com/watch?"):245 return None246 247 # Parse the file248 soup = None249 with open(local_path, "rt", encoding="utf-8") as fh:250 soup = BeautifulSoup(fh.read(), "html.parser")251 252 # Read the meta tags253 assert soup.title is not None and soup.title.string is not None254 metadata: Dict[str, str] = {"title": soup.title.string}255 for meta in soup(["meta"]):256 for a in meta.attrs:257 if a in ["itemprop", "property", "name"]:258 metadata[meta[a]] = meta.get("content", "")259 break260 261 # We can also try to read the full description. This is more prone to breaking, since it reaches into the page implementation262 try:263 for script in soup(["script"]):264 content = script.text265 if "ytInitialData" in content:266 lines = re.split(r"\r?\n", content)267 obj_start = lines[0].find("{")268 obj_end = lines[0].rfind("}")269 if obj_start >= 0 and obj_end >= 0:270 data = json.loads(lines[0][obj_start : obj_end + 1])271 attrdesc = self._findKey(data, "attributedDescriptionBodyText") # type: ignore272 if attrdesc:273 metadata["description"] = str(attrdesc["content"])274 break275 except Exception:276 pass277 278 # Start preparing the page279 webpage_text = "# YouTube\n"280 281 title = self._get(metadata, ["title", "og:title", "name"]) # type: ignore282 assert isinstance(title, str)283 284 if title:285 webpage_text += f"\n## {title}\n"286 287 stats = ""288 views = self._get(metadata, ["interactionCount"]) # type: ignore289 if views:290 stats += f"- **Views:** {views}\n"291 292 keywords = self._get(metadata, ["keywords"]) # type: ignore293 if keywords:294 stats += f"- **Keywords:** {keywords}\n"295 296 runtime = self._get(metadata, ["duration"]) # type: ignore297 if runtime:298 stats += f"- **Runtime:** {runtime}\n"299 300 if len(stats) > 0:301 webpage_text += f"\n### Video Metadata\n{stats}\n"302 303 description = self._get(metadata, ["description", "og:description"]) # type: ignore304 if description:305 webpage_text += f"\n### Description\n{description}\n"306 307 transcript_text = ""308 parsed_url = urlparse(url) # type: ignore309 params = parse_qs(parsed_url.query) # type: ignore310 if "v" in params:311 assert isinstance(params["v"][0], str)312 video_id = str(params["v"][0])313 try:314 # Must be a single transcript.315 transcript = YouTubeTranscriptApi.get_transcript(video_id) # type: ignore316 # transcript_text = " ".join([part["text"] for part in transcript]) # type: ignore317 # Alternative formatting:318 transcript_text = SRTFormatter().format_transcript(transcript)319 except Exception:320 pass321 if transcript_text:322 webpage_text += f"\n### Transcript\n{transcript_text}\n"323 324 title = title if title else soup.title.string325 assert isinstance(title, str)326 327 return DocumentConverterResult(328 title=title,329 text_content=webpage_text,330 )331 332 def _get(self, metadata: Dict[str, str], keys: List[str], default: Union[str, None] = None) -> Union[str, None]:333 for k in keys:334 if k in metadata:335 return metadata[k]336 return default337 338 def _findKey(self, json: Any, key: str) -> Union[str, None]: # TODO: Fix json type339 if isinstance(json, list):340 for elm in json:341 ret = self._findKey(elm, key)342 if ret is not None:343 return ret344 elif isinstance(json, dict):345 for k in json:346 if k == key:347 return json[k]348 else:349 ret = self._findKey(json[k], key)350 if ret is not None:351 return ret352 return None353 354 355class PdfConverter(DocumentConverter):356 """357 Converts PDFs to Markdown. Most style information is ignored, so the results are essentially plain-text.358 """359 360 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:361 # Bail if not a PDF362 extension = kwargs.get("file_extension", "")363 if extension.lower() != ".pdf":364 return None365 366 return DocumentConverterResult(367 title=None,368 text_content=pdfminer.high_level.extract_text(local_path),369 )370 371 372class DocxConverter(HtmlConverter):373 """374 Converts DOCX files to Markdown. Style information (e.g.m headings) and tables are preserved where possible.375 """376 377 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:378 # Bail if not a DOCX379 extension = kwargs.get("file_extension", "")380 if extension.lower() != ".docx":381 return None382 383 result = None384 with open(local_path, "rb") as docx_file:385 result = mammoth.convert_to_html(docx_file)386 html_content = result.value387 result = self._convert(html_content)388 389 return result390 391 392class XlsxConverter(HtmlConverter):393 """394 Converts XLSX files to Markdown, with each sheet presented as a separate Markdown table.395 """396 397 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:398 # Bail if not a XLSX399 extension = kwargs.get("file_extension", "")400 if extension.lower() not in [".xlsx", ".xls"]:401 return None402 403 sheets = pd.read_excel(local_path, sheet_name=None)404 md_content = ""405 for s in sheets:406 md_content += f"## {s}\n"407 html_content = sheets[s].to_html(index=False)408 md_content += self._convert(html_content).text_content.strip() + "\n\n"409 410 return DocumentConverterResult(411 title=None,412 text_content=md_content.strip(),413 )414 415 416class PptxConverter(HtmlConverter):417 """418 Converts PPTX files to Markdown. Supports heading, tables and images with alt text.419 """420 421 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:422 # Bail if not a PPTX423 extension = kwargs.get("file_extension", "")424 if extension.lower() != ".pptx":425 return None426 427 md_content = ""428 429 presentation = pptx.Presentation(local_path)430 slide_num = 0431 for slide in presentation.slides:432 slide_num += 1433 434 md_content += f"\n\n<!-- Slide number: {slide_num} -->\n"435 436 title = slide.shapes.title437 for shape in slide.shapes:438 # Pictures439 if self._is_picture(shape):440 # https://github.com/scanny/python-pptx/pull/512#issuecomment-1713100069441 alt_text = ""442 try:443 alt_text = shape._element._nvXxPr.cNvPr.attrib.get("descr", "")444 except Exception:445 pass446 447 # A placeholder name448 filename = re.sub(r"\W", "", shape.name) + ".jpg"449 md_content += "\n![" + (alt_text if alt_text else shape.name) + "](" + filename + ")\n"450 451 # Tables452 if self._is_table(shape):453 html_table = "<html><body><table>"454 first_row = True455 for row in shape.table.rows:456 html_table += "<tr>"457 for cell in row.cells:458 if first_row:459 html_table += "<th>" + html.escape(cell.text) + "</th>"460 else:461 html_table += "<td>" + html.escape(cell.text) + "</td>"462 html_table += "</tr>"463 first_row = False464 html_table += "</table></body></html>"465 md_content += "\n" + self._convert(html_table).text_content.strip() + "\n"466 467 # Text areas468 elif shape.has_text_frame:469 if shape == title:470 md_content += "# " + shape.text.lstrip() + "\n"471 else:472 md_content += shape.text + "\n"473 474 md_content = md_content.strip()475 476 if slide.has_notes_slide:477 md_content += "\n\n### Notes:\n"478 notes_frame = slide.notes_slide.notes_text_frame479 if notes_frame is not None:480 md_content += notes_frame.text481 md_content = md_content.strip()482 483 return DocumentConverterResult(484 title=None,485 text_content=md_content.strip(),486 )487 488 def _is_picture(self, shape):489 if shape.shape_type == pptx.enum.shapes.MSO_SHAPE_TYPE.PICTURE:490 return True491 if shape.shape_type == pptx.enum.shapes.MSO_SHAPE_TYPE.PLACEHOLDER:492 if hasattr(shape, "image"):493 return True494 return False495 496 def _is_table(self, shape):497 if shape.shape_type == pptx.enum.shapes.MSO_SHAPE_TYPE.TABLE:498 return True499 return False500 501 502class MediaConverter(DocumentConverter):503 """504 Abstract class for multi-modal media (e.g., images and audio)505 """506 507 def _get_metadata(self, local_path):508 exiftool = shutil.which("exiftool")509 if not exiftool:510 return None511 else:512 try:513 result = subprocess.run([exiftool, "-json", local_path], capture_output=True, text=True).stdout514 return json.loads(result)[0]515 except Exception:516 return None517 518 519class WavConverter(MediaConverter):520 """521 Converts WAV files to markdown via extraction of metadata (if `exiftool` is installed), and speech transcription (if `speech_recognition` is installed).522 """523 524 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:525 # Bail if not a XLSX526 extension = kwargs.get("file_extension", "")527 if extension.lower() != ".wav":528 return None529 530 md_content = ""531 532 # Add metadata533 metadata = self._get_metadata(local_path)534 if metadata:535 for f in [536 "Title",537 "Artist",538 "Author",539 "Band",540 "Album",541 "Genre",542 "Track",543 "DateTimeOriginal",544 "CreateDate",545 "Duration",546 ]:547 if f in metadata:548 md_content += f"{f}: {metadata[f]}\n"549 550 # Transcribe551 try:552 transcript = self._transcribe_audio(local_path)553 md_content += "\n\n### Audio Transcript:\n" + ("[No speech detected]" if transcript == "" else transcript)554 except Exception:555 md_content += "\n\n### Audio Transcript:\nError. Could not transcribe this audio."556 557 return DocumentConverterResult(558 title=None,559 text_content=md_content.strip(),560 )561 562 def _transcribe_audio(self, local_path) -> str:563 recognizer = sr.Recognizer()564 with sr.AudioFile(local_path) as source:565 audio = recognizer.record(source)566 return recognizer.recognize_google(audio).strip()567 568 569class Mp3Converter(WavConverter):570 """571 Converts MP3 files to markdown via extraction of metadata (if `exiftool` is installed), and speech transcription (if `speech_recognition` AND `pydub` are installed).572 """573 574 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:575 # Bail if not a MP3576 extension = kwargs.get("file_extension", "")577 if extension.lower() != ".mp3":578 return None579 580 md_content = ""581 582 # Add metadata583 metadata = self._get_metadata(local_path)584 if metadata:585 for f in [586 "Title",587 "Artist",588 "Author",589 "Band",590 "Album",591 "Genre",592 "Track",593 "DateTimeOriginal",594 "CreateDate",595 "Duration",596 ]:597 if f in metadata:598 md_content += f"{f}: {metadata[f]}\n"599 600 # Transcribe601 handle, temp_path = tempfile.mkstemp(suffix=".wav")602 os.close(handle)603 try:604 sound = pydub.AudioSegment.from_mp3(local_path)605 sound.export(temp_path, format="wav")606 607 _args = dict()608 _args.update(kwargs)609 _args["file_extension"] = ".wav"610 611 try:612 transcript = super()._transcribe_audio(temp_path).strip()613 md_content += "\n\n### Audio Transcript:\n" + (614 "[No speech detected]" if transcript == "" else transcript615 )616 except Exception:617 md_content += "\n\n### Audio Transcript:\nError. Could not transcribe this audio."618 619 finally:620 os.unlink(temp_path)621 622 # Return the result623 return DocumentConverterResult(624 title=None,625 text_content=md_content.strip(),626 )627 628 629class ImageConverter(MediaConverter):630 """631 Converts images to markdown via extraction of metadata (if `exiftool` is installed), OCR (if `easyocr` is installed), and description via a multimodal LLM (if an mlm_client is configured).632 """633 634 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:635 # Bail if not a XLSX636 extension = kwargs.get("file_extension", "")637 if extension.lower() not in [".jpg", ".jpeg", ".png"]:638 return None639 640 md_content = ""641 642 # Add metadata643 metadata = self._get_metadata(local_path)644 if metadata:645 for f in [646 "ImageSize",647 "Title",648 "Caption",649 "Description",650 "Keywords",651 "Artist",652 "Author",653 "DateTimeOriginal",654 "CreateDate",655 "GPSPosition",656 ]:657 if f in metadata:658 md_content += f"{f}: {metadata[f]}\n"659 660 # Try describing the image with GPTV661 mlm_client = kwargs.get("mlm_client")662 mlm_model = kwargs.get("mlm_model")663 if mlm_client is not None and mlm_model is not None:664 md_content += (665 "\n# Description:\n"666 + self._get_mlm_description(667 local_path, extension, mlm_client, mlm_model, prompt=kwargs.get("mlm_prompt")668 ).strip()669 + "\n"670 )671 672 return DocumentConverterResult(673 title=None,674 text_content=md_content,675 )676 677 def _get_mlm_description(self, local_path, extension, client, model, prompt=None):678 if prompt is None or prompt.strip() == "":679 prompt = "Write a detailed caption for this image."680 681 sys.stderr.write(f"MLM Prompt:\n{prompt}\n")682 683 data_uri = ""684 with open(local_path, "rb") as image_file:685 content_type, encoding = mimetypes.guess_type("_dummy" + extension)686 if content_type is None:687 content_type = "image/jpeg"688 image_base64 = base64.b64encode(image_file.read()).decode("utf-8")689 data_uri = f"data:{content_type};base64,{image_base64}"690 691 messages = [692 {693 "role": "user",694 "content": [695 {"type": "text", "text": prompt},696 {697 "type": "image_url",698 "image_url": {699 "url": data_uri,700 },701 },702 ],703 }704 ]705 706 response = client.chat.completions.create(model=model, messages=messages)707 return response.choices[0].message.content708 709 710class FileConversionException(BaseException):711 pass712 713 714class UnsupportedFormatException(BaseException):715 pass716 717 718class MarkdownConverter:719 """(In preview) An extremely simple text-based document reader, suitable for LLM use.720 This reader will convert common file-types or webpages to Markdown."""721 722 def __init__(723 self,724 requests_session: Optional[requests.Session] = None,725 mlm_client: Optional[Any] = None,726 mlm_model: Optional[Any] = None,727 ):728 if requests_session is None:729 self._requests_session = requests.Session()730 else:731 self._requests_session = requests_session732 733 self._mlm_client = mlm_client734 self._mlm_model = mlm_model735 736 self._page_converters: List[DocumentConverter] = []737 738 # Register converters for successful browsing operations739 # Later registrations are tried first / take higher priority than earlier registrations740 # To this end, the most specific converters should appear below the most generic converters741 self.register_page_converter(PlainTextConverter())742 self.register_page_converter(HtmlConverter())743 self.register_page_converter(WikipediaConverter())744 self.register_page_converter(YouTubeConverter())745 self.register_page_converter(DocxConverter())746 self.register_page_converter(XlsxConverter())747 self.register_page_converter(PptxConverter())748 self.register_page_converter(WavConverter())749 self.register_page_converter(Mp3Converter())750 self.register_page_converter(ImageConverter())751 self.register_page_converter(PdfConverter())752 753 def convert(754 self, source: Union[str, requests.Response], **kwargs: Any755 ) -> DocumentConverterResult: # TODO: deal with kwargs756 """757 Args:758 - source: can be a string representing a path or url, or a requests.response object759 - extension: specifies the file extension to use when interpreting the file. If None, infer from source (path, uri, content-type, etc.)760 """761 762 # Local path or url763 if isinstance(source, str):764 if source.startswith("http://") or source.startswith("https://") or source.startswith("file://"):765 return self.convert_url(source, **kwargs)766 else:767 return self.convert_local(source, **kwargs)768 # Request response769 elif isinstance(source, requests.Response):770 return self.convert_response(source, **kwargs)771 772 def convert_local(self, path: str, **kwargs: Any) -> DocumentConverterResult: # TODO: deal with kwargs773 # Prepare a list of extensions to try (in order of priority)774 ext = kwargs.get("file_extension")775 extensions = [ext] if ext is not None else []776 777 # Get extension alternatives from the path and puremagic778 base, ext = os.path.splitext(path)779 self._append_ext(extensions, ext)780 self._append_ext(extensions, self._guess_ext_magic(path))781 782 # Convert783 return self._convert(path, extensions, **kwargs)784 785 # TODO what should stream's type be?786 def convert_stream(self, stream: Any, **kwargs: Any) -> DocumentConverterResult: # TODO: deal with kwargs787 # Prepare a list of extensions to try (in order of priority)788 ext = kwargs.get("file_extension")789 extensions = [ext] if ext is not None else []790 791 # Save the file locally to a temporary file. It will be deleted before this method exits792 handle, temp_path = tempfile.mkstemp()793 fh = os.fdopen(handle, "wb")794 result = None795 try:796 # Write to the temporary file797 content = stream.read()798 if isinstance(content, str):799 fh.write(content.encode("utf-8"))800 else:801 fh.write(content)802 fh.close()803 804 # Use puremagic to check for more extension options805 self._append_ext(extensions, self._guess_ext_magic(temp_path))806 807 # Convert808 result = self._convert(temp_path, extensions, **kwargs)809 # Clean up810 finally:811 try:812 fh.close()813 except Exception:814 pass815 os.unlink(temp_path)816 817 return result818 819 def convert_url(self, url: str, **kwargs: Any) -> DocumentConverterResult: # TODO: fix kwargs type820 # Send a HTTP request to the URL821 user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/119.0.0.0 Safari/537.36 Edg/119.0.0.0"822 response = self._requests_session.get(url, stream=True, headers={"User-Agent": user_agent})823 response.raise_for_status()824 return self.convert_response(response, **kwargs)825 826 def convert_response(827 self, response: requests.Response, **kwargs: Any828 ) -> DocumentConverterResult: # TODO fix kwargs type829 # Prepare a list of extensions to try (in order of priority)830 ext = kwargs.get("file_extension")831 extensions = [ext] if ext is not None else []832 833 # Guess from the mimetype834 content_type = response.headers.get("content-type", "").split(";")[0]835 self._append_ext(extensions, mimetypes.guess_extension(content_type))836 837 # Read the content disposition if there is one838 content_disposition = response.headers.get("content-disposition", "")839 m = re.search(r"filename=([^;]+)", content_disposition)840 if m:841 base, ext = os.path.splitext(m.group(1).strip("\"'"))842 self._append_ext(extensions, ext)843 844 # Read from the extension from the path845 base, ext = os.path.splitext(urlparse(response.url).path)846 self._append_ext(extensions, ext)847 848 # Save the file locally to a temporary file. It will be deleted before this method exits849 handle, temp_path = tempfile.mkstemp()850 fh = os.fdopen(handle, "wb")851 result = None852 try:853 # Download the file854 for chunk in response.iter_content(chunk_size=512):855 fh.write(chunk)856 fh.close()857 858 # Use puremagic to check for more extension options859 self._append_ext(extensions, self._guess_ext_magic(temp_path))860 861 # Convert862 result = self._convert(temp_path, extensions, url=response.url)863 except Exception as e:864 print(f"Error in converting: {e}")865 866 # Clean up867 finally:868 try:869 fh.close()870 except Exception:871 pass872 os.unlink(temp_path)873 874 return result875 876 def _convert(self, local_path: str, extensions: List[Union[str, None]], **kwargs) -> DocumentConverterResult:877 error_trace = ""878 for ext in extensions + [None]: # Try last with no extension879 for converter in self._page_converters:880 _kwargs = copy.deepcopy(kwargs)881 882 # Overwrite file_extension appropriately883 if ext is None:884 if "file_extension" in _kwargs:885 del _kwargs["file_extension"]886 else:887 _kwargs.update({"file_extension": ext})888 889 # Copy any additional global options890 if "mlm_client" not in _kwargs and self._mlm_client is not None:891 _kwargs["mlm_client"] = self._mlm_client892 893 if "mlm_model" not in _kwargs and self._mlm_model is not None:894 _kwargs["mlm_model"] = self._mlm_model895 896 # If we hit an error log it and keep trying897 try:898 res = converter.convert(local_path, **_kwargs)899 except Exception:900 error_trace = ("\n\n" + traceback.format_exc()).strip()901 902 if res is not None:903 # Normalize the content904 res.text_content = "\n".join([line.rstrip() for line in re.split(r"\r?\n", res.text_content)])905 res.text_content = re.sub(r"\n{3,}", "\n\n", res.text_content)906 907 # Todo908 return res909 910 # If we got this far without success, report any exceptions911 if len(error_trace) > 0:912 raise FileConversionException(913 f"Could not convert '{local_path}' to Markdown. File type was recognized as {extensions}. While converting the file, the following error was encountered:\n\n{error_trace}"914 )915 916 # Nothing can handle it!917 raise UnsupportedFormatException(918 f"Could not convert '{local_path}' to Markdown. The formats {extensions} are not supported."919 )920 921 def _append_ext(self, extensions, ext):922 """Append a unique non-None, non-empty extension to a list of extensions."""923 if ext is None:924 return925 ext = ext.strip()926 if ext == "":927 return928 # if ext not in extensions:929 if True:930 extensions.append(ext)931 932 def _guess_ext_magic(self, path):933 """Use puremagic (a Python implementation of libmagic) to guess a file's extension based on the first few bytes."""934 # Use puremagic to guess935 try:936 guesses = puremagic.magic_file(path)937 if len(guesses) > 0:938 ext = guesses[0].extension.strip()939 if len(ext) > 0:940 return ext941 except FileNotFoundError:942 pass943 except IsADirectoryError:944 pass945 except PermissionError:946 pass947 return None948 949 def register_page_converter(self, converter: DocumentConverter) -> None:950 """Register a page text converter."""951 self._page_converters.insert(0, converter)952