Coverage for src/local_deep_research/library/download_management/retry_manager.py: 98%
102 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Retry Manager
4Core retry logic and cooldown management for download attempts.
5Prevents endless retry loops and implements intelligent retry strategies.
6"""
8from dataclasses import dataclass
9from datetime import datetime, timedelta, UTC
10from typing import List, Optional, Tuple
12from loguru import logger
13from .failure_classifier import FailureClassifier
14from .status_tracker import ResourceStatusTracker
17@dataclass
18class RetryDecision:
19 """Decision about whether to retry a resource"""
21 can_retry: bool
22 reason: Optional[str] = None
23 estimated_wait_time: Optional[timedelta] = None
26class ResourceFilterResult:
27 """Result of filtering a resource"""
29 def __init__(
30 self,
31 resource_id: int,
32 can_retry: bool,
33 status: str,
34 reason: str = "",
35 estimated_wait: Optional[timedelta] = None,
36 ):
37 self.resource_id = resource_id
38 self.can_retry = can_retry
39 self.status = status
40 self.reason = reason
41 self.estimated_wait = estimated_wait
44class FilterSummary:
45 """Summary of filtering results"""
47 def __init__(self):
48 self.total_count = 0
49 self.downloadable_count = 0
50 self.permanently_failed_count = 0
51 self.temporarily_failed_count = 0
52 self.available_count = 0
53 self.failure_type_counts = {}
55 def add_result(self, result: ResourceFilterResult):
56 """Add a filtering result to the summary"""
57 self.total_count += 1
59 if result.can_retry:
60 self.downloadable_count += 1
61 elif result.status == "permanently_failed":
62 self.permanently_failed_count += 1
63 elif result.status == "temporarily_failed":
64 self.temporarily_failed_count += 1
65 else:
66 self.available_count += 1
68 def to_dict(self) -> dict:
69 """Convert to dictionary"""
70 return {
71 "total_count": self.total_count,
72 "downloadable_count": self.downloadable_count,
73 "permanently_failed_count": self.permanently_failed_count,
74 "temporarily_failed_count": self.temporarily_failed_count,
75 "available_count": self.available_count,
76 "failure_type_counts": self.failure_type_counts,
77 }
80class RetryManager:
81 """Manage retry logic and prevent endless loops"""
83 def __init__(self, username: str, password: Optional[str] = None):
84 """
85 Initialize the retry manager.
87 Args:
88 username: Username for database access
89 password: Optional password for encrypted database
90 """
91 self.username = username
92 self.failure_classifier = FailureClassifier()
93 self.status_tracker = ResourceStatusTracker(username, password)
95 logger.info(f"Initialized for user: {username}")
97 def should_retry_resource(self, resource_id: int) -> RetryDecision:
98 """
99 Determine if a resource should be retried based on history.
101 Args:
102 resource_id: Resource identifier
104 Returns:
105 RetryDecision with can_retry flag and reasoning
106 """
107 can_retry, reason = self.status_tracker.can_retry(resource_id)
108 return RetryDecision(can_retry=can_retry, reason=reason)
110 def record_attempt(
111 self,
112 resource_id: int,
113 result: Tuple[bool, Optional[str]],
114 status_code: Optional[int] = None,
115 url: str = "",
116 details: str = "",
117 session=None,
118 ) -> None:
119 """
120 Record a download attempt result.
122 Args:
123 resource_id: Resource identifier
124 result: Tuple of (success, error_message)
125 status_code: HTTP status code if available
126 url: URL that was attempted
127 details: Additional error details
128 session: Optional database session to reuse
129 """
130 success, error_message = result
132 if success:
133 # Successful download
134 self.status_tracker.mark_success(resource_id, session=session)
135 logger.info(f"Resource {resource_id} marked as successful")
136 else:
137 # Failed download - classify the failure. ``error_message`` here
138 # is the caller-visible ``skip_reason`` string (PR #6564 fixed
139 # tokens like ``network_timeout``), so ``type(...).__name__`` is
140 # always the literal ``"str"``. Pass single-token values through
141 # as ``error_type`` so ``FailureClassifier`` can restore the
142 # pre-token granularity (``network_timeout`` -> ``timeout``
143 # 30-min instead of ``unknown_error`` 1h). Multi-word free text
144 # keeps the historic ``"str"`` shape; the classifier reads it via
145 # ``details``.
146 if error_message and isinstance(error_message, str): 146 ↛ 158line 146 didn't jump to line 158 because the condition on line 146 was always true
147 stripped = error_message.strip()
148 if (
149 stripped
150 and " " not in stripped
151 and "\n" not in stripped
152 and "\t" not in stripped
153 ):
154 resolved_error_type = stripped
155 else:
156 resolved_error_type = type(error_message).__name__
157 else:
158 resolved_error_type = (
159 type(error_message).__name__ if error_message else "unknown"
160 )
161 failure = self.failure_classifier.classify_failure(
162 error_type=resolved_error_type,
163 status_code=status_code,
164 url=url,
165 details=details or (error_message or "Unknown error"),
166 )
168 self.status_tracker.mark_failure(
169 resource_id, failure, session=session
170 )
171 logger.info(
172 f"Resource {resource_id} marked as failed: {failure.error_type}"
173 )
175 def filter_resources(self, resources: List) -> List[ResourceFilterResult]:
176 """
177 Filter resources based on their retry eligibility.
179 Args:
180 resources: List of resources to filter
182 Returns:
183 List of ResourceFilterResult objects
184 """
185 results = []
187 for resource in resources:
188 if not hasattr(resource, "id"):
189 # Skip resources without ID
190 continue
192 can_retry, reason = self.status_tracker.can_retry(resource.id)
193 status = self._get_resource_status(can_retry, reason)
195 # Get estimated wait time if temporarily failed
196 estimated_wait = None
197 if not can_retry and reason and "cooldown" in reason.lower():
198 try:
199 status_info = self.status_tracker.get_resource_status(
200 resource.id
201 )
202 if status_info and status_info["retry_after_timestamp"]: 202 ↛ 210line 202 didn't jump to line 210 because the condition on line 202 was always true
203 retry_time = datetime.fromisoformat(
204 status_info["retry_after_timestamp"]
205 )
206 estimated_wait = retry_time - datetime.now(UTC)
207 except Exception as e:
208 logger.debug(f"Error calculating wait time: {e}")
210 result = ResourceFilterResult(
211 resource_id=resource.id,
212 can_retry=can_retry,
213 status=status,
214 reason=reason or "",
215 estimated_wait=estimated_wait,
216 )
217 results.append(result)
219 logger.info(
220 f"Filtered {len(results)} resources: "
221 f"{sum(1 for r in results if r.can_retry)} downloadable, "
222 f"{sum(1 for r in results if r.status == 'permanently_failed')} permanently failed"
223 )
225 return results
227 def get_filter_summary(
228 self, results: List[ResourceFilterResult]
229 ) -> FilterSummary:
230 """
231 Generate a summary of filtering results.
233 Args:
234 results: List of ResourceFilterResult objects
236 Returns:
237 FilterSummary object with counts
238 """
239 summary = FilterSummary()
240 for result in results:
241 summary.add_result(result)
242 return summary
244 def _get_resource_status(
245 self, can_retry: bool, reason: Optional[str]
246 ) -> str:
247 """Get status string based on retry decision"""
248 if not can_retry:
249 # Every reason `ResourceStatusTracker.can_retry` produces starts
250 # with a capital ("Permanently failed: ...", "Cooldown active,
251 # ..."), so these must match case-insensitively — as the cooldown
252 # wait estimate in `filter_resources` already does. Matching
253 # lowercase only sent all of them to "unavailable", leaving
254 # `FilterSummary.permanently_failed_count` and
255 # `temporarily_failed_count` at zero however many resources were
256 # in either state.
257 reason_lower = (reason or "").lower()
258 if "permanently failed" in reason_lower:
259 return "permanently_failed"
260 if "cooldown" in reason_lower:
261 return "temporarily_failed"
262 return "unavailable"
263 return "available"
265 def reset_daily_retry_counters(self) -> int:
266 """
267 Reset daily retry counters (call this at midnight).
269 Returns:
270 Number of resources that had their daily counter reset
271 """
272 with self.status_tracker._get_session() as session:
273 # Reset all today_retry_count to 0
274 from .models import ResourceDownloadStatus
276 result = session.query(ResourceDownloadStatus).update(
277 {"today_retry_count": 0, "updated_at": datetime.now(UTC)}
278 )
279 session.commit()
280 logger.info(f"Reset daily retry counters for {result} resources")
281 return result
283 def clear_old_permanent_failures(self, days: int = 30) -> int:
284 """
285 Clear old permanent failure records.
287 Args:
288 days: Clear failures older than this many days
290 Returns:
291 Number of records cleared
292 """
293 return self.status_tracker.clear_permanent_failures(days)