Alpha108/GenerativeEngineOptimization
0
1"""
2Content Parsing Module
3Handles extraction of content from PDFs, text, and webpages
4"""
5
6import requests
7from bs4 import BeautifulSoup
8from urllib.parse import urljoin, urlparse
9from typing import List, Dict, Any
10import time
11from langchain_community.document_loaders import PyPDFLoader
12from langchain.schema import Document
13
14
15class BaseParser:
16 """Base class for all content parsers"""
17
18 def __init__(self):
19 self.supported_formats = []
20
21 def parse(self, source: str) -> List[Document]:
22 """Parse content from source and return LangChain Documents"""
23 raise NotImplementedError("Subclasses must implement parse method")
24
25 def validate_source(self, source: str) -> bool:
26 """Validate if the source can be processed"""
27 return True
28
29
30class PDFParser(BaseParser):
31 """Parser for PDF documents"""
32
33 def __init__(self):
34 super().__init__()
35 self.supported_formats = ['.pdf']
36
37 def parse(self, pdf_path: str) -> List[Document]:
38 """
39 Parse PDF file and return list of Document objects
40
41 Args:
42 pdf_path (str): Path to the PDF file
43
44 Returns:
45 List[Document]: List of parsed documents with metadata
46 """
47 try:
48 loader = PyPDFLoader(pdf_path)
49 documents = loader.load_and_split()
50
51 # Add additional metadata
52 for i, doc in enumerate(documents):
53 doc.metadata.update({
54 'source_type': 'pdf',
55 'page_number': i + 1,
56 'total_pages': len(documents),
57 'parser': 'PDFParser'
58 })
59
60 return documents
61
62 except Exception as e:
63 raise Exception(f"Error parsing PDF: {str(e)}")
64
65 def get_pdf_metadata(self, pdf_path: str) -> Dict[str, Any]:
66 """Extract metadata from PDF file"""
67 try:
68 loader = PyPDFLoader(pdf_path)
69 documents = loader.load()
70
71 total_pages = len(documents)
72 total_words = sum(len(doc.page_content.split()) for doc in documents)
73
74 return {
75 'total_pages': total_pages,
76 'total_words': total_words,
77 'average_words_per_page': total_words / total_pages if total_pages > 0 else 0,
78 'file_type': 'PDF',
79 'parser_used': 'PyPDFLoader'
80 }
81
82 except Exception as e:
83 return {'error': f"Could not extract metadata: {str(e)}"}
84
85
86class TextParser(BaseParser):
87 """Parser for plain text content"""
88
89 def __init__(self):
90 super().__init__()
91 self.supported_formats = ['.txt', 'plain_text']
92 self.chunk_size = 1000 # Default chunk size for long texts
93
94 def parse(self, text_content: str, chunk_size: int = None) -> List[Document]:
95 """
96 Parse text content and return list of Document objects
97
98 Args:
99 text_content (str): Raw text content
100 chunk_size (int): Optional chunk size for splitting long texts
101
102 Returns:
103 List[Document]: List of documents, potentially chunked
104 """
105 try:
106 if not text_content.strip():
107 raise ValueError("Empty text content provided")
108
109 chunk_size = chunk_size or self.chunk_size
110
111 # If text is short, return as single document
112 if len(text_content) <= chunk_size:
113 doc = Document(
114 page_content=text_content,
115 metadata={
116 'source_type': 'text',
117 'word_count': len(text_content.split()),
118 'char_count': len(text_content),
119 'chunk_index': 0,
120 'total_chunks': 1,
121 'parser': 'TextParser'
122 }
123 )
124 return [doc]
125
126 # Split long text into chunks
127 chunks = self._split_text_into_chunks(text_content, chunk_size)
128 documents = []
129
130 for i, chunk in enumerate(chunks):
131 doc = Document(
132 page_content=chunk,
133 metadata={
134 'source_type': 'text',
135 'word_count': len(chunk.split()),
136 'char_count': len(chunk),
137 'chunk_index': i,
138 'total_chunks': len(chunks),
139 'parser': 'TextParser'
140 }
141 )
142 documents.append(doc)
143
144 return documents
145
146 except Exception as e:
147 raise Exception(f"Error parsing text: {str(e)}")
148
149 def _split_text_into_chunks(self, text: str, chunk_size: int) -> List[str]:
150 """Split text into chunks while preserving sentence boundaries"""
151 sentences = text.split('. ')
152 chunks = []
153 current_chunk = ""
154
155 for sentence in sentences:
156 # Add sentence to current chunk if it fits
157 test_chunk = current_chunk + sentence + ". "
158
159 if len(test_chunk) <= chunk_size:
160 current_chunk = test_chunk
161 else:
162 # Start new chunk if current chunk has content
163 if current_chunk.strip():
164 chunks.append(current_chunk.strip())
165 current_chunk = sentence + ". "
166
167 # Add final chunk if it has content
168 if current_chunk.strip():
169 chunks.append(current_chunk.strip())
170
171 return chunks
172
173 def analyze_text_structure(self, text_content: str) -> Dict[str, Any]:
174 """Analyze the structure and characteristics of text content"""
175 try:
176 lines = text_content.split('\n')
177 words = text_content.split()
178 sentences = text_content.split('.')
179
180 # Count different elements
181 paragraphs = [p.strip() for p in text_content.split('\n\n') if p.strip()]
182
183 return {
184 'total_words': len(words),
185 'total_sentences': len([s for s in sentences if s.strip()]),
186 'total_lines': len(lines),
187 'total_paragraphs': len(paragraphs),
188 'average_words_per_sentence': len(words) / len(sentences) if sentences else 0,
189 'average_sentences_per_paragraph': len(sentences) / len(paragraphs) if paragraphs else 0,
190 'character_count': len(text_content),
191 'reading_time_minutes': len(words) / 200, # Assuming 200 words per minute
192 'complexity_score': self._calculate_text_complexity(text_content)
193 }
194
195 except Exception as e:
196 return {'error': f"Could not analyze text structure: {str(e)}"}
197
198 def _calculate_text_complexity(self, text: str) -> float:
199 """Calculate a simple text complexity score"""
200 words = text.split()
201 sentences = [s for s in text.split('.') if s.strip()]
202
203 if not sentences:
204 return 0.0
205
206 # Average words per sentence (higher = more complex)
207 avg_words_per_sentence = len(words) / len(sentences)
208
209 # Average characters per word (higher = more complex)
210 avg_chars_per_word = sum(len(word) for word in words) / len(words) if words else 0
211
212 # Simple complexity score (normalized to 1-10 scale)
213 complexity = (avg_words_per_sentence * 0.1) + (avg_chars_per_word * 0.5)
214 return min(complexity, 10.0)
215
216
217class WebpageParser(BaseParser):
218 """Parser for web content"""
219
220 def __init__(self):
221 super().__init__()
222 self.supported_formats = ['http', 'https']
223 self.headers = {
224 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
225 }
226 self.timeout = 10
227 self.max_retries = 3
228
229 def parse_website(self, url: str, max_pages: int = 1, include_subpages: bool = False) -> List[Dict[str, Any]]:
230 """
231 Parse website content and return structured data
232
233 Args:
234 url (str): Website URL to parse
235 max_pages (int): Maximum number of pages to parse
236 include_subpages (bool): Whether to include subpages
237
238 Returns:
239 List[Dict]: List of page data with content and metadata
240 """
241 try:
242 pages_data = []
243 urls_to_process = [url]
244 processed_urls = set()
245
246 # If including subpages, find additional URLs
247 if include_subpages and max_pages > 1:
248 subpage_urls = self._find_subpages(url, max_pages - 1)
249 urls_to_process.extend(subpage_urls)
250
251 # Process each URL
252 for current_url in urls_to_process[:max_pages]:
253 if current_url in processed_urls:
254 continue
255
256 page_data = self._parse_single_page(current_url)
257 if page_data:
258 pages_data.append(page_data)
259 processed_urls.add(current_url)
260
261 # Add small delay to be respectful
262 time.sleep(1)
263
264 return pages_data
265
266 except Exception as e:
267 raise Exception(f"Error parsing website: {str(e)}")
268
269 def _parse_single_page(self, url: str) -> Dict[str, Any]:
270 """Parse a single webpage and extract content"""
271 try:
272 # Make request with retries
273 response = None
274 for attempt in range(self.max_retries):
275 try:
276 response = requests.get(url, headers=self.headers, timeout=self.timeout)
277 response.raise_for_status()
278 break
279 except requests.RequestException as e:
280 if attempt == self.max_retries - 1:
281 raise e
282 time.sleep(2 ** attempt) # Exponential backoff
283
284 if not response:
285 return None
286
287 # Parse HTML content
288 soup = BeautifulSoup(response.content, 'html.parser')
289
290 # Remove unwanted elements
291 for element in soup(['script', 'style', 'nav', 'footer', 'header', 'aside']):
292 element.decompose()
293
294 # Extract main content
295 main_content = self._extract_main_content(soup)
296
297 # Extract metadata
298 title = self._extract_title(soup)
299 description = self._extract_description(soup)
300 headings = self._extract_headings(soup)
301 links = self._extract_links(soup, url)
302
303 # Clean and process text
304 cleaned_text = self._clean_text_content(main_content)
305
306 return {
307 'url': url,
308 'title': title,
309 'description': description,
310 'content': cleaned_text,
311 'headings': headings,
312 'internal_links': links['internal'],
313 'external_links': links['external'],
314 'word_count': len(cleaned_text.split()),
315 'char_count': len(cleaned_text),
316 'meta_keywords': self._extract_meta_keywords(soup),
317 'images': self._extract_images(soup, url),
318 'parser': 'WebpageParser',
319 'parsed_at': time.strftime('%Y-%m-%d %H:%M:%S')
320 }
321
322 except Exception as e:
323 return {'url': url, 'error': f"Failed to parse page: {str(e)}"}
324
325 def _extract_main_content(self, soup: BeautifulSoup) -> str:
326 """Extract the main content from the page"""
327 # Try to find main content in order of preference
328 content_selectors = [
329 'main',
330 'article',
331 '[role="main"]',
332 '.content',
333 '.main-content',
334 '#content',
335 '#main',
336 '.post-content',
337 '.entry-content'
338 ]
339
340 for selector in content_selectors:
341 element = soup.select_one(selector)
342 if element:
343 return element.get_text(separator=' ', strip=True)
344
345 # Fallback to body content
346 body = soup.find('body')
347 if body:
348 return body.get_text(separator=' ', strip=True)
349
350 return soup.get_text(separator=' ', strip=True)
351
352 def _extract_title(self, soup: BeautifulSoup) -> str:
353 """Extract page title"""
354 title_tag = soup.find('title')
355 if title_tag:
356 return title_tag.get_text().strip()
357
358 # Fallback to h1
359 h1 = soup.find('h1')
360 if h1:
361 return h1.get_text().strip()
362
363 return "No Title Found"
364
365 def _extract_description(self, soup: BeautifulSoup) -> str:
366 """Extract meta description"""
367 meta_desc = soup.find('meta', attrs={'name': 'description'})
368 if meta_desc and meta_desc.get('content'):
369 return meta_desc['content'].strip()
370
371 # Fallback to Open Graph description
372 og_desc = soup.find('meta', attrs={'property': 'og:description'})
373 if og_desc and og_desc.get('content'):
374 return og_desc['content'].strip()
375
376 return "No Description Found"
377
378 def _extract_headings(self, soup: BeautifulSoup) -> List[Dict[str, Any]]:
379 """Extract all headings with their hierarchy"""
380 headings = []
381
382 for i in range(1, 7): # h1 to h6
383 for heading in soup.find_all(f'h{i}'):
384 text = heading.get_text(strip=True)
385 if text:
386 headings.append({
387 'level': i,
388 'text': text,
389 'id': heading.get('id', ''),
390 'class': heading.get('class', [])
391 })
392
393 return headings
394
395 def _extract_links(self, soup: BeautifulSoup, base_url: str) -> Dict[str, List[str]]:
396 """Extract internal and external links"""
397 internal_links = []
398 external_links = []
399 base_domain = urlparse(base_url).netloc
400
401 for link in soup.find_all('a', href=True):
402 href = link['href']
403 full_url = urljoin(base_url, href)
404 parsed_url = urlparse(full_url)
405
406 if parsed_url.netloc == base_domain:
407 internal_links.append(full_url)
408 elif parsed_url.netloc: # External link with domain
409 external_links.append(full_url)
410
411 return {
412 'internal': list(set(internal_links)),
413 'external': list(set(external_links))
414 }
415
416 def _extract_meta_keywords(self, soup: BeautifulSoup) -> List[str]:
417 """Extract meta keywords if available"""
418 meta_keywords = soup.find('meta', attrs={'name': 'keywords'})
419 if meta_keywords and meta_keywords.get('content'):
420 keywords = meta_keywords['content'].split(',')
421 return [kw.strip() for kw in keywords if kw.strip()]
422 return []
423
424 def _extract_images(self, soup: BeautifulSoup, base_url: str) -> List[Dict[str, str]]:
425 """Extract image information"""
426 images = []
427
428 for img in soup.find_all('img'):
429 src = img.get('src')
430 if src:
431 full_url = urljoin(base_url, src)
432 images.append({
433 'src': full_url,
434 'alt': img.get('alt', ''),
435 'title': img.get('title', '')
436 })
437
438 return images
439
440 def _clean_text_content(self, text: str) -> str:
441 """Clean and normalize text content"""
442 if not text:
443 return ""
444
445 # Split into lines and clean each line
446 lines = text.split('\n')
447 cleaned_lines = []
448
449 for line in lines:
450 line = line.strip()
451 if line and len(line) > 1: # Skip empty lines and single characters
452 cleaned_lines.append(line)
453
454 # Join lines with single spaces
455 cleaned_text = ' '.join(cleaned_lines)
456
457 # Remove multiple spaces
458 while ' ' in cleaned_text:
459 cleaned_text = cleaned_text.replace(' ', ' ')
460
461 return cleaned_text
462
463 def _find_subpages(self, url: str, max_subpages: int) -> List[str]:
464 """Find subpages from the main page"""
465 try:
466 response = requests.get(url, headers=self.headers, timeout=self.timeout)
467 response.raise_for_status()
468
469 soup = BeautifulSoup(response.content, 'html.parser')
470 base_domain = urlparse(url).netloc
471 subpages = set()
472
473 # Find internal links
474 for link in soup.find_all('a', href=True):
475 href = link['href']
476 full_url = urljoin(url, href)
477 parsed_url = urlparse(full_url)
478
479 # Only include internal links from same domain
480 if (parsed_url.netloc == base_domain and
481 full_url != url and
482 not any(ext in full_url.lower() for ext in ['.pdf', '.jpg', '.png', '.gif', '.zip'])):
483 subpages.add(full_url)
484
485 if len(subpages) >= max_subpages:
486 break
487
488 return list(subpages)[:max_subpages]
489
490 except Exception:
491 return []
492
493 def validate_url(self, url: str) -> bool:
494 """Validate if URL is accessible"""
495 try:
496 response = requests.head(url, headers=self.headers, timeout=5)
497 return response.status_code == 200
498 except:
499 return False
500
501 def get_website_info(self, url: str) -> Dict[str, Any]:
502 """Get basic information about a website"""
503 try:
504 response = requests.get(url, headers=self.headers, timeout=self.timeout)
505 response.raise_for_status()
506
507 soup = BeautifulSoup(response.content, 'html.parser')
508
509 return {
510 'url': url,
511 'title': self._extract_title(soup),
512 'description': self._extract_description(soup),
513 'meta_keywords': self._extract_meta_keywords(soup),
514 'has_robots_meta': bool(soup.find('meta', attrs={'name': 'robots'})),
515 'has_viewport_meta': bool(soup.find('meta', attrs={'name': 'viewport'})),
516 'language': soup.get('lang', 'unknown'),
517 'status_code': response.status_code,
518 'content_type': response.headers.get('content-type', 'unknown'),
519 'server': response.headers.get('server', 'unknown')
520 }
521
522 except Exception as e:
523 return {'url': url, 'error': f"Could not get website info: {str(e)}"}
524
525
526class ParserFactory:
527 """Factory class to create appropriate parsers"""
528
529 @staticmethod
530 def get_parser(source_type: str):
531 """Get the appropriate parser for the source type"""
532 parsers = {
533 'pdf': PDFParser(),
534 'text': TextParser(),
535 'webpage': WebpageParser(),
536 'url': WebpageParser()
537 }
538
539 return parsers.get(source_type.lower())
540
541 @staticmethod
542 def detect_source_type(source: str) -> str:
543 """Detect the type of content source"""
544 if source.startswith(('http://', 'https://')):
545 return 'webpage'
546 elif source.endswith('.pdf'):
547 return 'pdf'
548 else:
549 return 'text'