add clean badge generation for each user from database
This commit is contained in:
@@ -0,0 +1,68 @@
|
||||
import logging
|
||||
import os
|
||||
import re
|
||||
import base64
|
||||
import subprocess
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
|
||||
logging.basicConfig(
|
||||
level=logging.INFO,
|
||||
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s"
|
||||
)
|
||||
|
||||
logger = logging.getLogger(__name__.split(".")[-1])
|
||||
|
||||
|
||||
class BadgeGeneratorService:
|
||||
|
||||
def __init__(self, svg_template: Path, chrome_path: Path):
|
||||
self.svg_template = svg_template
|
||||
self.chrome_path = chrome_path
|
||||
|
||||
def generate_badge(self, name, seat_id, picture) -> None:
|
||||
svg = self.generate_svg(name, seat_id, picture)
|
||||
self.svg_to_pdf(svg, f"files/{name}.pdf")
|
||||
|
||||
def generate_svg(self, username: str, seat: str, image_base64: bytes) -> str:
|
||||
with open(self.svg_template, "r", encoding="utf-8") as f:
|
||||
svg = f.read()
|
||||
# ToDo: Select font size for long names
|
||||
svg = svg.replace(">test_username</tspan>", f">{username}</tspan>")
|
||||
svg = svg.replace(">Platz: XYZ</tspan><", f">Platz: {seat}</tspan><")
|
||||
if image_base64:
|
||||
encoded = base64.b64encode(image_base64).decode("utf-8")
|
||||
svg = svg.replace('xlink:href="data:image/jpeg;base64,"', f'xlink:href="data:image/jpeg;base64,{encoded}"')
|
||||
logger.info(f"Generate svg string: Username: {username}, Seat: {seat}, Picture: {str(image_base64)[:15]}")
|
||||
return svg
|
||||
|
||||
def svg_to_pdf(self, svg_string: str, pdf_path: str):
|
||||
|
||||
width = re.search(r'width="([^"]+)"', svg_string).group(1)
|
||||
height = re.search(r'height="([^"]+)"', svg_string).group(1)
|
||||
|
||||
env = Environment(loader=FileSystemLoader("template"))
|
||||
template = env.get_template("badge_pdf.html")
|
||||
|
||||
html = template.render(
|
||||
width=width,
|
||||
height=height,
|
||||
svg_string=svg_string
|
||||
)
|
||||
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".html", mode="w", encoding="utf-8") as f:
|
||||
f.write(html)
|
||||
html_path = f.name
|
||||
|
||||
try:
|
||||
subprocess.run([
|
||||
self.chrome_path,
|
||||
"--headless",
|
||||
"--disable-gpu",
|
||||
f"--print-to-pdf={os.path.abspath(pdf_path)}",
|
||||
html_path
|
||||
], check=True)
|
||||
finally:
|
||||
os.remove(html_path)
|
||||
logger.info(f"Created: {pdf_path}")
|
||||
@@ -0,0 +1,38 @@
|
||||
import tomllib
|
||||
import aiomysql
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class DatabaseService:
|
||||
|
||||
def __init__(self, sql_config: str):
|
||||
with open(sql_config, "rb") as f:
|
||||
config = tomllib.load(f)
|
||||
|
||||
self.db_config = config["database"]
|
||||
self._pool: Optional[aiomysql.Pool] = None
|
||||
|
||||
async def init_db_pool(self):
|
||||
self._pool = await aiomysql.create_pool(
|
||||
host=self.db_config["db_host"],
|
||||
port=self.db_config["db_port"],
|
||||
user=self.db_config["db_user"],
|
||||
password=self.db_config["db_password"],
|
||||
db=self.db_config["db_name"],
|
||||
minsize=1,
|
||||
maxsize=40,
|
||||
autocommit=True
|
||||
)
|
||||
|
||||
async def get_user_badges(self) -> list:
|
||||
async with self._pool.acquire() as conn:
|
||||
async with conn.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(
|
||||
"""SELECT u.user_id, u.user_name, s.seat_id, upp.picture
|
||||
FROM users AS u
|
||||
LEFT JOIN seats AS s
|
||||
ON u.user_id = s.`user`
|
||||
LEFT JOIN user_profile_picture AS upp
|
||||
ON u.user_id = upp.user_id;"""
|
||||
)
|
||||
return await cursor.fetchall()
|
||||
Reference in New Issue
Block a user