"""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"""

Hallo {invoice['recipient_name']},

anbei erhalten Sie Ihre Rechnung {invoice['invoice_number']} über {invoice['amount_gross']:.2f} EUR.

Bitte überweisen Sie den Betrag innerhalb von 14 Tagen.

Verwendungszweck: {invoice['invoice_number']}

""" 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)