Spaces:
Running
Running
| import gradio as gr | |
| import os | |
| import shutil | |
| import requests | |
| from bs4 import BeautifulSoup | |
| from urllib.parse import urljoin, urlparse | |
| from queue import Queue | |
| import time | |
| import zipfile | |
| import tempfile | |
| import sys | |
| import logging | |
| import traceback | |
| import pypandoc | |
| # --- Configuration & Logging --- | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') | |
| USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/90.0.4430.212 Safari/537.36' | |
| REQUEST_TIMEOUT = 20 # seconds | |
| POLITENESS_DELAY = 0.3 # seconds between requests | |
| # --- Pandoc Check --- | |
| def check_pandoc_available(): | |
| """Checks if pypandoc can find a pandoc executable.""" | |
| try: | |
| pandoc_path = pypandoc.get_pandoc_path() | |
| logging.info(f"pypandoc found Pandoc executable at: {pandoc_path}") | |
| return True | |
| except OSError: | |
| logging.error("pypandoc could not find Pandoc executable.") | |
| logging.error("Please ensure Pandoc is installed OR install 'pypandoc_binary' (`pip install pypandoc_binary`)") | |
| return False | |
| except ImportError: | |
| logging.error("pypandoc library not found. Please install it (`pip install pypandoc_binary`).") | |
| return False | |
| # --- Core Functions --- | |
| def fetch_html(url): | |
| """Fetches HTML content from a given URL.""" | |
| try: | |
| headers = {'User-Agent': USER_AGENT} | |
| response = requests.get(url, timeout=REQUEST_TIMEOUT, headers=headers) | |
| response.raise_for_status() | |
| response.encoding = response.apparent_encoding if response.apparent_encoding else 'utf-8' | |
| logging.info(f"Successfully fetched: {url}") | |
| return response.text | |
| except requests.exceptions.Timeout: | |
| logging.error(f"Timeout fetching URL: {url}") | |
| return None | |
| except requests.exceptions.RequestException as e: | |
| logging.error(f"Error fetching URL {url}: {e}") | |
| return None | |
| except Exception as e: | |
| logging.error(f"Unexpected error fetching {url}: {e}") | |
| return None | |
| def convert_html_to_md(html_content, output_md_path, pandoc_output_format, pandoc_extra_args): | |
| """ | |
| Converts HTML content string to a Markdown file using pypandoc | |
| with specified format and arguments. | |
| """ | |
| if not html_content: | |
| logging.warning(f"Empty HTML content for {output_md_path}. Conversion skipped.") | |
| return False | |
| # Using html+smart enables better handling of typographic characters in source HTML | |
| input_format = 'html+smart' # Keep input format consistent | |
| try: | |
| logging.debug(f"pypandoc converting to {pandoc_output_format} with args: {pandoc_extra_args}") | |
| # Use pypandoc.convert_text to convert the HTML string | |
| # Specify input format ('html'), output format ('gfm'), and output file | |
| # pypandoc handles invoking pandoc correctly with the string input | |
| output = pypandoc.convert_text( | |
| source=html_content, | |
| to=pandoc_output_format, | |
| format=input_format, | |
| outputfile=output_md_path, | |
| extra_args=pandoc_extra_args, | |
| encoding='utf-8' | |
| ) | |
| # When using outputfile, convert_text returns an empty string on success | |
| if output == "": | |
| logging.info(f"Successfully converted using pypandoc -> {os.path.basename(output_md_path)}") | |
| return True | |
| else: | |
| logging.error(f"pypandoc conversion to {output_md_path} returned unexpected non-empty output.") | |
| if os.path.exists(output_md_path) and os.path.getsize(output_md_path) == 0: | |
| logging.warning(f"Output file {output_md_path} was created but is empty.") | |
| return False | |
| except Exception as e: | |
| logging.error(f"Error during pypandoc conversion for {output_md_path}: {e}") | |
| logging.error(traceback.format_exc()) | |
| if os.path.exists(output_md_path) and os.path.getsize(output_md_path) == 0: | |
| try: | |
| os.remove(output_md_path) | |
| logging.info(f"Removed empty/failed output file: {os.path.basename(output_md_path)}") | |
| except OSError as remove_err: | |
| logging.warning(f"Could not remove empty/failed output file {output_md_path}: {remove_err}") | |
| return False | |
| def create_zip_archive(source_dir, output_zip_path): | |
| """Creates a ZIP archive from the contents of source_dir.""" | |
| try: | |
| with zipfile.ZipFile(output_zip_path, 'w', zipfile.ZIP_DEFLATED) as zipf: | |
| for root, _, files in os.walk(source_dir): | |
| for file in files: | |
| file_path = os.path.join(root, file) | |
| # Arcname is the path inside the zip file (relative to source_dir) | |
| arcname = os.path.relpath(file_path, source_dir) | |
| zipf.write(file_path, arcname) | |
| logging.info(f"Successfully created ZIP archive: {output_zip_path}") | |
| return True | |
| except Exception as e: | |
| logging.error(f"Failed to create ZIP archive {output_zip_path}: {e}") | |
| return False | |
| # --- Main Gradio Function --- | |
| def process_conversion_request(start_url_str, restrict_path, use_aggressive_conversion, progress=gr.Progress(track_tqdm=True)): | |
| """The main function triggered by the Gradio interface.""" | |
| # --- 0. Check Pandoc via pypandoc --- | |
| if not check_pandoc_available(): | |
| return "Error: pypandoc could not find a Pandoc executable. Please ensure Pandoc is installed or install `pypandoc_binary`.", None | |
| # --- 1. Validate URL and Determine Restriction Path --- | |
| start_url_str = start_url_str.strip() | |
| start_path_dir_for_restriction = None # Initialize restriction path base | |
| if not start_url_str: | |
| return "Error: Starting URL cannot be empty.", None | |
| try: | |
| parsed_start_url = urlparse(start_url_str) | |
| if not parsed_start_url.scheme or not parsed_start_url.netloc: | |
| raise ValueError("Invalid URL format (missing scheme or domain).") | |
| base_netloc = parsed_start_url.netloc | |
| base_scheme = parsed_start_url.scheme | |
| # Calculate the base directory path for comparison if restriction is enabled | |
| start_path_cleaned = parsed_start_url.path.strip('/') | |
| if start_path_cleaned: # If not root path | |
| # Use os.path.dirname to get the directory part | |
| # dirname('main/index.html') -> 'main' | |
| # dirname('main') -> '' (This needs correction if start URL is like /main/) | |
| # Let's adjust: if no '/' it means it's the first level dir or a root file | |
| if '/' not in start_path_cleaned and '.' not in start_path_cleaned: | |
| start_path_dir_for_restriction = start_path_cleaned # e.g. 'main' | |
| else: | |
| start_path_dir_for_restriction = os.path.dirname(start_path_cleaned) # e.g. 'main' from main/index.html, or '' from /index.html | |
| if start_path_dir_for_restriction == '': # Handle case like /index.html correctly | |
| start_path_dir_for_restriction = None # Treat like root, don't restrict path based on this | |
| except ValueError as e: | |
| return f"Error: Invalid starting URL '{start_url_str}': {e}", None | |
| # Log restriction status | |
| restriction_msg = f"Path restriction enabled: limiting to paths starting like '{start_path_dir_for_restriction}/'." if restrict_path and start_path_dir_for_restriction else "Path restriction disabled or starting from root." | |
| # --- Determine Pandoc Settings based on Checkbox --- | |
| # wrap=none, Prevent auto-wrapping lines | |
| if use_aggressive_conversion: | |
| pandoc_format_to_use = 'gfm-raw_html+hard_line_breaks' | |
| pandoc_args_to_use = ['--wrap=none', '--markdown-headings=atx'] | |
| conversion_mode_msg = "Using aggressive Markdown conversion (less raw HTML, ATX headers)." | |
| else: | |
| # Using gfm+hard_line_breaks ensures GitHub compatibility and respects single newlines | |
| pandoc_format_to_use = 'gfm+hard_line_breaks' | |
| pandoc_args_to_use = ['--wrap=none'] | |
| conversion_mode_msg = "Using standard Markdown conversion (may preserve more raw HTML)." | |
| logging.info(conversion_mode_msg) # Log the mode | |
| # --- 2. Setup Temporary Directory & Crawler --- | |
| staging_dir = tempfile.mkdtemp(prefix="md_convert_") | |
| logging.info(f"Created temporary staging directory: {staging_dir}") | |
| output_zip_file = None | |
| urls_to_process = Queue() | |
| processed_urls = set() # Still needed to avoid duplicates | |
| failed_urls = set() | |
| converted_count = 0 | |
| url_count_estimate = 1 # Total unique URLs discovered so far (starts with the first one) | |
| dequeued_count = 0 | |
| urls_to_process.put(start_url_str) | |
| processed_urls.add(start_url_str) # Add start URL here | |
| log_messages = ["Process started...", restriction_msg, conversion_mode_msg] | |
| try: | |
| # --- 3. Crawl and Convert Loop --- | |
| while not urls_to_process.empty(): | |
| # --- Get URL and Increment Dequeued Count --- | |
| current_url = urls_to_process.get() | |
| dequeued_count += 1 # Increment when an item is taken for processing | |
| # --- Update Progress Bar --- | |
| # Calculate progress based on dequeued items vs. total discovered | |
| # Denominator is the total number of unique URLs added to processed_urls/queue so far | |
| denominator = max(1, url_count_estimate) # url_count_estimate increases when new links are found | |
| current_progress_value = dequeued_count / denominator | |
| # Update Gradio progress - use dequeued_count for user display | |
| # Display: Processed X / Total_Discovered Y | |
| progress(current_progress_value, desc=f"Processing {dequeued_count}/{url_count_estimate}. Queue: {urls_to_process.qsize()}") | |
| # --- Process the current URL --- | |
| log_message = f"\nProcessing ({dequeued_count}/{url_count_estimate}): {current_url}" | |
| logging.info(log_message) | |
| log_messages.append(log_message) | |
| # --- 3a. Fetch HTML --- | |
| time.sleep(POLITENESS_DELAY) | |
| html_content = fetch_html(current_url) | |
| if not html_content: | |
| failed_urls.add(current_url) | |
| log_message = f" -> Failed to fetch content." | |
| logging.warning(log_message) | |
| log_messages.append(log_message) | |
| continue | |
| # --- 3b. Determine Output Path --- | |
| parsed_current_url = urlparse(current_url) | |
| # Get the path part of the URL, removing leading/trailing slashes | |
| url_path_segment = parsed_current_url.path.strip('/') # e.g., "main/index.html", "HEAD/index.html", "" | |
| # If the path is empty (domain root like https://example.com/), use 'index' as the base name | |
| if not url_path_segment: | |
| path_in_zip_base = 'index' | |
| else: | |
| path_in_zip_base = url_path_segment # e.g., "main/index.html", "HEAD/index.html" | |
| # Now, determine the final .md filename based on the path base | |
| if path_in_zip_base.lower().endswith('.html'): | |
| relative_md_filename = os.path.splitext(path_in_zip_base)[0] + ".md" | |
| elif path_in_zip_base.endswith('/'): # Should not happen often with strip('/') but handle defensively | |
| # If URL was like /docs/, path_in_zip_base would be 'docs' after strip. | |
| # This case is less likely needed now, but safe to keep. | |
| relative_md_filename = os.path.join(path_in_zip_base, "index.md") | |
| else: | |
| # If it's not empty and doesn't end with .html, assume it's a directory path | |
| # Append 'index.md' to treat it like accessing a directory index | |
| # e.g., if URL path was /main, url_path_segment is 'main', output becomes 'main/index.md' | |
| # If URL path was /path/to/file (no .html), output becomes 'path/to/file.md' if '.' in basename, else 'path/to/file/index.md' | |
| basename = os.path.basename(path_in_zip_base) | |
| if '.' in basename: # Check if it looks like a file without .html extension | |
| relative_md_filename = path_in_zip_base + ".md" | |
| else: # Assume it's a directory reference | |
| relative_md_filename = os.path.join(path_in_zip_base, "index.md") | |
| # Construct full path within the temporary staging directory | |
| output_md_full_path = os.path.join(staging_dir, relative_md_filename) | |
| output_md_dir = os.path.dirname(output_md_full_path) | |
| # Create directories if they don't exist (check if output_md_dir is not empty) | |
| try: | |
| if output_md_dir and not os.path.exists(output_md_dir): | |
| os.makedirs(output_md_dir) | |
| except OSError as e: | |
| log_message = f" -> Error creating directory {output_md_dir}: {e}. Skipping conversion for this URL." | |
| logging.error(log_message) | |
| log_messages.append(log_message) | |
| failed_urls.add(current_url) | |
| continue # Skip to next URL | |
| # --- 3c. Convert HTML to Markdown --- | |
| if convert_html_to_md(html_content, output_md_full_path, pandoc_format_to_use, pandoc_args_to_use): | |
| converted_count += 1 | |
| log_message = f" -> Converted successfully to {os.path.relpath(output_md_full_path, staging_dir)}" | |
| logging.info(log_message) | |
| log_messages.append(log_message) | |
| else: | |
| failed_urls.add(current_url) | |
| log_message = f" -> Conversion failed." | |
| logging.warning(log_message) | |
| log_messages.append(log_message) | |
| # --- 3d. Find and Add New Links --- | |
| try: | |
| soup = BeautifulSoup(html_content, 'lxml') | |
| links_found_this_page = 0 | |
| links_skipped_due_to_path = 0 | |
| for link in soup.find_all('a', href=True): | |
| href = link['href'] | |
| absolute_url = urljoin(current_url, href) | |
| absolute_url = urlparse(absolute_url)._replace(fragment="").geturl() | |
| parsed_absolute_url = urlparse(absolute_url) | |
| # Basic Filtering (scheme, domain, looks like html) | |
| is_valid_target = ( | |
| parsed_absolute_url.scheme == base_scheme and | |
| parsed_absolute_url.netloc == base_netloc and | |
| (not parsed_absolute_url.path or | |
| parsed_absolute_url.path == '/' or | |
| parsed_absolute_url.path.lower().endswith('.html') or | |
| '.' not in os.path.basename(parsed_absolute_url.path.rstrip('/')) # Include directory links | |
| ) | |
| ) | |
| if not is_valid_target: | |
| continue # Skip invalid links early | |
| # --- Path Restriction Check --- | |
| path_restricted = False | |
| # Only apply if checkbox is checked AND we derived a non-root restriction path | |
| if restrict_path and start_path_dir_for_restriction is not None: | |
| candidate_path_clean = parsed_absolute_url.path.strip('/') | |
| # Check if the cleaned candidate path starts with the restriction dir + '/' | |
| # OR if the candidate path is exactly the restriction dir (e.g. /main matching main) | |
| if not (candidate_path_clean.startswith(start_path_dir_for_restriction + '/') or \ | |
| candidate_path_clean == start_path_dir_for_restriction): | |
| path_restricted = True | |
| links_skipped_due_to_path += 1 | |
| # --- End Path Restriction Check --- | |
| # Add to queue only if NOT restricted and NOT already processed | |
| if not path_restricted and absolute_url not in processed_urls: | |
| processed_urls.add(absolute_url) # Add to set immediately | |
| urls_to_process.put(absolute_url) | |
| links_found_this_page += 1 | |
| url_count_estimate += 1 | |
| # Log link discovery summary for the page | |
| log_links_msg = f" -> Found {links_found_this_page} new link(s) to process." | |
| if links_skipped_due_to_path > 0: | |
| log_links_msg += f" Skipped {links_skipped_due_to_path} link(s) due to path restriction." | |
| logging.info(log_links_msg) | |
| log_messages.append(log_links_msg) | |
| except Exception as e: | |
| log_message = f" -> Error parsing links on {current_url}: {e}" | |
| logging.error(log_message) | |
| log_messages.append(log_message) | |
| # --- 4. Create ZIP Archive --- | |
| progress(1.0, desc="Zipping files...") | |
| log_messages.append("\nCrawling complete. Creating ZIP file...") | |
| yield "\n".join(log_messages), None | |
| with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as temp_zip: | |
| output_zip_path = temp_zip.name | |
| if create_zip_archive(staging_dir, output_zip_path): | |
| log_messages.append(f"\nProcess finished successfully!") | |
| log_messages.append(f"Converted {converted_count} pages using {'aggressive' if use_aggressive_conversion else 'standard'} mode.") # Inform user of mode used | |
| if failed_urls: | |
| log_messages.append(f"Failed to process {len(failed_urls)} URLs (check logs).") | |
| log_messages.append(f"ZIP file ready: {os.path.basename(output_zip_path)}") | |
| yield "\n".join(log_messages), output_zip_path | |
| else: | |
| log_messages.append("\nError: Failed to create the final ZIP archive.") | |
| yield "\n".join(log_messages), None | |
| except KeyboardInterrupt: | |
| log_messages.append("\nProcess interrupted by user.") | |
| yield "\n".join(log_messages), None | |
| except Exception as e: | |
| log_messages.append(f"\nAn unexpected error occurred: {e}") | |
| logging.error("Unhandled exception in process_conversion_request:") | |
| logging.error(traceback.format_exc()) | |
| yield "\n".join(log_messages), None | |
| finally: | |
| # --- 5. Cleanup --- | |
| if os.path.exists(staging_dir): | |
| try: | |
| shutil.rmtree(staging_dir) | |
| logging.info(f"Cleaned up temporary directory: {staging_dir}") | |
| except Exception as e: | |
| logging.error(f"Error cleaning up temporary directory {staging_dir}: {e}") | |
| css = """ | |
| textarea[rows]:not([rows="1"]) { | |
| overflow-y: auto !important; | |
| scrollbar-width: thin !important; | |
| } | |
| textarea[rows]:not([rows="1"])::-webkit-scrollbar { | |
| all: initial !important; | |
| background: #f1f1f1 !important; | |
| } | |
| textarea[rows]:not([rows="1"])::-webkit-scrollbar-thumb { | |
| all: initial !important; | |
| background: #a8a8a8 !important; | |
| } | |
| """ | |
| # --- Gradio UI Definition --- | |
| with gr.Blocks(title="HTML Docs to Markdown Converter", css=css) as demo: | |
| gr.Markdown( | |
| """ | |
| # HTML Documentation to Markdown Converter (via pypandoc) | |
| Enter the starting `index.html` URL of an online documentation site. | |
| The script will crawl internal HTML links, convert pages to Markdown, and package results into a ZIP file. | |
| **Requires `pip install pypandoc_binary`**. | |
| """ | |
| ) | |
| with gr.Row(): | |
| url_input = gr.Textbox( | |
| label="Starting Index HTML URL", | |
| placeholder="e.g., https://dghs-imgutils.deepghs.org/main/index.html" | |
| ) | |
| with gr.Row(): | |
| restrict_path_checkbox = gr.Checkbox( | |
| label="Restrict crawl to starting path structure (e.g., if start is '/main/index.html', only crawl '/main/...' URLs)", | |
| value=True # Default to restricting path | |
| ) | |
| aggressive_md_checkbox = gr.Checkbox( | |
| label="Aggressive Markdown conversion (disable raw HTML, use ATX headers)", | |
| value=True # Default to aggressive conversion | |
| ) | |
| with gr.Row(): | |
| start_button = gr.Button("Start Conversion", variant="primary") | |
| with gr.Row(): | |
| log_output = gr.Textbox(label="Progress Logs", lines=15, interactive=False, show_copy_button=True) | |
| with gr.Row(): | |
| zip_output = gr.File(label="Download Markdown ZIP") | |
| start_button.click( | |
| fn=process_conversion_request, | |
| inputs=[url_input, restrict_path_checkbox, aggressive_md_checkbox], | |
| outputs=[log_output, zip_output], | |
| show_progress="full" | |
| ) | |
| # --- Launch App --- | |
| if __name__ == "__main__": | |
| demo.queue() | |
| demo.launch(inbrowser=True) | |