Compare commits
10 commits
c73aac7a50
...
ca79aed027
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
ca79aed027 | ||
|
|
b5bb18add6 | ||
|
|
80c8201a2d | ||
|
|
375fb3555f | ||
|
|
851012c2b6 | ||
|
|
cdffc5fe68 | ||
|
|
58ec27abd7 | ||
|
|
0172b7bf4a | ||
|
|
e97c8fcfec | ||
|
|
0dc32f063d |
26 changed files with 9169 additions and 16 deletions
6
.dockerignore
Normal file
6
.dockerignore
Normal file
|
|
@ -0,0 +1,6 @@
|
|||
.venv
|
||||
.git
|
||||
__pycache__
|
||||
*.pyc
|
||||
.ruff_cache
|
||||
.mypy_cache
|
||||
47
.github/workflows/build_docker.yml
vendored
Normal file
47
.github/workflows/build_docker.yml
vendored
Normal file
|
|
@ -0,0 +1,47 @@
|
|||
# This workflow uses actions that are not certified by GitHub.
|
||||
# They are provided by a third-party and are governed by
|
||||
# separate terms of service, privacy policy, and support
|
||||
# documentation.
|
||||
|
||||
# GitHub recommends pinning actions to a commit SHA.
|
||||
# To get a newer version, you will need to update the SHA.
|
||||
# You can also reference a tag or branch, but the action may change without warning.
|
||||
|
||||
name: Publish Docker image
|
||||
|
||||
on:
|
||||
push:
|
||||
branches:
|
||||
- main
|
||||
|
||||
jobs:
|
||||
push_to_registries:
|
||||
name: Push Docker image to multiple registries
|
||||
runs-on: ubuntu-latest
|
||||
permissions:
|
||||
packages: write
|
||||
contents: read
|
||||
steps:
|
||||
- name: Check out the repo
|
||||
uses: actions/checkout@v3
|
||||
|
||||
- name: Log in to the Container registry
|
||||
uses: docker/login-action@65b78e6e13532edd9afa3aa52ac7964289d1a9c1
|
||||
with:
|
||||
registry: ghcr.io
|
||||
username: ${{ github.actor }}
|
||||
password: ${{ secrets.GITHUB_TOKEN }}
|
||||
|
||||
- name: Extract metadata (tags, labels) for Docker
|
||||
id: meta
|
||||
uses: docker/metadata-action@9ec57ed1fcdbf14dcef7dfbe97b2010124a938b7
|
||||
with:
|
||||
images: ghcr.io/${{ github.repository }}
|
||||
|
||||
- name: Build and push Docker images
|
||||
uses: docker/build-push-action@3b5e8027fcad23fda98b2e3ac259d8d67585f671
|
||||
with:
|
||||
context: .
|
||||
push: true
|
||||
tags: ${{ steps.meta.outputs.tags }}
|
||||
labels: ${{ steps.meta.outputs.labels }}
|
||||
20
Dockerfile
Normal file
20
Dockerfile
Normal file
|
|
@ -0,0 +1,20 @@
|
|||
# Build stage
|
||||
FROM python:3.11 AS builder
|
||||
|
||||
COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /bin/
|
||||
|
||||
WORKDIR /app
|
||||
COPY pyproject.toml uv.lock ./
|
||||
RUN uv sync --frozen --no-install-project
|
||||
COPY . .
|
||||
RUN uv sync
|
||||
|
||||
# Runtime stage
|
||||
FROM python:3.11-slim
|
||||
|
||||
WORKDIR /app
|
||||
COPY --from=builder /app/ /app/
|
||||
|
||||
ENV PATH="/app/.venv/bin:$PATH"
|
||||
|
||||
CMD ["fastapi", "dev", "--host", "0.0.0.0", "--port", "31337", "labeler/adapter/fastapi_srv.py" ]
|
||||
12
docker-compose.yml
Normal file
12
docker-compose.yml
Normal file
|
|
@ -0,0 +1,12 @@
|
|||
version: "3.8"
|
||||
services:
|
||||
bot:
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile
|
||||
environment:
|
||||
- PRINTER_IP=192.168.1.93
|
||||
- TELEGRAM_TOKEN=6386069775:AAFOQL7lIsDe_5njXptfOwmHWEVo_yyVAi4
|
||||
command:
|
||||
- python
|
||||
- labeler/adapter/telegram_bot.py
|
||||
111
fegen/discourse.py
Normal file
111
fegen/discourse.py
Normal file
|
|
@ -0,0 +1,111 @@
|
|||
'''
|
||||
Class to generate a csv file based on data fetched via Discourse REST API
|
||||
'''
|
||||
import os
|
||||
import csv
|
||||
import json
|
||||
import requests
|
||||
from dotenv import load_dotenv
|
||||
|
||||
DISCOURSE_URL = "https://kb.hs3.pl" # Database is hosted here
|
||||
CATEGORY_ID = 9 # Database category ID
|
||||
|
||||
class DiscourseDatabase():
|
||||
def __init__(self):
|
||||
data = self.get_category_data()
|
||||
self.category_topics_csv(data)
|
||||
load_dotenv()
|
||||
|
||||
def get_headers(self, auth=False):
|
||||
"""Get request headers, optionally with auth data."""
|
||||
headers = {
|
||||
"content-type": "application/json",
|
||||
}
|
||||
if auth:
|
||||
headers["Api-Key"] = os.getenv("DISCOURSE_PAT")
|
||||
headers["Api-Username"] = os.getenv("DISCOURSE_USERNAME")
|
||||
return headers
|
||||
|
||||
def get_category_data(self) -> dict:
|
||||
"""Get all topics from a Discourse category with pagination"""
|
||||
url = f"{DISCOURSE_URL}/c/{CATEGORY_ID}.json"
|
||||
print(f"Fetching data from {url}")
|
||||
all_topics = []
|
||||
page = 0
|
||||
while True:
|
||||
params = {"per_page": 100, "page": page}
|
||||
res = requests.get(url, headers=self.get_headers(), params=params)
|
||||
res.raise_for_status()
|
||||
res_json = res.json()
|
||||
topics = res_json["topic_list"]["topics"]
|
||||
if not topics:
|
||||
break
|
||||
for topic in topics:
|
||||
if topic["category_id"] == CATEGORY_ID:
|
||||
all_topics.append(topic)
|
||||
print(f"Fetched page {page}: {len(topics)} topics, {len(all_topics)} total in category")
|
||||
page += 1
|
||||
return {"topic_list": {"topics": all_topics}}
|
||||
|
||||
def get_topic_content(self, topic_id: str):
|
||||
"""Get a single topic's content"""
|
||||
get_url = f"{DISCOURSE_URL}/posts/{topic_id}.json"
|
||||
res = requests.get(get_url, headers=self.get_headers(auth=True))
|
||||
res.raise_for_status()
|
||||
return res.json()
|
||||
|
||||
def category_topics_csv(self, category_data) -> None:
|
||||
"""Save category topics to a csv file"""
|
||||
columns = ["id", "title", "place", "tags"]
|
||||
records = category_data["topic_list"]["topics"]
|
||||
with open('zasoby.csv', 'w', encoding='UTF8') as f:
|
||||
write = csv.writer(f)
|
||||
write.writerow(columns)
|
||||
for topic in records:
|
||||
html_url = f'<a href="{DISCOURSE_URL}/t/{topic["id"]}">{topic["title"]}</a>'
|
||||
place = self.get_place(topic)
|
||||
write.writerow([topic["id"], html_url, place, topic["tags"]])
|
||||
print(f"New zasoby.csv generated with {len(records)} records")
|
||||
|
||||
def get_place(self, topic):
|
||||
"""Get place of a topic"""
|
||||
places = ["cow-work", "garage", "lab"]
|
||||
for place in places:
|
||||
if place in topic["tags"]:
|
||||
return f'<a href="https://kb.hs3.pl/tag/{place}">{place}</a>'
|
||||
return "unknown"
|
||||
|
||||
def replace_string_in_post(self, topic_id: str, old_string: str, new_string: str) -> dict:
|
||||
"""Replace a selected string within a topic's first post using Discourse REST API"""
|
||||
# Fetch the topic to get the first post ID
|
||||
topic_url = f"{DISCOURSE_URL}/t/{topic_id}.json"
|
||||
topic_res = requests.get(topic_url, headers=self.get_headers(auth=True))
|
||||
topic_res.raise_for_status()
|
||||
topic_data = topic_res.json()
|
||||
|
||||
# Get the first post ID from the topic
|
||||
first_post_id = topic_data["post_stream"]["posts"][0]["id"]
|
||||
|
||||
# Fetch the post content
|
||||
post_url = f"{DISCOURSE_URL}/posts/{first_post_id}.json"
|
||||
post_res = requests.get(post_url, headers=self.get_headers(auth=True))
|
||||
post_res.raise_for_status()
|
||||
post_data = post_res.json()
|
||||
|
||||
# Replace the string
|
||||
updated_raw = post_data["raw"].replace(old_string, new_string)
|
||||
|
||||
# Update the post
|
||||
payload = {"post": {"raw": updated_raw}}
|
||||
res = requests.put(post_url, json=payload, headers=self.get_headers(auth=True))
|
||||
res.raise_for_status()
|
||||
return res.json()
|
||||
|
||||
if __name__ == "__main__":
|
||||
disc = DiscourseDatabase()
|
||||
category = disc.get_category_data()
|
||||
records = category["topic_list"]["topics"]
|
||||
for topic in records:
|
||||
if "lab" in topic["tags"]:
|
||||
disc.replace_string_in_post(topic["id"], "[Workshop](https://kb.s.hs3.pl/tag/workshop)", "[Lab](https://kb.s.hs3.pl/tag/lab)")
|
||||
|
||||
7030
fegen/docs/index.html
Normal file
7030
fegen/docs/index.html
Normal file
File diff suppressed because one or more lines are too long
14
fegen/docs/static/css/style.css
vendored
Normal file
14
fegen/docs/static/css/style.css
vendored
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
footer {
|
||||
padding: 12px;
|
||||
margin-top: auto;
|
||||
}
|
||||
|
||||
/* || Sidenav */
|
||||
.sidenav {
|
||||
width: 250px;
|
||||
position: sticky;
|
||||
z-index: 1;
|
||||
top: 0;
|
||||
overflow-x: hidden;
|
||||
padding: 6px 8px 6px 16px;
|
||||
}
|
||||
44
fegen/main.py
Normal file
44
fegen/main.py
Normal file
|
|
@ -0,0 +1,44 @@
|
|||
import os, re, shutil
|
||||
from jinja2 import Environment, FileSystemLoader
|
||||
import pandas as pd
|
||||
|
||||
|
||||
def add_download_button(row):
|
||||
item_id = re.sub(r'<a href.*/([0-9]+)".*', "\\1", row[1])
|
||||
download_button = (
|
||||
f'<button id="btn_{item_id}"><i class="fa fa-download"></i> {item_id}</button>'
|
||||
)
|
||||
print_button = f'<a href="http://localhost:31337/print/{item_id}"><button id="prn_{item_id}"><i class="fa fa-print"></i> {item_id}</button></a>'
|
||||
return row + [download_button, print_button]
|
||||
|
||||
|
||||
def generate_dashboard():
|
||||
"""Generate dashboard from zasoby.csv file"""
|
||||
print("Generating HTML dashboard")
|
||||
website_folder = "fegen/docs"
|
||||
data = pd.read_csv("zasoby.csv")
|
||||
env = Environment(loader=FileSystemLoader("fegen/template"))
|
||||
print("Removing old website files")
|
||||
print("Creating a new website")
|
||||
shutil.copytree("fegen/template/static", f"{website_folder}/static", dirs_exist_ok=True)
|
||||
template = env.get_template("_main_layout.html")
|
||||
with open(f"{website_folder}/index.html", "w+", encoding="utf-8") as file:
|
||||
header_row = data.columns.values.tolist() + ["label", "print"]
|
||||
rows = map(
|
||||
add_download_button,
|
||||
data.values.tolist(),
|
||||
)
|
||||
html = template.render(
|
||||
title="Baza Zasobów Hackerspace Trójmiasto",
|
||||
t_header=header_row,
|
||||
t_body=rows,
|
||||
)
|
||||
file.write(html)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
from discourse import DiscourseDatabase
|
||||
|
||||
DiscourseDatabase()
|
||||
generate_dashboard()
|
||||
print("Done!")
|
||||
150
fegen/template/_base_template.html
Normal file
150
fegen/template/_base_template.html
Normal file
File diff suppressed because one or more lines are too long
29
fegen/template/_main_layout.html
Normal file
29
fegen/template/_main_layout.html
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
{% extends "_base_template.html" %} {% block body %}
|
||||
<div id="site-body" class="container-fluid">
|
||||
<div class="row flex-nonwrap">
|
||||
<div class="sidenav">{% block sidenav %}{% endblock sidenav %}</div>
|
||||
<div class="main">
|
||||
<h1>Baza Zasobów Hackerspace Trójmiasto</h1>
|
||||
<a href="/refresh"><button id="refresh"><i class="fa fa-arrows-rotate"></i> Refresh</button></a>
|
||||
<table id="dashboardTable">
|
||||
<thead>
|
||||
<tr>
|
||||
{% for cell in t_header %}
|
||||
<td>{{cell}}</td>
|
||||
{% endfor %}
|
||||
</tr>
|
||||
</thead>
|
||||
<tbody>
|
||||
{% for row in t_body %}
|
||||
<tr>
|
||||
{% for cell in row %}
|
||||
<td>{{cell}}</td>
|
||||
{% endfor %}
|
||||
</tr>
|
||||
{% endfor %}
|
||||
</tbody>
|
||||
</table>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
{% endblock body %}
|
||||
14
fegen/template/static/css/style.css
Normal file
14
fegen/template/static/css/style.css
Normal file
|
|
@ -0,0 +1,14 @@
|
|||
footer {
|
||||
padding: 12px;
|
||||
margin-top: auto;
|
||||
}
|
||||
|
||||
/* || Sidenav */
|
||||
.sidenav {
|
||||
width: 250px;
|
||||
position: sticky;
|
||||
z-index: 1;
|
||||
top: 0;
|
||||
overflow-x: hidden;
|
||||
padding: 6px 8px 6px 16px;
|
||||
}
|
||||
BIN
fonts/SourceCodePro-SemiBold.ttf
Normal file
BIN
fonts/SourceCodePro-SemiBold.ttf
Normal file
Binary file not shown.
BIN
img.png
Normal file
BIN
img.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 9.9 KiB |
BIN
img_1.png
Normal file
BIN
img_1.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 50 KiB |
75
labeler/adapter/cli.py
Normal file
75
labeler/adapter/cli.py
Normal file
|
|
@ -0,0 +1,75 @@
|
|||
import os
|
||||
|
||||
from labeler.app.labeler import Application
|
||||
from labeler.infra.e550w_printer.printer import E550W
|
||||
from labeler.infra.renderer import PILRenderer
|
||||
|
||||
|
||||
class LabelingBot:
|
||||
def __init__(self, app: Application):
|
||||
self.app = app
|
||||
|
||||
def media_info(self):
|
||||
media = self.app.get_installed_media()
|
||||
print(f"Installed medium: {media}")
|
||||
|
||||
def simple_label(self, label_text, label_length=0):
|
||||
try:
|
||||
label = self.app.print_label(text=label_text, length=label_length)
|
||||
except Exception as e:
|
||||
print(f"There was an exception: {e}")
|
||||
|
||||
def get_qrcode(self, label_text, label_length=0):
|
||||
label = self.app.render_qrcode_preview(text=label_text, length=label_length)
|
||||
return label
|
||||
|
||||
def print_qrcode(self, label_text, label_length=0):
|
||||
try:
|
||||
label = self.app.print_qrcode(text=label_text, length=label_length)
|
||||
except Exception as e:
|
||||
print(f"There was an exception: {e}")
|
||||
|
||||
# async def label_length(self):
|
||||
# await update.message.reply_text(
|
||||
# "Hello! Please tell me the length of the label, enter 0 for auto:"
|
||||
# )
|
||||
# return LABEL_LENGTH
|
||||
#
|
||||
# async def label_text(self, update: Update, context: CallbackContext) -> int:
|
||||
# user_input = update.message.text
|
||||
# context.user_data["length"] = int(user_input)
|
||||
# await update.message.reply_text("Now, please tell me the text of the label:")
|
||||
# return LABEL_TEXT
|
||||
#
|
||||
# async def simple_label(self, update: Update, context: CallbackContext) -> int:
|
||||
# user_input = update.message.text
|
||||
# context.user_data["label"] = user_input
|
||||
# try:
|
||||
# label = self.app.print_label(
|
||||
# text=context.user_data["label"], length=context.user_data["length"]
|
||||
# )
|
||||
# except Exception as e:
|
||||
# await update.message.reply_text(f"There was an exception: {e}")
|
||||
# return ConversationHandler.END
|
||||
#
|
||||
# await update.message.reply_photo(
|
||||
# label.bytes, f'Your label is: {context.user_data["label"]}'
|
||||
# )
|
||||
# return ConversationHandler.END
|
||||
#
|
||||
# async def cancel(self, update: Update, context: CallbackContext) -> int:
|
||||
# await update.message.reply_text("Cancelled.")
|
||||
# return ConversationHandler.END
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
application = Application(PILRenderer(), E550W(os.environ.get("PRINTER_IP")))
|
||||
bot = LabelingBot(application)
|
||||
|
||||
LABEL_LENGTH, LABEL_TEXT = range(2)
|
||||
|
||||
bot.media_info()
|
||||
label = bot.get_qrcode("512", 25)
|
||||
with open("label.png", "wb") as preview:
|
||||
preview.write(label.bytes)
|
||||
label = bot.print_qrcode("512", 25)
|
||||
98
labeler/adapter/fastapi_srv.py
Normal file
98
labeler/adapter/fastapi_srv.py
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
import os
|
||||
|
||||
from labeler.app.labeler import Application
|
||||
from labeler.infra.e550w_printer.printer import E550W
|
||||
from labeler.infra.renderer import PILRenderer
|
||||
|
||||
from fastapi import FastAPI
|
||||
from fastapi.responses import HTMLResponse, RedirectResponse
|
||||
from fastapi.staticfiles import StaticFiles
|
||||
|
||||
from fegen import discourse, main
|
||||
|
||||
app = FastAPI()
|
||||
|
||||
app.mount("/static", StaticFiles(directory="fegen/docs/static"), name="static")
|
||||
|
||||
|
||||
class LabelingBot:
|
||||
def __init__(self, app: Application):
|
||||
self.app = app
|
||||
|
||||
def media_info(self):
|
||||
media = self.app.get_installed_media()
|
||||
print(f"Installed medium: {media}")
|
||||
|
||||
def simple_label(self, label_text, label_length=0):
|
||||
try:
|
||||
label = self.app.print_label(text=label_text, length=label_length)
|
||||
except Exception as e:
|
||||
print(f"There was an exception: {e}")
|
||||
|
||||
def get_qrcode(self, label_text, label_length=0):
|
||||
label = self.app.render_qrcode_preview(text=label_text, length=label_length)
|
||||
return label
|
||||
|
||||
def print_qrcode(self, label_text, label_length=0):
|
||||
try:
|
||||
label = self.app.print_qrcode(text=label_text, length=label_length)
|
||||
except Exception as e:
|
||||
print(f"There was an exception: {e}")
|
||||
|
||||
# async def label_length(self):
|
||||
# await update.message.reply_text(
|
||||
# "Hello! Please tell me the length of the label, enter 0 for auto:"
|
||||
# )
|
||||
# return LABEL_LENGTH
|
||||
#
|
||||
# async def label_text(self, update: Update, context: CallbackContext) -> int:
|
||||
# user_input = update.message.text
|
||||
# context.user_data["length"] = int(user_input)
|
||||
# await update.message.reply_text("Now, please tell me the text of the label:")
|
||||
# return LABEL_TEXT
|
||||
#
|
||||
# async def simple_label(self, update: Update, context: CallbackContext) -> int:
|
||||
# user_input = update.message.text
|
||||
# context.user_data["label"] = user_input
|
||||
# try:
|
||||
# label = self.app.print_label(
|
||||
# text=context.user_data["label"], length=context.user_data["length"]
|
||||
# )
|
||||
# except Exception as e:
|
||||
# await update.message.reply_text(f"There was an exception: {e}")
|
||||
# return ConversationHandler.END
|
||||
#
|
||||
# await update.message.reply_photo(
|
||||
# label.bytes, f'Your label is: {context.user_data["label"]}'
|
||||
# )
|
||||
# return ConversationHandler.END
|
||||
#
|
||||
# async def cancel(self, update: Update, context: CallbackContext) -> int:
|
||||
# await update.message.reply_text("Cancelled.")
|
||||
# return ConversationHandler.END
|
||||
|
||||
@app.get("/", response_class=HTMLResponse)
|
||||
async def root():
|
||||
with open("fegen/docs/index.html", "r") as f:
|
||||
index = f.read()
|
||||
return index
|
||||
|
||||
@app.get("/print/{item_id}")
|
||||
def print_item(item_id: int, q: str | None = None):
|
||||
application = Application(PILRenderer(), E550W(os.environ.get("PRINTER_IP")))
|
||||
bot = LabelingBot(application)
|
||||
|
||||
LABEL_LENGTH, LABEL_TEXT = range(2)
|
||||
|
||||
bot.media_info()
|
||||
label = bot.get_qrcode(item_id, 25)
|
||||
with open("label.png", "wb") as preview:
|
||||
preview.write(label.bytes)
|
||||
bot.print_qrcode(item_id, 25)
|
||||
return RedirectResponse(url="/", status_code=302)
|
||||
|
||||
@app.get("/refresh")
|
||||
def refresh():
|
||||
discourse.DiscourseDatabase()
|
||||
main.generate_dashboard()
|
||||
return RedirectResponse(url="/", status_code=302)
|
||||
|
|
@ -1,6 +1,14 @@
|
|||
import os
|
||||
|
||||
from telegram.ext import CommandHandler, ApplicationBuilder
|
||||
from telegram import Update
|
||||
from telegram.ext import (
|
||||
CommandHandler,
|
||||
ApplicationBuilder,
|
||||
ConversationHandler,
|
||||
CallbackContext,
|
||||
filters,
|
||||
MessageHandler,
|
||||
)
|
||||
|
||||
from labeler.app.labeler import Application
|
||||
from labeler.infra.e550w_printer.printer import E550W
|
||||
|
|
@ -8,19 +16,67 @@ from labeler.infra.renderer import PILRenderer
|
|||
|
||||
|
||||
class LabelingBot:
|
||||
def __init__(self, app):
|
||||
def __init__(self, app: Application):
|
||||
self.app = app
|
||||
|
||||
async def media_info(self, update, context):
|
||||
media = self.app.get_installed_media()
|
||||
await update.message.reply_text(f"Installed media: {media.description}")
|
||||
|
||||
async def label_length(self, update, context):
|
||||
await update.message.reply_text(
|
||||
"Hello! Please tell me the length of the label, enter 0 for auto:"
|
||||
)
|
||||
return LABEL_LENGTH
|
||||
|
||||
async def label_text(self, update: Update, context: CallbackContext) -> int:
|
||||
user_input = update.message.text
|
||||
context.user_data["length"] = int(user_input)
|
||||
await update.message.reply_text("Now, please tell me the text of the label:")
|
||||
return LABEL_TEXT
|
||||
|
||||
async def simple_label(self, update: Update, context: CallbackContext) -> int:
|
||||
user_input = update.message.text
|
||||
context.user_data["label"] = user_input
|
||||
try:
|
||||
label = self.app.print_label(
|
||||
text=context.user_data["label"], length=context.user_data["length"]
|
||||
)
|
||||
except Exception as e:
|
||||
await update.message.reply_text(f"There was an exception: {e}")
|
||||
return ConversationHandler.END
|
||||
|
||||
await update.message.reply_photo(
|
||||
label.bytes, f'Your label is: {context.user_data["label"]}'
|
||||
)
|
||||
return ConversationHandler.END
|
||||
|
||||
async def cancel(self, update: Update, context: CallbackContext) -> int:
|
||||
await update.message.reply_text("Cancelled.")
|
||||
return ConversationHandler.END
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
application = Application(PILRenderer(), E550W(os.environ.get("PRINTER_IP")))
|
||||
bot = LabelingBot(application)
|
||||
|
||||
LABEL_LENGTH, LABEL_TEXT = range(2)
|
||||
|
||||
conv_handler = ConversationHandler(
|
||||
entry_points=[CommandHandler("simple_label", bot.label_length)],
|
||||
states={
|
||||
LABEL_LENGTH: [
|
||||
MessageHandler(filters.Text() & ~filters.Command(), bot.label_text)
|
||||
],
|
||||
LABEL_TEXT: [
|
||||
MessageHandler(filters.Text() & ~filters.Command(), bot.simple_label)
|
||||
],
|
||||
},
|
||||
fallbacks=[CommandHandler("cancel", bot.cancel)],
|
||||
)
|
||||
|
||||
app = ApplicationBuilder().token(os.environ["TELEGRAM_TOKEN"]).build()
|
||||
app.add_handler(CommandHandler("media_info", bot.media_info))
|
||||
app.add_handler(conv_handler)
|
||||
|
||||
app.run_polling()
|
||||
|
|
|
|||
|
|
@ -1,4 +1,11 @@
|
|||
from labeler.domain.objects import Label, LabelRequest, LabelDefinition, MediaDefinition
|
||||
from labeler.domain.objects import (
|
||||
Label,
|
||||
LabelRequest,
|
||||
LabelDefinition,
|
||||
MediaDefinition,
|
||||
Dimension,
|
||||
Image,
|
||||
)
|
||||
from labeler.interfaces import Renderer, Printer
|
||||
|
||||
|
||||
|
|
@ -7,16 +14,16 @@ class Application:
|
|||
self.renderer = renderer
|
||||
self.printer = printer
|
||||
|
||||
def render_preview(self, label_request: LabelRequest):
|
||||
def render_preview(self, text: str, length: int = None) -> Label:
|
||||
media = self.printer.get_installed_media()
|
||||
|
||||
if label_request.length is not None:
|
||||
label_length = label_request.length - 2 * media.minimal_margin_horizontal
|
||||
if length != 0:
|
||||
label_length = Dimension(mm=length) - 2 * media.minimal_margin_horizontal
|
||||
else:
|
||||
label_length = media.printable_length
|
||||
label_length = None
|
||||
|
||||
label_definition = LabelDefinition(
|
||||
text=label_request.text,
|
||||
text=text,
|
||||
length=label_length,
|
||||
width=media.printable_width,
|
||||
dpi=media.dpi,
|
||||
|
|
@ -24,5 +31,60 @@ class Application:
|
|||
|
||||
self.renderer.render_label(label_definition)
|
||||
|
||||
def render_qrcode_preview(self, text: str, length: int = None) -> Label:
|
||||
media = self.printer.get_installed_media()
|
||||
|
||||
if length != 0:
|
||||
label_length = Dimension(mm=length) - 2 * media.minimal_margin_horizontal
|
||||
else:
|
||||
label_length = None
|
||||
|
||||
label_definition = LabelDefinition(
|
||||
text=text,
|
||||
length=label_length,
|
||||
width=media.printable_width,
|
||||
dpi=media.dpi,
|
||||
)
|
||||
|
||||
return self.renderer.render_qrcode(label_definition)
|
||||
|
||||
def print_label(self, text: str, length: int = None) -> Image:
|
||||
media = self.printer.get_installed_media()
|
||||
|
||||
if length != 0:
|
||||
label_length = Dimension(mm=length) - 2 * media.minimal_margin_horizontal
|
||||
else:
|
||||
label_length = None
|
||||
|
||||
label_definition = LabelDefinition(
|
||||
text=text,
|
||||
length=label_length,
|
||||
width=media.printable_width,
|
||||
dpi=media.dpi,
|
||||
)
|
||||
|
||||
label = self.renderer.render_label(label_definition)
|
||||
self.printer.print_label(label)
|
||||
return label
|
||||
|
||||
def print_qrcode(self, text: str, length: int = None) -> Image:
|
||||
media = self.printer.get_installed_media()
|
||||
|
||||
if length != 0:
|
||||
label_length = Dimension(mm=length) - 2 * media.minimal_margin_horizontal
|
||||
else:
|
||||
label_length = None
|
||||
|
||||
label_definition = LabelDefinition(
|
||||
text=text,
|
||||
length=label_length,
|
||||
width=media.printable_width,
|
||||
dpi=media.dpi,
|
||||
)
|
||||
|
||||
label = self.renderer.render_qrcode(label_definition)
|
||||
self.printer.print_label(label)
|
||||
return label
|
||||
|
||||
def get_installed_media(self) -> MediaDefinition:
|
||||
return self.printer.get_installed_media()
|
||||
|
|
|
|||
|
|
@ -107,16 +107,17 @@ class MediaDefinition(BaseModel):
|
|||
length: Dimension
|
||||
minimal_margin_vertical: Dimension
|
||||
minimal_margin_horizontal: Dimension
|
||||
minimum_length: Dimension = Field(default_factory=lambda: Dimension(mm=5))
|
||||
dpi: int
|
||||
description: str
|
||||
|
||||
@property
|
||||
def printable_width(self) -> Dimension:
|
||||
return self.width - 2 * self.minimal_margin_horizontal
|
||||
return self.width - 2 * self.minimal_margin_vertical
|
||||
|
||||
@property
|
||||
def printable_length(self) -> Dimension:
|
||||
return self.length - 2 * self.minimal_margin_vertical
|
||||
return self.length - 2 * self.minimal_margin_horizontal
|
||||
|
||||
|
||||
class Label(BaseModel):
|
||||
|
|
|
|||
|
|
@ -1,10 +1,15 @@
|
|||
import io
|
||||
import logging
|
||||
from math import inf
|
||||
|
||||
from brother_ql import BrotherQLRaster, create_label
|
||||
from brother_ql.backends import guess_backend, backend_factory
|
||||
from brother_ql.conversion import convert
|
||||
from pysnmp.entity.engine import SnmpEngine
|
||||
from pysnmp.hlapi import getCmd, CommunityData, UdpTransportTarget, ContextData
|
||||
from pysnmp.smi.rfc1902 import ObjectType, ObjectIdentity
|
||||
|
||||
from labeler.domain.objects import MediaDefinition, Dimension
|
||||
from labeler.domain.objects import MediaDefinition, Dimension, Image
|
||||
from labeler.infra.e550w_printer.media_definitions import (
|
||||
media_width,
|
||||
tape_color,
|
||||
|
|
@ -16,6 +21,15 @@ from labeler.infra.e550w_printer.media_definitions import (
|
|||
TYPE_BYTE,
|
||||
)
|
||||
from labeler.interfaces import Printer
|
||||
from PIL import Image as PILImage
|
||||
|
||||
PRINTABLE_WIDTH = {
|
||||
6: Dimension.from_points(64, 360),
|
||||
9: Dimension.from_points(100, 360),
|
||||
12: Dimension.from_points(140, 360),
|
||||
18: Dimension.from_points(224, 360),
|
||||
24: Dimension.from_points(256, 360),
|
||||
}
|
||||
|
||||
|
||||
class E550W(Printer):
|
||||
|
|
@ -26,6 +40,45 @@ class E550W(Printer):
|
|||
def get_installed_media(self) -> MediaDefinition:
|
||||
return self.__get_printer_status()
|
||||
|
||||
def print_label(self, label: Image):
|
||||
im = PILImage.open(io.BytesIO(label.bytes))
|
||||
|
||||
qlr = BrotherQLRaster("PT-E550W")
|
||||
convert(
|
||||
qlr,
|
||||
[im],
|
||||
self.__media_width_to_type(label.height),
|
||||
red=False,
|
||||
threshold=70,
|
||||
cut=True,
|
||||
rotate=90,
|
||||
compress=True,
|
||||
dpi_600=True,
|
||||
hq=True,
|
||||
)
|
||||
|
||||
try:
|
||||
try:
|
||||
selected_backend = guess_backend(f"tcp://{self.ip_address}:9100")
|
||||
except ValueError:
|
||||
logging.error(
|
||||
"Couln't guess the backend to use from the printer string descriptor"
|
||||
)
|
||||
BACKEND_CLASS = backend_factory(selected_backend)["backend_class"]
|
||||
be = BACKEND_CLASS(f"tcp://{self.ip_address}:9100")
|
||||
be.write(qlr.data)
|
||||
be.dispose()
|
||||
del be
|
||||
except Exception as e:
|
||||
logging.exception("Exception happened: %s", e)
|
||||
|
||||
def __media_width_to_type(self, height: int):
|
||||
metric_width = Dimension.from_points(height, 360)
|
||||
for width, printable_width in PRINTABLE_WIDTH.items():
|
||||
if printable_width == metric_width:
|
||||
return f"pt5{width}"
|
||||
raise ValueError(f"Unsupported media width: {metric_width}")
|
||||
|
||||
def __get_printer_status(self):
|
||||
raw_snmp_data = self.__get_snmp_status().asNumbers()
|
||||
width = media_width(raw_snmp_data[WIDTH_BYTE])
|
||||
|
|
@ -36,9 +89,9 @@ class E550W(Printer):
|
|||
return MediaDefinition(
|
||||
width=Dimension(mm=width),
|
||||
length=Dimension(mm=inf),
|
||||
minimal_margin_vertical=Dimension(mm=1),
|
||||
minimal_margin_horizontal=Dimension(mm=2),
|
||||
dpi=600,
|
||||
minimal_margin_vertical=(Dimension(mm=width) - PRINTABLE_WIDTH[width]) / 2,
|
||||
minimal_margin_horizontal=Dimension(mm=1),
|
||||
dpi=360,
|
||||
description=f"{tape_type} - {width}mm, {media_text_color} on {media_tape_color} background",
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,3 +1,4 @@
|
|||
import qrcode
|
||||
import textwrap
|
||||
from string import ascii_letters
|
||||
|
||||
|
|
@ -7,9 +8,16 @@ from labeler.domain.objects import Image, LabelDefinition
|
|||
from labeler.interfaces import Renderer
|
||||
|
||||
|
||||
DPI = 360.0
|
||||
|
||||
|
||||
def points_to_pixels(point_size: float):
|
||||
return int(point_size * (72 / DPI))
|
||||
|
||||
|
||||
class PILRenderer(Renderer):
|
||||
def __init__(self):
|
||||
self.font_path = "/Library/Fonts/Arial.ttf"
|
||||
self.font_path = "/Library/Fonts/Arial Unicode.ttf"
|
||||
|
||||
def render_label(self, label_definition: LabelDefinition) -> Image:
|
||||
if label_definition.length is None:
|
||||
|
|
@ -47,7 +55,7 @@ class PILRenderer(Renderer):
|
|||
|
||||
while text_height > 0:
|
||||
font = ImageFont.truetype(
|
||||
"/Library/Fonts/Arial.ttf",
|
||||
"/Library/Fonts/Arial Unicode.ttf",
|
||||
text_height,
|
||||
)
|
||||
if lines_to_print > 1:
|
||||
|
|
@ -74,6 +82,42 @@ class PILRenderer(Renderer):
|
|||
)
|
||||
return im
|
||||
|
||||
def render_qrcode(self, label_definition: LabelDefinition) -> Image:
|
||||
width = label_definition.pixel_width
|
||||
length = label_definition.pixel_length
|
||||
text = label_definition.text
|
||||
font = self.__get_font_for_qr()
|
||||
qr = qrcode.QRCode(box_size=7)
|
||||
qr.add_data(f"https://hs3.pl/db/{text}")
|
||||
qr.make(fit=True)
|
||||
qr_img = qr.make_image()
|
||||
print(width, length)
|
||||
pil_image = PILImage.new("1", (width, length), 1)
|
||||
draw = ImageDraw.Draw(pil_image)
|
||||
pil_image.paste(qr_img)
|
||||
draw.text(
|
||||
(width / 2, 232),
|
||||
"HS3-DB",
|
||||
font=font,
|
||||
fill=0,
|
||||
anchor="mm",
|
||||
align="center",
|
||||
)
|
||||
draw.text(
|
||||
(20, 232 + 34),
|
||||
f"ID: {text}",
|
||||
font=font,
|
||||
fill=0,
|
||||
align="left",
|
||||
)
|
||||
|
||||
return Image.from_pil(pil_image.transpose(PILImage.Transpose.ROTATE_90))
|
||||
|
||||
def __get_font_for_qr(self):
|
||||
font_size = int(360 / 9)
|
||||
|
||||
return ImageFont.truetype("fonts/SourceCodePro-SemiBold.ttf", font_size)
|
||||
|
||||
def __get_font(self, text: str, max_width: int, max_length: int):
|
||||
font_size = max_width
|
||||
step = max_width // 2
|
||||
|
|
|
|||
|
|
@ -8,8 +8,16 @@ class Renderer(abc.ABC):
|
|||
def render_label(self, label_definition: LabelDefinition) -> Image:
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
def render_qrcode(self, label_definition: LabelDefinition) -> Image:
|
||||
pass
|
||||
|
||||
|
||||
class Printer(abc.ABC):
|
||||
@abc.abstractmethod
|
||||
def get_installed_media(self) -> MediaDefinition:
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
def print_label(self, label: Image):
|
||||
pass
|
||||
|
|
|
|||
37
pyproject.toml
Normal file
37
pyproject.toml
Normal file
|
|
@ -0,0 +1,37 @@
|
|||
[project]
|
||||
name = "python-scratchpad"
|
||||
version = "0.1.0"
|
||||
description = ""
|
||||
# authors = ["Hubert Bryłkowski <hubert@brylkowski.com>"]
|
||||
readme = "readme.md"
|
||||
requires-python = ">=3.11"
|
||||
dependencies = [
|
||||
"brother-ql @ git+https://github.com/hbrylkowski/brother_ql@4225d13d209e8e4a2c17e87a75f42809e0da8fda",
|
||||
"qrcode[pil]",
|
||||
# https://github.com/astral-sh/uv/issues/6384
|
||||
"setuptools<81",
|
||||
"jinja2>=3.1.2,<4",
|
||||
"pillow>=9.5.0,<10",
|
||||
"pysnmp>=4.4.12,<5",
|
||||
"pyasn1==0.4.8,<0.5",
|
||||
"python-telegram-bot>=20.3,<21",
|
||||
"fastapi[standard]>=0.114.0",
|
||||
"requests==2.32.5",
|
||||
"Jinja2==3.1.6",
|
||||
"pandas==2.3.3",
|
||||
"python-dotenv==1.2.1",
|
||||
]
|
||||
|
||||
[tool.poetry.group.dev.dependencies]
|
||||
pytest = "^7.4.0"
|
||||
black = "^23.3.0"
|
||||
|
||||
[tool.hatch]
|
||||
metadata.allow-direct-references = true
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["labeler"]
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
36
readme.md
Normal file
36
readme.md
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
## labeling telegram bot
|
||||
|
||||
This application is build to render and print labels sent to it via telegram, as well
|
||||
as provide info about printer status and other useful information.
|
||||
|
||||
|
||||
## Supported commands
|
||||
- `/media_info` - show info about currently installed media
|
||||

|
||||
- `/simple_label` - print a simple label
|
||||

|
||||
|
||||
### Usage example
|
||||
You need three things:
|
||||
1. A telegram bot token, you can write to [@BotFather](https://t.me/BotFather) to get one
|
||||
2. A compatible printer
|
||||
3. docker installed on your system
|
||||
|
||||
```yaml
|
||||
version: "3.8"
|
||||
services:
|
||||
bot:
|
||||
build:
|
||||
context: .
|
||||
dockerfile: Dockerfile
|
||||
environment:
|
||||
- PRINTER_IP=<printer_ip>
|
||||
- TELEGRAM_TOKEN=<telegram_bot_token>
|
||||
command:
|
||||
- python
|
||||
- labeler/adapter/telegram_bot.py
|
||||
|
||||
```
|
||||
|
||||
### Supported printers
|
||||
- Brother PT-E550W
|
||||
Loading…
Add table
Add a link
Reference in a new issue