html2md.py

#!/usr/bin/env python3
"""
HTML to Markdown Converter
Usage: ./html2md.py /path/to/folder

Behavior:
- Non-recursive: process only top-level *.html in the folder's subfolders
- All-or-nothing: if any file fails, no changes are made
- On success: write {parent}/{folder}.md and move the input folder to Trash
- Feedback: returns a message on success or failure
"""

from pathlib import Path

import argparse
import asyncio
import os
import sys
import time
import html as html_lib
import re
import shutil
import subprocess
from typing import List, Tuple

try:
    import aiolimiter
    from bs4 import BeautifulSoup, Comment
    from google import genai
    from google.genai import types
except ImportError as e:
    print(f"Error: Missing dependency - {e}")
    sys.exit(1)


# Default Configuration Variables
DEFAULT_MODEL = "gemini-2.5-flash-lite-preview-06-17"
DEFAULT_THINKING_BUDGET = -1
DEFAULT_MAX_RETRIES = 3
DEFAULT_RETRY_DELAY_BASE = 1.0
DEFAULT_MAX_CONCURRENT = 20
DEFAULT_RATE_LIMIT_PER_MIN = 875
DEFAULT_REMOVE_TAGS = ["script", "style", "nav", "header", "footer", "aside"]
DEFAULT_ADD_HEADERS = True
DEFAULT_SEPARATOR = "---"

# Runtime configuration (populated from CLI/env)
API_KEY = None
MODEL = DEFAULT_MODEL
THINKING_BUDGET = DEFAULT_THINKING_BUDGET
MAX_RETRIES = DEFAULT_MAX_RETRIES
RETRY_DELAY_BASE = DEFAULT_RETRY_DELAY_BASE
MAX_CONCURRENT = DEFAULT_MAX_CONCURRENT
RATE_LIMIT_PER_MIN = DEFAULT_RATE_LIMIT_PER_MIN
REMOVE_TAGS = DEFAULT_REMOVE_TAGS
ADD_HEADERS = DEFAULT_ADD_HEADERS
SEPARATOR = DEFAULT_SEPARATOR


def _escape_applescript_string(s: str) -> str:
    return s.replace("\\", "\\\\").replace('"', '\\"')


