Coverage for src/sankey_cashflow/settings.py: 95%

1import logging 

2from os import path 

3from typing import Union 

4 

5import pandas as pd 

6 

7from .utils import logger 

8 

9 

10class AppSettings: 

11 """ 

12 Application settings and defaults + validation, getters/setters, etc 

13 """ 

14 def __init__(self, args): 

15 # args will be from argparser 

16 self.DEFAULT_START_DATE = pd.to_datetime("10/1/2022") 

17 # A csv file or a google Sheets document containing transactions data. 

18 # (In the case of the latter, a sheet name must be provided as well) 

19 self.data_source = args.source 

20 self.audit_mode = args.audit 

21 # The name (or prefix plus wildcard) of the sheet containing the transactions data 

22 # (if data_source is a google Sheets document 

23 self.data_sheet = args.sheet or "Transactions_*" 

24 # A csv file or sheet name containing sources and targets 

25 self._labels_source = args.srcmap or "Sources-Targets" 

26 self.filter_dates = args.range 

27 self.separate_taxes = args.separate_tax 

28 self.verbose = args.verbose 

29 self._g_creds = args.creds or './google_service_account_key.json' 

30 self.distribute_amounts = args.distributions 

31 self.all_time = args.all_time 

32 self.recurring = args.recurring 

33 self.base_title = "Cashflow" 

34 self._date_filter_start = None 

35 self._date_filter_end = None 

36 self.tags = None 

37 self.feed_in = None 

38 self.exclude_tags = None 

39 self.stores = None 

40 self.tag_override = False 

41 self.hover = "Category" 

42 self.chart_resolution = None 

43 self.sales_tax_classification = "Taxes" 

44 self.tip_classification = "xTips" 

45 self.diagram_type = "sankey" 

46 if args.verbose: 

47 logger.setLevel(logging.DEBUG) 

48 logger.handlers[0].setLevel(logging.DEBUG) # Assumes only one handler 

49 if args.hover: 

50 if args.hover.lower() in ["desc", "stores", "description"]: 

51 self.hover = "Description" 

52 if args.hover == "tags": 

53 logger.warning("Tags in hovertext not yet implemented!") 

54 if args.hover.lower() in ["none", "no", "false"]: 

55 self.hover = None 

56 if args.dtype: 

57 if args.dtype.lower() in ["sankey", "line"]: 

58 self.diagram_type = args.dtype.lower() 

59 else: 

60 logger.warning(f"Unknown diagram type: {args.dtype}") 

61 if args.tags: 

62 self.tags = [i.strip() for i in args.tags.split(',')] 

63 if args.tag_override: 

64 self.tag_override = True 

65 if args.feed_in: 

66 self.feed_in = True 

67 if args.exclude: 

68 self.exclude_tags = [i.strip() for i in args.exclude.split(',')] 

69 if args.stores: 

70 self.stores = [i.strip() for i in args.stores.split(',')] 

71 self.colors = {} # label: [link, node] 

72 if self.tags and self.stores: 

73 raise Exception("Stores and tags visualizations should not be combined!") 

74 self.validate_sources() 

75 

76 @property 

77 def date_filter_start(self) -> Union[pd.Timestamp, None]: 

78 return self._date_filter_start 

79 

80 @date_filter_start.setter 

81 def date_filter_start(self, val: str) -> None: 

82 if not val or len(val) == 0: 

83 self._date_filter_start = None 

84 else: 

85 self._date_filter_start = pd.to_datetime(val) 

86 

87 @property 

88 def date_filter_end(self) -> Union[pd.Timestamp, None]: 

89 return self._date_filter_end 

90 

91 @date_filter_end.setter 

92 def date_filter_end(self, val: str) -> None: 

93 if not val or len(val) == 0: 

94 self._date_filter_end = None 

95 else: 

96 self._date_filter_end = pd.to_datetime(val) 

97 

98 @property 

99 def g_creds(self) -> str: 

100 return self._g_creds 

101 

102 @g_creds.setter 

103 def g_creds(self, val: str) -> None: 

104 if not val or len(val) == 0 or not path.isfile(val): 

105 raise Exception(f"Credentials file not found: {val}") 

106 self._g_creds = val 

107 

108 @property 

109 def labels_source(self) -> str: 

110 return self._labels_source 

111 

112 @labels_source.setter 

113 def labels_source(self, val: str) -> None: 

114 if not val or len(val) == 0 or (val.endswith('.csv') and not path.isfile(val)): 

115 raise Exception(f"Sources-targets file not found: {val}") 

116 self._labels_source = val 

117 

118 def source_data_location(self) -> str: 

119 if self.data_source.endswith('.csv'): 

120 return self.data_source 

121 else: 

122 return f"{self.data_source}: {self.data_sheet}" 

123 

124 def validate_sources(self) -> None: 

125 # check sources etc 

126 if not self.data_source or len(self.data_source) == 0: 

127 logger.warning("Please enter a valid data source!") 

128 raise Exception("Missing data source.") 

129 

130 if self.data_source.endswith(".csv"): 

131 logger.debug(f"Using csv data source: {self.data_source}") 

132 # Using csv data source 

133 # Note: additional data validation happens when loading this data 

134 if not self.labels_source or not self.labels_source.endswith(".csv"): 

135 raise Exception("A csv sources-targets sheet must be used when using csv source data.") 

136 if not path.isfile(self.data_source): 

137 raise Exception(f"Could not find provided data source: {self.data_source}") 

138 if not path.isfile(self.labels_source): 

139 raise Exception(f"Could not find provided sources-targets source: {self.labels_source}") 

140 else: 

141 # Using Google Sheets data source 

142 # Note: additional access/permissions/data validation happens when fetching and loading this data 

143 logger.debug(f"Using Google Sheets data source: {self.data_source}") 

144 if not self.data_sheet or len(self.data_sheet) == 0: 

145 raise Exception("Missing Google worksheet name.") 

146 if not self.labels_source or len(self.labels_source) == 0: 

147 raise Exception("A sources-targets sheet name must be supplied.") 

148 if not self.g_creds or len(self.g_creds) == 0: 

149 raise Exception("Google service account credentials must be provided.") 

150 if not path.isfile(self.g_creds): 

151 raise Exception(f"Invalid service account credential file provided: {self.g_creds}")