Improve number parsing and handle a few more events

README.md

@@ -39,7 +39,7 @@ To install uv on OSX/Linux, run:
 $ curl -LsSf https://astral.sh/uv/install.sh | sh
 ```
 
-On Window, run:
+On Windows, run:
 
 ```console
 > powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
@@ -61,7 +61,7 @@ $ uvx --with git+https://github.com/pytr-org/pytr.git pytr
 
 <!-- runcmd code:console COLUMNS=120 uv run --python 3.13 pytr -->
 ```console
-usage: pytr [-h] [-v {warning,info,debug}] [-V]
+usage: pytr [-h] [-V] [-v {warning,info,debug}] [--debug-logfile DEBUG_LOGFILE] [--debug-log-filter DEBUG_LOG_FILTER]
             {help,login,dl_docs,portfolio,details,get_price_alarms,set_price_alarms,export_transactions,completion} ...
 
 Use "pytr command_name --help" to get detailed help to a specific command
@@ -74,7 +74,7 @@ Commands:
                          device reset if needed
     dl_docs              Download all pdf documents from the timeline and sort them into folders. Also export account
                          transactions (account_transactions.csv) and JSON files with all events
-                         (events_with_documents.json and other_events.json
+                         (events_with_documents.json and other_events.json)
     portfolio            Show current portfolio
     details              Get details for an ISIN
     get_price_alarms     Get overview of current price alarms
@@ -84,9 +84,13 @@ Commands:
 
 Options:
   -h, --help             show this help message and exit
+  -V, --version          Print version information and quit
   -v, --verbosity {warning,info,debug}
                          Set verbosity level (default: info)
-  -V, --version          Print version information and quit
+  --debug-logfile DEBUG_LOGFILE
+                         Dump debug logs to a file
+  --debug-log-filter DEBUG_LOG_FILTER
+                         Filter debug log types
 ```
 <!-- end runcmd -->
 

pytr/event.py

@@ -1,3 +1,4 @@
+import json
 import re
 from dataclasses import dataclass
 from datetime import datetime
@@ -6,6 +7,8 @@
 
 from babel.numbers import NumberFormatError, parse_decimal
 
+from pytr.utils import get_logger
+
 
 class EventType(Enum):
     pass
@@ -51,15 +54,14 @@ class PPEventType(EventType):
     # Dividends
     "CREDIT": PPEventType.DIVIDEND,
     "ssp_corporate_action_invoice_cash": PPEventType.DIVIDEND,
-    # Failed card transactions
-    "card_failed_transaction": PPEventType.REMOVAL,
     # Interests
     "INTEREST_PAYOUT": PPEventType.INTEREST,
     "INTEREST_PAYOUT_CREATED": PPEventType.INTEREST,
     # Removals
     "OUTGOING_TRANSFER": PPEventType.REMOVAL,
     "OUTGOING_TRANSFER_DELEGATION": PPEventType.REMOVAL,
     "PAYMENT_OUTBOUND": PPEventType.REMOVAL,
+    "card_failed_transaction": PPEventType.REMOVAL,
     "card_order_billed": PPEventType.REMOVAL,
     "card_successful_atm_withdrawal": PPEventType.REMOVAL,
     "card_successful_transaction": PPEventType.REMOVAL,
@@ -72,10 +74,49 @@ class PPEventType(EventType):
     "ORDER_EXECUTED": ConditionalEventType.TRADE_INVOICE,
     "SAVINGS_PLAN_EXECUTED": ConditionalEventType.TRADE_INVOICE,
     "SAVINGS_PLAN_INVOICE_CREATED": ConditionalEventType.TRADE_INVOICE,
+    "trading_savingsplan_executed": ConditionalEventType.TRADE_INVOICE,
+    "trading_trade_executed": ConditionalEventType.TRADE_INVOICE,
     "benefits_spare_change_execution": ConditionalEventType.TRADE_INVOICE,
     "TRADE_INVOICE": ConditionalEventType.TRADE_INVOICE,
+    "TRADE_CORRECTED": ConditionalEventType.TRADE_INVOICE,
 }
 
