100 lines
4.4 KiB
Python
100 lines
4.4 KiB
Python
"""Geprüfte Formulardaten und private Vorlagen ohne Abhängigkeit von Qt."""
|
||
from datetime import date, timedelta
|
||
from pathlib import Path
|
||
import os
|
||
import re
|
||
import tempfile
|
||
|
||
from pydantic import Field, model_validator
|
||
import yaml
|
||
|
||
from .models import Invoice, Item, Party, StrictModel
|
||
from .presets import load_yaml
|
||
|
||
|
||
class InvoiceForm(StrictModel):
|
||
"""Vollständiger neuer Auftrag; die endgültige Nummer vergibt der Kern."""
|
||
|
||
buyer: Party
|
||
items: list[Item] = Field(min_length=1)
|
||
issued: date = Field(default_factory=date.today)
|
||
due_days: int = Field(default=30, ge=0, le=3650)
|
||
start: date
|
||
end: date
|
||
number: str | None = Field(default=None, pattern=r'^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$')
|
||
notes: str = 'Vielen Dank für Ihren Auftrag.'
|
||
buyer_reference: str | None = None
|
||
order_reference: str | None = None
|
||
|
||
@model_validator(mode='after')
|
||
def ordered_period(self):
|
||
if self.start > self.end:
|
||
raise ValueError('Der Leistungsbeginn muss vor dem Leistungsende liegen')
|
||
return self
|
||
|
||
def invoice(self, presets: Path) -> Invoice:
|
||
"""Auch Absender und Zahlungsdaten vor jeder Erstellung vollständig prüfen."""
|
||
seller = load_yaml(presets / 'seller.default.yaml')
|
||
if not seller.get('seller') or not seller.get('payment'):
|
||
raise ValueError(f'Absender und Zahlungsdaten fehlen: {presets / "seller.default.yaml"}')
|
||
return Invoice.model_validate({
|
||
'seller': seller['seller'], 'buyer': self.buyer,
|
||
'items': self.items,
|
||
'meta': {'invoice_number': self.number or 'AUTOMATISCH', 'invoice_date': self.issued,
|
||
'service_date': self.end, 'service_period_start': self.start,
|
||
'service_period_end': self.end, 'notes': self.notes,
|
||
'buyer_reference': self.buyer_reference, 'order_reference': self.order_reference},
|
||
'payment': {**seller['payment'], 'due_date': self.issued + timedelta(days=self.due_days),
|
||
'reference': f'Rechnung {self.number or "AUTOMATISCH"}'}})
|
||
|
||
|
||
class InvoiceTemplate(StrictModel):
|
||
"""Vorlagen enthalten bewusst keine alte Rechnungsnummer oder Leistungsdaten."""
|
||
|
||
schema_version: int = Field(default=1, ge=1, le=1)
|
||
buyer: Party
|
||
items: list[Item] = Field(min_length=1)
|
||
due_days: int = Field(default=30, ge=0, le=3650)
|
||
notes: str = 'Vielen Dank für Ihren Auftrag.'
|
||
buyer_reference: str | None = None
|
||
order_reference: str | None = None
|
||
|
||
@classmethod
|
||
def from_form(cls, form: InvoiceForm):
|
||
return cls.model_validate(form.model_dump(exclude={'issued', 'start', 'end', 'number'}))
|
||
|
||
|
||
def load_template(path: Path) -> InvoiceTemplate:
|
||
"""Eigene Vorlagen oder bestehende invoice.yaml als neue Rechnung übernehmen."""
|
||
data = load_yaml(path)
|
||
if 'meta' in data:
|
||
inv = Invoice.model_validate(data)
|
||
return InvoiceTemplate(buyer=inv.buyer, items=inv.items,
|
||
due_days=(inv.payment.due_date - inv.meta.invoice_date).days,
|
||
notes=inv.meta.notes or '', buyer_reference=inv.meta.buyer_reference,
|
||
order_reference=inv.meta.order_reference)
|
||
return InvoiceTemplate.model_validate(data)
|
||
|
||
|
||
def save_template(directory: Path, name: str, template: InvoiceTemplate, *, overwrite=False) -> Path:
|
||
"""Vorlagen geschützt speichern; vorhandene Namen nur ausdrücklich überschreiben."""
|
||
if not re.fullmatch(r'[\wÄÖÜäöüß][\wÄÖÜäöüß .-]{0,79}', name) or name.endswith(('.', ' ')):
|
||
raise ValueError('Vorlagenname: 1–80 Buchstaben/Ziffern, Leerzeichen, Punkt oder Bindestrich')
|
||
directory.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||
target = directory / f'{name}.yaml'
|
||
# Die temporäre Datei bleibt bei jedem Fehler außerhalb der Vorlagenauswahl.
|
||
fd, temporary = tempfile.mkstemp(prefix='.vorlage-', dir=directory)
|
||
try:
|
||
with os.fdopen(fd, 'w', encoding='utf-8') as stream:
|
||
yaml.safe_dump(template.model_dump(mode='json'), stream, allow_unicode=True, sort_keys=False)
|
||
stream.flush()
|
||
os.fsync(stream.fileno())
|
||
if overwrite:
|
||
os.replace(temporary, target)
|
||
else:
|
||
# link statt exists+rename verhindert auch gleichzeitiges Überschreiben.
|
||
os.link(temporary, target)
|
||
finally:
|
||
Path(temporary).unlink(missing_ok=True)
|
||
return target
|