Coverage for src/bayernwerk_client/map/cli.py: 93%

1"""`map` subcommand: Mein.Auftragsportal (bayernwerk-netz.de). 

2 

3 bayernwerk map login 

4 bayernwerk map orders 

5 bayernwerk -j map orders 

6 bayernwerk map order 2577481104 

7 bayernwerk map order-instances 2577481104 

8 bayernwerk map sync-documents 2577481104 "/path/to/customer/folder" 

9 

10Run `bayernwerk map login` once (reads MAP_EMAIL/MAP_PASSWORD from the 

11environment, or prompts interactively) to populate the local token cache - 

12see `auth.py` for why that's a real (visible by default) browser rather 

13than a plain HTTP request. Later commands reuse the cached token and 

14transparently re-run the login when it expires, provided MAP_EMAIL/ 

15MAP_PASSWORD are set in the environment; otherwise they'll ask you to run 

16`login` again. 

17 

18`-j`/`--json` is a top-level flag on `bayernwerk` itself (before `map`), 

19not per-command - see `bayernwerk_client.cli`. 

20""" 

21 

22from __future__ import annotations 

23 

24import argparse 

25import getpass 

26import json 

27import os 

28from typing import Any 

29 

30from bayernwerk_client.exceptions import BayernwerkClientError 

31from bayernwerk_client.formatting import format_result 

32from bayernwerk_client.map.client import MAP_TOKEN_PATH, MapClient 

33from bayernwerk_client.map.formatting import is_order_finished, render_instance_tree, render_orders 

34from bayernwerk_client.tokens import TokenSet, TokenStore 

35 

36 

37def _relogin_from_env(_old_tokens: TokenSet) -> TokenSet: 

38 from bayernwerk_client.map.auth import login_interactive 

39 

40 email = os.environ.get("MAP_EMAIL") 

41 password = os.environ.get("MAP_PASSWORD") 

42 if not email or not password: 

43 raise BayernwerkClientError( 

44 "Access-Token abgelaufen und MAP_EMAIL/MAP_PASSWORD sind nicht gesetzt - " 

45 "bitte 'bayernwerk map login' erneut ausfuehren." 

46 ) 

47 return login_interactive(email, password, headless=False) 

48 

49 

50def _make_client() -> MapClient: 

51 return MapClient.from_token_store(TokenStore(MAP_TOKEN_PATH), on_token_expired=_relogin_from_env) 

52 

53 

54def _output(data: Any, *, as_json: bool) -> None: 

55 if as_json: 

56 print(json.dumps(data, ensure_ascii=False, indent=2)) 

57 else: 

58 print(format_result(data)) 

59 

60 

61def cmd_login(args: argparse.Namespace) -> None: 

62 from bayernwerk_client.map.auth import login_interactive 

63 

64 email = os.environ.get("MAP_EMAIL") or input("E-Mail: ") 

65 password = os.environ.get("MAP_PASSWORD") or getpass.getpass("Passwort: ") 

66 tokens = login_interactive(email, password, headless=args.headless) 

67 TokenStore(MAP_TOKEN_PATH).save(tokens) 

68 print("Login erfolgreich, Token gespeichert.") 

69 

70 

71def cmd_orders(args: argparse.Namespace) -> None: 

72 with _make_client() as client: 

73 orders = client.list_orders() 

74 

75 if args.finished: 

76 orders = [order for order in orders if is_order_finished(order)] 

77 elif args.open: 

78 orders = [order for order in orders if not is_order_finished(order)] 

79 

80 if args.json: 

81 _output(orders, as_json=True) 

82 else: 

83 print(render_orders(orders, short=args.short)) 

84 

85 

86def cmd_order(args: argparse.Namespace) -> None: 

87 with _make_client() as client: 

88 _output(client.get_order(args.order_id), as_json=args.json) 

89 

90 

91def cmd_order_details(args: argparse.Namespace) -> None: 

92 with _make_client() as client: 

93 _output(client.get_order_details(args.order_id), as_json=args.json) 

94 

95 

96def cmd_order_documents(args: argparse.Namespace) -> None: 

97 with _make_client() as client: 

98 _output(client.get_order_documents(args.order_id), as_json=args.json) 

99 

100 

101def cmd_order_notes(args: argparse.Namespace) -> None: 

102 with _make_client() as client: 

103 _output(client.get_order_notes(args.order_id), as_json=args.json) 

104 

105 

106def cmd_order_instances(args: argparse.Namespace) -> None: 

107 with _make_client() as client: 

108 if args.json: 

109 _output(list(client.iter_order_instance_items(args.order_id)), as_json=True) 

110 else: 

111 print(render_instance_tree(client.get_order_instances(args.order_id))) 

112 

113 

114def cmd_sync_documents(args: argparse.Namespace) -> None: 

115 with _make_client() as client: 

116 downloaded = client.sync_order_documents(args.order_id, args.folder) 

117 

118 if args.json: 

119 print(json.dumps({"downloaded": downloaded}, ensure_ascii=False, indent=2)) 

