Coverage for src/local_deep_research/research_library/downloaders/direct_pdf.py: 98%

1""" 

2Direct PDF Link Downloader 

3""" 

4 

5from typing import Dict, Optional 

6from urllib.parse import urlparse 

7from loguru import logger 

8 

9from .base import BaseDownloader, ContentType, DownloadResult 

10from ...security.ssrf_validator import redact_url_for_log 

11 

12 

13class DirectPDFDownloader(BaseDownloader): 

14 """Downloader for direct PDF links.""" 

15 

16 def can_handle(self, url: str) -> bool: 

17 """Check if URL is a direct PDF link using proper URL parsing.""" 

18 try: 

19 # Parse the URL 

20 parsed = urlparse(url.lower()) 

21 path = parsed.path or "" 

22 query = parsed.query or "" 

23 

24 # Check for .pdf extension in path 

25 if path.endswith(".pdf"): 

26 return True 

27 

28 # Check for .pdf with query parameters 

29 if ".pdf?" in url.lower(): 29 ↛ 30line 29 didn't jump to line 30 because the condition on line 29 was never true

30 return True 

31 

32 # Check for /pdf/ in path 

33 if "/pdf/" in path: 

34 return True 

35 

36 # Check query parameters for PDF format 

37 if "type=pdf" in query or "format=pdf" in query: 

38 return True 

39 

40 return False 

41 

42 except Exception: 

43 logger.warning(f"Error parsing URL {redact_url_for_log(url)}") 

44 return False 

45 

46 def download( 

47 self, url: str, content_type: ContentType = ContentType.PDF 

48 ) -> Optional[bytes]: 

49 """Download PDF directly from URL.""" 

50 if content_type == ContentType.TEXT: 

51 # Download PDF and extract text 

52 pdf_content = self._download_pdf(url) 

53 if pdf_content: 

54 text = self.extract_text_from_pdf(pdf_content) 

55 if text: 

56 return text.encode("utf-8") 

57 return None 

58 return self._download_pdf(url) 

59 

60 def download_with_result( 

61 self, url: str, content_type: ContentType = ContentType.PDF 

62 ) -> DownloadResult: 

63 """Download content and return detailed result with skip reason.""" 

64 if content_type == ContentType.TEXT: 

65 # Download PDF and extract text 

66 pdf_content = self._download_pdf(url) 

67 if pdf_content: 

68 text = self.extract_text_from_pdf(pdf_content) 

69 if text: 

70 return DownloadResult( 

71 content=text.encode("utf-8"), is_success=True 

72 ) 

73 return DownloadResult( 

74 skip_reason="PDF downloaded but text extraction failed" 

75 ) 

76 return DownloadResult( 

77 skip_reason="Could not download PDF from direct link" 

78 ) 

79 # Try to download PDF directly 

80 logger.info( 

81 f"Attempting direct PDF download from {redact_url_for_log(url)}" 

82 ) 

83 pdf_content = super()._download_pdf(url) 

84 

85 if pdf_content: 

86 logger.info( 

87 f"Successfully downloaded PDF directly from " 

88 f"{redact_url_for_log(url)}" 

89 ) 

90 return DownloadResult(content=pdf_content, is_success=True) 

91 # Try to determine specific reason for failure 

92 try: 

93 response = self.session.head(url, timeout=5, allow_redirects=True) 

94 if response.status_code == 404: 

95 return DownloadResult( 

96 skip_reason="PDF file not found (404) - link may be broken", 

97 status_code=404, 

98 ) 

99 if response.status_code == 403: 

100 return DownloadResult( 

101 skip_reason="Access denied (403) - PDF requires authentication", 

102 status_code=403, 

103 ) 

104 if response.status_code >= 500: 

105 return DownloadResult( 

106 skip_reason=f"Server error ({response.status_code}) - try again later", 

107 status_code=response.status_code, 

108 ) 

109 return DownloadResult( 

110 skip_reason=f"Could not download PDF - server returned status {response.status_code}", 

111 status_code=response.status_code, 

112 ) 

113 except Exception: 

114 return DownloadResult( 

115 skip_reason="Failed to download PDF from direct link" 

116 ) 

117 

118 def _download_pdf( 

119 self, url: str, headers: Optional[Dict[str, str]] = None 

120 ) -> Optional[bytes]: 

121 """Download PDF directly from URL.""" 

122 logger.info(f"Downloading PDF directly from: {redact_url_for_log(url)}") 

123 return super()._download_pdf(url)