122 lines
4.1 KiB
Python
122 lines
4.1 KiB
Python
from flask import Flask, render_template, request, jsonify, url_for
|
|
import pandas as pd
|
|
import os
|
|
import logging
|
|
import numpy as np
|
|
from datetime import datetime, timedelta
|
|
from dotenv import load_dotenv
|
|
import requests
|
|
from collections import defaultdict
|
|
|
|
app = Flask(__name__, static_folder='static')
|
|
logging.basicConfig(level=logging.DEBUG)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# Version der Anwendung
|
|
VERSION = "1.0.6"
|
|
|
|
# Pfad zur CSV-Datei
|
|
CSV_FILE = "data/customers.csv"
|
|
|
|
# Lade Umgebungsvariablen
|
|
load_dotenv()
|
|
|
|
def clean_dataframe(df):
|
|
"""Konvertiert NaN-Werte in None für JSON-Kompatibilität"""
|
|
return df.replace({np.nan: None})
|
|
|
|
# CSV-Datei laden
|
|
def load_data():
|
|
try:
|
|
logger.info("Versuche CSV-Datei zu laden...")
|
|
if not os.path.exists(CSV_FILE):
|
|
logger.error(f"CSV-Datei '{CSV_FILE}' nicht gefunden!")
|
|
return None
|
|
|
|
# Lade CSV mit Komma als Trennzeichen
|
|
df = pd.read_csv(CSV_FILE, sep=',', encoding='utf-8', quotechar='"')
|
|
# Entferne Anführungszeichen aus den Spaltennamen
|
|
df.columns = df.columns.str.strip('"')
|
|
# Entferne Anführungszeichen aus den Werten
|
|
for col in df.columns:
|
|
if df[col].dtype == 'object':
|
|
df[col] = df[col].str.strip('"')
|
|
df = clean_dataframe(df)
|
|
logger.info(f"CSV-Datei erfolgreich geladen. {len(df)} Einträge gefunden.")
|
|
return df
|
|
except Exception as e:
|
|
logger.error(f"Fehler beim Laden der CSV-Datei: {str(e)}")
|
|
return None
|
|
|
|
@app.route('/')
|
|
def index():
|
|
return render_template('index.html')
|
|
|
|
@app.route('/search')
|
|
def search():
|
|
try:
|
|
# CSV-Datei laden
|
|
df = load_data()
|
|
if df is None:
|
|
return jsonify({"error": "Datenbank konnte nicht geladen werden"}), 500
|
|
|
|
# Suchparameter aus der URL holen
|
|
name = request.args.get('name', '').strip()
|
|
ort = request.args.get('ort', '').strip()
|
|
kundennummer = request.args.get('kundennummer', '').strip()
|
|
fachrichtung = request.args.get('fachrichtung', '').strip()
|
|
telefon = request.args.get('telefon', '').strip()
|
|
query = request.args.get('q', '').strip()
|
|
|
|
# Initialisiere die Maske für die Filterung
|
|
mask = pd.Series(True, index=df.index)
|
|
|
|
# Wenn eine allgemeine Suche angegeben ist
|
|
if query:
|
|
query_mask = (
|
|
df['Vorname'].str.contains(query, case=False, na=False) |
|
|
df['Nachname'].str.contains(query, case=False, na=False) |
|
|
df['Ort'].str.contains(query, case=False, na=False) |
|
|
df['Nummer'].astype(str).str.contains(query, case=False, na=False) |
|
|
df['Fachrichtung'].str.contains(query, case=False, na=False) |
|
|
df['Tel'].astype(str).str.contains(query, case=False, na=False)
|
|
)
|
|
mask &= query_mask
|
|
|
|
# Spezifische Suchkriterien anwenden
|
|
if name:
|
|
name_mask = (
|
|
df['Vorname'].str.contains(name, case=False, na=False) |
|
|
df['Nachname'].str.contains(name, case=False, na=False)
|
|
)
|
|
mask &= name_mask
|
|
|
|
if ort:
|
|
ort_mask = df['Ort'].str.contains(ort, case=False, na=False)
|
|
mask &= ort_mask
|
|
|
|
if kundennummer:
|
|
kundennummer_mask = df['Nummer'].astype(str).str.contains(kundennummer, case=False, na=False)
|
|
mask &= kundennummer_mask
|
|
|
|
if fachrichtung:
|
|
fachrichtung_mask = df['Fachrichtung'].str.contains(fachrichtung, case=False, na=False)
|
|
mask &= fachrichtung_mask
|
|
|
|
if telefon:
|
|
telefon_mask = df['Tel'].astype(str).str.contains(telefon, case=False, na=False)
|
|
mask &= telefon_mask
|
|
|
|
results = df[mask].to_dict('records')
|
|
logger.info(f"{len(results)} Ergebnisse gefunden")
|
|
|
|
return jsonify({
|
|
'results': results,
|
|
'total': len(results)
|
|
})
|
|
except Exception as e:
|
|
logger.error(f"Fehler bei der Suche: {str(e)}")
|
|
return jsonify({"error": str(e)}), 500
|
|
|
|
if __name__ == '__main__':
|
|
app.run(debug=True, port=5001) |