+logger = None
+
+
+def get_event_logger():
+    global logger
+    if logger is None:
+        logger = get_logger(__name__)
+    return logger
+
+
+events_known_ignored = [
+    "CUSTOMER_CREATED",
+    "DEVICE_RESET",
+    "DOCUMENTS_ACCEPTED",
+    "DOCUMENTS_CREATED",
+    "EMAIL_VALIDATED",
+    "ORDER_CANCELED",
+    "ORDER_CREATED",
+    "ORDER_EXPIRED",
+    "ORDER_REJECTED",
+    "PUK_CREATED",
+    "REFERENCE_ACCOUNT_CHANGED",
+    "SECURITIES_ACCOUNT_CREATED",
+    "VERIFICATION_TRANSFER_ACCEPTED",
+    "card_failed_verification",
+    "card_successful_verification",
+    "current_account_activated",
+    "new_tr_iban",
+    "trading_order_cancelled",
+    "trading_order_created",
+    "trading_savingsplan_execution_failed",
+    "ssp_corporate_action_informative_notification",
+    "ssp_corporate_action_invoice_shares",
+    "ssp_dividend_option_customer_instruction",
+]
+
 
 @dataclass
 class Event:
@@ -110,9 +151,15 @@ def from_dict(cls, event_dict: Dict[Any, Any]):
 
     @staticmethod
     def _parse_type(event_dict: Dict[Any, Any]) -> Optional[EventType]:
-        event_type: Optional[EventType] = tr_event_type_mapping.get(event_dict.get("eventType", ""), None)
-        if event_dict.get("status", "").lower() == "canceled":
-            event_type = None
+        eventTypeStr = event_dict.get("eventType", "")
+        event_type: Optional[EventType] = tr_event_type_mapping.get(eventTypeStr, None)
+        if event_type is not None:
+            if event_dict.get("status", "").lower() == "canceled":
+                event_type = None
+        else:
+            if eventTypeStr not in events_known_ignored:
+                get_event_logger().warning(f"Ignoring unknown event {eventTypeStr}")
+                get_event_logger().debug("Unknown event %s: %s", eventTypeStr, json.dumps(event_dict, indent=4))
         return event_type
 
     @classmethod
