Coverage for src / local_deep_research / research_scheduler / document_scheduler.py: 100%

1""" 

2Document scheduler utility for automatic processing of research history. 

3This utility class interfaces with the NewsScheduler to provide document processing functionality. 

4""" 

5 

6from typing import Any, Dict 

7 

8from loguru import logger 

9from ..news.subscription_manager.scheduler import get_news_scheduler 

10 

11 

12class DocumentSchedulerUtil: 

13 """ 

14 Utility class for document processing that interfaces with NewsScheduler. 

15 

16 This utility provides a simple interface for the API routes to interact with 

17 the NewsScheduler's document processing functionality without exposing the 

18 underlying scheduler implementation details. 

19 """ 

20 

21 def __init__(self): 

22 """Initialize the document scheduler utility.""" 

23 logger.debug("Document scheduler utility initialized") 

24 

25 def get_status(self, username: str) -> Dict[str, Any]: 

26 """ 

27 Get document processing status for a user. 

28 

29 Args: 

30 username: The username to get status for 

31 

32 Returns: 

33 Dictionary containing document processing status information 

34 """ 

35 try: 

36 scheduler = get_news_scheduler() 

37 return scheduler.get_document_scheduler_status(username) 

38 

39 except Exception: 

40 logger.exception( 

41 f"Error getting document scheduler status for {username}" 

42 ) 

43 return { 

44 "error": "Failed to get scheduler status", 

45 "is_running": False, 

46 "last_run_time": None, 

47 "next_run_time": None, 

48 "total_processed": 0, 

49 "currently_processing": 0, 

50 "processing_ids": [], 

51 "settings": {}, 

52 } 

53 

54 def trigger_manual_run(self, username: str) -> tuple[bool, str]: 

55 """ 

56 Trigger a manual document processing run for a user. 

57 

58 Args: 

59 username: The username to trigger processing for 

60 

61 Returns: 

62 Tuple of (success: bool, message: str) 

63 """ 

64 try: 

65 scheduler = get_news_scheduler() 

66 success = scheduler.trigger_document_processing(username) 

67 

68 if success: 

69 return True, "Manual document processing triggered successfully" 

70 else: 

71 return ( 

72 False, 

73 "Failed to trigger document processing - user may not be active or processing disabled", 

74 ) 

75 

76 except Exception: 

77 logger.exception( 

78 f"Error triggering manual document processing for {username}" 

79 ) 

80 return False, "Failed to trigger manual processing" 

81 

82 

83# Singleton instance getter 

84_scheduler_util_instance = None 

85 

86 

87def get_document_scheduler() -> DocumentSchedulerUtil: 

88 """ 

89 Get the singleton document scheduler utility instance. 

90 

91 Note: This returns the utility class that interfaces with NewsScheduler, 

92 not a standalone scheduler instance. 

93 """ 

94 global _scheduler_util_instance 

95 if _scheduler_util_instance is None: 

96 _scheduler_util_instance = DocumentSchedulerUtil() 

97 return _scheduler_util_instance