62 lines
1.8 KiB
Python
62 lines
1.8 KiB
Python
import csv
|
|
import re
|
|
from enum import Enum, auto
|
|
from pathlib import Path
|
|
|
|
from src.models import B4wExport
|
|
|
|
|
|
def reader_raw(path: Path):
|
|
with path.open() as fid:
|
|
return list(csv.DictReader(fid))
|
|
|
|
|
|
def reader_b4w(path: Path):
|
|
raw = reader_raw(path)
|
|
return list(map(B4wExport.model_validate, raw))
|
|
|
|
|
|
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()
|
|
|
|
with path.open(encoding="latin1") as fid:
|
|
lines = fid.readlines()
|
|
|
|
expect_next = EXPECT_LINE.ACCOUNT_NAME
|
|
account_name = None
|
|
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]
|
|
result[account_name] = {}
|
|
expect_next = EXPECT_LINE.BLANK
|
|
elif expect_next == EXPECT_LINE.BLANK:
|
|
if line == "\n":
|
|
expect_next = EXPECT_LINE.HEADER
|
|
elif expect_next == EXPECT_LINE.HEADER:
|
|
header = re.findall(PATTERN_HEADER, line)
|
|
if header:
|
|
result[account_name]["data"] = []
|
|
expect_next = EXPECT_LINE.DATA
|
|
elif expect_next == EXPECT_LINE.DATA:
|
|
data = re.findall(PATTERN_DATA, line)
|
|
if data:
|
|
result[account_name]["data"].append(
|
|
{column: value for column, value in zip(header, data)}
|
|
)
|
|
expect_next = EXPECT_LINE.ACCOUNT_NAME
|
|
else:
|
|
expect_next = EXPECT_LINE.ACCOUNT_NAME
|
|
return result
|