rene
b68a12587a
- DB-Migration: invoices + invoice_items Tabellen inkl. Indizes - routes/invoices.py: vollständiger Admin-Router (prefix /api/admin/invoices) - CRUD: Liste, Detail, Erstellen, Senden, Bezahlen, Stornieren - PDF-Generierung via fpdf2 mit §14-UStG-Pflichtangaben (Kleinunternehmer-Hinweis) - Cashflow-Übersicht und Quartalsbericht - PDF-Download-Endpunkt - Speicherung in /scaninput + optionaler Paperless-Upload - mailer.py: send_email() + Backends um optionale PDF-Anhänge erweitert (Brevo: base64, SMTP: MIMEApplication) - main.py: invoices_router registriert - docker-compose.yml: /volume1/scaninput:/scaninput Volume hinzugefügt
489 lines
18 KiB
Python
489 lines
18 KiB
Python
"""BAN YARO — Rechnungs-System (Admin)"""
|
||
|
||
import os
|
||
import logging
|
||
from datetime import datetime
|
||
from typing import Optional, List
|
||
from fastapi import APIRouter, Depends, HTTPException
|
||
from fastapi.responses import Response
|
||
from pydantic import BaseModel
|
||
from database import db
|
||
from auth import require_admin
|
||
import mailer
|
||
|
||
router = APIRouter(prefix="/api/admin/invoices", tags=["invoices"])
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Schemas
|
||
# ------------------------------------------------------------------
|
||
class InvoiceItem(BaseModel):
|
||
description: str
|
||
quantity: float = 1.0
|
||
unit_price: float
|
||
|
||
|
||
class InvoiceCreate(BaseModel):
|
||
user_id: Optional[int] = None
|
||
recipient_name: str
|
||
recipient_email: str
|
||
recipient_address: Optional[str] = None
|
||
items: List[InvoiceItem]
|
||
discount_pct: Optional[float] = 0.0
|
||
service_period: Optional[str] = None
|
||
notes: Optional[str] = None
|
||
|
||
|
||
class PayBody(BaseModel):
|
||
paid_at: str
|
||
paid_amount: float
|
||
|
||
|
||
class CancelBody(BaseModel):
|
||
reason: str
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Hilfsfunktionen
|
||
# ------------------------------------------------------------------
|
||
def _next_invoice_number(conn, prefix="RG"):
|
||
year = datetime.now().year
|
||
last = conn.execute(
|
||
"SELECT invoice_number FROM invoices WHERE invoice_number LIKE ? ORDER BY id DESC LIMIT 1",
|
||
(f"{prefix}-{year}-%",)
|
||
).fetchone()
|
||
if last:
|
||
n = int(last[0].split("-")[-1]) + 1
|
||
else:
|
||
n = 1
|
||
return f"{prefix}-{year}-{n:04d}"
|
||
|
||
|
||
def _generate_pdf(invoice, items) -> bytes:
|
||
KLEINUNTERNEHMER = os.getenv("KLEINUNTERNEHMER", "true").lower() == "true"
|
||
STEUERNUMMER = os.getenv("STEUERNUMMER", "")
|
||
INHABER = os.getenv("RECHNUNG_INHABER", "René Degelmann")
|
||
GESCHAEFTSNAME = os.getenv("RECHNUNG_GESCHAEFTSNAME", "Ban Yaro")
|
||
STRASSE = os.getenv("RECHNUNG_STRASSE", "")
|
||
PLZ_ORT = os.getenv("RECHNUNG_PLZ_ORT", "")
|
||
EMAIL_ABSENDER = os.getenv("RECHNUNG_EMAIL", "hallo@banyaro.app")
|
||
BANK_IBAN = os.getenv("RECHNUNG_IBAN", "")
|
||
BANK_BIC = os.getenv("RECHNUNG_BIC", "")
|
||
BANK_BANK = os.getenv("RECHNUNG_BANK", "")
|
||
|
||
from fpdf import FPDF
|
||
pdf = FPDF()
|
||
pdf.add_page()
|
||
pdf.set_margins(20, 20, 20)
|
||
pdf.set_auto_page_break(auto=True, margin=20)
|
||
|
||
pdf.set_font("Helvetica", "B", 18)
|
||
pdf.cell(0, 10, GESCHAEFTSNAME, new_x="LMARGIN", new_y="NEXT")
|
||
pdf.set_font("Helvetica", "", 10)
|
||
pdf.cell(0, 5, INHABER, new_x="LMARGIN", new_y="NEXT")
|
||
if STRASSE:
|
||
pdf.cell(0, 5, STRASSE, new_x="LMARGIN", new_y="NEXT")
|
||
if PLZ_ORT:
|
||
pdf.cell(0, 5, PLZ_ORT, new_x="LMARGIN", new_y="NEXT")
|
||
pdf.cell(0, 5, EMAIL_ABSENDER, new_x="LMARGIN", new_y="NEXT")
|
||
pdf.ln(10)
|
||
|
||
pdf.set_font("Helvetica", "B", 10)
|
||
pdf.cell(0, 5, "Rechnungsempfänger:", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.set_font("Helvetica", "", 10)
|
||
pdf.cell(0, 5, invoice["recipient_name"], new_x="LMARGIN", new_y="NEXT")
|
||
if invoice["recipient_address"]:
|
||
for line in invoice["recipient_address"].split("\n"):
|
||
pdf.cell(0, 5, line, new_x="LMARGIN", new_y="NEXT")
|
||
pdf.cell(0, 5, invoice["recipient_email"], new_x="LMARGIN", new_y="NEXT")
|
||
pdf.ln(8)
|
||
|
||
pdf.set_font("Helvetica", "B", 10)
|
||
pdf.cell(0, 5, f"Rechnungsnummer: {invoice['invoice_number']}", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.set_font("Helvetica", "", 10)
|
||
rg_date = invoice["created_at"][:10] if invoice["created_at"] else ""
|
||
pdf.cell(0, 5, f"Rechnungsdatum: {rg_date}", new_x="LMARGIN", new_y="NEXT")
|
||
if invoice["service_period"]:
|
||
pdf.cell(0, 5, f"Leistungszeitraum: {invoice['service_period']}", new_x="LMARGIN", new_y="NEXT")
|
||
if STEUERNUMMER:
|
||
pdf.cell(0, 5, f"Steuernummer: {STEUERNUMMER}", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.ln(8)
|
||
|
||
pdf.set_font("Helvetica", "B", 14)
|
||
pdf.cell(0, 10, "RECHNUNG", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.ln(4)
|
||
|
||
pdf.set_fill_color(240, 240, 240)
|
||
pdf.set_font("Helvetica", "B", 9)
|
||
pdf.cell(90, 7, "Beschreibung", border=1, fill=True)
|
||
pdf.cell(20, 7, "Menge", border=1, fill=True, align="C")
|
||
pdf.cell(35, 7, "Einzelpreis", border=1, fill=True, align="R")
|
||
pdf.cell(35, 7, "Gesamt", border=1, fill=True, align="R", new_x="LMARGIN", new_y="NEXT")
|
||
|
||
pdf.set_font("Helvetica", "", 9)
|
||
for item in items:
|
||
qty_str = f"{item['quantity']:.2f}".rstrip("0").rstrip(".")
|
||
pdf.cell(90, 7, str(item["description"])[:60], border=1)
|
||
pdf.cell(20, 7, qty_str, border=1, align="C")
|
||
pdf.cell(35, 7, f"{item['unit_price']:.2f} EUR", border=1, align="R")
|
||
pdf.cell(35, 7, f"{item['total']:.2f} EUR", border=1, align="R", new_x="LMARGIN", new_y="NEXT")
|
||
|
||
pdf.ln(4)
|
||
|
||
def right_row(label, value, bold=False):
|
||
pdf.set_font("Helvetica", "B" if bold else "", 10)
|
||
pdf.cell(120, 6, "")
|
||
pdf.cell(40, 6, label, align="R")
|
||
pdf.cell(10, 6, "")
|
||
pdf.cell(0, 6, value, align="R", new_x="LMARGIN", new_y="NEXT")
|
||
|
||
right_row("Nettobetrag:", f"{invoice['amount_net']:.2f} EUR")
|
||
if invoice["discount_pct"] and invoice["discount_pct"] > 0:
|
||
right_row(f"Rabatt ({invoice['discount_pct']:.0f}%):", f"-{invoice['discount_amount']:.2f} EUR")
|
||
right_row("Nach Rabatt:", f"{invoice['amount_after_discount']:.2f} EUR")
|
||
|
||
if not KLEINUNTERNEHMER:
|
||
right_row(f"MwSt. {invoice['tax_rate']:.0f}%:", f"{invoice['tax_amount']:.2f} EUR")
|
||
|
||
pdf.ln(2)
|
||
pdf.set_draw_color(0, 0, 0)
|
||
right_row("Gesamtbetrag:", f"{invoice['amount_gross']:.2f} EUR", bold=True)
|
||
|
||
pdf.ln(8)
|
||
|
||
if KLEINUNTERNEHMER:
|
||
pdf.set_font("Helvetica", "I", 9)
|
||
pdf.multi_cell(0, 5, "Gemäß § 19 UStG wird keine Umsatzsteuer berechnet.")
|
||
pdf.ln(4)
|
||
|
||
if BANK_IBAN:
|
||
pdf.set_font("Helvetica", "", 9)
|
||
pdf.cell(0, 5, "Bitte überweisen Sie den Betrag innerhalb von 14 Tagen auf:", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.cell(0, 5, f"IBAN: {BANK_IBAN}", new_x="LMARGIN", new_y="NEXT")
|
||
if BANK_BIC:
|
||
pdf.cell(0, 5, f"BIC: {BANK_BIC}", new_x="LMARGIN", new_y="NEXT")
|
||
if BANK_BANK:
|
||
pdf.cell(0, 5, f"Bank: {BANK_BANK}", new_x="LMARGIN", new_y="NEXT")
|
||
pdf.cell(0, 5, f"Verwendungszweck: {invoice['invoice_number']}", new_x="LMARGIN", new_y="NEXT")
|
||
|
||
if invoice["notes"]:
|
||
pdf.ln(4)
|
||
pdf.set_font("Helvetica", "I", 9)
|
||
pdf.multi_cell(0, 5, invoice["notes"])
|
||
|
||
return bytes(pdf.output())
|
||
|
||
|
||
async def _save_to_paperless(pdf_bytes: bytes, invoice_number: str, filename: str):
|
||
scaninput = os.getenv("SCANINPUT_DIR", "/scaninput")
|
||
os.makedirs(scaninput, exist_ok=True)
|
||
path = os.path.join(scaninput, filename)
|
||
with open(path, "wb") as f:
|
||
f.write(pdf_bytes)
|
||
|
||
paperless_url = os.getenv("PAPERLESS_URL", "")
|
||
paperless_token = os.getenv("PAPERLESS_TOKEN", "")
|
||
if paperless_url and paperless_token:
|
||
try:
|
||
import httpx
|
||
async with httpx.AsyncClient(timeout=30) as client:
|
||
await client.post(
|
||
f"{paperless_url}/api/documents/post_document/",
|
||
headers={"Authorization": f"Token {paperless_token}"},
|
||
files={"document": (filename, pdf_bytes, "application/pdf")},
|
||
data={"title": invoice_number, "tags": "banyaro,Rechnung"},
|
||
)
|
||
except Exception as e:
|
||
logger.warning(f"Paperless upload failed: {e}")
|
||
|
||
|
||
def _row_to_dict(row) -> dict:
|
||
return dict(row)
|
||
|
||
|
||
def _fetch_items(conn, invoice_id: int) -> list:
|
||
rows = conn.execute(
|
||
"SELECT * FROM invoice_items WHERE invoice_id=? ORDER BY id",
|
||
(invoice_id,)
|
||
).fetchall()
|
||
return [dict(r) for r in rows]
|
||
|
||
|
||
# ------------------------------------------------------------------
|
||
# Endpoints
|
||
# ------------------------------------------------------------------
|
||
@router.get("")
|
||
def list_invoices(status: Optional[str] = None, admin=Depends(require_admin)):
|
||
with db() as conn:
|
||
if status:
|
||
rows = conn.execute(
|
||
"SELECT * FROM invoices WHERE status=? ORDER BY id DESC",
|
||
(status,)
|
||
).fetchall()
|
||
else:
|
||
rows = conn.execute(
|
||
"SELECT * FROM invoices ORDER BY id DESC"
|
||
).fetchall()
|
||
return [_row_to_dict(r) for r in rows]
|
||
|
||
|
||
@router.get("/cashflow")
|
||
def get_cashflow(admin=Depends(require_admin)):
|
||
with db() as conn:
|
||
monthly = conn.execute("""
|
||
SELECT substr(created_at, 1, 7) AS month,
|
||
SUM(amount_gross) AS revenue,
|
||
COUNT(*) AS count
|
||
FROM invoices
|
||
WHERE status IN ('sent', 'paid')
|
||
GROUP BY month
|
||
ORDER BY month DESC
|
||
""").fetchall()
|
||
|
||
year = datetime.now().year
|
||
total_year = conn.execute(
|
||
"SELECT COALESCE(SUM(amount_gross),0) FROM invoices WHERE status IN ('sent','paid') AND created_at LIKE ?",
|
||
(f"{year}%",)
|
||
).fetchone()[0]
|
||
|
||
total_outstanding = conn.execute(
|
||
"SELECT COALESCE(SUM(amount_gross),0) FROM invoices WHERE status='sent'"
|
||
).fetchone()[0]
|
||
|
||
total_paid = conn.execute(
|
||
"SELECT COALESCE(SUM(COALESCE(paid_amount, amount_gross)),0) FROM invoices WHERE status='paid'"
|
||
).fetchone()[0]
|
||
|
||
counts_rows = conn.execute(
|
||
"SELECT status, COUNT(*) AS n FROM invoices GROUP BY status"
|
||
).fetchall()
|
||
|
||
counts = {r["status"]: r["n"] for r in counts_rows}
|
||
return {
|
||
"monthly": [_row_to_dict(r) for r in monthly],
|
||
"total_year": round(total_year, 2),
|
||
"total_outstanding": round(total_outstanding, 2),
|
||
"total_paid": round(total_paid, 2),
|
||
"counts": counts,
|
||
}
|
||
|
||
|
||
@router.get("/quarterly/{year}/{q}")
|
||
def get_quarterly(year: int, q: int, admin=Depends(require_admin)):
|
||
if q not in (1, 2, 3, 4):
|
||
raise HTTPException(400, "Quartal muss 1–4 sein.")
|
||
month_start = (q - 1) * 3 + 1
|
||
month_end = month_start + 2
|
||
from_date = f"{year}-{month_start:02d}-01"
|
||
import calendar
|
||
last_day = calendar.monthrange(year, month_end)[1]
|
||
to_date = f"{year}-{month_end:02d}-{last_day:02d}"
|
||
|
||
labels = {1: "01.01.", 2: "01.04.", 3: "01.07.", 4: "01.10."}
|
||
ends = {1: "31.03.", 2: "30.06.", 3: "30.09.", 4: "31.12."}
|
||
period = f"Q{q} {year} ({labels[q]} – {ends[q]})"
|
||
|
||
with db() as conn:
|
||
rows = conn.execute(
|
||
"SELECT * FROM invoices WHERE created_at >= ? AND created_at <= ? ORDER BY id",
|
||
(from_date, to_date + "T23:59:59Z")
|
||
).fetchall()
|
||
|
||
total_net = sum(r["amount_net"] for r in rows)
|
||
total_tax = sum(r["tax_amount"] for r in rows)
|
||
total_gross = sum(r["amount_gross"] for r in rows)
|
||
|
||
return {
|
||
"period": period,
|
||
"invoices": [_row_to_dict(r) for r in rows],
|
||
"total_net": round(total_net, 2),
|
||
"total_tax": round(total_tax, 2),
|
||
"total_gross": round(total_gross, 2),
|
||
"count": len(rows),
|
||
}
|
||
|
||
|
||
@router.get("/{invoice_id}")
|
||
def get_invoice(invoice_id: int, admin=Depends(require_admin)):
|
||
with db() as conn:
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Rechnung nicht gefunden.")
|
||
items = _fetch_items(conn, invoice_id)
|
||
result = _row_to_dict(row)
|
||
result["items"] = items
|
||
return result
|
||
|
||
|
||
@router.post("", status_code=201)
|
||
def create_invoice(data: InvoiceCreate, admin=Depends(require_admin)):
|
||
if not data.items:
|
||
raise HTTPException(400, "Mindestens eine Position erforderlich.")
|
||
|
||
KLEINUNTERNEHMER = os.getenv("KLEINUNTERNEHMER", "true").lower() == "true"
|
||
TAX_RATE = 0.0 if KLEINUNTERNEHMER else float(os.getenv("RECHNUNG_MWST", "19"))
|
||
|
||
amount_net = round(sum(i.quantity * i.unit_price for i in data.items), 2)
|
||
discount_pct = data.discount_pct or 0.0
|
||
discount_amount = round(amount_net * discount_pct / 100, 2)
|
||
amount_after_discount = round(amount_net - discount_amount, 2)
|
||
tax_amount = round(amount_after_discount * TAX_RATE / 100, 2)
|
||
amount_gross = round(amount_after_discount + tax_amount, 2)
|
||
|
||
description = data.items[0].description if len(data.items) == 1 else f"{len(data.items)} Positionen"
|
||
|
||
with db() as conn:
|
||
invoice_number = _next_invoice_number(conn)
|
||
conn.execute("""
|
||
INSERT INTO invoices
|
||
(invoice_number, user_id, recipient_name, recipient_email, recipient_address,
|
||
description, service_period, amount_net, discount_pct, discount_amount,
|
||
amount_after_discount, tax_rate, tax_amount, amount_gross, notes)
|
||
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
|
||
""", (
|
||
invoice_number, data.user_id, data.recipient_name, data.recipient_email,
|
||
data.recipient_address, description, data.service_period,
|
||
amount_net, discount_pct, discount_amount,
|
||
amount_after_discount, TAX_RATE, tax_amount, amount_gross,
|
||
data.notes,
|
||
))
|
||
invoice_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||
|
||
for item in data.items:
|
||
total = round(item.quantity * item.unit_price, 2)
|
||
conn.execute(
|
||
"INSERT INTO invoice_items (invoice_id, description, quantity, unit_price, total) VALUES (?,?,?,?,?)",
|
||
(invoice_id, item.description, item.quantity, item.unit_price, total)
|
||
)
|
||
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
items = _fetch_items(conn, invoice_id)
|
||
|
||
result = _row_to_dict(row)
|
||
result["items"] = items
|
||
return result
|
||
|
||
|
||
@router.post("/{invoice_id}/send")
|
||
async def send_invoice(invoice_id: int, admin=Depends(require_admin)):
|
||
with db() as conn:
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Rechnung nicht gefunden.")
|
||
if row["status"] == "cancelled":
|
||
raise HTTPException(400, "Stornierte Rechnung kann nicht gesendet werden.")
|
||
items = _fetch_items(conn, invoice_id)
|
||
|
||
invoice = _row_to_dict(row)
|
||
|
||
try:
|
||
pdf_bytes = _generate_pdf(invoice, items)
|
||
except Exception as e:
|
||
logger.error(f"PDF-Generierung fehlgeschlagen: {e}")
|
||
raise HTTPException(500, f"PDF-Generierung fehlgeschlagen: {e}")
|
||
|
||
filename = f"{invoice['invoice_number']}_banyaro.pdf"
|
||
|
||
try:
|
||
await _save_to_paperless(pdf_bytes, invoice["invoice_number"], filename)
|
||
except Exception as e:
|
||
logger.warning(f"Paperless-Speicherung fehlgeschlagen: {e}")
|
||
|
||
import base64
|
||
body_html = f"""
|
||
<p style="margin:0 0 16px">Hallo <b>{invoice['recipient_name']}</b>,</p>
|
||
<p style="margin:0 0 16px">
|
||
anbei erhalten Sie Ihre Rechnung <b>{invoice['invoice_number']}</b>
|
||
über <b>{invoice['amount_gross']:.2f} EUR</b>.
|
||
</p>
|
||
<p style="margin:0 0 8px">Bitte überweisen Sie den Betrag innerhalb von 14 Tagen.</p>
|
||
<p style="margin:0;font-size:13px;color:#888">Verwendungszweck: {invoice['invoice_number']}</p>
|
||
"""
|
||
html = mailer.email_html(body_html)
|
||
plain = (
|
||
f"Hallo {invoice['recipient_name']},\n\n"
|
||
f"anbei Ihre Rechnung {invoice['invoice_number']} über {invoice['amount_gross']:.2f} EUR.\n\n"
|
||
f"Bitte überweisen Sie den Betrag innerhalb von 14 Tagen.\n"
|
||
f"Verwendungszweck: {invoice['invoice_number']}\n"
|
||
)
|
||
|
||
attachments = [{
|
||
"filename": filename,
|
||
"content": pdf_bytes,
|
||
"content_type": "application/pdf",
|
||
}]
|
||
|
||
try:
|
||
await mailer.send_email(
|
||
to=invoice["recipient_email"],
|
||
subject=f"Ihre Rechnung {invoice['invoice_number']} von Ban Yaro",
|
||
html=html,
|
||
plain=plain,
|
||
attachments=attachments,
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"Mail-Versand fehlgeschlagen: {e}")
|
||
raise HTTPException(500, f"Mail-Versand fehlgeschlagen: {e}")
|
||
|
||
now = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
||
with db() as conn:
|
||
conn.execute(
|
||
"UPDATE invoices SET status='sent', sent_at=? WHERE id=?",
|
||
(now, invoice_id)
|
||
)
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
|
||
return _row_to_dict(row)
|
||
|
||
|
||
@router.get("/{invoice_id}/pdf")
|
||
def download_pdf(invoice_id: int, admin=Depends(require_admin)):
|
||
with db() as conn:
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Rechnung nicht gefunden.")
|
||
items = _fetch_items(conn, invoice_id)
|
||
|
||
invoice = _row_to_dict(row)
|
||
pdf_bytes = _generate_pdf(invoice, items)
|
||
filename = f"{invoice['invoice_number']}_banyaro.pdf"
|
||
return Response(
|
||
content=pdf_bytes,
|
||
media_type="application/pdf",
|
||
headers={"Content-Disposition": f'attachment; filename="{filename}"'},
|
||
)
|
||
|
||
|
||
@router.post("/{invoice_id}/pay")
|
||
def pay_invoice(invoice_id: int, data: PayBody, admin=Depends(require_admin)):
|
||
with db() as conn:
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Rechnung nicht gefunden.")
|
||
if row["status"] == "cancelled":
|
||
raise HTTPException(400, "Stornierte Rechnung kann nicht als bezahlt markiert werden.")
|
||
conn.execute(
|
||
"UPDATE invoices SET status='paid', paid_at=?, paid_amount=? WHERE id=?",
|
||
(data.paid_at, data.paid_amount, invoice_id)
|
||
)
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
return _row_to_dict(row)
|
||
|
||
|
||
@router.post("/{invoice_id}/cancel")
|
||
def cancel_invoice(invoice_id: int, data: CancelBody, admin=Depends(require_admin)):
|
||
with db() as conn:
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "Rechnung nicht gefunden.")
|
||
if row["status"] == "cancelled":
|
||
raise HTTPException(400, "Rechnung ist bereits storniert.")
|
||
cancellation_number = _next_invoice_number(conn, "ST")
|
||
now = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.000Z")
|
||
conn.execute(
|
||
"UPDATE invoices SET status='cancelled', cancelled_at=?, cancellation_reason=?, cancellation_number=? WHERE id=?",
|
||
(now, data.reason, cancellation_number, invoice_id)
|
||
)
|
||
row = conn.execute("SELECT * FROM invoices WHERE id=?", (invoice_id,)).fetchone()
|
||
return _row_to_dict(row)
|