Compare commits
No commits in common. "ca79aed027e1d56c585b41d2fe397f0f4c4216c2" and "c73aac7a50305db2de1a4e9953fee48a3823b1ab" have entirely different histories.
ca79aed027
...
c73aac7a50
26 changed files with 16 additions and 9169 deletions
|
|
@ -1,6 +0,0 @@
|
||||||
.venv
|
|
||||||
.git
|
|
||||||
__pycache__
|
|
||||||
*.pyc
|
|
||||||
.ruff_cache
|
|
||||||
.mypy_cache
|
|
||||||
47
.github/workflows/build_docker.yml
vendored
47
.github/workflows/build_docker.yml
vendored
|
|
@ -1,47 +0,0 @@
|
||||||
# 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
20
Dockerfile
|
|
@ -1,20 +0,0 @@
|
||||||
# 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" ]
|
|
||||||
|
|
@ -1,12 +0,0 @@
|
||||||
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
|
|
||||||
|
|
@ -1,111 +0,0 @@
|
||||||
'''
|
|
||||||
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)")
|
|
||||||
|
|
||||||
File diff suppressed because one or more lines are too long
14
fegen/docs/static/css/style.css
vendored
14
fegen/docs/static/css/style.css
vendored
|
|
@ -1,14 +0,0 @@
|
||||||
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;
|
|
||||||
}
|
|
||||||
|
|
@ -1,44 +0,0 @@
|
||||||
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!")
|
|
||||||
File diff suppressed because one or more lines are too long
|
|
@ -1,29 +0,0 @@
|
||||||
{% 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 %}
|
|
||||||
|
|
@ -1,14 +0,0 @@
|
||||||
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;
|
|
||||||
}
|
|
||||||
Binary file not shown.
BIN
img.png
BIN
img.png
Binary file not shown.
|
Before Width: | Height: | Size: 9.9 KiB |
BIN
img_1.png
BIN
img_1.png
Binary file not shown.
|
Before Width: | Height: | Size: 50 KiB |
|
|
@ -1,75 +0,0 @@
|
||||||
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)
|
|
||||||
|
|
@ -1,98 +0,0 @@
|
||||||
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,14 +1,6 @@
|
||||||
import os
|
import os
|
||||||
|
|
||||||
from telegram import Update
|
from telegram.ext import CommandHandler, ApplicationBuilder
|
||||||
from telegram.ext import (
|
|
||||||
CommandHandler,
|
|
||||||
ApplicationBuilder,
|
|
||||||
ConversationHandler,
|
|
||||||
CallbackContext,
|
|
||||||
filters,
|
|
||||||
MessageHandler,
|
|
||||||
)
|
|
||||||
|
|
||||||
from labeler.app.labeler import Application
|
from labeler.app.labeler import Application
|
||||||
from labeler.infra.e550w_printer.printer import E550W
|
from labeler.infra.e550w_printer.printer import E550W
|
||||||
|
|
@ -16,67 +8,19 @@ from labeler.infra.renderer import PILRenderer
|
||||||
|
|
||||||
|
|
||||||
class LabelingBot:
|
class LabelingBot:
|
||||||
def __init__(self, app: Application):
|
def __init__(self, app):
|
||||||
self.app = app
|
self.app = app
|
||||||
|
|
||||||
async def media_info(self, update, context):
|
async def media_info(self, update, context):
|
||||||
media = self.app.get_installed_media()
|
media = self.app.get_installed_media()
|
||||||
await update.message.reply_text(f"Installed media: {media.description}")
|
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__":
|
if __name__ == "__main__":
|
||||||
application = Application(PILRenderer(), E550W(os.environ.get("PRINTER_IP")))
|
application = Application(PILRenderer(), E550W(os.environ.get("PRINTER_IP")))
|
||||||
bot = LabelingBot(application)
|
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 = ApplicationBuilder().token(os.environ["TELEGRAM_TOKEN"]).build()
|
||||||
app.add_handler(CommandHandler("media_info", bot.media_info))
|
app.add_handler(CommandHandler("media_info", bot.media_info))
|
||||||
app.add_handler(conv_handler)
|
|
||||||
|
|
||||||
app.run_polling()
|
app.run_polling()
|
||||||
|
|
|
||||||
|
|
@ -1,11 +1,4 @@
|
||||||
from labeler.domain.objects import (
|
from labeler.domain.objects import Label, LabelRequest, LabelDefinition, MediaDefinition
|
||||||
Label,
|
|
||||||
LabelRequest,
|
|
||||||
LabelDefinition,
|
|
||||||
MediaDefinition,
|
|
||||||
Dimension,
|
|
||||||
Image,
|
|
||||||
)
|
|
||||||
from labeler.interfaces import Renderer, Printer
|
from labeler.interfaces import Renderer, Printer
|
||||||
|
|
||||||
|
|
||||||
|
|
@ -14,16 +7,16 @@ class Application:
|
||||||
self.renderer = renderer
|
self.renderer = renderer
|
||||||
self.printer = printer
|
self.printer = printer
|
||||||
|
|
||||||
def render_preview(self, text: str, length: int = None) -> Label:
|
def render_preview(self, label_request: LabelRequest):
|
||||||
media = self.printer.get_installed_media()
|
media = self.printer.get_installed_media()
|
||||||
|
|
||||||
if length != 0:
|
if label_request.length is not None:
|
||||||
label_length = Dimension(mm=length) - 2 * media.minimal_margin_horizontal
|
label_length = label_request.length - 2 * media.minimal_margin_horizontal
|
||||||
else:
|
else:
|
||||||
label_length = None
|
label_length = media.printable_length
|
||||||
|
|
||||||
label_definition = LabelDefinition(
|
label_definition = LabelDefinition(
|
||||||
text=text,
|
text=label_request.text,
|
||||||
length=label_length,
|
length=label_length,
|
||||||
width=media.printable_width,
|
width=media.printable_width,
|
||||||
dpi=media.dpi,
|
dpi=media.dpi,
|
||||||
|
|
@ -31,60 +24,5 @@ class Application:
|
||||||
|
|
||||||
self.renderer.render_label(label_definition)
|
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:
|
def get_installed_media(self) -> MediaDefinition:
|
||||||
return self.printer.get_installed_media()
|
return self.printer.get_installed_media()
|
||||||
|
|
|
||||||
|
|
@ -107,17 +107,16 @@ class MediaDefinition(BaseModel):
|
||||||
length: Dimension
|
length: Dimension
|
||||||
minimal_margin_vertical: Dimension
|
minimal_margin_vertical: Dimension
|
||||||
minimal_margin_horizontal: Dimension
|
minimal_margin_horizontal: Dimension
|
||||||
minimum_length: Dimension = Field(default_factory=lambda: Dimension(mm=5))
|
|
||||||
dpi: int
|
dpi: int
|
||||||
description: str
|
description: str
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def printable_width(self) -> Dimension:
|
def printable_width(self) -> Dimension:
|
||||||
return self.width - 2 * self.minimal_margin_vertical
|
return self.width - 2 * self.minimal_margin_horizontal
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def printable_length(self) -> Dimension:
|
def printable_length(self) -> Dimension:
|
||||||
return self.length - 2 * self.minimal_margin_horizontal
|
return self.length - 2 * self.minimal_margin_vertical
|
||||||
|
|
||||||
|
|
||||||
class Label(BaseModel):
|
class Label(BaseModel):
|
||||||
|
|
|
||||||
|
|
@ -1,15 +1,10 @@
|
||||||
import io
|
|
||||||
import logging
|
|
||||||
from math import inf
|
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.entity.engine import SnmpEngine
|
||||||
from pysnmp.hlapi import getCmd, CommunityData, UdpTransportTarget, ContextData
|
from pysnmp.hlapi import getCmd, CommunityData, UdpTransportTarget, ContextData
|
||||||
from pysnmp.smi.rfc1902 import ObjectType, ObjectIdentity
|
from pysnmp.smi.rfc1902 import ObjectType, ObjectIdentity
|
||||||
|
|
||||||
from labeler.domain.objects import MediaDefinition, Dimension, Image
|
from labeler.domain.objects import MediaDefinition, Dimension
|
||||||
from labeler.infra.e550w_printer.media_definitions import (
|
from labeler.infra.e550w_printer.media_definitions import (
|
||||||
media_width,
|
media_width,
|
||||||
tape_color,
|
tape_color,
|
||||||
|
|
@ -21,15 +16,6 @@ from labeler.infra.e550w_printer.media_definitions import (
|
||||||
TYPE_BYTE,
|
TYPE_BYTE,
|
||||||
)
|
)
|
||||||
from labeler.interfaces import Printer
|
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):
|
class E550W(Printer):
|
||||||
|
|
@ -40,45 +26,6 @@ class E550W(Printer):
|
||||||
def get_installed_media(self) -> MediaDefinition:
|
def get_installed_media(self) -> MediaDefinition:
|
||||||
return self.__get_printer_status()
|
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):
|
def __get_printer_status(self):
|
||||||
raw_snmp_data = self.__get_snmp_status().asNumbers()
|
raw_snmp_data = self.__get_snmp_status().asNumbers()
|
||||||
width = media_width(raw_snmp_data[WIDTH_BYTE])
|
width = media_width(raw_snmp_data[WIDTH_BYTE])
|
||||||
|
|
@ -89,9 +36,9 @@ class E550W(Printer):
|
||||||
return MediaDefinition(
|
return MediaDefinition(
|
||||||
width=Dimension(mm=width),
|
width=Dimension(mm=width),
|
||||||
length=Dimension(mm=inf),
|
length=Dimension(mm=inf),
|
||||||
minimal_margin_vertical=(Dimension(mm=width) - PRINTABLE_WIDTH[width]) / 2,
|
minimal_margin_vertical=Dimension(mm=1),
|
||||||
minimal_margin_horizontal=Dimension(mm=1),
|
minimal_margin_horizontal=Dimension(mm=2),
|
||||||
dpi=360,
|
dpi=600,
|
||||||
description=f"{tape_type} - {width}mm, {media_text_color} on {media_tape_color} background",
|
description=f"{tape_type} - {width}mm, {media_text_color} on {media_tape_color} background",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -1,4 +1,3 @@
|
||||||
import qrcode
|
|
||||||
import textwrap
|
import textwrap
|
||||||
from string import ascii_letters
|
from string import ascii_letters
|
||||||
|
|
||||||
|
|
@ -8,16 +7,9 @@ from labeler.domain.objects import Image, LabelDefinition
|
||||||
from labeler.interfaces import Renderer
|
from labeler.interfaces import Renderer
|
||||||
|
|
||||||
|
|
||||||
DPI = 360.0
|
|
||||||
|
|
||||||
|
|
||||||
def points_to_pixels(point_size: float):
|
|
||||||
return int(point_size * (72 / DPI))
|
|
||||||
|
|
||||||
|
|
||||||
class PILRenderer(Renderer):
|
class PILRenderer(Renderer):
|
||||||
def __init__(self):
|
def __init__(self):
|
||||||
self.font_path = "/Library/Fonts/Arial Unicode.ttf"
|
self.font_path = "/Library/Fonts/Arial.ttf"
|
||||||
|
|
||||||
def render_label(self, label_definition: LabelDefinition) -> Image:
|
def render_label(self, label_definition: LabelDefinition) -> Image:
|
||||||
if label_definition.length is None:
|
if label_definition.length is None:
|
||||||
|
|
@ -55,7 +47,7 @@ class PILRenderer(Renderer):
|
||||||
|
|
||||||
while text_height > 0:
|
while text_height > 0:
|
||||||
font = ImageFont.truetype(
|
font = ImageFont.truetype(
|
||||||
"/Library/Fonts/Arial Unicode.ttf",
|
"/Library/Fonts/Arial.ttf",
|
||||||
text_height,
|
text_height,
|
||||||
)
|
)
|
||||||
if lines_to_print > 1:
|
if lines_to_print > 1:
|
||||||
|
|
@ -82,42 +74,6 @@ class PILRenderer(Renderer):
|
||||||
)
|
)
|
||||||
return im
|
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):
|
def __get_font(self, text: str, max_width: int, max_length: int):
|
||||||
font_size = max_width
|
font_size = max_width
|
||||||
step = max_width // 2
|
step = max_width // 2
|
||||||
|
|
|
||||||
|
|
@ -8,16 +8,8 @@ class Renderer(abc.ABC):
|
||||||
def render_label(self, label_definition: LabelDefinition) -> Image:
|
def render_label(self, label_definition: LabelDefinition) -> Image:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def render_qrcode(self, label_definition: LabelDefinition) -> Image:
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class Printer(abc.ABC):
|
class Printer(abc.ABC):
|
||||||
@abc.abstractmethod
|
@abc.abstractmethod
|
||||||
def get_installed_media(self) -> MediaDefinition:
|
def get_installed_media(self) -> MediaDefinition:
|
||||||
pass
|
pass
|
||||||
|
|
||||||
@abc.abstractmethod
|
|
||||||
def print_label(self, label: Image):
|
|
||||||
pass
|
|
||||||
|
|
|
||||||
|
|
@ -1,37 +0,0 @@
|
||||||
[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
36
readme.md
|
|
@ -1,36 +0,0 @@
|
||||||
## 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