Files
kontor/kontor-scripts/read_queue.py
T
Thomas Peetz 5655bc5322
Gitea Actions Demo / Explore-Gitea-Actions (push) Successful in 3s
use msgspec to parse message
2026-07-31 04:05:23 +02:00

84 lines
2.9 KiB
Python

import logging
import stomp
import logging.config
import time
import yaml
import msgspec
import sys
from pathlib import Path
from argparse import ArgumentParser, ArgumentDefaultsHelpFormatter
from typing import Any, Dict, List, Optional
from platformdirs import PlatformDirs
from api import Server, get_api_config
from log import get_logger
parser = ArgumentParser(formatter_class=ArgumentDefaultsHelpFormatter)
parser.add_argument('--verbose', '-v', action='count', default=0)
parser.add_argument("--config", "-c", default="kontor-api")
parser.add_argument("--server", "-s", default="127.0.0.1")
parser.add_argument("--port", "-p", default="61616")
args = parser.parse_args()
class Link(msgspec.Struct):
url: str
class AddLinkListener(stomp.ConnectionListener):
def __init__(self, log, conn):
self.log = log
self.conn = conn
def on_error(self, frame):
self.log.info("received an error %s", frame.body)
def on_message(self, frame):
self.log.info("received a message %s", frame.body)
link = msgspec.json.decode(frame.body, type=Link)
self.log.info("found link: %s", link.url)
json_bytes = msgspec.json.encode(link)
self.conn.send(body=json_bytes, destination="add_link_accepted")
self.conn.send(body=json_bytes, destination="update_title")
class UpdateTitleListener(stomp.ConnectionListener):
def __init__(self, log, conn):
self.log = log
self.conn = conn
def on_error(self, frame):
self.log.info("received an error %s", frame.body)
def on_message(self, frame):
self.log.info("received a message %s", frame.body)
link = msgspec.json.decode(frame.body, type=Link)
self.log.info("found link: %s", link.url)
json_bytes = msgspec.json.encode(link)
self.conn.send(body=json_bytes, destination="update_title_accepted")
if __name__ == '__main__':
logger = get_logger(args.verbose, __file__)
logger.info("kontor.read_queue started")
APICONFIG = get_api_config(logger, args.config)
first_server: Optional[Server] = APICONFIG.get_server("inky")
if not first_server:
sys.exit(2)
data = server.request(log=log, table="media_file")
host = [(args.server, args.port)]
conn_add = stomp.Connection(host_and_ports=host)
conn_add.set_listener('', AddLinkListener(logger, conn_add))
conn_add.start()
conn_add.connect(username='artemis', passcode='artemis', wait=True)
conn_add.subscribe(destination='add_link', id=1, ack='auto', headers={})
conn_update = stomp.Connection(host_and_ports=host)
conn_update.set_listener('', UpdateTitleListener(logger, conn_update))
conn_update.start()
conn_update.connect(username='artemis', passcode='artemis', wait=True)
conn_update.subscribe(destination='update_title', id=1, ack='auto', headers={})
time.sleep(5)
conn_add.disconnect()
conn_update.disconnect()
logger.info("kontor.read_queue finished")