Coverage for src/local_deep_research/library/download_management/filters/resource_filter.py: 100%

1""" 

2Resource Filter 

3 

4Smart filtering logic for downloadable resources based on failure history, 

5cooldowns, and retry policies. Replaces simple file existence checks with 

6intelligent retry management. 

7""" 

8 

9from typing import List, Optional 

10 

11from loguru import logger 

12 

13from ..retry_manager import RetryManager, ResourceFilterResult, FilterSummary 

14 

15 

16class ResourceFilter: 

17 """Filter resources for download based on history and policies""" 

18 

19 def __init__(self, username: str, password: Optional[str] = None): 

20 """ 

21 Initialize the resource filter. 

22 

23 Args: 

24 username: Username for database access 

25 password: Optional password for encrypted database 

26 """ 

27 self.username = username 

28 self.retry_manager = RetryManager(username, password) 

29 logger.info(f"Initialized for user: {username}") 

30 

31 def filter_downloadable_resources( 

32 self, resources: List 

33 ) -> List[ResourceFilterResult]: 

34 """ 

35 Filter resources that are available for download. 

36 

37 Args: 

38 resources: List of ResearchResource objects to filter 

39 

40 Returns: 

41 List of ResourceFilterResult objects with retry decisions 

42 """ 

43 logger.info(f"Filtering {len(resources)} resources") 

44 

45 # Use retry manager to filter based on failure history 

46 return self.retry_manager.filter_resources(resources) 

47 

48 def get_filter_summary(self, resources: List) -> FilterSummary: 

49 """ 

50 Get a summary of filtering results. 

51 

52 Args: 

53 resources: List of resources that were filtered 

54 

55 Returns: 

56 FilterSummary object with detailed counts 

57 """ 

58 results = self.filter_downloadable_resources(resources) 

59 return self.retry_manager.get_filter_summary(results) 

60 

61 def get_skipped_resources_info(self, resources: List) -> dict: 

62 """ 

63 Get detailed information about skipped resources for UI display. 

64 

65 Args: 

66 resources: List of all resources 

67 

68 Returns: 

69 Dictionary with detailed skip information 

70 """ 

71 results = self.filter_downloadable_resources(resources) 

72 

73 skipped_resources = [] 

74 for result in results: 

75 if not result.can_retry: 

76 status_info = ( 

77 self.retry_manager.status_tracker.get_resource_status( 

78 result.resource_id 

79 ) 

80 ) 

81 skipped_resources.append( 

82 { 

83 "resource_id": result.resource_id, 

84 "status": result.status, 

85 "reason": result.reason, 

86 "estimated_wait_minutes": result.estimated_wait.total_seconds() 

87 // 60 

88 if result.estimated_wait 

89 else None, 

90 "status_info": status_info, 

91 } 

92 ) 

93 

94 return { 

95 "total_skipped": len(skipped_resources), 

96 "permanently_failed": [ 

97 r 

98 for r in skipped_resources 

99 if r["status"] == "permanently_failed" 

100 ], 

101 "temporarily_failed": [ 

102 r 

103 for r in skipped_resources 

104 if r["status"] == "temporarily_failed" 

105 ], 

106 "other_skipped": [ 

107 r 

108 for r in skipped_resources 

109 if r["status"] 

110 not in ["permanently_failed", "temporarily_failed"] 

111 ], 

112 "skipped_resources": skipped_resources, 

113 } 

114 

115 def should_skip_resource(self, resource_id: int) -> tuple[bool, str]: 

116 """ 

117 Quick check if a specific resource should be skipped. 

118 

119 Args: 

120 resource_id: Resource identifier 

121 

122 Returns: 

123 Tuple of (should_skip, reason) 

124 """ 

125 decision = self.retry_manager.should_retry_resource(resource_id) 

126 return ( 

127 not decision.can_retry, 

128 decision.reason or "Resource not available for retry", 

129 )