comdirect reader tweak
This commit is contained in:
+25
-22
@@ -1,7 +1,9 @@
|
||||
import csv
|
||||
from dataclasses import dataclass
|
||||
import re
|
||||
from enum import Enum, auto
|
||||
from pathlib import Path
|
||||
from typing import Callable
|
||||
|
||||
from src.models import B4wExport
|
||||
|
||||
@@ -17,42 +19,43 @@ def reader_b4w(path: Path):
|
||||
|
||||
|
||||
def reader_comdirect(path: Path):
|
||||
PATTERN_ACCOUNT_NAME = r'^"Umsätze ([^"]+)";"Zeitraum: .*$'
|
||||
PATTERN_HEADER = r'"([^"]+)"'
|
||||
PATTERN_DATA = r'"([^"]+)"'
|
||||
|
||||
class EXPECT_LINE(Enum):
|
||||
ACCOUNT_NAME = auto()
|
||||
BLANK = auto()
|
||||
HEADER = auto()
|
||||
DATA = auto()
|
||||
@dataclass
|
||||
class Pattern:
|
||||
pattern: str
|
||||
|
||||
@classmethod
|
||||
def extract_pattern(cls, txt):
|
||||
return re.findall(cls.pattern, txt)
|
||||
|
||||
class EXPECT_LINE(Enum, Pattern):
|
||||
ACCOUNT_NAME = Pattern(r'^"Umsätze ([^"]+)";"Zeitraum: .*$')
|
||||
BLANK = Pattern(r"\n")
|
||||
HEADER = Pattern(r'"([^"]+)"')
|
||||
DATA = Pattern(r'"([^"]+)"')
|
||||
|
||||
with path.open(encoding="latin1") as fid:
|
||||
lines = fid.readlines()
|
||||
|
||||
expect_next = EXPECT_LINE.ACCOUNT_NAME
|
||||
account_name = None
|
||||
account_name = ""
|
||||
header = ""
|
||||
result = {}
|
||||
for line in lines:
|
||||
if expect_next == EXPECT_LINE.ACCOUNT_NAME:
|
||||
account_name = re.findall(PATTERN_ACCOUNT_NAME, line)
|
||||
if account_name:
|
||||
account_name = account_name[0]
|
||||
parsed = expect_next.extract_pattern(line)
|
||||
if expect_next == EXPECT_LINE.ACCOUNT_NAME and parsed:
|
||||
account_name = parsed[0]
|
||||
result[account_name] = {}
|
||||
expect_next = EXPECT_LINE.BLANK
|
||||
elif expect_next == EXPECT_LINE.BLANK:
|
||||
if line == "\n":
|
||||
elif expect_next == EXPECT_LINE.BLANK and parsed:
|
||||
expect_next = EXPECT_LINE.HEADER
|
||||
elif expect_next == EXPECT_LINE.HEADER:
|
||||
header = re.findall(PATTERN_HEADER, line)
|
||||
if header:
|
||||
elif expect_next == EXPECT_LINE.HEADER and parsed:
|
||||
header = parsed
|
||||
result[account_name]["data"] = []
|
||||
expect_next = EXPECT_LINE.DATA
|
||||
elif expect_next == EXPECT_LINE.DATA:
|
||||
data = re.findall(PATTERN_DATA, line)
|
||||
if data:
|
||||
elif expect_next == EXPECT_LINE.DATA and parsed:
|
||||
result[account_name]["data"].append(
|
||||
{column: value for column, value in zip(header, data)}
|
||||
{column: value for column, value in zip(header, parsed)}
|
||||
)
|
||||
expect_next = EXPECT_LINE.ACCOUNT_NAME
|
||||
else:
|
||||
|
||||
Reference in New Issue
Block a user