Feature: Rechnungs-System (invoices) — Backend komplett
- 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
This commit is contained in:
parent
9c359bb07e
commit
b68a12587a
5 changed files with 570 additions and 10 deletions
489
backend/routes/invoices.py
Normal file
489
backend/routes/invoices.py
Normal file
|
|
@ -0,0 +1,489 @@
|
|||
"""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)
|
||||
Loading…
Add table
Add a link
Reference in a new issue