Upload files to "/"
This commit is contained in:
@@ -0,0 +1,185 @@
|
||||
import os
|
||||
import re
|
||||
import camelot
|
||||
import pandas as pd
|
||||
|
||||
# --- CONFIGURATION ---
|
||||
|
||||
PDF_PATH = r"MSDataReport (1).pdf" # PDF filename in the current directory
|
||||
OUTPUT_CSV = r"transactions.csv" # Output semicolon-delimited CSV
|
||||
PAGES = "1-end" # All pages; change if needed
|
||||
|
||||
# These are the logical columns we want in the final CSV
|
||||
TARGET_COLUMNS = [
|
||||
"value date",
|
||||
"Creditor / Debtor",
|
||||
"account",
|
||||
"payment details",
|
||||
"debit",
|
||||
"credit",
|
||||
]
|
||||
|
||||
# If the header row in the PDF is slightly different, you can map it here.
|
||||
# Keys are patterns you expect in the PDF header; values are our normalized names.
|
||||
HEADER_MAP = {
|
||||
# left side is a regex pattern (case-insensitive)
|
||||
r"value\s*date": "value date",
|
||||
r"creditor\s*/\s*debtor": "Creditor / Debtor",
|
||||
r"account": "account",
|
||||
r"payment\s*details": "payment details",
|
||||
r"debit": "debit",
|
||||
r"credit": "credit",
|
||||
}
|
||||
|
||||
|
||||
def normalize_header(col_name: str) -> str:
|
||||
"""
|
||||
Normalize a raw column header from the PDF to one of TARGET_COLUMNS (if possible).
|
||||
Uses HEADER_MAP regex patterns.
|
||||
"""
|
||||
if not isinstance(col_name, str):
|
||||
return col_name
|
||||
|
||||
name = col_name.strip().lower()
|
||||
for pattern, target in HEADER_MAP.items():
|
||||
if re.search(pattern, name, flags=re.IGNORECASE):
|
||||
return target
|
||||
return col_name.strip()
|
||||
|
||||
|
||||
def is_header_row(row_values):
|
||||
"""
|
||||
Heuristically determine if a row in the table is the header row:
|
||||
- If it contains at least two of the expected header keywords.
|
||||
"""
|
||||
header_text = " ".join(str(x) for x in row_values if isinstance(x, str)).lower()
|
||||
hits = 0
|
||||
for pattern in HEADER_MAP.keys():
|
||||
if re.search(pattern, header_text, re.IGNORECASE):
|
||||
hits += 1
|
||||
return hits >= 2
|
||||
|
||||
|
||||
def clean_amount(value: str) -> str:
|
||||
"""
|
||||
Clean debit/credit amount strings:
|
||||
- Strip spaces
|
||||
- Convert localized formats to plain decimal (e.g., "1 234,56-" -> "-1234.56")
|
||||
"""
|
||||
if not isinstance(value, str):
|
||||
return ""
|
||||
|
||||
v = value.strip()
|
||||
if not v:
|
||||
return ""
|
||||
|
||||
# Move trailing minus to the front for easier numeric parsing later
|
||||
if v.endswith("-"):
|
||||
v = "-" + v[:-1].strip()
|
||||
|
||||
# Remove thousand separators (spaces or dots depending on locale)
|
||||
v = v.replace(" ", "")
|
||||
# Common European formatting: "1.234,56"
|
||||
# Replace thousand '.' with nothing, and decimal ',' with '.'
|
||||
if "," in v and "." in v and v.rfind(",") > v.rfind("."):
|
||||
v = v.replace(".", "").replace(",", ".")
|
||||
else:
|
||||
# If only ',' present, assume it's decimal separator
|
||||
if "," in v and "." not in v:
|
||||
v = v.replace(",", ".")
|
||||
|
||||
# Final sanity strip
|
||||
return v
|
||||
|
||||
|
||||
def main():
|
||||
if not os.path.exists(PDF_PATH):
|
||||
raise FileNotFoundError(f"PDF file not found: {PDF_PATH}")
|
||||
|
||||
print(f"Reading tables from {PDF_PATH} on pages: {PAGES} ...")
|
||||
# flavor="lattice" works best when there are visible cell borders
|
||||
# If results are poor, try flavor="stream"
|
||||
tables = camelot.read_pdf(
|
||||
PDF_PATH,
|
||||
pages=PAGES,
|
||||
flavor="lattice",
|
||||
strip_text="\n",
|
||||
)
|
||||
|
||||
print(f"Found {len(tables)} table(s). Processing...")
|
||||
|
||||
all_rows = []
|
||||
|
||||
for idx, table in enumerate(tables):
|
||||
df = table.df.copy()
|
||||
print(f"Processing table {idx+1}/{len(tables)} with shape {df.shape}")
|
||||
|
||||
if df.empty:
|
||||
continue
|
||||
|
||||
# Detect header row (row index where header is located)
|
||||
header_row_idx = None
|
||||
for i in range(min(5, len(df))): # search first few rows
|
||||
if is_header_row(df.iloc[i].tolist()):
|
||||
header_row_idx = i
|
||||
break
|
||||
|
||||
if header_row_idx is None:
|
||||
print(f" Warning: No header row detected in table {idx+1}; skipping this table.")
|
||||
continue
|
||||
|
||||
# Use that row as header, drop all rows up to that row
|
||||
df.columns = [normalize_header(c) for c in df.iloc[header_row_idx]]
|
||||
df = df.iloc[header_row_idx + 1:].reset_index(drop=True)
|
||||
|
||||
# Keep only columns we care about (if they exist)
|
||||
# But first, ensure uniqueness of column names
|
||||
df = df.loc[:, ~df.columns.duplicated()]
|
||||
|
||||
present_cols = [c for c in df.columns if c in TARGET_COLUMNS]
|
||||
missing_cols = [c for c in TARGET_COLUMNS if c not in present_cols]
|
||||
|
||||
if missing_cols:
|
||||
print(f" Note: in table {idx+1}, these target columns are missing: {missing_cols}")
|
||||
|
||||
# Reindex with our target columns, missing will be filled with empty strings
|
||||
df = df.reindex(columns=TARGET_COLUMNS)
|
||||
|
||||
# Drop rows that are completely empty
|
||||
df = df.replace(r"^\s*$", pd.NA, regex=True)
|
||||
df = df.dropna(how="all")
|
||||
|
||||
# Clean debit/credit formats
|
||||
if "debit" in df.columns:
|
||||
df["debit"] = df["debit"].apply(lambda x: clean_amount(str(x)) if pd.notna(x) else "")
|
||||
if "credit" in df.columns:
|
||||
df["credit"] = df["credit"].apply(lambda x: clean_amount(str(x)) if pd.notna(x) else "")
|
||||
|
||||
# Append to master list, preserving the order Camelot returns tables (which is page order)
|
||||
all_rows.extend(df.to_dict(orient="records"))
|
||||
|
||||
if not all_rows:
|
||||
raise RuntimeError("No data rows extracted. Check that the PDF has tables and try flavor='stream'.")
|
||||
|
||||
# Convert to DataFrame and sort by "value date" to enforce chronological order
|
||||
out_df = pd.DataFrame(all_rows, columns=TARGET_COLUMNS)
|
||||
|
||||
# Try parsing the date to sort chronologically; if it fails, preserve original order
|
||||
try:
|
||||
out_df["__parsed_date"] = pd.to_datetime(out_df["value date"], dayfirst=True, errors="coerce")
|
||||
if out_df["__parsed_date"].notna().any():
|
||||
out_df = out_df.sort_values(["__parsed_date", "value date"]).reset_index(drop=True)
|
||||
out_df = out_df.drop(columns=["__parsed_date"])
|
||||
except Exception as e:
|
||||
print(f"Warning: could not parse dates for sorting: {e}")
|
||||
# leave out_df as-is
|
||||
|
||||
# Export as semicolon-delimited CSV with UTF-8 encoding
|
||||
out_df.to_csv(OUTPUT_CSV, sep=";", index=False, encoding="utf-8")
|
||||
|
||||
print(f"Done. Extracted {len(out_df)} rows into {OUTPUT_CSV}")
|
||||
print("Columns:", list(out_df.columns))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user