257 lines
9.8 KiB
Python
257 lines
9.8 KiB
Python
"""LeoCRM - Minimal CRM with auth, companies, contacts."""
|
|
import os
|
|
import sqlite3
|
|
from datetime import datetime
|
|
from functools import wraps
|
|
|
|
from flask import Flask, render_template, request, redirect, url_for, flash, session, g
|
|
from werkzeug.security import generate_password_hash, check_password_hash
|
|
|
|
app = Flask(__name__)
|
|
app.secret_key = os.urandom(24).hex()
|
|
DATABASE = os.path.join(os.path.dirname(__file__), 'leocrm.db')
|
|
|
|
|
|
def get_db():
|
|
if 'db' not in g:
|
|
g.db = sqlite3.connect(DATABASE)
|
|
g.db.row_factory = sqlite3.Row
|
|
g.db.execute("PRAGMA foreign_keys = ON")
|
|
return g.db
|
|
|
|
|
|
@app.teardown_appcontext
|
|
def close_db(exception):
|
|
db = g.pop('db', None)
|
|
if db is not None:
|
|
db.close()
|
|
|
|
|
|
def init_db():
|
|
db = sqlite3.connect(DATABASE)
|
|
db.executescript('''
|
|
CREATE TABLE IF NOT EXISTS users (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
username TEXT UNIQUE NOT NULL,
|
|
password_hash TEXT NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
|
|
);
|
|
CREATE TABLE IF NOT EXISTS companies (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
name TEXT NOT NULL,
|
|
address TEXT,
|
|
phone TEXT,
|
|
email TEXT,
|
|
website TEXT,
|
|
notes TEXT,
|
|
user_id INTEGER NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (user_id) REFERENCES users(id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS contacts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
first_name TEXT NOT NULL,
|
|
last_name TEXT NOT NULL,
|
|
email TEXT,
|
|
phone TEXT,
|
|
position TEXT,
|
|
notes TEXT,
|
|
company_id INTEGER NOT NULL,
|
|
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
|
|
FOREIGN KEY (company_id) REFERENCES companies(id) ON DELETE CASCADE
|
|
);
|
|
''')
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
def login_required(f):
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
if 'user_id' not in session:
|
|
flash('Bitte melde dich an.', 'warning')
|
|
return redirect(url_for('login'))
|
|
return f(*args, **kwargs)
|
|
return decorated
|
|
|
|
|
|
@app.route('/')
|
|
def index():
|
|
if 'user_id' in session:
|
|
return redirect(url_for('dashboard'))
|
|
return redirect(url_for('login'))
|
|
|
|
|
|
@app.route('/register', methods=['GET', 'POST'])
|
|
def register():
|
|
if request.method == 'POST':
|
|
username = request.form['username'].strip()
|
|
password = request.form['password']
|
|
if not username or not password:
|
|
flash('Benutzername und Passwort sind erforderlich.', 'danger')
|
|
return render_template('register.html')
|
|
db = get_db()
|
|
existing = db.execute('SELECT id FROM users WHERE username = ?', (username,)).fetchone()
|
|
if existing:
|
|
flash('Benutzername bereits vergeben.', 'danger')
|
|
return render_template('register.html')
|
|
db.execute('INSERT INTO users (username, password_hash) VALUES (?, ?)',
|
|
(username, generate_password_hash(password)))
|
|
db.commit()
|
|
flash('Registrierung erfolgreich! Bitte melde dich an.', 'success')
|
|
return redirect(url_for('login'))
|
|
return render_template('register.html')
|
|
|
|
|
|
@app.route('/login', methods=['GET', 'POST'])
|
|
def login():
|
|
if request.method == 'POST':
|
|
username = request.form['username'].strip()
|
|
password = request.form['password']
|
|
db = get_db()
|
|
user = db.execute('SELECT * FROM users WHERE username = ?', (username,)).fetchone()
|
|
if user and check_password_hash(user['password_hash'], password):
|
|
session['user_id'] = user['id']
|
|
session['username'] = user['username']
|
|
flash(f'Willkommen, {username}!', 'success')
|
|
return redirect(url_for('dashboard'))
|
|
flash('Ungültige Anmeldedaten.', 'danger')
|
|
return render_template('login.html')
|
|
|
|
|
|
@app.route('/logout')
|
|
def logout():
|
|
session.clear()
|
|
flash('Du wurdest abgemeldet.', 'info')
|
|
return redirect(url_for('login'))
|
|
|
|
|
|
@app.route('/dashboard')
|
|
@login_required
|
|
def dashboard():
|
|
db = get_db()
|
|
companies = db.execute(
|
|
'SELECT * FROM companies WHERE user_id = ? ORDER BY name',
|
|
(session['user_id'],)
|
|
).fetchall()
|
|
return render_template('dashboard.html', companies=companies)
|
|
|
|
|
|
@app.route('/companies/create', methods=['GET', 'POST'])
|
|
@login_required
|
|
def company_create():
|
|
if request.method == 'POST':
|
|
db = get_db()
|
|
db.execute('INSERT INTO companies (name, address, phone, email, website, notes, user_id) VALUES (?,?,?,?,?,?,?)',
|
|
(request.form['name'], request.form.get('address',''), request.form.get('phone',''),
|
|
request.form.get('email',''), request.form.get('website',''), request.form.get('notes',''),
|
|
session['user_id']))
|
|
db.commit()
|
|
flash('Firma erstellt.', 'success')
|
|
return redirect(url_for('dashboard'))
|
|
return render_template('company_form.html', company=None)
|
|
|
|
|
|
@app.route('/companies/<int:id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def company_edit(id):
|
|
db = get_db()
|
|
company = db.execute('SELECT * FROM companies WHERE id = ? AND user_id = ?',
|
|
(id, session['user_id'])).fetchone()
|
|
if not company:
|
|
flash('Firma nicht gefunden.', 'danger')
|
|
return redirect(url_for('dashboard'))
|
|
if request.method == 'POST':
|
|
db.execute('UPDATE companies SET name=?, address=?, phone=?, email=?, website=?, notes=? WHERE id=?',
|
|
(request.form['name'], request.form.get('address',''), request.form.get('phone',''),
|
|
request.form.get('email',''), request.form.get('website',''), request.form.get('notes',''), id))
|
|
db.commit()
|
|
flash('Firma aktualisiert.', 'success')
|
|
return redirect(url_for('dashboard'))
|
|
return render_template('company_form.html', company=company)
|
|
|
|
|
|
@app.route('/companies/<int:id>/delete', methods=['POST'])
|
|
@login_required
|
|
def company_delete(id):
|
|
db = get_db()
|
|
db.execute('DELETE FROM companies WHERE id = ? AND user_id = ?', (id, session['user_id']))
|
|
db.commit()
|
|
flash('Firma gelöscht.', 'info')
|
|
return redirect(url_for('dashboard'))
|
|
|
|
|
|
@app.route('/companies/<int:company_id>/contacts')
|
|
@login_required
|
|
def contact_list(company_id):
|
|
db = get_db()
|
|
company = db.execute('SELECT * FROM companies WHERE id = ? AND user_id = ?',
|
|
(company_id, session['user_id'])).fetchone()
|
|
if not company:
|
|
flash('Firma nicht gefunden.', 'danger')
|
|
return redirect(url_for('dashboard'))
|
|
contacts = db.execute('SELECT * FROM contacts WHERE company_id = ? ORDER BY last_name, first_name',
|
|
(company_id,)).fetchall()
|
|
return render_template('contact_list.html', company=company, contacts=contacts)
|
|
|
|
|
|
@app.route('/companies/<int:company_id>/contacts/create', methods=['GET', 'POST'])
|
|
@login_required
|
|
def contact_create(company_id):
|
|
db = get_db()
|
|
company = db.execute('SELECT * FROM companies WHERE id = ? AND user_id = ?',
|
|
(company_id, session['user_id'])).fetchone()
|
|
if not company:
|
|
flash('Firma nicht gefunden.', 'danger')
|
|
return redirect(url_for('dashboard'))
|
|
if request.method == 'POST':
|
|
db.execute('INSERT INTO contacts (first_name, last_name, email, phone, position, notes, company_id) VALUES (?,?,?,?,?,?,?)',
|
|
(request.form['first_name'], request.form['last_name'], request.form.get('email',''),
|
|
request.form.get('phone',''), request.form.get('position',''), request.form.get('notes',''), company_id))
|
|
db.commit()
|
|
flash('Kontaktperson erstellt.', 'success')
|
|
return redirect(url_for('contact_list', company_id=company_id))
|
|
return render_template('contact_form.html', company=company, contact=None)
|
|
|
|
|
|
@app.route('/contacts/<int:id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
def contact_edit(id):
|
|
db = get_db()
|
|
contact = db.execute('''SELECT c.*, co.name as company_name, co.user_id
|
|
FROM contacts c JOIN companies co ON c.company_id = co.id
|
|
WHERE c.id = ? AND co.user_id = ?''',
|
|
(id, session['user_id'])).fetchone()
|
|
if not contact:
|
|
flash('Kontakt nicht gefunden.', 'danger')
|
|
return redirect(url_for('dashboard'))
|
|
if request.method == 'POST':
|
|
db.execute('UPDATE contacts SET first_name=?, last_name=?, email=?, phone=?, position=?, notes=? WHERE id=?',
|
|
(request.form['first_name'], request.form['last_name'], request.form.get('email',''),
|
|
request.form.get('phone',''), request.form.get('position',''), request.form.get('notes',''), id))
|
|
db.commit()
|
|
flash('Kontakt aktualisiert.', 'success')
|
|
return redirect(url_for('contact_list', company_id=contact['company_id']))
|
|
return render_template('contact_form.html', company={'id': contact['company_id'], 'name': contact['company_name']}, contact=contact)
|
|
|
|
|
|
@app.route('/contacts/<int:id>/delete', methods=['POST'])
|
|
@login_required
|
|
def contact_delete(id):
|
|
db = get_db()
|
|
contact = db.execute('''SELECT c.company_id, co.user_id
|
|
FROM contacts c JOIN companies co ON c.company_id = co.id
|
|
WHERE c.id = ?''', (id,)).fetchone()
|
|
if contact and contact['user_id'] == session['user_id']:
|
|
db.execute('DELETE FROM contacts WHERE id = ?', (id,))
|
|
db.commit()
|
|
flash('Kontakt gelöscht.', 'info')
|
|
return redirect(url_for('contact_list', company_id=contact['company_id']))
|
|
flash('Kontakt nicht gefunden.', 'danger')
|
|
return redirect(url_for('dashboard'))
|
|
|
|
|
|
if __name__ == '__main__':
|
|
init_db()
|
|
app.run(host='0.0.0.0', port=5000, debug=True)
|