@@ -182,17 +229,37 @@ def _parse_shares_and_fees(cls, event_dict: Dict[Any, Any]) -> Tuple[Optional[fl
         Returns:
             Tuple[Optional[float]]: shares, fees
         """
-        return_vals = {}
+        shares, fees = None, None
+        dump_dict = {"eventType": event_dict["eventType"], "id": event_dict["id"]}
+
         sections = event_dict.get("details", {}).get("sections", [{}])
-        for section in sections:
-            if section.get("title") == "Transaktion":
-                data = section["data"]
-                shares_dicts = list(filter(lambda x: x["title"] in ["Aktien", "Anteile"], data))
-                fees_dicts = list(filter(lambda x: x["title"] == "Gebühr", data))
-                titles = ["shares"] * len(shares_dicts) + ["fees"] * len(fees_dicts)
-                for key, elem_dict in zip(titles, shares_dicts + fees_dicts):
-                    return_vals[key] = cls._parse_float_from_detail(elem_dict)
-        return return_vals.get("shares"), return_vals.get("fees")
+        transaction_dicts = filter(lambda x: x["title"] in ["Transaktion"], sections)
+        for transaction_dict in transaction_dicts:
+            dump_dict["maintitle"] = transaction_dict["title"]
+            data = transaction_dict.get("data", [{}])
+            shares_dicts = filter(lambda x: x["title"] in ["Aktien", "Anteile"], data)
+            for shares_dict in shares_dicts:
+                dump_dict["subtitle"] = shares_dict["title"]
+                dump_dict["type"] = "shares"
+                pref_locale = (
+                    "en"
+                    if event_dict["eventType"] in ["benefits_saveback_execution", "benefits_spare_change_execution"]
+                    and shares_dict["title"] == "Aktien"
+                    else "de"
+                )
+                shares = cls._parse_float_from_detail(shares_dict, dump_dict, pref_locale)
+                break
+
+            fees_dicts = filter(lambda x: x["title"] == "Gebühr", data)
+            for fees_dict in fees_dicts:
+                dump_dict["subtitle"] = fees_dict["title"]
+                dump_dict["type"] = "fees"
+                fees = cls._parse_float_from_detail(fees_dict, dump_dict)
+                break
+
+            break
+
+        return shares, fees
 
     @classmethod
     def _parse_taxes(cls, event_dict: Dict[Any, Any]) -> Optional[float]:
@@ -204,23 +271,26 @@ def _parse_taxes(cls, event_dict: Dict[Any, Any]) -> Optional[float]:
         Returns:
             Optional[float]: taxes
         """
-        # taxes keywords
-        taxes_keys = {"Steuer", "Steuern"}
-        # Gather all section dicts
+        parsed_taxes_val = None
+        dump_dict = {"eventType": event_dict["eventType"], "id": event_dict["id"]}
+        pref_locale = "en" if event_dict["eventType"] in ["INTEREST_PAYOUT"] else "de"
+
         sections = event_dict.get("details", {}).get("sections", [{}])
-        # Gather all dicts pertaining to transactions
-        transaction_dicts = filter(lambda x: x["title"] in {"Transaktion", "Geschäft"}, sections)
+        transaction_dicts = filter(lambda x: x["title"] in ["Transaktion", "Geschäft"], sections)
         for transaction_dict in transaction_dicts:
             # Filter for taxes dicts
+            dump_dict["maintitle"] = transaction_dict["title"]
             data = transaction_dict.get("data", [{}])
-            taxes_dicts = filter(lambda x: x["title"] in taxes_keys, data)
+            taxes_dicts = filter(lambda x: x["title"] in ["Steuer", "Steuern"], data)
             # Iterate over dicts containing tax information and parse each one
             for taxes_dict in taxes_dicts:
-                parsed_taxes_val = cls._parse_float_from_detail(taxes_dict)
-                if parsed_taxes_val is not None:
-                    return parsed_taxes_val
+                dump_dict["subtitle"] = taxes_dict["title"]
+                dump_dict["type"] = "taxes"
+                parsed_taxes_val = cls._parse_float_from_detail(taxes_dict, dump_dict, pref_locale)
+                break
+            break
 
-        return None
+        return parsed_taxes_val
 
     @staticmethod
     def _parse_card_note(event_dict: Dict[Any, Any]) -> Optional[str]:
@@ -237,7 +307,11 @@ def _parse_card_note(event_dict: Dict[Any, Any]) -> Optional[str]:
         return None
 
     @staticmethod
-    def _parse_float_from_detail(elem_dict: Dict[str, Any]) -> Optional[float]:
+    def _parse_float_from_detail(
+        elem_dict: Dict[str, Any],
+        dump_dict={"eventType": "Unknown", "id": "Unknown", "type": "Unknown"},
+        pref_locale="de",
+    ) -> Optional[float]:
         """Parses a "detail" dictionary potentially containing a float in a certain locale format
 
         Args:
@@ -247,13 +321,15 @@ def _parse_float_from_detail(elem_dict: Dict[str, Any]) -> Optional[float]:
             Optional[float]: parsed float value or None
         """
         unparsed_val = elem_dict.get("detail", {}).get("text", "")
+        if unparsed_val == "":
+            return None
         parsed_val = re.sub(r"[^\,\.\d-]", "", unparsed_val)
 
-        # Try the locale that will fail more likely first.
-        if "." not in parsed_val:
-            locales = ("en", "de")
-        else:
+        # Try the preferred locale first
+        if pref_locale == "de":
             locales = ("de", "en")
+        else:
+            locales = ("en", "de")
 
         try:
             result = float(parse_decimal(parsed_val, locales[0], strict=True))
@@ -262,4 +338,33 @@ def _parse_float_from_detail(elem_dict: Dict[str, Any]) -> Optional[float]:
                 result = float(parse_decimal(parsed_val, locales[1], strict=True))
             except NumberFormatError:
                 return None
+            get_event_logger().warning(
+                "Number %s parsed as %s although preference was %s: %s",
+                parsed_val,
+                locales[1],
+                locales[0],
+                json.dumps(dump_dict, indent=4),
+            )
+            return None if result == 0.0 else result
+
+        alternative_result = None
+        if "," in parsed_val or "." in parsed_val:
+            try:
+                alternative_result = float(parse_decimal(parsed_val, locales[1], strict=True))
+            except NumberFormatError:
+                pass
+
+        if alternative_result is None:
+            get_event_logger().debug(
+                "Number %s parsed as %s: %s", parsed_val, locales[0], json.dumps(dump_dict, indent=4)
+            )
+        else:
+            get_event_logger().debug(
+                "Number %s parsed as %s but could also be parsed as %s: %s",
+                parsed_val,
+                locales[0],
+                locales[1],
+                json.dumps(dump_dict, indent=4),
+            )
+
         return None if result == 0.0 else result

pytr/main.py

@@ -32,6 +32,12 @@ def formatter(prog):
         elif grp.title == "positional arguments":
             grp.title = "Commands"
 
+    parser.add_argument(
+        "-V",
+        "--version",
+        help="Print version information and quit",
+        action="store_true",
+    )
     parser.add_argument(
         "-v",
         "--verbosity",
@@ -40,11 +46,17 @@ def formatter(prog):
         default="info",
     )
     parser.add_argument(
-        "-V",
-        "--version",
-        help="Print version information and quit",
-        action="store_true",
+        "--debug-logfile",
+        help="Dump debug logs to a file",
+        type=Path,
+        default=None,
     )
+    parser.add_argument(
+        "--debug-log-filter",
+        help="Filter debug log types",
+        default=None,
+    )
+
     parser_cmd = parser.add_subparsers(help="Desired action to perform", dest="command")
 
     # help
@@ -93,7 +105,7 @@ def formatter(prog):
     info = (
         "Download all pdf documents from the timeline and sort them into folders."
         + " Also export account transactions (account_transactions.csv)"
-        + " and JSON files with all events (events_with_documents.json and other_events.json"
+        + " and JSON files with all events (events_with_documents.json and other_events.json)"
     )
     parser_dl_docs = parser_cmd.add_parser(
         "dl_docs",
@@ -120,7 +132,6 @@ def formatter(prog):
     parser_dl_docs.add_argument(
         "--workers",
         help="Number of workers for parallel downloading",
-        metavar="WORKERS",
         default=8,
         type=int,
     )
@@ -134,7 +145,7 @@ def formatter(prog):
         help=info,
         description=info,
     )
-    parser_portfolio.add_argument("-o", "--output", help="Output path of CSV file", metavar="OUTPUT", type=Path)
+    parser_portfolio.add_argument("-o", "--output", help="Output path of CSV file", type=Path)
     # details
     info = "Get details for an ISIN"
     parser_details = parser_cmd.add_parser(
@@ -183,10 +194,9 @@ def formatter(prog):
     parser_export_transactions.add_argument(
         "input",
         help="Input path to JSON (use other_events.json from dl_docs)",
-        metavar="INPUT",
         type=Path,
     )
-    parser_export_transactions.add_argument("output", help="Output path of CSV file", metavar="OUTPUT", type=Path)
+    parser_export_transactions.add_argument("output", help="Output path of CSV file", type=Path)
     parser_export_transactions.add_argument(
         "-l",
         "--lang",
@@ -238,9 +248,9 @@ def main():
     args = parser.parse_args()
     # print(vars(args))
 
-    log = get_logger(__name__, args.verbosity)
-    log.setLevel(args.verbosity.upper())
-    log.debug("logging is set to debug")
+    log = get_logger(__name__, args.verbosity, args.debug_logfile, args.debug_log_filter)
+    if args.verbosity.upper() == "DEBUG":
+        log.debug("logging is set to debug")
 
     if args.command == "login":
         login(