Add parser engine: base with subprocess isolation, HTML/PDF/email/cable/text parsers
This commit is contained in:
@@ -0,0 +1,125 @@
|
||||
"""HTML document parser for Mosaic — BS4 with html.parser backend."""
|
||||
import re
|
||||
import logging
|
||||
|
||||
from bs4 import BeautifulSoup
|
||||
|
||||
from .base import ParseResult
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Elements to remove (navigation, scripts, styles)
|
||||
REMOVE_TAGS = ['script', 'style', 'nav', 'footer', 'header', 'aside',
|
||||
'noscript', 'iframe', 'svg']
|
||||
|
||||
# ANSI escape sequence pattern
|
||||
ANSI_ESCAPE = re.compile(r'\x1b\[[0-9;]*[a-zA-Z]|\x1b\].*?\x07')
|
||||
|
||||
|
||||
class HTMLDocParser:
|
||||
"""Parse HTML documents into structured text."""
|
||||
|
||||
def parse(self, file_path: str) -> ParseResult:
|
||||
"""Parse an HTML file.
|
||||
|
||||
Uses BeautifulSoup with 'html.parser' backend (explicitly specified,
|
||||
not auto-detected) to extract structured text.
|
||||
"""
|
||||
try:
|
||||
with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
|
||||
raw_html = f.read()
|
||||
|
||||
soup = BeautifulSoup(raw_html, 'html.parser')
|
||||
|
||||
# Extract title
|
||||
title = ""
|
||||
title_tag = soup.find('title')
|
||||
if title_tag:
|
||||
title = title_tag.get_text(strip=True)
|
||||
|
||||
# Extract metadata
|
||||
metadata = self._extract_metadata(soup)
|
||||
|
||||
# Remove unwanted elements
|
||||
for tag_name in REMOVE_TAGS:
|
||||
for tag in soup.find_all(tag_name):
|
||||
tag.decompose()
|
||||
|
||||
# Extract text preserving structure
|
||||
text = self._extract_structured_text(soup)
|
||||
|
||||
# Strip ANSI escape sequences
|
||||
text = ANSI_ESCAPE.sub('', text)
|
||||
|
||||
# Extract tables separately
|
||||
tables = self._extract_tables(soup)
|
||||
if tables:
|
||||
metadata['tables'] = tables
|
||||
|
||||
char_count = len(text)
|
||||
|
||||
return ParseResult(
|
||||
success=True,
|
||||
text=text,
|
||||
metadata=metadata,
|
||||
doc_type='html',
|
||||
title=title,
|
||||
char_count=char_count,
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
return ParseResult(error=f"HTML parse error: {str(e)}")
|
||||
|
||||
def _extract_metadata(self, soup: BeautifulSoup) -> dict:
|
||||
"""Extract metadata from HTML meta tags."""
|
||||
metadata = {}
|
||||
for meta in soup.find_all('meta'):
|
||||
name = meta.get('name', meta.get('property', ''))
|
||||
content = meta.get('content', '')
|
||||
if name and content:
|
||||
metadata[name] = content
|
||||
return metadata
|
||||
|
||||
def _extract_structured_text(self, soup: BeautifulSoup) -> str:
|
||||
"""Extract text preserving heading structure."""
|
||||
parts = []
|
||||
|
||||
for element in soup.find_all(['h1', 'h2', 'h3', 'h4', 'h5', 'h6', 'p',
|
||||
'li', 'td', 'th', 'pre', 'blockquote', 'div']):
|
||||
text = element.get_text(separator=' ', strip=True)
|
||||
if not text:
|
||||
continue
|
||||
|
||||
tag = element.name
|
||||
if tag.startswith('h'):
|
||||
level = int(tag[1])
|
||||
prefix = '#' * level
|
||||
parts.append(f"\n{prefix} {text}\n")
|
||||
elif tag == 'li':
|
||||
parts.append(f" - {text}")
|
||||
elif tag == 'pre':
|
||||
parts.append(f"\n```\n{text}\n```\n")
|
||||
elif tag == 'blockquote':
|
||||
parts.append(f"> {text}")
|
||||
else:
|
||||
parts.append(text)
|
||||
|
||||
result = '\n'.join(parts)
|
||||
# Collapse excessive whitespace
|
||||
result = re.sub(r'\n{3,}', '\n\n', result)
|
||||
return result.strip()
|
||||
|
||||
def _extract_tables(self, soup: BeautifulSoup) -> list[dict]:
|
||||
"""Extract tables as structured data."""
|
||||
tables = []
|
||||
for table in soup.find_all('table'):
|
||||
rows = []
|
||||
for tr in table.find_all('tr'):
|
||||
cells = []
|
||||
for td in tr.find_all(['td', 'th']):
|
||||
cells.append(td.get_text(strip=True))
|
||||
if cells:
|
||||
rows.append(cells)
|
||||
if rows:
|
||||
tables.append({'rows': rows})
|
||||
return tables
|
||||
Reference in New Issue
Block a user