186 lines
9.0 KiB
Python
186 lines
9.0 KiB
Python
"""One-command invoice creation with reusable private customer records."""
|
||
from dataclasses import dataclass
|
||
from datetime import date, timedelta
|
||
from decimal import Decimal, InvalidOperation
|
||
from pathlib import Path
|
||
import calendar
|
||
import hashlib
|
||
import json
|
||
import re
|
||
import shutil
|
||
import tempfile
|
||
|
||
import yaml
|
||
|
||
from .calc import compute_totals
|
||
from .utils import money_fmt
|
||
from .folder_naming import default_invoice_folder_name
|
||
from .invoice_number import invoice_root_lock
|
||
from .models import Invoice, Party
|
||
from .pipeline import build_in_folder, BuildResult
|
||
from .presets import load_yaml
|
||
|
||
MONTHS = 'januar februar märz april mai juni juli august september oktober november dezember'.split()
|
||
|
||
|
||
def decimal_input(value: str) -> Decimal:
|
||
"""Accept 85,00 €, 1.234,56 EUR or ungrouped decimal-point notation."""
|
||
raw = re.sub(r'\s*(?:€|EUR)\s*$', '', value.strip(), flags=re.I)
|
||
raw = raw.replace('\u00a0', '').strip()
|
||
if ',' in raw:
|
||
if not re.fullmatch(r'[+-]?(?:\d+|\d{1,3}(?:\.\d{3})+),\d+', raw):
|
||
raise ValueError(f'Ungültige Dezimalzahl: {value!r}')
|
||
raw = raw.replace('.', '').replace(',', '.')
|
||
elif not re.fullmatch(r'[+-]?\d+(?:\.\d+)?', raw):
|
||
raise ValueError(f'Ungültige Dezimalzahl: {value!r}')
|
||
try:
|
||
result = Decimal(raw)
|
||
except InvalidOperation as exc:
|
||
raise ValueError(f'Ungültige Dezimalzahl: {value!r}') from exc
|
||
if not result.is_finite():
|
||
raise ValueError('Zahl muss endlich sein')
|
||
return result
|
||
|
||
|
||
def customer_key(name: str) -> str:
|
||
# Hash prevents punctuation/transliteration collisions and path traversal.
|
||
readable = re.sub(r'[^a-z0-9]+', '-', name.casefold()).strip('-')[:32] or 'kunde'
|
||
return f'{readable}-{hashlib.sha256(name.casefold().encode()).hexdigest()[:8]}'
|
||
|
||
|
||
def find_customer(presets: Path, name: str) -> tuple[str, dict] | None:
|
||
wanted = name.strip().casefold()
|
||
matches = []
|
||
for path in sorted((presets / 'customers').glob('*.yaml')):
|
||
data = load_yaml(path)
|
||
if wanted in (path.stem.casefold(), str(data.get('buyer', {}).get('name', '')).casefold()):
|
||
matches.append((path.stem, data))
|
||
if len(matches) > 1:
|
||
raise ValueError('Kundenname ist mehrdeutig; bitte den eindeutigen Kundenschlüssel verwenden')
|
||
return matches[0] if matches else None
|
||
|
||
|
||
def save_customer(presets: Path, name: str, buyer: dict) -> tuple[str, dict]:
|
||
buyer = Party.model_validate(buyer).model_dump(mode='json')
|
||
key = customer_key(name)
|
||
directory = presets / 'customers'
|
||
directory.mkdir(parents=True, exist_ok=True, mode=0o700)
|
||
target = directory / f'{key}.yaml'
|
||
data = {'buyer': buyer}
|
||
# Exclusive create; never replace an existing customer's stored details silently.
|
||
with target.open('x', encoding='utf-8') as file:
|
||
target.chmod(0o600)
|
||
yaml.safe_dump(data, file, allow_unicode=True, sort_keys=False)
|
||
return key, data
|
||
|
||
|
||
def service_period(entry: str, month: str | None, service_date: date | None, issued: date):
|
||
if month and service_date:
|
||
raise ValueError('--Month und --ServiceDate können nicht kombiniert werden')
|
||
if service_date:
|
||
return service_date, None, None
|
||
if month is None:
|
||
match = re.search(r'\b(' + '|'.join(MONTHS) + r')\s+(\d{4})\b', entry.casefold())
|
||
if match:
|
||
month = f'{match[2]}-{MONTHS.index(match[1]) + 1:02d}'
|
||
if month:
|
||
if not re.fullmatch(r'\d{4}-\d{2}', month):
|
||
raise ValueError('--Month muss YYYY-MM sein')
|
||
year, m = map(int, month.split('-'))
|
||
first = date(year, m, 1)
|
||
last = date(year, m, calendar.monthrange(year, m)[1])
|
||
return last, first, last
|
||
return issued, None, None
|
||
|
||
|
||
@dataclass
|
||
class CreatedInvoice:
|
||
build: BuildResult
|
||
attachment: Path
|
||
invoice: Invoice
|
||
|
||
|
||
def create_invoice(*, root: Path, resources: Path, presets: Path, customer: str,
|
||
entry: str, volume: Decimal, price: Decimal, jar: Path, logger,
|
||
number: str | None = None, issued: date | None = None,
|
||
month: str | None = None, service_date: date | None = None,
|
||
unit: str = 'HUR', tax: Decimal = Decimal('19'), due_days: int = 30,
|
||
buyer_reference: str | None = None, order_reference: str | None = None,
|
||
progress=lambda message: None) -> CreatedInvoice:
|
||
if due_days < 0:
|
||
raise ValueError('Zahlungsziel darf nicht negativ sein')
|
||
if not jar.is_file():
|
||
raise FileNotFoundError('Mustang fehlt. Einmalig jr-einvoice setup-mustang ausführen')
|
||
seller_data = load_yaml(presets / 'seller.default.yaml')
|
||
if not seller_data.get('seller') or not seller_data.get('payment'):
|
||
raise ValueError(f'Eigene Absender- und Zahlungsdaten in {presets / "seller.default.yaml"} hinterlegen')
|
||
match = find_customer(presets, customer)
|
||
if match is None:
|
||
raise ValueError(f'Kunde {customer!r} ist noch nicht angelegt')
|
||
key, record = match
|
||
issued = issued or date.today()
|
||
delivery, start, end = service_period(entry, month, service_date, issued)
|
||
root = root.expanduser().resolve()
|
||
with invoice_root_lock(root):
|
||
# Only dedicated invoice directories; unknown/corrupt records stop allocation.
|
||
used = set()
|
||
for path in root.glob('*/invoice.yaml'):
|
||
existing = load_yaml(path).get('meta', {}).get('invoice_number')
|
||
if not isinstance(existing, str) or not existing.strip():
|
||
raise ValueError(f'Rechnungsnummer im vorhandenen Entwurf unklar: {path}')
|
||
used.add(existing)
|
||
if not number:
|
||
prefix = f'JR-{issued.year}-'
|
||
sequence = max((int(n[len(prefix):]) for n in used
|
||
if n.startswith(prefix) and n[len(prefix):].isdigit()), default=0) + 1
|
||
number = f'{prefix}{sequence:04d}'
|
||
if number in used:
|
||
raise ValueError(f'Rechnungsnummer bereits vergeben: {number}')
|
||
if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_-]{0,63}', number):
|
||
raise ValueError('Rechnungsnummer: 1–64 Buchstaben/Ziffern, Bindestrich oder Unterstrich')
|
||
target = root / default_invoice_folder_name(number, key)
|
||
if target.exists():
|
||
raise FileExistsError(f'Rechnungsordner existiert bereits: {target}')
|
||
inv = Invoice.model_validate({
|
||
'meta': {'invoice_number': number, 'invoice_date': issued, 'service_date': delivery,
|
||
'service_period_start': start, 'service_period_end': end,
|
||
'buyer_reference': buyer_reference, 'order_reference': order_reference,
|
||
'notes': 'Vielen Dank für Ihren Auftrag.'},
|
||
'seller': seller_data['seller'], 'buyer': record['buyer'],
|
||
'payment': {**seller_data['payment'], 'due_date': issued + timedelta(days=due_days),
|
||
'reference': f'Rechnung {number}'},
|
||
'items': [{'name': entry, 'quantity': volume, 'unit': unit,
|
||
'unit_price_net': price, 'tax_rate': tax}],
|
||
})
|
||
with tempfile.TemporaryDirectory(prefix='.create-', dir=root) as staging:
|
||
work = Path(staging) / 'invoice'
|
||
work.mkdir(mode=0o700)
|
||
(work / 'invoice.yaml').write_text(yaml.safe_dump(inv.model_dump(mode='json'),
|
||
allow_unicode=True, sort_keys=False), encoding='utf-8')
|
||
(work / 'assets').mkdir()
|
||
logo = presets / 'logo.svg'
|
||
if not logo.is_file():
|
||
logo = resources / 'assets/logo.svg'
|
||
if logo.is_file():
|
||
shutil.copy2(logo, work / 'assets/logo.svg')
|
||
work.rename(target)
|
||
# Retain the reserved number and input on build failures for diagnosis/retry.
|
||
from .logging_utils import setup_logging
|
||
logger = setup_logging('WARNING', target / 'out/logs/run.log')
|
||
progress(f'Rechnungsdaten vollständig: {number} · {inv.buyer.name}')
|
||
progress('PDF/A und strukturierte Rechnungsdaten werden erzeugt …')
|
||
try:
|
||
build = build_in_folder(target, 'out', resources / 'templates/invoice.html.j2', logger, jar, True, True)
|
||
except Exception as exc:
|
||
raise RuntimeError(f'Erstellung fehlgeschlagen. Entwurf: {target}. Log: {target / "out/logs/run.log"}. {exc}') from exc
|
||
progress('Rechnung validiert: PDF/A und EN16931-XML technisch gültig.')
|
||
attachment = target / f'Rechnung_{number}.pdf'
|
||
shutil.copy2(build.facturx_pdf, attachment)
|
||
manifest = {str(p.relative_to(target)): hashlib.sha256(p.read_bytes()).hexdigest()
|
||
for p in [target / 'invoice.yaml', build.facturx_xml, build.facturx_pdf,
|
||
build.validate_report, attachment]}
|
||
(target / 'out/manifest.json').write_text(json.dumps(manifest, indent=2) + '\n', encoding='utf-8')
|
||
totals = compute_totals(inv)
|
||
progress(f'Rechnung erzeugt · Netto: {money_fmt(totals.net_total)} · USt: {money_fmt(totals.tax_total)} · Brutto: {money_fmt(totals.gross_total)}')
|
||
return CreatedInvoice(build, attachment, inv)
|