Add subcommand to list savings plans

README.md

@@ -62,12 +62,12 @@ If you want to use the cutting-edge version, use this command instead:
 <!-- runcmd code:console uv run --python 3.13 pytr help --for-readme -->
 ```console
 usage: pytr [-h] [-V] [-v {warning,info,debug}] [--debug-logfile DEBUG_LOGFILE] [--debug-log-filter DEBUG_LOG_FILTER]
-            {help,login,portfolio,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,completion} ...
+            {help,login,portfolio,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion} ...
 
 Use "pytr command_name --help" to get detailed help to a specific command
 
 Commands:
-  {help,login,portfolio,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,completion}
+  {help,login,portfolio,details,dl_docs,export_transactions,get_price_alarms,set_price_alarms,get_savings_plans,completion}
                                         Desired action to perform
     help                                Print this help message
     login                               Check if credentials file exists. If not create it and ask for input. Try to
@@ -81,6 +81,7 @@ Commands:
                                         into account_transactions.csv.
     get_price_alarms                    Get current price alarms
     set_price_alarms                    Set new price alarms
+    get_savings_plans                   Get current savings plans
     completion                          Print shell tab completion
 
 Options:

pytr/main.py

@@ -17,6 +17,7 @@
 from pytr.dl import DL
 from pytr.event import Event
 from pytr.portfolio import PORTFOLIO_COLUMNS, Portfolio
+from pytr.savings_plans import SavingsPlans
 from pytr.timeline import Timeline
 from pytr.transactions import SUPPORTED_LANGUAGES, TransactionExporter
 from pytr.utils import check_version, get_logger
@@ -381,6 +382,23 @@ def formatter(prog):
         nargs="?",
     )
 
+    # get_savings_plans
+    info = "Get current savings plans"
+    parser_get_savings_plans = parser_cmd.add_parser(
+        "get_savings_plans",
+        formatter_class=formatter,
+        parents=[parser_login_args, parser_lang, parser_decimal_localization],
+        help=info,
+        description=info,
+    )
+    parser_get_savings_plans.add_argument(
+        "--outputfile",
+        help="Output file path",
+        type=argparse.FileType("w", encoding="utf-8"),
+        default="-",
+        nargs="?",
+    )
+
     # completion
     info = "Print shell tab completion"
     parser_completion = parser_cmd.add_parser(
@@ -447,7 +465,7 @@ def main():
             waf_token=args.waf_token,
         )
     elif args.command == "portfolio":
-        p = Portfolio(
+        Portfolio(
             login(
                 phone_no=args.phone_no,
                 pin=args.pin,
@@ -460,8 +478,7 @@ def main():
             output=args.output,
             sort_by_column=args.sort_by_column,
             sort_descending=not args.sort_ascending,
-        )
-        p.get()
+        ).get()
     elif args.command == "details":
         Details(
             login(
@@ -565,6 +582,18 @@ def main():
         except ValueError as e:
             print(e)
             return -1
+    elif args.command == "get_savings_plans":
+        SavingsPlans(
+            login(
+                phone_no=args.phone_no,
+                pin=args.pin,
+                store_credentials=args.store_credentials,
+                waf_token=args.waf_token,
+            ),
+            args.outputfile,
+            decimal_localization=args.decimal_localization,
+            lang=args.lang,
+        ).get()
     elif args.version:
         installed_version = version("pytr")
         print(installed_version)

pytr/savings_plans.py

@@ -0,0 +1,93 @@
+import asyncio
+import csv
+import platform
+import sys
+from locale import getdefaultlocale
+
+from babel.numbers import format_decimal
+
+from pytr.utils import get_logger, preview
+
+
+class SavingsPlans:
+    def __init__(self, tr, fp=None, decimal_localization=False, lang="en"):
+        self.tr = tr
+        self.fp = fp
+        self.decimal_localization = decimal_localization
+        self.log = get_logger(__name__)
+        self.savings_plans = []
+
+        self.lang = lang
+        if self.lang == "auto":
+            default_locale = getdefaultlocale()[0]
+            self.lang = default_locale.split("_")[0] if default_locale else "en"
+
+    async def savings_plans_loop(self):
+        await self.tr.savings_plan_overview()
+        while True:
+            _, subscription, response = await self.tr.recv()
+
+            if subscription["type"] == "savingsPlans":
+                self.savings_plans = response.get("savingsPlans", [])
+                return
+            else:
+                print(f"unmatched subscription of type '{subscription['type']}':\n{preview(response)}")
+
+    def _format_amount(self, value):
+        if value is None:
+            return ""
+        if self.decimal_localization:
+            return format_decimal(value, format="#,##0.##", locale=self.lang)
+        return str(value)
+
+    def overview(self):
+        if not self.savings_plans:
+            print("No savings plans found.")
+            return
+
+        fieldnames = [
+            "instrumentId",
+            "amount",
+            "interval",
+            "nextExecutionDate",
+            "previousExecutionDate",
+            "paused",
+        ]
+
+        def format_plan(plan):
+            row = {}
+            for f in fieldnames:
+                val = plan.get(f, "")
+                if f == "amount":
+                    val = self._format_amount(val)
+                row[f] = val
+            return row
+
+        if self.fp == sys.stdout:
+            header = "  ".join(f"{f}" for f in fieldnames)
+            print(header)
+            for plan in self.savings_plans:
+                formatted = format_plan(plan)
+                row = "  ".join(str(formatted.get(f, "")) for f in fieldnames)
+                print(row)
+        else:
+            print(f"Writing savings plans to file {self.fp.name}...")
+            lineterminator = "\n" if platform.system() == "Windows" else "\r\n"
+            writer = csv.DictWriter(
+                self.fp,
+                fieldnames=fieldnames,
+                delimiter=";",
+                lineterminator=lineterminator,
+                extrasaction="ignore",
+            )
+            writer.writeheader()
+            writer.writerows([format_plan(plan) for plan in self.savings_plans])
+            self.fp.close()
+
+    def get(self):
+        async def get_and_close():
+            await self.savings_plans_loop()
+            await self.tr.close()
+
+        asyncio.run(get_and_close())
+        self.overview()

pytr/transactions.py

@@ -75,11 +75,8 @@ def __post_init__(self):
         self._log = get_logger(__name__)
 
         if self.lang == "auto":
-            locale = getdefaultlocale()[0]
-            if locale is None:
-                self.lang = "en"
-            else:
-                self.lang = locale.split("_")[0]
+            default_locale = getdefaultlocale()[0]
+            self.lang = default_locale.split("_")[0] if default_locale else "en"
 
         if self.lang not in SUPPORTED_LANGUAGES:
             self._log.info(f'Language not yet supported "{self.lang}", defaulting to "en"')