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)