def move_to_trash(path: Path) -> None:
    abs_path = str(path.resolve())
    script = f'tell application "Finder" to delete POSIX file "{_escape_applescript_string(abs_path)}"'
    result = subprocess.run(["osascript", "-e", script], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
    if result.returncode == 0:
        return
    # Fallback: move to user's Trash folder
    try:
        trash_dir = Path(os.path.expanduser("~/.Trash"))
        trash_dir.mkdir(parents=True, exist_ok=True)
        target = trash_dir / path.name
        # ensure unique name
        counter = 1
        while target.exists():
            target = trash_dir / f"{path.stem} {counter}{path.suffix}"
            counter += 1
        shutil.move(str(path), str(target))
    except Exception as e:
        raise RuntimeError(result.stderr.strip() or f"Failed to move to Trash: {e}")


def clean_html(content: str) -> str:
    soup = BeautifulSoup(content, 'html.parser')
    for tag_name in REMOVE_TAGS:
        for element in soup.find_all(tag_name):
            element.decompose()
    for comment in soup.find_all(string=lambda t: isinstance(t, Comment)):
        comment.extract()
    for tag in soup.find_all(True):
        if 'style' in tag.attrs:
            del tag['style']
    return str(soup)


def clean_filename(filename: str) -> str:
    name = Path(filename).stem
    name = name.replace('_', ' ')
    name = html_lib.unescape(name)
    name = re.sub(r'[<>:"/\\|?*]', '', name)
    name = re.sub(r'\s+', ' ', name).strip()
    return name


async def _make_api_call(client: genai.Client, prompt: str) -> str:
    response = await asyncio.to_thread(
        client.models.generate_content,
        model=MODEL,
        contents=prompt,
        config=types.GenerateContentConfig(
            thinking_config=types.ThinkingConfig(
                thinking_budget=THINKING_BUDGET,
                include_thoughts=False
            )
        )
    )
    return response.text


async def convert_html_to_markdown(limiter: aiolimiter.AsyncLimiter, client: genai.Client, html_content: str) -> str:
    prompt = f"""Convert this HTML content into clean Markdown format.

Guidelines:
- Focus on main content, ignore navigation, UI elements, metadata, and boilerplate HTML/CSS/JavaScript
- Preserve structure, order, and formatting of the content
- Use the title for the level 1 heading
- Use level 2 headings for other sub-sections (Description, Solution, Test Suite, Library, etc.)
- Convert HTML elements to appropriate Markdown equivalents, strip HTML tags, attributes, styling, etc.
- Use language identifiers for code blocks (```java, ```sql, etc.)
- Preserve ASCII art diagrams and visual representations
- Keep mathematical expressions intact using LaTeX if appropriate
- Maintain the pages flow (e.g., Description → Solution → Test Suite → Library
- Do NOT use horizontal rules

Output pure Markdown without HTML remnants.

HTML Content:
{html_content}"""
    async with limiter:
        for attempt in range(MAX_RETRIES):
            try:
                return await _make_api_call(client, prompt)
            except Exception as e:
                msg = str(e)
                if attempt < MAX_RETRIES - 1:
                    # crude 429 handling: sleep if retryDelay present
                    retry_match = re.search(r"'retryDelay': '(\d+)s'", msg)
                    delay = int(retry_match.group(1)) if retry_match else int(RETRY_DELAY_BASE * (2 ** attempt))
                    await asyncio.sleep(delay)
                else:
                    raise


async def _process_one(semaphore: asyncio.Semaphore, limiter: aiolimiter.AsyncLimiter, client: genai.Client, file_path: Path, index: int) -> Tuple[int, str, str]:
    async with semaphore:
        content = file_path.read_text(encoding='utf-8', errors='ignore')
        cleaned = clean_html(content)
        md = await convert_html_to_markdown(limiter, client, cleaned)
        header = clean_filename(file_path.name)
        return index, header, md


def _compose_markdown(ordered_results: List[Tuple[str, str]]) -> str:
    parts: List[str] = []
    for i, (header, body) in enumerate(ordered_results):
        if ADD_HEADERS:
            parts.append(f"# {header}\n\n")
        parts.append(body.strip())
        parts.append("\n\n")
        if i < len(ordered_results) - 1:
            parts.append(f"{SEPARATOR}\n\n")
    return ''.join(parts)


def _process_single_folder(input_dir: Path) -> None:
    if not input_dir.is_dir():
        raise RuntimeError("Input path is not a directory")

    html_files_with_mtime = []
    for p in sorted(input_dir.glob("*.html")):
        if p.is_file():
            html_files_with_mtime.append((p, p.stat().st_mtime))
    if not html_files_with_mtime:
        raise RuntimeError("No HTML files found")

    # Sort chronologically (by mtime)
    html_files_with_mtime.sort(key=lambda t: t[1])
    html_files = [fp for fp, _ in html_files_with_mtime]

    async def _run_batch(indices: List[int]):
        client = genai.Client(api_key=API_KEY)
        limiter = aiolimiter.AsyncLimiter(RATE_LIMIT_PER_MIN, 60)
        semaphore = asyncio.Semaphore(MAX_CONCURRENT)
        tasks = [
            _process_one(semaphore, limiter, client, html_files[i], i)
            for i in indices
        ]
        gathered = await asyncio.gather(*tasks, return_exceptions=True)
        return list(zip(indices, gathered))

    # First attempt for all files
    initial = asyncio.run(_run_batch(list(range(len(html_files)))))
    results_map: dict[int, object] = {idx: res for idx, res in initial}

    # Retry up to 2 more times for failed indices
    for _ in range(2):
        failed_indices = [i for i, r in results_map.items() if isinstance(r, Exception)]
        if not failed_indices:
            break
        retry_pairs = asyncio.run(_run_batch(failed_indices))
        for idx, res in retry_pairs:
            results_map[idx] = res

    if any(isinstance(r, Exception) for r in results_map.values()):
        raise RuntimeError("One or more files failed to convert")

    # Order results by original chronological order
    ordered: List[Tuple[str, str]] = []
    for i in range(len(html_files)):
        res_obj = results_map[i]
        idx_ret, header, md = res_obj  # type: ignore[misc]
        ordered.append((header, md))
    content = _compose_markdown(ordered)

    # Prepare paths
    parent = input_dir.parent
    final_output = parent / f"{input_dir.name}.md"
    temp_output = parent / f".{input_dir.name}.md.tmp-{int(time.time())}"

    # Write temp output only
    temp_output.write_text(content, encoding='utf-8')

    backup_dir = parent / f".fileproc_backup_{int(time.time())}"
    backup_dir.mkdir(parents=True, exist_ok=False)

    moved_items: List[Tuple[Path, Path]] = []

    try:
        # Move existing output to backup if present
        if final_output.exists():
            dst = backup_dir / final_output.name
            shutil.move(str(final_output), str(dst))
            moved_items.append((dst, final_output))  # record for potential restore (dst->final_output)

        # Move input folder to backup
        src_folder_backup = backup_dir / input_dir.name
        shutil.move(str(input_dir), str(src_folder_backup))
        moved_items.append((src_folder_backup, input_dir))

        # Put new output in place (atomic move)
        shutil.move(str(temp_output), str(final_output))

        # Trash the backup dir (contains old output and the source folder)
        try:
            move_to_trash(backup_dir)
        except Exception as trash_err:
            # Revert
            if final_output.exists():
                try:
                    final_output.unlink()
                except Exception:
                    pass
            # Move items back in reverse order
            for src, dst in reversed(moved_items):
                if src.exists():
                    shutil.move(str(src), str(dst))
            # Cleanup temp if still present
            if temp_output.exists():
                try:
                    temp_output.unlink()
                except Exception:
                    pass
            raise RuntimeError(f"Failed to move items to Trash: {trash_err}")
    except Exception as commit_err:
        # On commit error, revert any moves
        if final_output.exists():
            try:
                final_output.unlink()
            except Exception:
                pass
        for src, dst in reversed(moved_items):
            if src.exists():
                try:
                    shutil.move(str(src), str(dst))
                except Exception:
                    pass
        if temp_output.exists():
            try:
                temp_output.unlink()
            except Exception:
                pass
        # Attempt to remove empty backup dir
        try:
            if backup_dir.exists():
                os.rmdir(backup_dir)
        except Exception:
            pass
        raise RuntimeError(str(commit_err))


def run_html2md(directory: str) -> None:
    try:
        input_dir = Path(directory)
        if not input_dir.is_dir():
            print("Error: Input path is not a directory")
            sys.exit(1)

        # Always process immediate subfolders
        subdirs = sorted([p for p in input_dir.iterdir() if p.is_dir()])
        if not subdirs:
            print("Error: No subfolders found")
            sys.exit(1)

        succeeded: List[str] = []
        failed: List[Tuple[str, str]] = []
        for sub in subdirs:
            try:
                _process_single_folder(sub)
                succeeded.append(sub.name)
            except Exception as e:
                failed.append((sub.name, str(e)))

        if not succeeded and failed:
            # Nothing succeeded at all
            print("Error: One or more folders failed to convert")
            # List failures for visibility
            for name, reason in failed:
                print(f"- {name}: {reason}")
            sys.exit(1)

        # Success summary, even if partial
        print("Success: html2md completed")
        print(f"Processed: {len(succeeded)}/{len(subdirs)} subfolders")
        if failed:
            print("Unsuccessful items:")
            for name, reason in failed:
                print(f"- {name}: {reason}")
    except Exception as e:
        print(f"Error: {e}")
        sys.exit(1)


def _configure_from_args(args: argparse.Namespace) -> None:
    global API_KEY, MODEL, THINKING_BUDGET, MAX_RETRIES, RETRY_DELAY_BASE, MAX_CONCURRENT, RATE_LIMIT_PER_MIN, REMOVE_TAGS, ADD_HEADERS, SEPARATOR

    API_KEY = args.api_key
    MODEL = args.model
    THINKING_BUDGET = args.thinking_budget
    MAX_RETRIES = args.max_retries
    RETRY_DELAY_BASE = args.retry_delay_base
    MAX_CONCURRENT = args.max_concurrent
    RATE_LIMIT_PER_MIN = args.rate_limit_per_min
    REMOVE_TAGS = args.remove_tags
    ADD_HEADERS = args.add_headers
    SEPARATOR = args.separator


def _parse_args(argv: List[str]) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Convert HTML folders to Markdown")
    parser.add_argument("directory", help="Path to folder containing subfolders of HTML files")
    parser.add_argument("--api-key", dest="api_key", default=os.getenv("HTML2MD_API_KEY"), help="Google GenAI API key (env HTML2MD_API_KEY)")
    parser.add_argument("--model", default=DEFAULT_MODEL, help="Model name to use for generation")
    parser.add_argument("--thinking-budget", dest="thinking_budget", type=int, default=DEFAULT_THINKING_BUDGET, help="Thinking budget to supply to the model")
    parser.add_argument("--max-retries", type=int, default=DEFAULT_MAX_RETRIES, help="Maximum retries per request")
    parser.add_argument("--retry-delay-base", type=float, default=DEFAULT_RETRY_DELAY_BASE, help="Base delay for exponential backoff")
    parser.add_argument("--max-concurrent", type=int, default=DEFAULT_MAX_CONCURRENT, help="Maximum concurrent conversions")
    parser.add_argument("--rate-limit-per-min", type=int, default=DEFAULT_RATE_LIMIT_PER_MIN, help="API calls per minute")
    parser.add_argument("--remove-tags", nargs="+", default=DEFAULT_REMOVE_TAGS, help="HTML tags to remove before conversion")
    parser.add_argument("--add-headers", dest="add_headers", action="store_true", help="Prefix each section with a header")
    parser.add_argument("--no-add-headers", dest="add_headers", action="store_false", help="Disable automatic headers")
    parser.add_argument("--separator", default=DEFAULT_SEPARATOR, help="Separator to insert between sections")
    parser.set_defaults(add_headers=DEFAULT_ADD_HEADERS)
    return parser.parse_args(argv)


def main():
    args = _parse_args(sys.argv[1:])
    _configure_from_args(args)
    if not API_KEY:
        print("Error: API key is required (pass --api-key or set HTML2MD_API_KEY)")
        sys.exit(1)
    run_html2md(args.directory)


if __name__ == '__main__':
    main()