Coverage for hledger_lots/prompt.py: 28%

1import subprocess 

2from dataclasses import dataclass 

3from datetime import datetime 

4from typing import List, Optional, Tuple 

5 

6import questionary 

7from prompt_toolkit.shortcuts import CompleteStyle 

8 

9from .avg_info import AllAvgInfo 

10from .fifo_info import AllFifoInfo 

11from .info import LotsInfo 

12from .lib import get_files_comm 

13 

14 

15class PromptError(BaseException): 

16 def __init__(self, message: str) -> None: 

17 self.message = message 

18 super().__init__(self.message) 

19 

20 

21@dataclass 

22class Tradeinfo: 

23 date: str 

24 quantity: float 

25 commodity: str 

26 cash_account: str 

27 commodity_account: str 

28 price: float 

29 value: float 

30 

31 

32def custom_autocomplete(name: str, choices: List[str]): 

33 question = questionary.autocomplete( 

34 f"{name} (TAB to autocomplete)", 

35 choices=choices, 

36 ignore_case=True, 

37 match_middle=True, 

38 style=questionary.Style([("answer", "fg:#f71b07")]), 

39 complete_style=CompleteStyle.MULTI_COLUMN, 

40 ) 

41 return question 

42 

43 

44def get_append_file(default_file: str): 

45 confirm = questionary.confirm( 

46 "Add sale transaction to a journal", default=False, auto_enter=True 

47 ).ask() 

48 

49 if confirm: 

50 file_append: str = questionary.path( 

51 "File to append transaction", default=default_file 

52 ).ask() 

53 comm = ["hledger", "-f", file_append, "check"] 

54 subprocess.run(comm, check=True) 

55 return file_append 

56 

57 

58def select_commodities_text(commodities: List[str]): 

59 answer = questionary.select( 

60 "Commodity", 

61 choices=commodities, 

62 use_shortcuts=True, 

63 ).ask() 

64 return answer 

65 

66 

67def ask_commodities_text(commodities: List[str]): 

68 answer: str = custom_autocomplete("Commodity", commodities).ask() 

69 return answer 

70 

71 

72def val_date(date: str): 

73 try: 

74 datetime.strptime(date, "%Y-%m-%d") 

75 return True 

76 except ValueError: 

77 return "Invalid date format" 

78 

79 

80def val_sell_qtty(answer: str, available: float): 

81 try: 

82 answer_float = float(answer) 

83 except ValueError: 

84 return "Invalid number" 

85 

86 if answer_float <= 0: 

87 return "Quantity should be positive" 

88 

89 if answer_float > available: 

90 return "Quantity should be less than available" 

91 

92 return True 

93 

94 

95def val_price(answer: str): 

96 if answer == "": 

97 return True 

98 

99 try: 

100 answer_float = float(answer) 

101 except ValueError: 

102 return "Invalid number" 

103 

104 if answer_float < 0: 

105 return "Price should be positive" 

106 

107 return True 

108 

109 

110def val_total(answer: str): 

111 try: 

112 answer_float = float(answer) 

113 except ValueError: 

114 return "Invalid number" 

115 

116 if answer_float < 0: 

117 return "Amount should be positive" 

118 

119 return True 

120 

121 

122class Prompt: 

123 def __init__( 

124 self, 

125 file: Tuple[str, ...], 

126 avg_cost: bool, 

127 check: bool, 

128 no_desc: Optional[str] = None, 

129 ) -> None: 

130 self.file = file 

131 self.check = check 

132 self.no_desc = no_desc 

133 self.avg_cost = avg_cost 

134 

135 self.files_comm = get_files_comm(file) 

136 self.infos = self.get_infos() 

137 self.commodities = [info["comm"] for info in self.get_infos()] 

138 

139 def run_hledger(self, *comm: str): 

140 command = ["hledger", *self.files_comm, *comm] 

141 

142 if self.no_desc and self.no_desc!= "": 

143 command = [*command, f"not:desc:{self.no_desc}"] 

144 

145 proc = subprocess.run(command, capture_output=True) 

146 if proc.returncode != 0: 

147 raise subprocess.SubprocessError(proc.stderr.decode("utf8")) 

148 

