uweziegenhagen.de

Hier noch ein funktionierendes Beispiel, wie man PDF Dateien OCR-technisch auswerten kann. Im Gegensatz zu pdf2image besteht hier keine Abhängigkeit zu poppler.

import fitz
import numpy as np
from paddleocr import PaddleOCR
 
ocr = PaddleOCR(lang="de", enable_mkldnn=False)
 
doc = fitz.open("Widerspruch.pdf")
 
for page_number, page in enumerate(doc, start=1):
 
    # Render PDF page
    pix = page.get_pixmap(dpi=300)
 
    # Convert directly to NumPy
    image = np.frombuffer(
        pix.samples,
        dtype=np.uint8
    ).reshape(pix.height, pix.width, pix.n)
 
    # PaddleOCR expects RGB, not RGBA
    if pix.n == 4:
        image = image[:, :, :3]
 
    result = ocr.predict(image)
 
    print(f"\n--- Page {page_number} ---")
 
    for page_result in result:
        for text, score in zip(
            page_result["rec_texts"],
            page_result["rec_scores"],
        ):
            print(f"{score:.3f}: {text}")

PaddleOCR (Github) ist ein quelloffenes Framework für die Erkennung von Text (OCR). Hier ein kurzes Beispiel, wie man aus Python PaddleOCR nutzen kann.

# pip install paddlepaddle paddleocr
from paddleocr import PaddleOCR
 
# Initialize OCR
ocr = PaddleOCR(
    lang="de",          # "de" for German, "en" for English, etc.
    use_doc_orientation_classify=True,
    use_doc_unwarping=True,
    use_textline_orientation=True,
    enable_mkldnn=False  # prevents MKLDNN/PIR crash
)
 
# Perform OCR
result = ocr.predict("wident.png")
 
# Print results
with open('wident_paddle.txt','w') as o:
    for page in result:
        for line in page["rec_texts"]:
            print(line)
            o.writelines(line)

In this final article of this series I will add another data mart. In the models/marts subfolder we add fx_current_rates.sql with the following SQL code:

{{ config(materialized='view') }}
 
SELECT
    currency,
    valid_from,
    rate_vs_eur
FROM {{ REF('fx_rates_with_validity') }}
WHERE is_current = TRUE

This will be a new view on top of the „fx_rates_with_validity“ mart. As this view ‚materializes‘ in the Postgres DB, we can use HeidiSQL to retrieve data from it:

In this article we will continue our dbt journey. With dbt-core installed we can ask dbt to set up a new project for us.

Simply go to the root folder where your project folder shall be created and run dbt init projectname. dbt will then ask you for the database adapter and the connection settings as well as for the schema dbt is to build its objects in.

After these questions and answers dbt tells you that you may run dbt debug now to check if everything is fine, BUT: you need to switch first to the subdirectory you created otherwise there will be a meaningless error message:


Encountered an error:
Internal Error
Profile should not be None if loading profile completed


If run in the correct subdirectory dbt debug should not produce any errors. If it does, run the error message through the search engine of your choice.

C:\Users\UWe\Desktop\fx_project>dbt debug
16:52:20  Running with dbt=1.10.13
16:52:20  dbt version: 1.10.13
16:52:20  python version: 3.12.6
16:52:20  python path: C:\Python312\python.exe
16:52:20  os info: Windows-11-10.0.26200-SP0
16:52:20  Using profiles dir at C:\Users\UWe\.dbt
16:52:20  Using profiles.yml file at C:\Users\UWe\.dbt\profiles.yml
16:52:20  Using dbt_project.yml file at C:\Users\UWe\Desktop\fx_project\dbt_project.yml
16:52:20  adapter type: postgres
16:52:20  adapter version: 1.9.1
16:52:20  Configuration:
16:52:20    profiles.yml file [OK found and valid]
16:52:20    dbt_project.yml file [OK found and valid]
16:52:20  Required dependencies:
16:52:20   - git [OK found]

