forked from marlanperumal/pdf_statement_reader
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathparse.py
More file actions
118 lines (88 loc) · 3.1 KB
/
Copy pathparse.py
File metadata and controls
118 lines (88 loc) · 3.1 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from tabula import read_pdf
from pikepdf import Pdf
import pandas as pd
import numpy as np
def get_raw_df(filename, num_pages, config):
dfs = []
for i in range(num_pages):
if i == 0 and "first" in config["layout"]:
area = config["layout"]["first"]["area"]
columns = config["layout"]["first"]["columns"]
else:
area = config["layout"]["default"]["area"]
columns = config["layout"]["default"]["columns"]
df = read_pdf(
filename,
pages=i + 1,
area=area,
columns=columns,
stream=True,
guess=False,
pandas_options={"dtype": str},
java_options=[
"-Dorg.slf4j.simpleLogger.defaultLogLevel=off",
"-Dorg.apache.commons.logging.Log=org.apache.commons.logging.impl.NoOpLog"
]
)
if df is not None and len(df) > 0:
dfs.extend(df)
statement = pd.concat(dfs, sort=False).reset_index(drop=True)
return statement
def format_negatives(s):
s = str(s)
if s.endswith("-"):
return "-" + s[:-1]
else:
return s
def clean_numeric(df, config):
numeric_cols = [config["columns"][col] for col in config["cleaning"]["numeric"]]
for col in numeric_cols:
df[col] = df[col].apply(format_negatives)
df[col] = df[col].str.replace(" ", "")
df[col] = pd.to_numeric(
df[col],
errors="coerce"
)
def clean_date(df, config):
date_cols = [config["columns"][col] for col in config["cleaning"]["date"]]
if "date_format" in config["cleaning"]:
date_format = config["cleaning"]["date_format"]
else:
date_format = None
for col in date_cols:
df[col] = pd.to_datetime(
df[col],
errors="coerce",
format=date_format
)
def clean_trans_detail(df, config):
trans_detail = config["columns"]["trans_detail"]
trans_type = config["columns"]["trans_type"]
balance = config["columns"]["balance"]
df[trans_detail] = ""
for i, row in df.iterrows():
if i == 0:
continue
if np.isnan(row[balance]):
df.loc[i - 1, trans_detail] = row[trans_type]
def clean_dropna(df, config):
drop_cols = [config["columns"][col] for col in config["cleaning"]["dropna"]]
df.dropna(subset=drop_cols, inplace=True)
def reorder_columns(df, config):
columns = [config["columns"][col] for col in config["order"]]
return df[columns]
def parse_statement(filename, config):
pdf = Pdf.open(filename)
num_pages = len(pdf.pages)
statement = get_raw_df(filename, num_pages, config)
if "numeric" in config["cleaning"]:
clean_numeric(statement, config)
if "trans_detail" in config["cleaning"]:
clean_trans_detail(statement, config)
if "date" in config["cleaning"]:
clean_date(statement, config)
if "dropna" in config["cleaning"]:
clean_dropna(statement, config)
if "order" in config:
statement = reorder_columns(statement, config)
return statement