149 result = proc.stdout.decode("utf8") 

150 return result 

151 

152 def run_hledger_no_query_desc(self, *comm: str): 

153 command = ["hledger", *self.files_comm, *comm] 

154 proc = subprocess.run(command, capture_output=True) 

155 if proc.returncode != 0: 

156 raise subprocess.SubprocessError(proc.stderr.decode("utf8")) 

157 

158 result = proc.stdout.decode("utf8") 

159 return result 

160 

161 def get_infos(self): 

162 if self.avg_cost: 

163 infos = AllAvgInfo(self.file, self.no_desc or "", self.check) 

164 else: 

165 infos = AllFifoInfo(self.file, self.no_desc or "", self.check) 

166 

167 valid_infos = [info for info in infos.infos if float(info["qtty"]) > 0] 

168 return valid_infos 

169 

170 def get_last_purchase(self, info: LotsInfo): 

171 commodity = info["comm"] 

172 reg = self.run_hledger("reg", f"cur:{commodity}", "amt:>0") 

173 rows_list = [row for row in reg.split("\n") if row != ""] 

174 

175 if len(rows_list) > 0: 

176 last_date = rows_list[-1][0:10] 

177 return last_date 

178 

179 def get_append_file(self): 

180 default_file = self.file[0] 

181 

182 confirm = questionary.confirm( 

183 "Add sale transaction to a journal", default=False, auto_enter=True 

184 ).ask() 

185 

186 if confirm: 

187 file_append: str = questionary.path( 

188 "File to append transaction", default=default_file 

189 ).ask() 

190 comm = ["hledger", "-f", file_append, "check"] 

191 subprocess.run(comm, check=True) 

192 return file_append 

193 

194 def ask_date(self, last_purchase: Optional[str]): 

195 last_purchase = last_purchase 

196 

197 answer: str = questionary.text( 

198 f"Date YYYY-MM-DD", 

199 validate=val_date, 

200 instruction=f"(Last Purchase: {last_purchase})", 

201 ).ask() 

202 return answer 

203 

204 def ask_sell_qtty(self, info: LotsInfo): 

205 available = float(info["qtty"]) 

206 

207 answer_str: str = questionary.text( 

208 f"Quantity (available {available})", 

209 validate=lambda answer: val_sell_qtty(answer, available), 

210 instruction="", 

211 ).ask() 

212 return answer_str 

213 

214 def ask_price(self, info: LotsInfo): 

215 cost_str = info["avg_cost"].replace(",", ".") 

216 cost = float(cost_str) 

217 

218 answer_str: str = questionary.text( 

219 f"Price (avg cost: {cost})", 

220 validate=val_price, 

221 instruction="Empty to input total amount", 

222 ).ask() 

223 return answer_str 

224 

225 def ask_total(self, qtty: float, info: LotsInfo): 

226 available = qtty * float(info["avg_cost"]) 

227 available_str = f"{available:,.2f}" 

228 

229 answer_str: str = questionary.text( 

230 f"Total (available {available_str})", 

231 validate=val_total, 

232 instruction="Amount received", 

233 ).ask() 

234 return answer_str 

235 

236 def ask_cash_account(self): 

237 accts_txt = self.run_hledger("accounts") 

238 accts = [acct for acct in accts_txt.split("\n") if acct != ""] 

239 answer: str = custom_autocomplete("Cash Account", accts).ask() 

240 return answer 

241 

242 def ask_revenue_account(self): 

243 accts_txt = self.run_hledger("accounts") 

244 accts = [acct for acct in accts_txt.split("\n") if acct != ""] 

245 answer: str = custom_autocomplete("Revenue Account", accts).ask() 

246 return answer 

247 

248 @property 

249 def initial_info(self): 

250 cost_method_text = "Average Cost" if self.avg_cost else "Fifo" 

251 no_desc_text = self.no_desc or "None" 

252 check_text = "Checking" if self.check else "Not Checking" 

253 

254 files = [f if f != "-" else "stdin" for f in self.file] 

255 files_text = " ".join(files) 

256 

257 result = f""" 

258Files : {files_text} 

259Cost Method : {cost_method_text} - {check_text} 

260Remove description : {no_desc_text} 

261""" 

262 return result