16:52:20  Connection:
16:52:20    host: localhost
16:52:20    port: 5432
16:52:20    user: postgres
16:52:20    database: FX
16:52:20    schema: dbt
16:52:20    connect_timeout: 10
16:52:20    role: None
16:52:20    search_path: None
16:52:20    keepalives_idle: 0
16:52:20    sslmode: None
16:52:20    sslcert: None
16:52:20    sslkey: None
16:52:20    sslrootcert: None
16:52:20    application_name: dbt
16:52:20    retries: 1
16:52:20  Registered adapter: postgres=1.9.1
16:52:20    Connection test: [OK connection ok]

16:52:20  All checks passed!

In the project folder you find a set of subfolders, remove the models/example subfolder, as it may cause error messages.

Next we will create some basic models, using YAML files.

In models, create sources.yml with the following content:

version: 2
sources:
  - name: raw
    schema: raw
    tables:
      - name: ecb_fx_rates

In models, create a subfolder staging and add stg_fx_rates.sql to this staging subfolder with the following content:

{{ config(materialized='view') }}
SELECT
    fx_date,
    currency,
    rate,
    created_at
FROM {{ SOURCE('raw', 'ecb_fx_rates') }}

This will simply select all the data we have in our raw DB schema.

Next comes is the interesting part. Each row gets a valid_from and valid_to date using a window function.

In models, create a subfolder marts and add fx_rates_with_validity.sql to this marts subfolder with the following content:

{{ config(materialized='table') }}
 
WITH base AS (
    SELECT
        fx_date,
        currency,
        rate
    FROM {{ REF('stg_fx_rates') }}
),
 
with_validity AS (
    SELECT
        currency,
        fx_date                        AS valid_from,
        -- valid_to is the day before the next rate, or null if it's the latest
        lead(fx_date) OVER (
            partition BY currency
            ORDER BY fx_date
        ) - INTERVAL '1 day'             AS valid_to,
        rate,
        -- flag the currently active rate
        CASE
            WHEN lead(fx_date) OVER (
                partition BY currency ORDER BY fx_date
            ) IS NULL THEN TRUE
            ELSE FALSE
        END                              AS is_current
    FROM base
)
 
SELECT * FROM with_validity
ORDER BY currency, valid_from

This will give uzs data as the following

It also makes sense now to add some data checks.

In models, create schema.yml with the following content:

version: 2

models:
  - name: stg_fx_rates
    columns:
      - name: rate_date
        tests: [not_null]
      - name: currency
        tests: [not_null]
      - name: rate_vs_eur
        tests: [not_null]

  - name: fx_rates_with_validity
    columns:
      - name: currency
        tests: [not_null]
      - name: valid_from
        tests: [not_null]
      - name: is_current
        tests: [not_null]

We can now run the whole setup:

dbt run
dbt test
dbt docs generate
dbt docs serve

Going forward

Since ECB publishes the latest rate every data at around 16:00 CET the Python job to fetch them should be run each day (apparently after 16:00 CET). Once the raw database is updated you can update the dbt models:

dbt run --select stg_fx_rates fx_rates_with_validity
dbt test --select stg_fx_rates fx_rates_with_validity

In the final part of the series we will add another data market for the latest rates.

Für ein kleines Projekt brauchte ich die Möglichkeit, flexible Zahlungspläne zu erzeugen. Der folgende Python-Code tut genau das und erzeugt auch gleich passende INSERT Statements für Postgres.

Das CREATE TABLE ist wie folgt:

CREATE TABLE paymentplan (
    contract INTEGER NOT NULL,
    paymentdate INTEGER NOT NULL,
    paymenttype INTEGER NOT NULL,
    amount NUMERIC(12, 2) NOT NULL
);

import pandas as pd
from datetime import datetime
from dateutil.relativedelta import relativedelta
 
def genPaymentPlan(contract, principal, paymentsperyear, interestrate, startdate, years):
    interest = interestrate/100*principal/paymentsperyear
    date = datetime.strptime(startdate, '%Y%m%d').date()
    output = pd.DataFrame(columns=['Contract', 'Paymentdate', 'Paymenttype', 'Amount'])
 
    # Add a row to the DataFrame
    start = {'Contract': contract, 'Paymentdate': startdate, 'Paymenttype': 1, 'Amount' : principal}
    output = pd.concat([output, pd.DataFrame([start])], ignore_index=True)  
 
    for payment in range(paymentsperyear*years):
        new_date = (date + relativedelta(months=12/paymentsperyear))
        date = new_date
        interest_line = {'Contract': contract, 'Paymentdate': new_date.strftime('%Y%m%d'),'Paymenttype': 2, 'Amount' : -1*interest}
        output = pd.concat([output, pd.DataFrame([interest_line])], ignore_index=True)  
 
    end = {'Contract': contract, 'Paymentdate': date.strftime('%Y%m%d'), 'Paymenttype': 3, 'Amount' : -principal}
    output = pd.concat([output, pd.DataFrame([end])], ignore_index=True)          
    return output
 
 
test = genPaymentPlan(123458, 2000, 2, 12, '20260101', 4)
#print(test)
 
 
 
# Generate INSERT statements
print("Generated SQL INSERT statements:\n")
 
for index, row in test.iterrows():
    contract = int(row['Contract'])
    date = row['Paymentdate']  # already string in format YYYY-MM-DD
    type =  row['Paymenttype'] 
    amount = float(row['Amount'])
 
 
    insert = f"INSERT INTO paymentplan (contract, paymentdate,paymenttype, amount) VALUES ({contract}, '{date}', {type} , {amount});"
    print(insert)

Hier ein einfacher Code-Schnipsel, um aus CSV-Dateien Excel-Dateien zu machen.

Es empfiehlt sich auch, die Engine explizit zu setzen, mit der die Excel-Datei geschrieben wird. Standardmäßig nutzt pandas openpyxl, xlsxwriter scheint nach meinen Tests aber um ca. 30% schneller zu sein.

import pandas as pd
from openpyxl import Workbook
from openpyxl.utils.dataframe import dataframe_to_rows
 
 
# output control
files = ['']
 
wb = Workbook()
wb.remove(wb['Sheet'])
 
for index, file in enumerate(files,0):
    temp = pd.read_csv(file + '.csv',sep='\t')
    ws1 = wb.create_sheet()
    ws1.title = file
    rows = dataframe_to_rows(temp, index=False, header=True)
 
    for r_idx, row in enumerate(rows, 1):
        for c_idx, value in enumerate(row, 1):
            ws1.cell(row=r_idx, column=c_idx, value=value)
 
wb.save('files45646.xlsx')
wb.close()

Aus aktuellem Anlass hier ein einfaches Beispiel, wie man Werte in pandas Dataframes ersetzen kann:

import pandas as pd
 
# initialize data of lists.
data = {'Hersteller': ['VW', 'BMW', 'VW', 'Porsche'],
        'Modell': ['Golf', '1er', 'Polo', '911']}
 
 
df = pd.DataFrame(data)
 
 
print(df,'\n')
 
 
df['Hersteller'] = df['Hersteller'].replace(
    {"VW": "Volkswagen", "Horch": "Audi"})
 
 
print(df)

Hier ein Code-Beispiel für die wichtigsten Funktionen von DuckDB.

import duckdb as ddb
import pandas as pd
 
con = ddb.connect(':memory:')
 
con_p = ddb.connect('my_database.db')
 
con_p.execute('CREATE OR REPLACE TABLE telefonnummern(fullname VARCHAR,phone VARCHAR);')
 
 
con_p.execute("INSERT INTO telefonnummern VALUES ('Max Mustermann', '0123-4567890')")
 
print(con_p.sql('SHOW ALL TABLES'))
 
print(con_p.sql('SELECT * FROM telefonnummern;'))
 
 
ddb_object = con_p.sql('SELECT * FROM telefonnummern;')
 
df = ddb_object.to_df()
 
ddb_tuple = ddb_object.fetchall()
 
print(df)
 
print(ddb_tuple)

Hier ein paar Beispiele, wie man mit icecream print() Ausgaben ersetzen kann.

"""
icecream examples
"""
 
 
from icecream import ic
 
# define some function
def addiere(x, y):
    return x + y
 
# call ice
ic(addiere(1, 2))
 
# Output:
# ic| addiere(1, 2): 3
 
 
d = {'i': 2, 'j': 3, 'k': 4711}
ic(d['k'])
 
 
struct = {
    "hersteller": "VW",
    "modell": "Golf",
    "Farben": ["gelb", "rot"]
}
 
ic(struct)
 
ic.disable()
ic(struct) # no output
ic.enable()
 
 
def logstuff(text):
    # log to output file
    print(text)
 
 
ic.configureOutput(prefix="Hallo| ", outputFunction=logstuff)
 
ic(addiere(7, 7))
 
 
ic.configureOutput(prefix="Welt| ", outputFunction=logstuff)
 
ic(addiere(7, 7))

Für Dante e.V. bestand die Notwendigkeit, aus MT940 Dateien moderne CAMT.053 zu erzeugen, dank Python wurde das eine lösbare Aufgabe.

Schritt 1

Die MT940 Datei parsen und die Transaktionen in einen pandas DataFrame überführen.

import mt940
import pprint
import pandas as pd

df = pd.DataFrame()

transactions = mt940.parse('Umsaetze_2310007_22.07.2024.mta')

print('Transactions:')
pprint.pprint(transactions.data)

for transaction in transactions:
    print('Transaction: ', transaction)
    pprint.pprint(transaction.data)
    t = transaction.data
    tt = pd.DataFrame(t, index=[0])

    df = pd.concat([df,tt],ignore_index=True)
    
df.to_excel('AllBookings.xlsx',index=False)

Schritt 2

Aus dem DataFrame das XML befüllen, die für den Kopf der XML-Datei notwendigen Kontostandsinformationen holen ich dazu aus der MT940 Datei.

import pandas as pd # data wrangling
import jinja2 # template engine
import os # for file-related stuff
import mt940
from datetime import datetime


today = datetime.today()
now = today.strftime("%Y-%m-%d")



transactions = mt940.parse('Umsaetze_2310007_22.07.2024.mta')

opening = transactions.data['final_opening_balance']
openingamount = str(opening.amount)[:-4]
openingdate = opening.date

closing = transactions.data['final_closing_balance']
closingamount = str(closing.amount)[:-4]
closingdate = closing.date

 
# create jinja env that can load template from filesystem
jinja_env = jinja2.Environment(loader = jinja2.FileSystemLoader(os.path.abspath('.')))
 
df = pd.read_excel('AllBookings.xlsx', dtype={'date': str,'amount':str})

df['CreditDebit'] = ''
df['CreditDebit'] = df['CreditDebit'].where(df['amount'].str.get(0).isin(['-']), 'CRDT')
df['CreditDebit'] = df['CreditDebit'].where(~df['amount'].str.get(0).isin(['-']), 'DBIT')

df['date'] = df['date'].str[:-9]


# Währung weg
df['amount'] = df['amount'].str[:-4]
# Vorzeichen weg
df['amount'] = df['amount'].str.replace('-','')

#df['amount'].replace('-','',inplace=True)
#df["amount"] = df["amount"].apply(lambda x: x.str.replace("-", ""))

template = jinja_env.get_template('Ntry.xml')
 
with open('FertigesXML.xml','w') as output:
    output.write(template.render(data=df,
                                 openingamount=openingamount,
                                 openingdate=openingdate,
                                 closingamount=closingamount,
                                 closingdate=closingdate
                                 ))

Jinja2 XML-Template

Das XML-Template für Jinja2 findet ihr hier:

Ntry_blog