Akjava/open_Deep-Research-DuckDuckGo
4
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 # TODO options105 src="#"106 return "" % (alt, src, title_part)107 108 def convert_soup(self, soup: Any) -> str:109 return super().convert_soup(soup) # type: ignore110 111 112class DocumentConverterResult:113 """The result of converting a document to text."""114 115 def __init__(self, title: Union[str, None] = None, text_content: str = ""):116 self.title: Union[str, None] = title117 self.text_content: str = text_content118 119 120class DocumentConverter:121 """Abstract superclass of all DocumentConverters."""122 123 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:124 raise NotImplementedError()125 126 127class PlainTextConverter(DocumentConverter):128 """Anything with content type text/plain"""129 130 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:131 # Guess the content type from any file extension that might be around132 content_type, _ = mimetypes.guess_type("__placeholder" + kwargs.get("file_extension", ""))133 134 # Only accept text files135 if content_type is None:136 return None137 # elif "text/" not in content_type.lower():138 # return None139 140 text_content = ""141 with open(local_path, "rt", encoding="utf-8") as fh:142 text_content = fh.read()143 return DocumentConverterResult(144 title=None,145 text_content=text_content,146 )147 148 149class HtmlConverter(DocumentConverter):150 """Anything with content type text/html"""151 152 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:153 # Bail if not html154 extension = kwargs.get("file_extension", "")155 if extension.lower() not in [".html", ".htm"]:156 return None157 158 result = None159 with open(local_path, "rt", encoding="utf-8") as fh:160 result = self._convert(fh.read())161 162 return result163 164 def _convert(self, html_content: str) -> Union[None, DocumentConverterResult]:165 """Helper function that converts and HTML string."""166 167 # Parse the string168 soup = BeautifulSoup(html_content, "html.parser")169 170 # Remove javascript and style blocks171 for script in soup(["script", "style"]):172 script.extract()173 174 # Print only the main content175 body_elm = soup.find("body")176 webpage_text = ""177 if body_elm:178 webpage_text = _CustomMarkdownify().convert_soup(body_elm)179 else:180 webpage_text = _CustomMarkdownify().convert_soup(soup)181 182 assert isinstance(webpage_text, str)183 184 return DocumentConverterResult(185 title=None if soup.title is None else soup.title.string, text_content=webpage_text186 )187 188 189class WikipediaConverter(DocumentConverter):190 """Handle Wikipedia pages separately, focusing only on the main document content."""191 192 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:193 # Bail if not Wikipedia194 extension = kwargs.get("file_extension", "")195 if extension.lower() not in [".html", ".htm"]:196 return None197 url = kwargs.get("url", "")198 if not re.search(r"^https?:\/\/[a-zA-Z]{2,3}\.wikipedia.org\/", url):199 return None200 201 # Parse the file202 soup = None203 with open(local_path, "rt", encoding="utf-8") as fh:204 soup = BeautifulSoup(fh.read(), "html.parser")205 206 # Remove javascript and style blocks207 for script in soup(["script", "style"]):208 script.extract()209 210 # Print only the main content211 body_elm = soup.find("div", {"id": "mw-content-text"})212 title_elm = soup.find("span", {"class": "mw-page-title-main"})213 214 webpage_text = ""215 main_title = None if soup.title is None else soup.title.string216 217 if body_elm:218 # What's the title219 if title_elm and len(title_elm) > 0:220 main_title = title_elm.string # type: ignore221 assert isinstance(main_title, str)222 223 # Convert the page224 webpage_text = f"# {main_title}\n\n" + _CustomMarkdownify().convert_soup(body_elm)225 else:226 webpage_text = _CustomMarkdownify().convert_soup(soup)227 228 return DocumentConverterResult(229 title=main_title,230 text_content=webpage_text,231 )232 233 234class YouTubeConverter(DocumentConverter):235 """Handle YouTube specially, focusing on the video title, description, and transcript."""236 237 def convert(self, local_path: str, **kwargs: Any) -> Union[None, DocumentConverterResult]:238 # Bail if not YouTube239 extension = kwargs.get("file_extension", "")240 if extension.lower() not in [".html", ".htm"]:241 return None242 url = kwargs.get("url", "")243 if not url.startswith("https://www.youtube.com/watch?"):244 return None245 246 # Parse the file247 soup = None248 with open(local_path, "rt", encoding="utf-8") as fh:249 soup = BeautifulSoup(fh.read(), "html.parser")250 251 # Read the meta tags252 assert soup.title is not None and soup.title.string is not None253 metadata: Dict[str, str] = {"title": soup.title.string}254 for meta in soup(["meta"]):255 for a in meta.attrs:256 if a in ["itemprop", "property", "name"]:257 metadata[meta[a]] = meta.get("content", "")258 break259 260 # We can also try to read the full description. This is more prone to breaking, since it reaches into the page implementation261 try:262 for script in soup(["script"]):263 content = script.text264 if "ytInitialData" in content:265 lines = re.split(r"\r?\n", content)266 obj_start = lines[0].find("{")267 obj_end = lines[0].rfind("}")268 if obj_start >= 0 and obj_end >= 0:269 data = json.loads(lines[0][obj_start : obj_end + 1])270 attrdesc = self._findKey(data, "attributedDescriptionBodyText") # type: ignore271 if attrdesc:272 metadata["description"] = str(attrdesc["content"])273 break274 except Exception:275 pass276 277 # Start preparing the page278 webpage_text = "# YouTube\n"279 280 title = self._get(metadata, ["title", "og:title", "name"]) # type: ignore281 assert isinstance(title, str)282 283 if title:284 webpage_text += f"\n## {title}\n"285 286 stats = ""287 views = self._get(metadata, ["interactionCount"]) # type: ignore288 if views:289 stats += f"- **Views:** {views}\n"290 291 keywords = self._get(metadata, ["keywords"]) # type: ignore292 if keywords:293 stats += f"- **Keywords:** {keywords}\n"294 295 runtime = self._get(metadata, ["duration"]) # type: ignore296 if runtime:297 stats += f"- **Runtime:** {runtime}\n"298 299 if len(stats) > 0:300 webpage_text += f"\n### Video Metadata\n{stats}\n"301 302 description = self._get(metadata, ["description", "og:description"]) # type: ignore303 if description:304 webpage_text += f"\n### Description\n{description}\n"305 306 transcript_text = ""307 parsed_url = urlparse(url) # type: ignore308 params = parse_qs(parsed_url.query) # type: ignore309 if "v" in params:310 assert isinstance(params["v"][0], str)311 video_id = str(params["v"][0])312 try:313 # Must be a single transcript.314 transcript = YouTubeTranscriptApi.get_transcript(video_id) # type: ignore315 # transcript_text = " ".join([part["text"] for part in transcript]) # type: ignore316 # Alternative formatting:317 transcript_text = SRTFormatter().format_transcript(transcript)318 except Exception:319 pass320 if transcript_text:321 webpage_text += f"\n### Transcript\n{transcript_text}\n"322 323 title = title if title else soup.title.string324 assert isinstance(title, str)325 326 return DocumentConverterResult(327 title=title,328 text_content=webpage_text,329 )330 331 def _get(self, metadata: Dict[str, str], keys: List[str], default: Union[str, None] = None) -> Union[str, None]:332 for k in keys:333 if k in metadata:334 return metadata[k]335 return default336 337 def _findKey(self, json: Any, key: str) -> Union[str, None]: # TODO: Fix json type338 if isinstance(json, list):339 for elm in json:340 ret = self._findKey(elm, key)341 if ret is not None:342 return ret343 elif isinstance(json, dict):344 for k in json:345 if k == key:346 return json[k]347 else:348 ret = self._findKey(json[k], key)349 if ret is not None:350 return ret351 return None352 353 354class PdfConverter(DocumentConverter):355 """356 Converts PDFs to Markdown. Most style information is ignored, so the results are essentially plain-text.357 """358 359 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:360 # Bail if not a PDF361 extension = kwargs.get("file_extension", "")362 if extension.lower() != ".pdf":363 return None364 365 return DocumentConverterResult(366 title=None,367 text_content=pdfminer.high_level.extract_text(local_path),368 )369 370 371class DocxConverter(HtmlConverter):372 """373 Converts DOCX files to Markdown. Style information (e.g.m headings) and tables are preserved where possible.374 """375 376 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:377 # Bail if not a DOCX378 extension = kwargs.get("file_extension", "")379 if extension.lower() != ".docx":380 return None381 382 result = None383 with open(local_path, "rb") as docx_file:384 result = mammoth.convert_to_html(docx_file)385 html_content = result.value386 result = self._convert(html_content)387 388 return result389 390 391class XlsxConverter(HtmlConverter):392 """393 Converts XLSX files to Markdown, with each sheet presented as a separate Markdown table.394 """395 396 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:397 # Bail if not a XLSX398 extension = kwargs.get("file_extension", "")399 if extension.lower() not in [".xlsx", ".xls"]:400 return None401 402 sheets = pd.read_excel(local_path, sheet_name=None)403 md_content = ""404 for s in sheets:405 md_content += f"## {s}\n"406 html_content = sheets[s].to_html(index=False)407 md_content += self._convert(html_content).text_content.strip() + "\n\n"408 409 return DocumentConverterResult(410 title=None,411 text_content=md_content.strip(),412 )413 414 415class PptxConverter(HtmlConverter):416 """417 Converts PPTX files to Markdown. Supports heading, tables and images with alt text.418 """419 420 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:421 # Bail if not a PPTX422 extension = kwargs.get("file_extension", "")423 if extension.lower() != ".pptx":424 return None425 426 md_content = ""427 428 presentation = pptx.Presentation(local_path)429 slide_num = 0430 for slide in presentation.slides:431 slide_num += 1432 433 md_content += f"\n\n<!-- Slide number: {slide_num} -->\n"434 435 title = slide.shapes.title436 for shape in slide.shapes:437 # Pictures438 if self._is_picture(shape):439 # https://github.com/scanny/python-pptx/pull/512#issuecomment-1713100069440 alt_text = ""441 try:442 alt_text = shape._element._nvXxPr.cNvPr.attrib.get("descr", "")443 except Exception:444 pass445 446 # A placeholder name447 filename = re.sub(r"\W", "", shape.name) + ".jpg"448 md_content += "\n![" + (alt_text if alt_text else shape.name) + "](" + filename + ")\n"449 450 # Tables451 if self._is_table(shape):452 html_table = "<html><body><table>"453 first_row = True454 for row in shape.table.rows:455 html_table += "<tr>"456 for cell in row.cells:457 if first_row:458 html_table += "<th>" + html.escape(cell.text) + "</th>"459 else:460 html_table += "<td>" + html.escape(cell.text) + "</td>"461 html_table += "</tr>"462 first_row = False463 html_table += "</table></body></html>"464 md_content += "\n" + self._convert(html_table).text_content.strip() + "\n"465 466 # Text areas467 elif shape.has_text_frame:468 if shape == title:469 md_content += "# " + shape.text.lstrip() + "\n"470 else:471 md_content += shape.text + "\n"472 473 md_content = md_content.strip()474 475 if slide.has_notes_slide:476 md_content += "\n\n### Notes:\n"477 notes_frame = slide.notes_slide.notes_text_frame478 if notes_frame is not None:479 md_content += notes_frame.text480 md_content = md_content.strip()481 482 return DocumentConverterResult(483 title=None,484 text_content=md_content.strip(),485 )486 487 def _is_picture(self, shape):488 if shape.shape_type == pptx.enum.shapes.MSO_SHAPE_TYPE.PICTURE:489 return True490 if shape.shape_type == pptx.enum.shapes.MSO_SHAPE_TYPE.PLACEHOLDER:491 if hasattr(shape, "image"):492 return True493 return False494 495 def _is_table(self, shape):496 if shape.shape_type == pptx.enum.shapes.MSO_SHAPE_TYPE.TABLE:497 return True498 return False499 500 501class MediaConverter(DocumentConverter):502 """503 Abstract class for multi-modal media (e.g., images and audio)504 """505 506 def _get_metadata(self, local_path):507 exiftool = shutil.which("exiftool")508 if not exiftool:509 return None510 else:511 try:512 result = subprocess.run([exiftool, "-json", local_path], capture_output=True, text=True).stdout513 return json.loads(result)[0]514 except Exception:515 return None516 517 518class WavConverter(MediaConverter):519 """520 Converts WAV files to markdown via extraction of metadata (if `exiftool` is installed), and speech transcription (if `speech_recognition` is installed).521 """522 523 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:524 # Bail if not a XLSX525 extension = kwargs.get("file_extension", "")526 if extension.lower() != ".wav":527 return None528 529 md_content = ""530 531 # Add metadata532 metadata = self._get_metadata(local_path)533 if metadata:534 for f in [535 "Title",536 "Artist",537 "Author",538 "Band",539 "Album",540 "Genre",541 "Track",542 "DateTimeOriginal",543 "CreateDate",544 "Duration",545 ]:546 if f in metadata:547 md_content += f"{f}: {metadata[f]}\n"548 549 # Transcribe550 try:551 transcript = self._transcribe_audio(local_path)552 md_content += "\n\n### Audio Transcript:\n" + ("[No speech detected]" if transcript == "" else transcript)553 except Exception:554 md_content += "\n\n### Audio Transcript:\nError. Could not transcribe this audio."555 556 return DocumentConverterResult(557 title=None,558 text_content=md_content.strip(),559 )560 561 def _transcribe_audio(self, local_path) -> str:562 recognizer = sr.Recognizer()563 with sr.AudioFile(local_path) as source:564 audio = recognizer.record(source)565 return recognizer.recognize_google(audio).strip()566 567 568class Mp3Converter(WavConverter):569 """570 Converts MP3 files to markdown via extraction of metadata (if `exiftool` is installed), and speech transcription (if `speech_recognition` AND `pydub` are installed).571 """572 573 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:574 # Bail if not a MP3575 extension = kwargs.get("file_extension", "")576 if extension.lower() != ".mp3":577 return None578 579 md_content = ""580 581 # Add metadata582 metadata = self._get_metadata(local_path)583 if metadata:584 for f in [585 "Title",586 "Artist",587 "Author",588 "Band",589 "Album",590 "Genre",591 "Track",592 "DateTimeOriginal",593 "CreateDate",594 "Duration",595 ]:596 if f in metadata:597 md_content += f"{f}: {metadata[f]}\n"598 599 # Transcribe600 handle, temp_path = tempfile.mkstemp(suffix=".wav")601 os.close(handle)602 try:603 sound = pydub.AudioSegment.from_mp3(local_path)604 sound.export(temp_path, format="wav")605 606 _args = dict()607 _args.update(kwargs)608 _args["file_extension"] = ".wav"609 610 try:611 transcript = super()._transcribe_audio(temp_path).strip()612 md_content += "\n\n### Audio Transcript:\n" + (613 "[No speech detected]" if transcript == "" else transcript614 )615 except Exception:616 md_content += "\n\n### Audio Transcript:\nError. Could not transcribe this audio."617 618 finally:619 os.unlink(temp_path)620 621 # Return the result622 return DocumentConverterResult(623 title=None,624 text_content=md_content.strip(),625 )626 627 628class ImageConverter(MediaConverter):629 """630 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).631 """632 633 def convert(self, local_path, **kwargs) -> Union[None, DocumentConverterResult]:634 # Bail if not a XLSX635 extension = kwargs.get("file_extension", "")636 if extension.lower() not in [".jpg", ".jpeg", ".png"]:637 return None638 639 md_content = ""640 641 # Add metadata642 metadata = self._get_metadata(local_path)643 if metadata:644 for f in [645 "ImageSize",646 "Title",647 "Caption",648 "Description",649 "Keywords",650 "Artist",651 "Author",652 "DateTimeOriginal",653 "CreateDate",654 "GPSPosition",655 ]:656 if f in metadata:657 md_content += f"{f}: {metadata[f]}\n"658 659 # Try describing the image with GPTV660 mlm_client = kwargs.get("mlm_client")661 mlm_model = kwargs.get("mlm_model")662 if mlm_client is not None and mlm_model is not None:663 md_content += (664 "\n# Description:\n"665 + self._get_mlm_description(666 local_path, extension, mlm_client, mlm_model, prompt=kwargs.get("mlm_prompt")667 ).strip()668 + "\n"669 )670 671 return DocumentConverterResult(672 title=None,673 text_content=md_content,674 )675 676 def _get_mlm_description(self, local_path, extension, client, model, prompt=None):677 if prompt is None or prompt.strip() == "":678 prompt = "Write a detailed caption for this image."679 680 sys.stderr.write(f"MLM Prompt:\n{prompt}\n")681 682 data_uri = ""683 with open(local_path, "rb") as image_file:684 content_type, encoding = mimetypes.guess_type("_dummy" + extension)685 if content_type is None:686 content_type = "image/jpeg"687 image_base64 = base64.b64encode(image_file.read()).decode("utf-8")688 data_uri = f"data:{content_type};base64,{image_base64}"689 690 messages = [691 {692 "role": "user",693 "content": [694 {"type": "text", "text": prompt},695 {696 "type": "image_url",697 "image_url": {698 "url": data_uri,699 },700 },701 ],702 }703 ]704 705 response = client.chat.completions.create(model=model, messages=messages)706 return response.choices[0].message.content707 708 709class FileConversionException(BaseException):710 pass711 712 713class UnsupportedFormatException(BaseException):714 pass715 716 717class MarkdownConverter:718 """(In preview) An extremely simple text-based document reader, suitable for LLM use.719 This reader will convert common file-types or webpages to Markdown."""720 721 def __init__(722 self,723 requests_session: Optional[requests.Session] = None,724 mlm_client: Optional[Any] = None,725 mlm_model: Optional[Any] = None,726 ):727 if requests_session is None:728 self._requests_session = requests.Session()729 else:730 self._requests_session = requests_session731 732 self._mlm_client = mlm_client733 self._mlm_model = mlm_model734 735 self._page_converters: List[DocumentConverter] = []736 737 # Register converters for successful browsing operations738 # Later registrations are tried first / take higher priority than earlier registrations739 # To this end, the most specific converters should appear below the most generic converters740 self.register_page_converter(PlainTextConverter())741 self.register_page_converter(HtmlConverter())742 self.register_page_converter(WikipediaConverter())743 self.register_page_converter(YouTubeConverter())744 self.register_page_converter(DocxConverter())745 self.register_page_converter(XlsxConverter())746 self.register_page_converter(PptxConverter())747 self.register_page_converter(WavConverter())748 self.register_page_converter(Mp3Converter())749 self.register_page_converter(ImageConverter())750 self.register_page_converter(PdfConverter())751 752 def convert(753 self, source: Union[str, requests.Response], **kwargs: Any754 ) -> DocumentConverterResult: # TODO: deal with kwargs755 """756 Args:757 - source: can be a string representing a path or url, or a requests.response object758 - extension: specifies the file extension to use when interpreting the file. If None, infer from source (path, uri, content-type, etc.)759 """760 761 # Local path or url762 if isinstance(source, str):763 if source.startswith("http://") or source.startswith("https://") or source.startswith("file://"):764 return self.convert_url(source, **kwargs)765 else:766 return self.convert_local(source, **kwargs)767 # Request response768 elif isinstance(source, requests.Response):769 return self.convert_response(source, **kwargs)770 771 def convert_local(self, path: str, **kwargs: Any) -> DocumentConverterResult: # TODO: deal with kwargs772 # Prepare a list of extensions to try (in order of priority)773 ext = kwargs.get("file_extension")774 extensions = [ext] if ext is not None else []775 776 # Get extension alternatives from the path and puremagic777 base, ext = os.path.splitext(path)778 self._append_ext(extensions, ext)779 self._append_ext(extensions, self._guess_ext_magic(path))780 781 # Convert782 return self._convert(path, extensions, **kwargs)783 784 # TODO what should stream's type be?785 def convert_stream(self, stream: Any, **kwargs: Any) -> DocumentConverterResult: # TODO: deal with kwargs786 # Prepare a list of extensions to try (in order of priority)787 ext = kwargs.get("file_extension")788 extensions = [ext] if ext is not None else []789 790 # Save the file locally to a temporary file. It will be deleted before this method exits791 handle, temp_path = tempfile.mkstemp()792 fh = os.fdopen(handle, "wb")793 result = None794 try:795 # Write to the temporary file796 content = stream.read()797 if isinstance(content, str):798 fh.write(content.encode("utf-8"))799 else:800 fh.write(content)801 fh.close()802 803 # Use puremagic to check for more extension options804 self._append_ext(extensions, self._guess_ext_magic(temp_path))805 806 # Convert807 result = self._convert(temp_path, extensions, **kwargs)808 # Clean up809 finally:810 try:811 fh.close()812 except Exception:813 pass814 os.unlink(temp_path)815 816 return result817 818 def convert_url(self, url: str, **kwargs: Any) -> DocumentConverterResult: # TODO: fix kwargs type819 # Send a HTTP request to the URL820 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"821 response = self._requests_session.get(url, stream=True, headers={"User-Agent": user_agent})822 response.raise_for_status()823 return self.convert_response(response, **kwargs)824 825 def convert_response(826 self, response: requests.Response, **kwargs: Any827 ) -> DocumentConverterResult: # TODO fix kwargs type828 # Prepare a list of extensions to try (in order of priority)829 ext = kwargs.get("file_extension")830 extensions = [ext] if ext is not None else []831 832 # Guess from the mimetype833 content_type = response.headers.get("content-type", "").split(";")[0]834 self._append_ext(extensions, mimetypes.guess_extension(content_type))835 836 # Read the content disposition if there is one837 content_disposition = response.headers.get("content-disposition", "")838 m = re.search(r"filename=([^;]+)", content_disposition)839 if m:840 base, ext = os.path.splitext(m.group(1).strip("\"'"))841 self._append_ext(extensions, ext)842 843 # Read from the extension from the path844 base, ext = os.path.splitext(urlparse(response.url).path)845 self._append_ext(extensions, ext)846 847 # Save the file locally to a temporary file. It will be deleted before this method exits848 handle, temp_path = tempfile.mkstemp()849 fh = os.fdopen(handle, "wb")850 result = None851 try:852 # Download the file853 for chunk in response.iter_content(chunk_size=512):854 fh.write(chunk)855 fh.close()856 857 # Use puremagic to check for more extension options858 self._append_ext(extensions, self._guess_ext_magic(temp_path))859 860 # Convert861 result = self._convert(temp_path, extensions, url=response.url)862 except Exception as e:863 print(f"Error in converting: {e}")864 865 # Clean up866 finally:867 try:868 fh.close()869 except Exception:870 pass871 os.unlink(temp_path)872 873 return result874 875 def _convert(self, local_path: str, extensions: List[Union[str, None]], **kwargs) -> DocumentConverterResult:876 error_trace = ""877 for ext in extensions + [None]: # Try last with no extension878 for converter in self._page_converters:879 _kwargs = copy.deepcopy(kwargs)880 881 # Overwrite file_extension appropriately882 if ext is None:883 if "file_extension" in _kwargs:884 del _kwargs["file_extension"]885 else:886 _kwargs.update({"file_extension": ext})887 888 # Copy any additional global options889 if "mlm_client" not in _kwargs and self._mlm_client is not None:890 _kwargs["mlm_client"] = self._mlm_client891 892 if "mlm_model" not in _kwargs and self._mlm_model is not None:893 _kwargs["mlm_model"] = self._mlm_model894 895 # If we hit an error log it and keep trying896 try:897 res = converter.convert(local_path, **_kwargs)898 except Exception:899 error_trace = ("\n\n" + traceback.format_exc()).strip()900 901 if res is not None:902 # Normalize the content903 res.text_content = "\n".join([line.rstrip() for line in re.split(r"\r?\n", res.text_content)])904 res.text_content = re.sub(r"\n{3,}", "\n\n", res.text_content)905 906 # Todo907 return res908 909 # If we got this far without success, report any exceptions910 if len(error_trace) > 0:911 raise FileConversionException(912 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}"913 )914 915 # Nothing can handle it!916 raise UnsupportedFormatException(917 f"Could not convert '{local_path}' to Markdown. The formats {extensions} are not supported."918 )919 920 def _append_ext(self, extensions, ext):921 """Append a unique non-None, non-empty extension to a list of extensions."""922 if ext is None:923 return924 ext = ext.strip()925 if ext == "":926 return927 # if ext not in extensions:928 if True:929 extensions.append(ext)930 931 def _guess_ext_magic(self, path):932 """Use puremagic (a Python implementation of libmagic) to guess a file's extension based on the first few bytes."""933 # Use puremagic to guess934 try:935 guesses = puremagic.magic_file(path)936 if len(guesses) > 0:937 ext = guesses[0].extension.strip()938 if len(ext) > 0:939 return ext940 except FileNotFoundError:941 pass942 except IsADirectoryError:943 pass944 except PermissionError:945 pass946 return None947 948 def register_page_converter(self, converter: DocumentConverter) -> None:949 """Register a page text converter."""950 self._page_converters.insert(0, converter)951