Feature: QR-Kontingente für Partner — Bestellung, Übergabe, Rückverfolgung
Partner verteilen gedruckte QR-Codes (Sticker/Flyer); jeder physische Code
ist einzeln rückverfolgbar von Scan bis Registrierung.
Backend:
- partner_qr_batches + partner_qr_codes (Token 8-stellig, ohne 0/O/1/l/I),
users.referred_qr, partner_codes.owner_user_id (+Backfill über referred_by)
- /q/{token}: Scan zählen (scans, first/last_scan_at) → Redirect
/?ref=CODE&qr=TOKEN — dockt am bestehenden Referral-Flow an
- Registrierung: qr_token wird nur zugeordnet, wenn er zum eingelösten
Partner-Code gehört (Manipulationsschutz)
- Admin: Kontingent bestellen (max 500), Liste mit Scans/Registrierungen,
Löschen (Zweiklick), druckfertiges A4-PDF (segno+fpdf2, 3×4 Grid mit
Kurz-URL + laufender Nummer), Code-Besitzer zuordnen
- Partner-Self-Service: /partner/my-qr (+PDF) für Code-Besitzer
Frontend:
- Admin-Partner-Tab: Karte 'QR-Kontingente' (Bestellung, Stats, PDF, Besitzer)
- Partner-Profil: 'Meine QR-Codes' mit Scans/Registrierungen + PDF-Download
- boot.js/app.js speichern ?qr=, Registrierung schickt qr_token mit
Neu: segno==1.6.6 (pure-python QR). Tests: 5 neue (PDF, Scan-Zählung,
Attribution, Fremd-Token-Schutz, Self-Service). Suite: 51 passed.
This commit is contained in:
parent
cadfb24a8d
commit
f604ab7c4f
16 changed files with 621 additions and 23 deletions
|
|
@ -153,6 +153,7 @@ class RegisterRequest(BaseModel):
|
|||
password: str = Field(..., min_length=8, max_length=200)
|
||||
name: str = Field(..., min_length=2, max_length=40)
|
||||
ref_code: Optional[str] = Field(None, max_length=50)
|
||||
qr_token: Optional[str] = Field(None, max_length=20) # physischer Partner-QR (Sticker/Flyer)
|
||||
|
||||
|
||||
def _gen_referral_code() -> str:
|
||||
|
|
@ -227,6 +228,16 @@ async def register(data: RegisterRequest, response: Response, request: Request):
|
|||
|
||||
if redeemed:
|
||||
updates = {"referred_by": -partner["id"]}
|
||||
# QR-Rückverfolgung: Token muss zu einem Kontingent DIESES Codes gehören
|
||||
if data.qr_token:
|
||||
qr = conn.execute(
|
||||
"""SELECT q.token FROM partner_qr_codes q
|
||||
JOIN partner_qr_batches b ON b.id = q.batch_id
|
||||
WHERE q.token=? AND b.partner_code_id=?""",
|
||||
(data.qr_token.strip(), partner["id"])
|
||||
).fetchone()
|
||||
if qr:
|
||||
updates["referred_qr"] = qr["token"]
|
||||
if partner["grants_founder"]:
|
||||
total_founders = conn.execute(
|
||||
"SELECT COUNT(*) FROM users WHERE is_founder=1"
|
||||
|
|
|
|||
|
|
@ -37,10 +37,12 @@ def list_partner_codes(user=Depends(require_admin)):
|
|||
with db() as conn:
|
||||
rows = conn.execute(
|
||||
"""SELECT pc.id, pc.code, pc.label, pc.grants_founder,
|
||||
pc.max_uses, pc.uses, pc.created_at,
|
||||
u.name AS created_by_name
|
||||
pc.max_uses, pc.uses, pc.created_at, pc.owner_user_id,
|
||||
u.name AS created_by_name,
|
||||
o.name AS owner_name
|
||||
FROM partner_codes pc
|
||||
LEFT JOIN users u ON u.id = pc.created_by
|
||||
LEFT JOIN users o ON o.id = pc.owner_user_id
|
||||
ORDER BY pc.created_at DESC"""
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
|
@ -546,3 +548,220 @@ def review_partner_profile(user_id: int, data: PartnerProfileReview, user=Depend
|
|||
)
|
||||
profile = _pp_get_or_empty(conn, user_id)
|
||||
return {"profile": profile}
|
||||
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# QR-Kontingente — gedruckte Sticker/Flyer mit Scan- und
|
||||
# Registrierungs-Rückverfolgung pro Einzelcode und Kontingent.
|
||||
# Bestellung: Admin legt Kontingent für einen Partner-Code an.
|
||||
# Übergabe: PDF-Download (Admin + Partner im eigenen Profil).
|
||||
# ------------------------------------------------------------------
|
||||
|
||||
_QR_MAX_QUANTITY = 500
|
||||
# Ohne verwechselbare Zeichen (0/O, 1/l/I) — Tokens landen gedruckt auf Stickern
|
||||
_QR_ALPHABET = "abcdefghjkmnpqrstuvwxyzABCDEFGHJKMNPQRSTUVWXYZ23456789"
|
||||
_QR_BASE_URL = os.getenv("APP_URL", "https://banyaro.app")
|
||||
|
||||
|
||||
def _qr_new_token(conn) -> str:
|
||||
import secrets
|
||||
for _ in range(20):
|
||||
token = "".join(secrets.choice(_QR_ALPHABET) for _ in range(8))
|
||||
if not conn.execute(
|
||||
"SELECT 1 FROM partner_qr_codes WHERE token=?", (token,)
|
||||
).fetchone():
|
||||
return token
|
||||
raise HTTPException(500, "Token-Generierung fehlgeschlagen.")
|
||||
|
||||
|
||||
def _qr_batch_stats(conn, batch_id: int) -> dict:
|
||||
row = conn.execute(
|
||||
"""SELECT COUNT(*) AS codes, COALESCE(SUM(q.scans),0) AS scans,
|
||||
(SELECT COUNT(*) FROM users u
|
||||
JOIN partner_qr_codes q2 ON q2.token = u.referred_qr
|
||||
WHERE q2.batch_id = ?) AS registrations
|
||||
FROM partner_qr_codes q WHERE q.batch_id = ?""",
|
||||
(batch_id, batch_id)
|
||||
).fetchone()
|
||||
return dict(row)
|
||||
|
||||
|
||||
def _qr_list_batches(conn, where_sql: str, params: tuple) -> list:
|
||||
rows = conn.execute(
|
||||
f"""SELECT b.id, b.label, b.quantity, b.created_at,
|
||||
pc.code, pc.label AS code_label, pc.id AS partner_code_id
|
||||
FROM partner_qr_batches b
|
||||
JOIN partner_codes pc ON pc.id = b.partner_code_id
|
||||
{where_sql}
|
||||
ORDER BY b.created_at DESC""",
|
||||
params
|
||||
).fetchall()
|
||||
result = []
|
||||
for r in rows:
|
||||
d = dict(r)
|
||||
d.update(_qr_batch_stats(conn, r["id"]))
|
||||
result.append(d)
|
||||
return result
|
||||
|
||||
|
||||
def _qr_batch_pdf(conn, batch_id: int) -> bytes:
|
||||
"""Druckfertiges A4-PDF: 3×4 QR-Codes pro Seite mit Kurz-URL + laufender Nummer."""
|
||||
import io as _io
|
||||
import segno
|
||||
from fpdf import FPDF
|
||||
|
||||
batch = conn.execute(
|
||||
"""SELECT b.id, b.label, b.quantity, pc.code, pc.label AS code_label
|
||||
FROM partner_qr_batches b
|
||||
JOIN partner_codes pc ON pc.id = b.partner_code_id
|
||||
WHERE b.id=?""",
|
||||
(batch_id,)
|
||||
).fetchone()
|
||||
if not batch:
|
||||
raise HTTPException(404, "Kontingent nicht gefunden.")
|
||||
codes = conn.execute(
|
||||
"SELECT token, seq FROM partner_qr_codes WHERE batch_id=? ORDER BY seq",
|
||||
(batch_id,)
|
||||
).fetchall()
|
||||
|
||||
pdf = FPDF(format="A4")
|
||||
pdf.set_auto_page_break(False)
|
||||
pdf.set_title(f"Ban Yaro QR-Kontingent — {batch['label']}")
|
||||
|
||||
COLS, ROWS = 3, 4
|
||||
CELL_W, CELL_H = 60, 64 # mm — Zelle inkl. Beschriftung
|
||||
MARGIN_X = (210 - COLS * CELL_W) / 2
|
||||
MARGIN_Y = 18
|
||||
QR_SIZE = 42 # mm
|
||||
|
||||
def _latin1(s: str) -> str:
|
||||
return s.encode("latin-1", "replace").decode("latin-1")
|
||||
|
||||
for i, c in enumerate(codes):
|
||||
pos = i % (COLS * ROWS)
|
||||
if pos == 0:
|
||||
pdf.add_page()
|
||||
pdf.set_font("Helvetica", "B", 11)
|
||||
pdf.set_text_color(60)
|
||||
pdf.cell(0, 6, _latin1(f"Ban Yaro — {batch['code_label']} · Kontingent: {batch['label']} ({batch['quantity']} Stk.)"),
|
||||
align="C", new_x="LMARGIN", new_y="NEXT")
|
||||
col, row_ = pos % COLS, pos // COLS
|
||||
x = MARGIN_X + col * CELL_W
|
||||
y = MARGIN_Y + 10 + row_ * CELL_H
|
||||
|
||||
url = f"{_QR_BASE_URL}/q/{c['token']}"
|
||||
buf = _io.BytesIO()
|
||||
segno.make(url, error="m").save(buf, kind="png", scale=8, border=2)
|
||||
buf.seek(0)
|
||||
pdf.image(buf, x=x + (CELL_W - QR_SIZE) / 2, y=y, w=QR_SIZE, h=QR_SIZE)
|
||||
|
||||
pdf.set_xy(x, y + QR_SIZE + 1)
|
||||
pdf.set_font("Helvetica", "", 8)
|
||||
pdf.set_text_color(90)
|
||||
pdf.cell(CELL_W, 4, _latin1(f"banyaro.app/q/{c['token']}"), align="C", new_x="LEFT", new_y="NEXT")
|
||||
pdf.set_x(x)
|
||||
pdf.set_font("Helvetica", "B", 8)
|
||||
pdf.cell(CELL_W, 4, f"#{c['seq']}", align="C")
|
||||
|
||||
return bytes(pdf.output())
|
||||
|
||||
|
||||
class QrBatchCreate(BaseModel):
|
||||
label: str = Field(..., min_length=1, max_length=100)
|
||||
quantity: int = Field(..., ge=1, le=_QR_MAX_QUANTITY)
|
||||
|
||||
|
||||
@router.post("/admin/partner/codes/{code_id}/qr-batches", status_code=201)
|
||||
def create_qr_batch(code_id: int, data: QrBatchCreate, user=Depends(require_admin)):
|
||||
"""Bestellung: neues QR-Kontingent für einen Partner-Code anlegen."""
|
||||
with db() as conn:
|
||||
code = conn.execute(
|
||||
"SELECT id FROM partner_codes WHERE id=?", (code_id,)
|
||||
).fetchone()
|
||||
if not code:
|
||||
raise HTTPException(404, "Partner-Code nicht gefunden.")
|
||||
conn.execute(
|
||||
"INSERT INTO partner_qr_batches (partner_code_id, label, quantity, created_by) VALUES (?,?,?,?)",
|
||||
(code_id, data.label.strip(), data.quantity, user["id"])
|
||||
)
|
||||
batch_id = conn.execute("SELECT last_insert_rowid()").fetchone()[0]
|
||||
for seq in range(1, data.quantity + 1):
|
||||
conn.execute(
|
||||
"INSERT INTO partner_qr_codes (batch_id, token, seq) VALUES (?,?,?)",
|
||||
(batch_id, _qr_new_token(conn), seq)
|
||||
)
|
||||
batches = _qr_list_batches(conn, "WHERE b.id=?", (batch_id,))
|
||||
return batches[0]
|
||||
|
||||
|
||||
@router.get("/admin/partner/qr-batches")
|
||||
def list_qr_batches(user=Depends(require_admin)):
|
||||
"""Alle QR-Kontingente mit Stats (Scans, Registrierungen)."""
|
||||
with db() as conn:
|
||||
return _qr_list_batches(conn, "", ())
|
||||
|
||||
|
||||
@router.delete("/admin/partner/qr-batches/{batch_id}", status_code=204)
|
||||
def delete_qr_batch(batch_id: int, user=Depends(require_admin)):
|
||||
"""Kontingent löschen (z. B. Fehlbestellung) — Codes via CASCADE mit weg."""
|
||||
with db() as conn:
|
||||
if not conn.execute(
|
||||
"SELECT id FROM partner_qr_batches WHERE id=?", (batch_id,)
|
||||
).fetchone():
|
||||
raise HTTPException(404, "Kontingent nicht gefunden.")
|
||||
conn.execute("DELETE FROM partner_qr_batches WHERE id=?", (batch_id,))
|
||||
return None
|
||||
|
||||
|
||||
@router.get("/admin/partner/qr-batches/{batch_id}/pdf")
|
||||
def qr_batch_pdf_admin(batch_id: int, user=Depends(require_admin)):
|
||||
from fastapi.responses import Response
|
||||
with db() as conn:
|
||||
pdf = _qr_batch_pdf(conn, batch_id)
|
||||
return Response(content=pdf, media_type="application/pdf",
|
||||
headers={"Content-Disposition": f'attachment; filename="banyaro-qr-{batch_id}.pdf"'})
|
||||
|
||||
|
||||
@router.get("/partner/my-qr")
|
||||
def my_qr_batches(user=Depends(require_partner)):
|
||||
"""Übergabe/Self-Service: eigene Kontingente mit Stats (Code-Besitzer)."""
|
||||
with db() as conn:
|
||||
return _qr_list_batches(
|
||||
conn, "WHERE pc.owner_user_id = ?", (user["id"],)
|
||||
)
|
||||
|
||||
|
||||
@router.get("/partner/my-qr/{batch_id}/pdf")
|
||||
def qr_batch_pdf_partner(batch_id: int, user=Depends(require_partner)):
|
||||
from fastapi.responses import Response
|
||||
with db() as conn:
|
||||
own = conn.execute(
|
||||
"""SELECT b.id FROM partner_qr_batches b
|
||||
JOIN partner_codes pc ON pc.id = b.partner_code_id
|
||||
WHERE b.id=? AND pc.owner_user_id=?""",
|
||||
(batch_id, user["id"])
|
||||
).fetchone()
|
||||
if not own and user.get("rolle") != "admin":
|
||||
raise HTTPException(403, "Kein Zugriff auf dieses Kontingent.")
|
||||
pdf = _qr_batch_pdf(conn, batch_id)
|
||||
return Response(content=pdf, media_type="application/pdf",
|
||||
headers={"Content-Disposition": f'attachment; filename="banyaro-qr-{batch_id}.pdf"'})
|
||||
|
||||
|
||||
class CodeOwnerSet(BaseModel):
|
||||
user_id: int
|
||||
|
||||
|
||||
@router.post("/admin/partner/codes/{code_id}/owner")
|
||||
def set_code_owner(code_id: int, data: CodeOwnerSet, user=Depends(require_admin)):
|
||||
"""Partner-Code einem User zuordnen (für Self-Service-QR-Zugriff)."""
|
||||
with db() as conn:
|
||||
if not conn.execute("SELECT id FROM partner_codes WHERE id=?", (code_id,)).fetchone():
|
||||
raise HTTPException(404, "Partner-Code nicht gefunden.")
|
||||
if not conn.execute("SELECT id FROM users WHERE id=?", (data.user_id,)).fetchone():
|
||||
raise HTTPException(404, "User nicht gefunden.")
|
||||
conn.execute(
|
||||
"UPDATE partner_codes SET owner_user_id=? WHERE id=?",
|
||||
(data.user_id, code_id)
|
||||
)
|
||||
return {"ok": True}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue