pig.py/neopig/html2md.py

734 lines
27 KiB
Python

#!/usr/bin/env python3
# This is free software for the public good of a permacomputer hosted at
# permacomputer.com, an always-on computer by the people, for the people.
# One which is durable, easy to repair, & distributed like tap water
# for machine learning intelligence.
#
# The permacomputer is community-owned infrastructure optimized around
# four values:
#
# TRUTH First principles, math & science, open source code freely distributed
# FREEDOM Voluntary partnerships, freedom from tyranny & corporate control
# HARMONY Minimal waste, self-renewing systems with diverse thriving connections
# LOVE Be yourself without hurting others, cooperation through natural law
#
# This software contributes to that vision by archiving the web, preserving digital knowledge before it disappears.
# Code is seeds to sprout on any abandoned technology.
"""
Smart HTML to Markdown converter.
Detects content structure (forums, blogs, Q&A, e-commerce) and generates
clean markdown that preserves semantic meaning.
"""
from bs4 import BeautifulSoup, Tag
from typing import Optional, List, Dict, Tuple
from urllib.parse import urljoin
import re
class SmartMarkdownConverter:
"""Convert HTML to markdown with semantic structure preservation."""
def __init__(self, base_url: str = ''):
self.base_url = base_url
def convert(self, html: str, hint: str = None) -> str:
"""Convert HTML to markdown, auto-detecting content type.
Args:
html: Raw HTML content
hint: Optional content type hint ('forum', 'blog', 'qa', 'ecommerce') to skip detection
"""
# lxml is 2-5x faster than html.parser, fall back if not installed
try:
soup = BeautifulSoup(html, 'lxml')
except Exception:
soup = BeautifulSoup(html, 'html.parser')
# Remove noise
self._remove_noise(soup)
# Fast path: if hint provided, skip detection
if hint == 'forum':
posts = soup.select('.post_container, .topic-post, .post-stream > article')
if posts:
return self._render_forum(self._extract_forum_posts(soup, posts))
# Detect content type and extract accordingly
content_type, content = self._detect_and_extract(soup)
if content_type == 'forum':
return self._render_forum(content)
elif content_type == 'blog':
return self._render_blog(content)
elif content_type == 'qa':
return self._render_qa(content)
elif content_type == 'ecommerce':
return self._render_ecommerce(content)
else:
# Fallback: clean conversion of main content
return self._render_generic(soup)
def _remove_noise(self, soup: BeautifulSoup):
"""Remove navigation, scripts, styles, and other noise."""
# Remove script, style, and other non-content elements
for tag in soup(['script', 'style', 'noscript', 'iframe', 'svg']):
tag.decompose()
# Remove common navigation/chrome elements - single combined selector
noise_selector = ', '.join([
'nav', 'header', 'footer',
'.sidebar', '.nav', '.navigation', '.menu', '.breadcrumb',
'.header', '.footer', '.logo', '.site-logo',
'#header', '#footer', '#nav', '#sidebar',
'.d-header', '.d-footer',
'.pagination', '.pager',
'[role="banner"]', '[role="navigation"]', '[role="contentinfo"]',
'.cookie-banner', '.modal', '.popup', '.overlay',
'.ad', '.ads', '.advertisement', '.sponsored',
'.social-share', '.share-buttons',
# Table of contents (various CMS patterns)
'.contents.topic', '#contents', '.toc-wrapper',
])
for tag in soup.select(noise_selector):
tag.decompose()
def _detect_and_extract(self, soup: BeautifulSoup) -> Tuple[str, any]:
"""Detect content type and extract structured content."""
# Forum detection (Discourse, phpBB, vBulletin, Reddit, etc.)
forum_indicators = [
'.post_container', # Discourse archived (has avatar_container + post)
'.topic-post', # Discourse live
'.post-stream > article', # Discourse alt
'.message', # XenForo
'.postcontainer', # vBulletin
'article.boxed', # Generic
'.thing.comment', # Reddit old
'shreddit-comment', # Reddit new
'.Comment', # Reddit redesign
'.remarkbox-comment', # Remarkbox
'.comment', # Generic comments
]
for post_sel in forum_indicators:
posts = soup.select(post_sel)
if len(posts) >= 2: # At least 2 posts to be a forum thread
return 'forum', self._extract_forum_posts(soup, posts)
# Q&A detection (Stack Exchange, etc.)
qa_indicators = [
('.question', '.answer'),
('#question', '.answer'),
('.question-page', '.answercell'),
]
for q_sel, a_sel in qa_indicators:
questions = soup.select(q_sel)
answers = soup.select(a_sel)
if questions or answers:
return 'qa', self._extract_qa(soup, questions, answers)
# Blog detection
blog_indicators = [
'article', '.post', '.entry', '.blog-post',
'.hentry', '[itemtype*="BlogPosting"]',
]
for selector in blog_indicators:
articles = soup.select(selector)
if articles and len(articles) <= 10: # Not a listing page
return 'blog', self._extract_blog(soup, articles)
# E-commerce detection
ecom_indicators = [
'.product', '.product-detail', '[itemtype*="Product"]',
'.product-info', '.pdp-main',
]
for selector in ecom_indicators:
products = soup.select(selector)
if products:
return 'ecommerce', self._extract_ecommerce(soup, products)
return 'generic', soup
def _extract_forum_posts(self, soup: BeautifulSoup, posts: List[Tag]) -> List[Dict]:
"""Extract forum posts with avatar, username, content, timestamp."""
extracted = []
for post in posts:
post_data = {
'avatar': None,
'username': None,
'timestamp': None,
'content': None,
'quotes': [],
}
# Avatar - look for common patterns (may be sibling or child)
avatar = (
post.select_one('.avatar_container img.avatar') or # Discourse archived
post.select_one('.avatar_container img') or
post.select_one('.avatar-container img') or
post.select_one('img.avatar') or
post.select_one('.avatar img') or
post.select_one('.avatar-flair img') or
post.select_one('.user-avatar img') or
post.select_one('.postprofile img') or
post.select_one('.author img') or
post.select_one('img[alt*="avatar"]') or # Reddit
post.select_one('.flair img') or # Reddit flair
post.select_one('.remarkbox-avatar img') # Remarkbox
)
if avatar:
src = avatar.get('src', '')
# Skip placeholder avatars
if src and '{size}' not in src:
post_data['avatar'] = self._resolve_url(src)
# Username - Discourse archived uses .user_name, live uses .username
username_el = (
post.select_one('.user_name') or # Discourse archived
post.select_one('.username') or
post.select_one('.author') or
post.select_one('.creator a') or
post.select_one('.user-card-name') or
post.select_one('a[data-user-card]') or
post.select_one('.names .name') or
post.select_one('.postprofile dt') or
post.select_one('strong.username') or
post.select_one('a[href*="/user/"]') or # Reddit
post.select_one('a[href*="/u/"]') or # Reddit shorthand
post.select_one('.remarkbox-author') # Remarkbox
)
if username_el:
post_data['username'] = username_el.get_text(strip=True)
# Timestamp
time_el = (
post.select_one('time') or
post.select_one('.post-date') or
post.select_one('.timestamp') or
post.select_one('.date') or
post.select_one('.relative-date')
)
if time_el:
post_data['timestamp'] = time_el.get('title') or time_el.get_text(strip=True)
# Content - the main post body
content_el = (
post.select_one('.post_content') or # Discourse archived
post.select_one('.cooked') or # Discourse live
post.select_one('.post-content') or
post.select_one('.message-body') or
post.select_one('.postcontent') or
post.select_one('.post_body') or
post.select_one('.entry-content') or
post.select_one('.content') or
post.select_one('.post') or # Fallback to inner .post div
post.select_one('.md') or # Reddit markdown content
post.select_one('.usertext-body') or # Reddit old
post.select_one('.remarkbox-content') # Remarkbox
)
if content_el:
post_data['content'] = self._element_to_markdown(content_el)
else:
# Fallback: use whole post but try to exclude metadata
clone = BeautifulSoup(str(post), 'html.parser')
for sel in ['.avatar', '.avatar_container', '.user_name', '.username', '.author', '.date', '.post-actions', '.post-menu']:
for el in clone.select(sel):
el.decompose()
post_data['content'] = self._element_to_markdown(clone)
if post_data['content']:
extracted.append(post_data)
return extracted
def _extract_qa(self, soup: BeautifulSoup, questions: List[Tag], answers: List[Tag]) -> Dict:
"""Extract Q&A content with votes, user info."""
qa_data = {'question': None, 'answers': []}
if questions:
q = questions[0]
qa_data['question'] = {
'title': self._get_text(q, '.question-title, h1'),
'votes': self._get_text(q, '.vote-count, .js-vote-count'),
'content': self._element_to_markdown(q.select_one('.post-text, .s-prose, .question-body')),
'author': self._get_text(q, '.user-info .user-details a, .author'),
}
for a in answers:
answer_data = {
'votes': self._get_text(a, '.vote-count, .js-vote-count'),
'content': self._element_to_markdown(a.select_one('.post-text, .s-prose, .answer-body')),
'author': self._get_text(a, '.user-info .user-details a, .author'),
'accepted': bool(a.select_one('.accepted-answer, .is-accepted')),
}
if answer_data['content']:
qa_data['answers'].append(answer_data)
return qa_data
def _extract_blog(self, soup: BeautifulSoup, articles: List[Tag]) -> List[Dict]:
"""Extract blog articles with metadata."""
extracted = []
for article in articles:
article_data = {
'title': self._get_text(article, 'h1, h2, .entry-title, .post-title'),
'author': self._get_text(article, '.author, .byline, [rel="author"]'),
'date': self._get_text(article, 'time, .date, .published, .post-date'),
'content': None,
}
content_el = (
article.select_one('.entry-content') or
article.select_one('.post-content') or
article.select_one('.article-body') or
article.select_one('.article_text') or
article.select_one('.article-text') or
article.select_one('.content') or
article
)
article_data['content'] = self._element_to_markdown(content_el)
if article_data['content']:
extracted.append(article_data)
return extracted
def _extract_ecommerce(self, soup: BeautifulSoup, products: List[Tag]) -> List[Dict]:
"""Extract product information."""
extracted = []
for product in products:
product_data = {
'name': self._get_text(product, '.product-name, .product-title, h1, h2'),
'price': self._get_text(product, '.price, .product-price, [itemprop="price"]'),
'description': self._element_to_markdown(
product.select_one('.description, .product-description, [itemprop="description"]')
),
'image': None,
'rating': self._get_text(product, '.rating, .stars, [itemprop="ratingValue"]'),
}
img = product.select_one('.product-image img, .gallery img, [itemprop="image"]')
if img:
product_data['image'] = self._resolve_url(img.get('src', ''))
if product_data['name']:
extracted.append(product_data)
return extracted
def _element_to_markdown(self, el: Optional[Tag]) -> str:
"""Convert a single element to markdown."""
if not el:
return ''
lines = []
self._walk_element(el, lines)
return '\n'.join(lines).strip()
def _walk_element(self, el: Tag, lines: List[str], depth: int = 0):
"""Recursively walk element and build markdown lines."""
if isinstance(el, str):
text = el.strip()
if text:
lines.append(text)
return
if not isinstance(el, Tag):
return
tag = el.name.lower() if el.name else ''
# Block elements
if tag in ('p', 'div'):
# Check if contains block-level elements (including nested divs)
block_tags = {'div', 'p', 'pre', 'ul', 'ol', 'blockquote', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'table', 'section', 'article'}
has_blocks = any(
isinstance(c, Tag) and c.name and c.name.lower() in block_tags
for c in el.children
)
if has_blocks:
# Recursively process children to preserve block structure
for child in el.children:
self._walk_element(child, lines, depth)
else:
content = self._inline_content(el)
if content:
lines.append(content)
lines.append('')
elif tag in ('h1', 'h2', 'h3', 'h4', 'h5', 'h6'):
level = int(tag[1])
content = self._inline_content(el)
if content:
lines.append(f"{'#' * level} {content}")
lines.append('')
elif tag == 'blockquote':
quote_lines = []
for child in el.children:
self._walk_element(child, quote_lines, depth + 1)
for line in quote_lines:
if line:
lines.append(f"> {line}")
else:
lines.append('>')
lines.append('')
elif tag in ('ul', 'ol'):
for i, li in enumerate(el.find_all('li', recursive=False)):
prefix = f"{i+1}." if tag == 'ol' else "-"
content = self._inline_content(li)
lines.append(f"{prefix} {content}")
lines.append('')
elif tag == 'pre':
code = el.get_text() or ''
# Try to detect language from class
lang = ''
code_el = el.select_one('code')
if code_el:
classes = code_el.get('class', [])
for cls in classes:
if cls.startswith('language-') or cls.startswith('lang-'):
lang = cls.split('-', 1)[1]
break
lines.append(f"```{lang}")
lines.append(code.strip())
lines.append("```")
lines.append('')
elif tag == 'code' and el.parent and el.parent.name != 'pre':
# Inline code handled in _inline_content
pass
elif tag == 'hr':
lines.append('---')
lines.append('')
elif tag == 'br':
lines.append('')
elif tag == 'img':
src = self._resolve_url(el.get('src', ''))
alt = el.get('alt', '')
if src:
lines.append(f"![{alt}]({src})")
lines.append('')
elif tag == 'a':
# Links are handled inline
pass
elif tag == 'table':
lines.extend(self._table_to_markdown(el))
lines.append('')
else:
# Generic container - recurse
for child in el.children:
self._walk_element(child, lines, depth)
def _inline_content(self, el: Tag) -> str:
"""Extract inline content from an element, handling formatting."""
parts = []
for child in el.children:
if isinstance(child, str):
text = child.strip()
if text:
# Collapse whitespace
text = re.sub(r'\s+', ' ', text)
parts.append(text)
elif isinstance(child, Tag):
tag = child.name.lower() if child.name else ''
if tag in ('strong', 'b'):
inner = self._inline_content(child)
if inner:
parts.append(f"**{inner}**")
elif tag in ('em', 'i'):
inner = self._inline_content(child)
if inner:
parts.append(f"*{inner}*")
elif tag == 'code':
code = child.get_text()
if code:
parts.append(f"`{code}`")
elif tag == 'a':
href = self._resolve_url(child.get('href', ''))
text = self._inline_content(child) or href
if href:
parts.append(f"[{text}]({href})")
else:
parts.append(text)
elif tag == 'img':
src = self._resolve_url(child.get('src', ''))
alt = child.get('alt', '')
if src:
# Emoji images (alt like :smile:) stay inline
if alt.startswith(':') and alt.endswith(':'):
parts.append(f"![{alt}]({src})")
else:
# Regular images are block-level
parts.append(f"\n\n![{alt}]({src})\n\n")
elif tag == 'br':
parts.append('\n')
elif tag in ('span', 'small', 'mark'):
# Pass through
inner = self._inline_content(child)
if inner:
parts.append(inner)
else:
# Unknown inline - just get text
inner = self._inline_content(child)
if inner:
parts.append(inner)
return ' '.join(parts).strip()
def _table_to_markdown(self, table: Tag) -> List[str]:
"""Convert HTML table to markdown."""
lines = []
rows = table.find_all('tr')
if not rows:
return lines
# Extract headers
header_row = rows[0]
headers = [self._inline_content(th) for th in header_row.find_all(['th', 'td'])]
if headers:
lines.append('| ' + ' | '.join(headers) + ' |')
lines.append('| ' + ' | '.join(['---'] * len(headers)) + ' |')
# Extract body rows
for row in rows[1:]:
cells = [self._inline_content(td) for td in row.find_all(['td', 'th'])]
if cells:
# Pad cells if needed
while len(cells) < len(headers):
cells.append('')
lines.append('| ' + ' | '.join(cells) + ' |')
return lines
def _get_text(self, el: Tag, selectors: str) -> str:
"""Get text from first matching selector."""
for selector in selectors.split(','):
found = el.select_one(selector.strip())
if found:
return found.get_text(strip=True)
return ''
def _resolve_url(self, url: str) -> str:
"""Resolve relative URL to absolute, keeping same-page anchors as fragments."""
if not url or url.startswith('data:'):
return url
# Keep anchor-only links as-is
if url.startswith('#'):
return url
if self.base_url:
# Check if this is a same-page link with anchor
if '#' in url:
from miniuri import Uri
resolved = urljoin(self.base_url, url)
base_uri = Uri(self.base_url)
resolved_uri = Uri(resolved)
# Same host and path = same page, keep just the fragment
if (base_uri.hostname == resolved_uri.hostname and
base_uri.path.rstrip('/') == resolved_uri.path.rstrip('/')):
return '#' + resolved.split('#', 1)[1]
return resolved
if not url.startswith(('http://', 'https://', '//')):
return urljoin(self.base_url, url)
return url
# Renderers for different content types
def _render_forum(self, posts: List[Dict]) -> str:
"""Render forum posts to markdown."""
lines = []
for i, post in enumerate(posts):
if i > 0:
lines.append('')
lines.append('---')
lines.append('')
# Post header with avatar and username
# Always include avatar (placeholder if none) to maintain grid layout
if post['avatar']:
avatar_md = f"![avatar]({post['avatar']})"
else:
# Use # as placeholder - JS will detect and replace with initial
avatar_md = "![avatar](#)"
header_parts = [avatar_md]
if post['username']:
header_parts.append(f"**{post['username']}**")
if post['timestamp']:
header_parts.append(f"*{post['timestamp']}*")
lines.append(' '.join(header_parts))
lines.append('')
# Post content
if post['content']:
lines.append(post['content'])
return '\n'.join(lines)
def _render_qa(self, qa: Dict) -> str:
"""Render Q&A content to markdown."""
lines = []
if qa['question']:
q = qa['question']
if q['title']:
lines.append(f"# {q['title']}")
lines.append('')
if q['votes']:
lines.append(f"**{q['votes']} votes**")
if q['author']:
lines.append(f"*Asked by {q['author']}*")
lines.append('')
if q['content']:
lines.append(q['content'])
lines.append('')
if qa['answers']:
lines.append('---')
lines.append('')
lines.append(f"## {len(qa['answers'])} Answers")
lines.append('')
for answer in qa['answers']:
if answer['accepted']:
lines.append('### ✓ Accepted Answer')
else:
lines.append('### Answer')
if answer['votes']:
lines.append(f"**{answer['votes']} votes**")
if answer['author']:
lines.append(f"*By {answer['author']}*")
lines.append('')
if answer['content']:
lines.append(answer['content'])
lines.append('')
lines.append('---')
lines.append('')
return '\n'.join(lines)
def _render_blog(self, articles: List[Dict]) -> str:
"""Render blog articles to markdown."""
lines = []
for article in articles:
content = article['content'] or ''
# Only add title if it's not already at the start of content
if article['title']:
title_md = f"# {article['title']}"
if not content.startswith(title_md) and not content.startswith(f"# [{article['title']}"):
lines.append(title_md)
lines.append('')
meta = []
if article['author']:
meta.append(f"By {article['author']}")
if article['date']:
meta.append(article['date'])
if meta:
lines.append(f"*{' | '.join(meta)}*")
lines.append('')
if content:
lines.append(content)
return '\n'.join(lines)
def _render_ecommerce(self, products: List[Dict]) -> str:
"""Render product info to markdown."""
lines = []
for product in products:
if product['name']:
lines.append(f"# {product['name']}")
lines.append('')
if product['image']:
lines.append(f"![{product['name']}]({product['image']})")
lines.append('')
if product['price']:
lines.append(f"**Price:** {product['price']}")
lines.append('')
if product['rating']:
lines.append(f"**Rating:** {product['rating']}")
lines.append('')
if product['description']:
lines.append(product['description'])
return '\n'.join(lines)
def _render_generic(self, soup: BeautifulSoup) -> str:
"""Fallback generic rendering."""
# Find main content area
main = (
soup.select_one('main') or
soup.select_one('article') or
soup.select_one('.content') or
soup.select_one('#content') or
soup.select_one('.main') or
soup.body or
soup
)
return self._element_to_markdown(main)
def html_to_markdown(html: str, base_url: str = '', hint: str = None) -> str:
"""Convert HTML to markdown with smart structure detection.
Args:
html: Raw HTML content
base_url: Base URL for resolving relative links
hint: Content type hint ('forum', 'blog', 'qa', 'ecommerce') to skip auto-detection
Returns:
Clean markdown with preserved semantic structure
"""
converter = SmartMarkdownConverter(base_url)
return converter.convert(html, hint=hint)
if __name__ == '__main__':
import sys
if len(sys.argv) < 2:
print("Usage: python html2md.py <file.html> [base_url]")
sys.exit(1)
with open(sys.argv[1]) as f:
html = f.read()
base_url = sys.argv[2] if len(sys.argv) > 2 else ''
print(html_to_markdown(html, base_url))