Coverage for functions \ flipdare \ task \ report \ core \ time_series_report.py: 78%

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14from typing import Literal, override 

15from collections.abc import Callable 

16from flipdare.app_types import ReportListType 

17from flipdare.task.report.core._admin_report import AdminReport 

18from flipdare.backend.app_logger import AppLogger 

19from flipdare.analysis.data._time_series_protocol import TimeSeriesProtocol 

20from flipdare.analysis.plot.time_series_plotter import TimeSeriesPlotter 

21from flipdare.app_log import LOG 

22from flipdare.constants import IS_DEBUG 

23from flipdare.mailer.core.email_composer import EmailComposer 

24from flipdare.mailer.email_image import EmailImage 

25from flipdare.mailer.admin_mailer import AdminMailer 

26from flipdare.generated.shared.backend.app_job_type import AppJobType 

27 

28type TimeSeriesReportType = Literal[ 

29 AppJobType.REPORT_LOG_STATS, 

30 AppJobType.REPORT_ERROR_STATS, 

31 AppJobType.REPORT_JOB_TYPE_STATS, 

32 AppJobType.REPORT_PAYMENT_STATS, 

33] 

34 

35 

36class TimeSeriesReport(AdminReport): 

37 

38 __slots__ = ("_plot_fn", "_query_fn") 

39 

40 def __init__( 

41 self, 

42 job_type: TimeSeriesReportType, 

43 query_fn: Callable[..., TimeSeriesProtocol], 

44 plot_fn: Callable[[TimeSeriesProtocol], TimeSeriesPlotter], 

45 app_logger: AppLogger, 

46 mailer: AdminMailer, 

47 ) -> None: 

48 super().__init__( 

49 job_type=job_type, 

50 app_logger=app_logger, 

51 mailer=mailer, 

52 ) 

53 self._query_fn = query_fn 

54 self._plot_fn = plot_fn 

55 

56 @override 

57 def compose(self) -> EmailComposer | None: 

58 label = self.job_type.value 

59 

60 try: 

61 stats = self._query_fn() 

62 if IS_DEBUG: 

63 LOG().debug(f"Report {label} generated {stats.count} results") 

64 except Exception as ex: 

65 msg = f"Failed to generate report data for {label}: {ex}" 

66 self.add_error(msg) 

67 return None 

68 

69 image_builder = self._plot_fn(stats) 

70 plots: list[EmailImage] = [] 

71 try: 

72 result = image_builder.create() 

73 if not result.is_error: 

74 plots = result.plots 

75 else: 

76 error_str = "; ".join(result.errors) 

77 msg = f"Report {label} encountered errors during image creation: {error_str}" 

78 self.add_error(msg) 

79 

80 except Exception as ex: 

81 msg = f"Failed to create images for report {label}: {ex}" 

82 self.add_error(msg) 

83 

84 return self._compose_result( 

85 data=stats.table_data(), 

86 headers=stats.headers, 

87 plots=plots, 

88 ) 

89 

90 def _compose_result( 

91 self, 

92 data: ReportListType, 

93 headers: list[str], 

94 plots: list[EmailImage] | None = None, 

95 ) -> EmailComposer | None: 

96 try: 

97 return ( 

98 EmailComposer.table( 

99 data=data, 

100 headers=headers, 

101 priority=self.priority, 

102 ) 

103 if plots is None 

104 else EmailComposer.gallery( 

105 data=data, 

106 headers=headers, 

107 priority=self.priority, 

108 images=plots, 

109 ) 

110 ) 

111 except Exception as ex: 

112 msg = f"Failed to build HTML report: {ex}" 

113 self.add_error(msg) 

114 return None