Improve locale selection for number parsing and add possibility to dump data

Author: RealCLanger

pytr/event.py

@@ -1,3 +1,4 @@
+import pprint
 import re
 from dataclasses import dataclass
 from datetime import datetime
@@ -6,7 +7,7 @@
 
 from babel.numbers import NumberFormatError, parse_decimal
 
-from pytr.utils import get_logger
+from pytr.utils import dump, dump_enabled, get_logger
 
 
 class EventType(Enum):
@@ -145,7 +146,9 @@ def _parse_type(event_dict: Dict[Any, Any]) -> Optional[EventType]:
                 event_type = None
         else:
             if eventTypeStr not in events_known_ignored:
-                log.warning(f"Ignoring event {eventTypeStr}")
+                log.warning(f"Ignoring unknown event {eventTypeStr}")
+                if dump_enabled():
+                    dump(f"Unknown event {eventTypeStr}: {pprint.pformat(event_dict, indent=4)}")
         return event_type
 
     @classmethod
@@ -215,17 +218,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]:
@@ -237,23 +260,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", "trading_savingsplan_executed"] 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]:
@@ -270,7 +296,7 @@ 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, pref_locale="de") -> Optional[float]:
         """Parses a "detail" dictionary potentially containing a float in a certain locale format
 
         Args:
@@ -284,8 +310,11 @@ def _parse_float_from_detail(elem_dict: Dict[str, Any]) -> Optional[float]:
             return None
         parsed_val = re.sub(r"[^\,\.\d-]", "", unparsed_val)
 
-        # Prefer german locale.
-        locales = ("de", "en")
+        # 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))
@@ -294,4 +323,32 @@ 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
+            log.warning(
+                f"Number {parsed_val} parsed as {locales[1]} although preference was {locales[0]}. ({dump_dict['eventType']}, {dump_dict['id']}, {dump_dict['type']})"
+            )
+            if dump_enabled():
+                dump(
+                    f"Number {parsed_val} parsed as as {locales[1]} although preference was {locales[0]}: {pprint.pformat(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:
+            if dump_enabled():
+                dump(f"Number {parsed_val} parsed as {locales[0]}: {pprint.pformat(dump_dict, indent=4)}")
+        else:
+            log.debug(
+                f"Number {parsed_val} as {locales[0]} but could also be parsed as {locales[1]}. ({dump_dict['eventType']}, {dump_dict['id']}, {dump_dict['type']})"
+            )
+            if dump_enabled():
+                dump(
+                    f"Number {parsed_val} as {locales[0]} but could also be parsed as {locales[1]}: {pprint.pformat(dump_dict, indent=4)}"
+                )
+
         return None if result == 0.0 else result

pytr/main.py

@@ -15,7 +15,7 @@
 from pytr.dl import DL
 from pytr.portfolio import Portfolio
 from pytr.transactions import export_transactions
-from pytr.utils import check_version, get_logger
+from pytr.utils import check_version, enable_debug_dump, get_logger
 
 
 def get_main_parser():
@@ -45,6 +45,14 @@ def formatter(prog):
         help="Print version information and quit",
         action="store_true",
     )
+    parser.add_argument(
+        "--dump-debug-data-to-file",
+        help="Dump debug data to a given file",
+        metavar="DEBUG_DATA_FILE",
+        type=Path,
+        default=None,
+    )
+
     parser_cmd = parser.add_subparsers(help="Desired action to perform", dest="command")
 
     # help
@@ -238,6 +246,7 @@ def main():
     args = parser.parse_args()
     # print(vars(args))
 
+    enable_debug_dump(args.dump_debug_data_to_file)
     log = get_logger(__name__, args.verbosity)
     log.setLevel(args.verbosity.upper())
     log.debug("logging is set to debug")

pytr/utils.py

@@ -7,6 +7,28 @@
 import requests
 from packaging import version
 
+dump_file = None
+
+
+def enable_debug_dump(df):
+    global dump_file
+    dump_file = df
+    if df is not None:
+        with open(dump_file, "w"):
+            pass
+
+
+def dump_enabled():
+    return dump_file is not None
+
+
+def dump(str):
+    if dump_file is None:
+        return
+    with open(dump_file, "a") as file:
+        file.write(str + "\n")
+
+
 log_level = None