Compare commits
11 Commits
374c242051
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dd30c10801 | ||
|
|
50cbd08950 | ||
|
|
2545ab84fd | ||
|
|
799fdbd905 | ||
|
|
49e2f16dbb | ||
| 800ccd8376 | |||
|
|
fca9462bc7 | ||
|
|
081915dacf | ||
|
|
1e21730bd4 | ||
|
|
62613a2f39 | ||
|
|
998574f26a |
9
.gitignore
vendored
9
.gitignore
vendored
@@ -16,4 +16,11 @@ __pycache__/
|
||||
.DS_Store
|
||||
|
||||
# vCards
|
||||
vcards
|
||||
vcards/
|
||||
|
||||
# Logs
|
||||
logs/
|
||||
|
||||
#
|
||||
anmeldung.sqbpro
|
||||
docs/CHAT_SUMMARY.md
|
||||
|
||||
41
README.md
41
README.md
@@ -23,3 +23,44 @@ python app.py
|
||||
```
|
||||
|
||||
Das Formular ist dann unter http://127.0.0.1:5000/ erreichbar.
|
||||
|
||||
Migrations (Flask-Migrate / Alembic)
|
||||
---------------------------------
|
||||
|
||||
Das Projekt verwendet jetzt Flask-Migrate (Alembic) für Datenbank-Migrationen. Ablauf lokal:
|
||||
|
||||
```bash
|
||||
# virtuelles Environment aktivieren
|
||||
source .venv/bin/activate
|
||||
|
||||
# Migrationen initialisieren (nur einmal)
|
||||
export FLASK_APP=app.py
|
||||
flask db init
|
||||
|
||||
# Neue Migration auf Basis der Models erzeugen
|
||||
flask db migrate -m "Beschreibung der Änderung"
|
||||
|
||||
# Migration anwenden
|
||||
flask db upgrade
|
||||
```
|
||||
|
||||
Hinweis: Wenn die Datenbank bereits Tabellen enthält, benutze `flask db stamp head`, damit Alembic den aktuellen Stand als angewendet markiert.
|
||||
|
||||
Tests
|
||||
-----
|
||||
|
||||
Die Tests laufen mit pytest:
|
||||
|
||||
```bash
|
||||
source .venv/bin/activate
|
||||
python -m pytest -q
|
||||
```
|
||||
|
||||
vCard Export
|
||||
-----------
|
||||
|
||||
Nach dem Absenden eines Anmeldeformulars wird eine vCard 4.0 im Ordner `vcards/` erzeugt. Dateiname-Muster:
|
||||
|
||||
```
|
||||
Nachname_Vorname_<id>.vcf
|
||||
```
|
||||
|
||||
77
app.py
77
app.py
@@ -2,8 +2,10 @@ from flask import Flask, render_template, request, redirect, url_for
|
||||
from flask_sqlalchemy import SQLAlchemy
|
||||
import os
|
||||
import re
|
||||
import pathlib
|
||||
import unicodedata
|
||||
from utils import generate_vcard
|
||||
import logging
|
||||
from logging.handlers import RotatingFileHandler
|
||||
from flask_migrate import Migrate
|
||||
|
||||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||||
DB_PATH = os.path.join(BASE_DIR, 'anmeldung.db')
|
||||
@@ -13,6 +15,20 @@ app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{DB_PATH}'
|
||||
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
|
||||
|
||||
db = SQLAlchemy(app)
|
||||
# Flask-Migrate (Alembic) integration
|
||||
migrate = Migrate(app, db)
|
||||
|
||||
# --- logging setup -------------------------------------------------
|
||||
LOG_DIR = os.path.join(BASE_DIR, 'logs')
|
||||
os.makedirs(LOG_DIR, exist_ok=True)
|
||||
log_file = os.path.join(LOG_DIR, 'app.log')
|
||||
handler = RotatingFileHandler(log_file, maxBytes=5 * 1024 * 1024, backupCount=3)
|
||||
formatter = logging.Formatter('%(asctime)s %(levelname)s %(name)s: %(message)s')
|
||||
handler.setFormatter(formatter)
|
||||
handler.setLevel(logging.INFO)
|
||||
app.logger.addHandler(handler)
|
||||
app.logger.setLevel(logging.INFO)
|
||||
# --------------------------------------------------------------------
|
||||
|
||||
|
||||
class Adresse(db.Model):
|
||||
@@ -98,45 +114,12 @@ def index():
|
||||
|
||||
# vCard 4.0 erzeugen und speichern
|
||||
try:
|
||||
vcards_dir = os.path.join(BASE_DIR, 'vcards')
|
||||
os.makedirs(vcards_dir, exist_ok=True)
|
||||
|
||||
# sanitize filename: remove diacritics and unsafe chars
|
||||
def slug(s):
|
||||
s = unicodedata.normalize('NFKD', s)
|
||||
s = ''.join(c for c in s if not unicodedata.combining(c))
|
||||
s = ''.join(c for c in s if c.isalnum() or c in (' ', '_', '-'))
|
||||
return s.replace(' ', '_')
|
||||
|
||||
filename = f"{slug(adresse.nachname)}_{slug(adresse.vorname)}_{adresse.id}.vcf"
|
||||
filepath = os.path.join(vcards_dir, filename)
|
||||
|
||||
# build vCard 4.0 content
|
||||
lines = [
|
||||
'BEGIN:VCARD',
|
||||
'VERSION:4.0',
|
||||
f'N:{adresse.nachname};{adresse.vorname};;;',
|
||||
f'FN:{adresse.vorname} {adresse.nachname}',
|
||||
]
|
||||
# ADR: PO Box;Extended;Street;Locality;Region;PostalCode;Country
|
||||
street = adresse.strasse or ''
|
||||
if adresse.hausnummer:
|
||||
street = f"{street} {adresse.hausnummer}".strip()
|
||||
adr = f'ADR:;;{street};{adresse.ort};;{adresse.plz};{adresse.land}'
|
||||
lines.append(adr)
|
||||
if adresse.email:
|
||||
lines.append(f'EMAIL;TYPE=internet:{adresse.email}')
|
||||
phone = ''
|
||||
if adresse.telefon_vorwahl or adresse.telefon_nummer:
|
||||
phone = f"+{adresse.telefon_vorwahl}{adresse.telefon_nummer}".replace('++', '+')
|
||||
lines.append(f'TEL;TYPE=voice:{phone}')
|
||||
lines.append('END:VCARD')
|
||||
|
||||
with open(filepath, 'w', encoding='utf-8') as f:
|
||||
f.write('\n'.join(lines))
|
||||
except Exception:
|
||||
# nicht kritisch: bei Fehlern nicht die ganze Anfrage abbrechen
|
||||
pass
|
||||
# determine base dir: prefer app.config, then app attribute, then module BASE_DIR
|
||||
base_dir = app.config.get('BASE_DIR') if app.config.get('BASE_DIR') else getattr(app, 'BASE_DIR', BASE_DIR)
|
||||
generate_vcard(adresse, base_dir)
|
||||
except Exception as e:
|
||||
# Log the exception with stack trace but don't abort the request
|
||||
app.logger.exception('Fehler beim Erzeugen der vCard for adresse id=%s: %s', adresse.id if hasattr(adresse, 'id') else 'unknown', e)
|
||||
|
||||
# Nach erfolgreichem Speichern weiterleiten
|
||||
return redirect(url_for('danke', id=adresse.id))
|
||||
@@ -149,12 +132,18 @@ def index():
|
||||
@app.route('/danke')
|
||||
def danke():
|
||||
id = request.args.get('id')
|
||||
adresse = Adresse.query.get(id)
|
||||
# Use session.get to avoid SQLAlchemy LegacyAPIWarning and cast id to int if present
|
||||
adresse = db.session.get(Adresse, int(id)) if id else None
|
||||
return render_template('danke.html', adresse=adresse)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
# Ensure DB exists
|
||||
if not os.path.exists(DB_PATH):
|
||||
# In development only: if there is no migrations directory, create tables automatically.
|
||||
# In all other cases prefer using Flask-Migrate (flask db upgrade).
|
||||
migrations_dir = os.path.join(BASE_DIR, 'migrations')
|
||||
if not os.path.exists(migrations_dir):
|
||||
app.logger.info('No migrations directory found; creating database tables with db.create_all()')
|
||||
db.create_all()
|
||||
else:
|
||||
app.logger.info('Migrations directory present; please use "flask db upgrade" to update the database schema')
|
||||
app.run(debug=True)
|
||||
|
||||
1
migrations/README
Normal file
1
migrations/README
Normal file
@@ -0,0 +1 @@
|
||||
Single-database configuration for Flask.
|
||||
50
migrations/alembic.ini
Normal file
50
migrations/alembic.ini
Normal file
@@ -0,0 +1,50 @@
|
||||
# A generic, single database configuration.
|
||||
|
||||
[alembic]
|
||||
# template used to generate migration files
|
||||
# file_template = %%(rev)s_%%(slug)s
|
||||
|
||||
# set to 'true' to run the environment during
|
||||
# the 'revision' command, regardless of autogenerate
|
||||
# revision_environment = false
|
||||
|
||||
|
||||
# Logging configuration
|
||||
[loggers]
|
||||
keys = root,sqlalchemy,alembic,flask_migrate
|
||||
|
||||
[handlers]
|
||||
keys = console
|
||||
|
||||
[formatters]
|
||||
keys = generic
|
||||
|
||||
[logger_root]
|
||||
level = WARN
|
||||
handlers = console
|
||||
qualname =
|
||||
|
||||
[logger_sqlalchemy]
|
||||
level = WARN
|
||||
handlers =
|
||||
qualname = sqlalchemy.engine
|
||||
|
||||
[logger_alembic]
|
||||
level = INFO
|
||||
handlers =
|
||||
qualname = alembic
|
||||
|
||||
[logger_flask_migrate]
|
||||
level = INFO
|
||||
handlers =
|
||||
qualname = flask_migrate
|
||||
|
||||
[handler_console]
|
||||
class = StreamHandler
|
||||
args = (sys.stderr,)
|
||||
level = NOTSET
|
||||
formatter = generic
|
||||
|
||||
[formatter_generic]
|
||||
format = %(levelname)-5.5s [%(name)s] %(message)s
|
||||
datefmt = %H:%M:%S
|
||||
113
migrations/env.py
Normal file
113
migrations/env.py
Normal file
@@ -0,0 +1,113 @@
|
||||
import logging
|
||||
from logging.config import fileConfig
|
||||
|
||||
from flask import current_app
|
||||
|
||||
from alembic import context
|
||||
|
||||
# this is the Alembic Config object, which provides
|
||||
# access to the values within the .ini file in use.
|
||||
config = context.config
|
||||
|
||||
# Interpret the config file for Python logging.
|
||||
# This line sets up loggers basically.
|
||||
fileConfig(config.config_file_name)
|
||||
logger = logging.getLogger('alembic.env')
|
||||
|
||||
|
||||
def get_engine():
|
||||
try:
|
||||
# this works with Flask-SQLAlchemy<3 and Alchemical
|
||||
return current_app.extensions['migrate'].db.get_engine()
|
||||
except (TypeError, AttributeError):
|
||||
# this works with Flask-SQLAlchemy>=3
|
||||
return current_app.extensions['migrate'].db.engine
|
||||
|
||||
|
||||
def get_engine_url():
|
||||
try:
|
||||
return get_engine().url.render_as_string(hide_password=False).replace(
|
||||
'%', '%%')
|
||||
except AttributeError:
|
||||
return str(get_engine().url).replace('%', '%%')
|
||||
|
||||
|
||||
# add your model's MetaData object here
|
||||
# for 'autogenerate' support
|
||||
# from myapp import mymodel
|
||||
# target_metadata = mymodel.Base.metadata
|
||||
config.set_main_option('sqlalchemy.url', get_engine_url())
|
||||
target_db = current_app.extensions['migrate'].db
|
||||
|
||||
# other values from the config, defined by the needs of env.py,
|
||||
# can be acquired:
|
||||
# my_important_option = config.get_main_option("my_important_option")
|
||||
# ... etc.
|
||||
|
||||
|
||||
def get_metadata():
|
||||
if hasattr(target_db, 'metadatas'):
|
||||
return target_db.metadatas[None]
|
||||
return target_db.metadata
|
||||
|
||||
|
||||
def run_migrations_offline():
|
||||
"""Run migrations in 'offline' mode.
|
||||
|
||||
This configures the context with just a URL
|
||||
and not an Engine, though an Engine is acceptable
|
||||
here as well. By skipping the Engine creation
|
||||
we don't even need a DBAPI to be available.
|
||||
|
||||
Calls to context.execute() here emit the given string to the
|
||||
script output.
|
||||
|
||||
"""
|
||||
url = config.get_main_option("sqlalchemy.url")
|
||||
context.configure(
|
||||
url=url, target_metadata=get_metadata(), literal_binds=True
|
||||
)
|
||||
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
||||
|
||||
def run_migrations_online():
|
||||
"""Run migrations in 'online' mode.
|
||||
|
||||
In this scenario we need to create an Engine
|
||||
and associate a connection with the context.
|
||||
|
||||
"""
|
||||
|
||||
# this callback is used to prevent an auto-migration from being generated
|
||||
# when there are no changes to the schema
|
||||
# reference: http://alembic.zzzcomputing.com/en/latest/cookbook.html
|
||||
def process_revision_directives(context, revision, directives):
|
||||
if getattr(config.cmd_opts, 'autogenerate', False):
|
||||
script = directives[0]
|
||||
if script.upgrade_ops.is_empty():
|
||||
directives[:] = []
|
||||
logger.info('No changes in schema detected.')
|
||||
|
||||
conf_args = current_app.extensions['migrate'].configure_args
|
||||
if conf_args.get("process_revision_directives") is None:
|
||||
conf_args["process_revision_directives"] = process_revision_directives
|
||||
|
||||
connectable = get_engine()
|
||||
|
||||
with connectable.connect() as connection:
|
||||
context.configure(
|
||||
connection=connection,
|
||||
target_metadata=get_metadata(),
|
||||
**conf_args
|
||||
)
|
||||
|
||||
with context.begin_transaction():
|
||||
context.run_migrations()
|
||||
|
||||
|
||||
if context.is_offline_mode():
|
||||
run_migrations_offline()
|
||||
else:
|
||||
run_migrations_online()
|
||||
24
migrations/script.py.mako
Normal file
24
migrations/script.py.mako
Normal file
@@ -0,0 +1,24 @@
|
||||
"""${message}
|
||||
|
||||
Revision ID: ${up_revision}
|
||||
Revises: ${down_revision | comma,n}
|
||||
Create Date: ${create_date}
|
||||
|
||||
"""
|
||||
from alembic import op
|
||||
import sqlalchemy as sa
|
||||
${imports if imports else ""}
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = ${repr(up_revision)}
|
||||
down_revision = ${repr(down_revision)}
|
||||
branch_labels = ${repr(branch_labels)}
|
||||
depends_on = ${repr(depends_on)}
|
||||
|
||||
|
||||
def upgrade():
|
||||
${upgrades if upgrades else "pass"}
|
||||
|
||||
|
||||
def downgrade():
|
||||
${downgrades if downgrades else "pass"}
|
||||
2
requirements-dev.txt
Normal file
2
requirements-dev.txt
Normal file
@@ -0,0 +1,2 @@
|
||||
pytest
|
||||
Flask-Migrate>=4.0
|
||||
@@ -1,2 +1,4 @@
|
||||
Flask>=2.0
|
||||
Flask-SQLAlchemy>=3.0
|
||||
Flask-Migrate>=4.0
|
||||
alembic>=1.9
|
||||
|
||||
29
tests/test_utils_vcard.py
Normal file
29
tests/test_utils_vcard.py
Normal file
@@ -0,0 +1,29 @@
|
||||
from types import SimpleNamespace
|
||||
from pathlib import Path
|
||||
|
||||
from utils import generate_vcard
|
||||
|
||||
|
||||
def test_generate_vcard_writes_file(tmp_path):
|
||||
addr = SimpleNamespace(
|
||||
vorname='Anna',
|
||||
nachname='Muster',
|
||||
strasse='Beispielweg',
|
||||
hausnummer='5a',
|
||||
plz='54321',
|
||||
ort='Beispielstadt',
|
||||
land='Deutschland',
|
||||
email='anna@example.com',
|
||||
telefon_vorwahl='49',
|
||||
telefon_nummer='7654321',
|
||||
id=42,
|
||||
)
|
||||
|
||||
path = generate_vcard(addr, str(tmp_path))
|
||||
assert Path(path).exists()
|
||||
|
||||
content = Path(path).read_text(encoding='utf-8')
|
||||
assert 'BEGIN:VCARD' in content
|
||||
assert 'FN:Anna Muster' in content
|
||||
assert 'EMAIL;TYPE=internet:anna@example.com' in content
|
||||
assert 'TEL;TYPE=voice:+497654321' in content
|
||||
65
tests/test_vcard_export.py
Normal file
65
tests/test_vcard_export.py
Normal file
@@ -0,0 +1,65 @@
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
|
||||
from app import app, db, Frage
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client(tmp_path):
|
||||
# Use a temporary directory for vcards and a temporary sqlite db
|
||||
# temp DB file
|
||||
db_file = tmp_path / "test.db"
|
||||
|
||||
# Configure app for testing
|
||||
app.config['TESTING'] = True
|
||||
app.config['SQLALCHEMY_DATABASE_URI'] = f'sqlite:///{db_file}'
|
||||
|
||||
# Use temp vcards dir by setting app.BASE_DIR
|
||||
app.BASE_DIR = str(tmp_path)
|
||||
|
||||
# ensure clean DB and create tables
|
||||
with app.app_context():
|
||||
db.drop_all()
|
||||
db.create_all()
|
||||
# create a sample question so POST processing loops over it
|
||||
q = Frage(text='Testfrage?')
|
||||
db.session.add(q)
|
||||
db.session.commit()
|
||||
# record question id for use in test POST data
|
||||
app.config['TEST_QUESTION_ID'] = q.id
|
||||
|
||||
with app.test_client() as test_client:
|
||||
yield test_client
|
||||
|
||||
|
||||
def test_vcard_created_after_submit(client):
|
||||
data = {
|
||||
'vorname': 'Max',
|
||||
'nachname': 'Mustermann',
|
||||
'strasse': 'Musterstraße',
|
||||
'hausnummer': '1',
|
||||
'plz': '12345',
|
||||
'ort': 'Musterstadt',
|
||||
'land': 'Deutschland',
|
||||
'telefon_vorwahl': '49',
|
||||
'telefon_nummer': '1234567',
|
||||
'email': 'max@example.com',
|
||||
# question id set in fixture
|
||||
f'frage_{app.config.get("TEST_QUESTION_ID")}': 'Antwort'
|
||||
}
|
||||
|
||||
# Submit the form
|
||||
res = client.post('/', data=data, follow_redirects=True)
|
||||
assert res.status_code == 200
|
||||
|
||||
# find vcards dir under tmp_path (app.BASE_DIR)
|
||||
vcards_dir = Path(app.BASE_DIR) / 'vcards'
|
||||
files = list(vcards_dir.glob('*.vcf'))
|
||||
assert len(files) == 1
|
||||
|
||||
content = files[0].read_text(encoding='utf-8')
|
||||
assert 'BEGIN:VCARD' in content
|
||||
assert 'VERSION:4.0' in content
|
||||
assert 'FN:Max Mustermann' in content
|
||||
assert 'EMAIL;TYPE=internet:max@example.com' in content
|
||||
assert 'ADR:' in content
|
||||
57
utils.py
Normal file
57
utils.py
Normal file
@@ -0,0 +1,57 @@
|
||||
import os
|
||||
import unicodedata
|
||||
|
||||
|
||||
def _slug(s: str) -> str:
|
||||
if not s:
|
||||
return ''
|
||||
s = unicodedata.normalize('NFKD', s)
|
||||
s = ''.join(c for c in s if not unicodedata.combining(c))
|
||||
s = ''.join(c for c in s if c.isalnum() or c in (' ', '_', '-'))
|
||||
return s.replace(' ', '_')
|
||||
|
||||
|
||||
def generate_vcard(adresse, base_dir: str):
|
||||
"""Generate a vCard 4.0 file for the given adresse object.
|
||||
|
||||
adresse: object with attributes vorname, nachname, strasse, hausnummer,
|
||||
plz, ort, land, email, telefon_vorwahl, telefon_nummer, id
|
||||
base_dir: directory where 'vcards/' will be created
|
||||
|
||||
Returns the path to the written vcard file.
|
||||
"""
|
||||
vcards_dir = os.path.join(base_dir, 'vcards')
|
||||
os.makedirs(vcards_dir, exist_ok=True)
|
||||
|
||||
filename = f"{_slug(adresse.nachname)}_{_slug(adresse.vorname)}_{adresse.id}.vcf"
|
||||
filepath = os.path.join(vcards_dir, filename)
|
||||
|
||||
# build vCard 4.0 content
|
||||
lines = [
|
||||
'BEGIN:VCARD',
|
||||
'VERSION:4.0',
|
||||
f'N:{adresse.nachname};{adresse.vorname};;;',
|
||||
f'FN:{adresse.vorname} {adresse.nachname}',
|
||||
]
|
||||
|
||||
street = getattr(adresse, 'strasse', '') or ''
|
||||
hausnummer = getattr(adresse, 'hausnummer', '') or ''
|
||||
if hausnummer:
|
||||
street = f"{street} {hausnummer}".strip()
|
||||
|
||||
adr = f'ADR:;;{street};{getattr(adresse, "ort", "")};;{getattr(adresse, "plz", "")};{getattr(adresse, "land", "")}'
|
||||
lines.append(adr)
|
||||
|
||||
if getattr(adresse, 'email', None):
|
||||
lines.append(f'EMAIL;TYPE=internet:{adresse.email}')
|
||||
|
||||
if getattr(adresse, 'telefon_vorwahl', None) or getattr(adresse, 'telefon_nummer', None):
|
||||
tel = f"+{getattr(adresse, 'telefon_vorwahl', '')}{getattr(adresse, 'telefon_nummer', '')}".replace('++', '+')
|
||||
lines.append(f'TEL;TYPE=voice:{tel}')
|
||||
|
||||
lines.append('END:VCARD')
|
||||
|
||||
with open(filepath, 'w', encoding='utf-8') as f:
|
||||
f.write('\n'.join(lines))
|
||||
|
||||
return filepath
|
||||
Reference in New Issue
Block a user