120 elif downloaded: 

121 print(f"{len(downloaded)} neue(s) Dokument(e) heruntergeladen:") 

122 for filename in downloaded: 

123 print(" -", filename) 

124 else: 

125 print("Alles bereits vorhanden, nichts heruntergeladen.") 

126 

127 

128def cmd_installer(args: argparse.Namespace) -> None: 

129 with _make_client() as client: 

130 _output(client.list_installers(), as_json=args.json) 

131 

132 

133def cmd_inverters(args: argparse.Namespace) -> None: 

134 with _make_client() as client: 

135 _output(client.list_inverters(primary_energy_forms=args.primary_energy_forms), as_json=args.json) 

136 

137 

138def cmd_storages(args: argparse.Namespace) -> None: 

139 with _make_client() as client: 

140 _output(client.list_storages(), as_json=args.json) 

141 

142 

143def cmd_product_orders(args: argparse.Namespace) -> None: 

144 with _make_client() as client: 

145 _output(client.list_product_orders(), as_json=args.json) 

146 

147 

148def register(subparsers: argparse._SubParsersAction) -> None: 

149 """Add the `map` subcommand (and its own sub-subcommands) to `subparsers`.""" 

150 map_parser = subparsers.add_parser("map", help="Mein.Auftragsportal (bayernwerk-netz.de)") 

151 map_sub = map_parser.add_subparsers(dest="map_command", required=True) 

152 

153 def add(name: str, *, help: str) -> argparse.ArgumentParser: 

154 """`add_parser` with `help` also used as `description`, so `--help` 

155 on the leaf command itself shows more than a bare usage line.""" 

156 return map_sub.add_parser(name, help=help, description=help) 

157 

158 p = add("login", help="Einmalig einloggen und Token cachen") 

159 p.add_argument("--headless", action="store_true", help="Browser beim Login nicht anzeigen (Standard: sichtbar)") 

160 p.set_defaults(func=cmd_login) 

161 

162 p = add("orders", help="Alle Auftraege auflisten") 

163 p.add_argument( 

164 "-s", 

165 "--short", 

166 action="store_true", 

167 help="Nur die Kopfzeile pro Auftrag ausgeben (wirkt nur auf die Textausgabe, nicht auf -j/--json)", 

168 ) 

169 status_group = p.add_mutually_exclusive_group() 

170 status_group.add_argument( 

171 "-f", "--finished", action="store_true", help="Nur abgeschlossene Auftraege (Status FINISHED)" 

172 ) 

173 status_group.add_argument("-o", "--open", action="store_true", help="Nur offene (nicht abgeschlossene) Auftraege") 

174 p.set_defaults(func=cmd_orders) 

175 

176 p = add("order", help="Einen Auftrag anzeigen") 

177 p.add_argument("order_id", help="Auftragsnummer, z.B. 2577481104") 

178 p.set_defaults(func=cmd_order) 

179 

180 p = add("order-details", help="Details eines Auftrags anzeigen") 

181 p.add_argument("order_id", help="Auftragsnummer, z.B. 2577481104") 

182 p.set_defaults(func=cmd_order_details) 

183 

184 p = add("order-documents", help="Dokumente eines Auftrags auflisten") 

185 p.add_argument("order_id", help="Auftragsnummer, z.B. 2577481104") 

186 p.set_defaults(func=cmd_order_documents) 

187 

188 p = add("order-notes", help="Notizen eines Auftrags auflisten") 

189 p.add_argument("order_id", help="Auftragsnummer, z.B. 2577481104") 

190 p.set_defaults(func=cmd_order_notes) 

191 

192 p = add("order-instances", help="Anschluss-Baum eines Auftrags anzeigen") 

193 p.add_argument("order_id", help="Auftragsnummer, z.B. 2577481104") 

194 p.set_defaults(func=cmd_order_instances) 

195 

196 p = add("sync-documents", help="Fehlende Dokumente eines Auftrags in einen Ordner laden") 

197 p.add_argument("order_id", help="Auftragsnummer, z.B. 2577481104") 

198 p.add_argument("folder", help="Zielordner, wird bei Bedarf angelegt") 

199 p.set_defaults(func=cmd_sync_documents) 

200 

201 add("installer", help="Eigene Installateur-Stammdaten anzeigen").set_defaults(func=cmd_installer) 

202 

203 p = add("inverters", help="Wechselrichter-Stammdaten auflisten") 

204 p.add_argument( 

205 "--primary-energy-form", 

206 action="append", 

207 required=True, 

208 dest="primary_energy_forms", 

209 metavar="FORM", 

210 help="z.B. PV oder AC_STORAGE (mehrfach angebbar)", 

211 ) 

212 p.set_defaults(func=cmd_inverters) 

213 

214 add("storages", help="Speicher-Stammdaten auflisten").set_defaults(func=cmd_storages) 

215 add("product-orders", help="Produktauftraege auflisten").set_defaults(func=cmd_product_orders)