sch/Slixprint
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

slixprint.py

Browse source
This commit is contained in:
Schimon Jehudah 2024-03-20 16:18:20 +00:00
commit 64b8cc167c
1 changed files with 322 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

322
slixprint.py Normal file
View file

@ -0,0 +1,322 @@
#!/usr/bin/env python3
# Slixprint: The XMPP Printer Bot
# Copyright (C) 2024 Schimon Zackary
# This file is part of Slixmark.
# See the file LICENSE for copying permission.
from getpass import getpass
from argparse import ArgumentParser
import cups
import logging
import magic
import pdfkit
# from reportlab.pdfgen import canvas
import requests
import slixmpp
import time
import xml.sax.saxutils as saxutils
class Actions:
def determine_mimetype(content):
m = magic.Magic(mime=True)
mimetype = m.from_buffer(content)
return mimetype
def download_document(url):
response = requests.get(url)
content = response.content
return content
def export_to_pdf(filename):
# text = content.decode("utf-8")
# pdfkit.from_string(text, filename + '.pdf')
pdfkit.from_file(filename, filename + '.pdf')
# c = canvas.Canvas("Content.pdf")
# # Add content to the canvas according to your requirements here
# c.showPage()
# c.drawString(100, 750, "Hello World")
# c.save()
def write_document(content, filename):
with open(filename, 'wb') as f:
f.write(content)
def print_document(filename):
try:
conn = cups.Connection()
default_printer = conn.getDefault()
conn.printFile(default_printer, filename, "Slixprint Document Print", {})
except Exception as e:
return str(e)
def get_printers():
conn = cups.Connection()
printers = conn.getPrinters()
return printers
class Documentation:
def about():
return ('Slixprint'
'\n'
'Jabber/XMPP Bookmark Manager'
'\n\n'
'Slixprint is an XMPP bot that will print documents sent to it.'
'\n\n'
'Slixprint is written with slixmppp in Python.'
'\n\n'
'https://gitgud.io/sjehuda/slixprint'
'\n\n'
'Copyright 2024 Schimon Zackary'
'\n\n'
'Made in Switzerland'
'\n\n'
'🇨🇭️')
def commands():
return ("basic usage"
"\n"
" Send a URL to print."
"\n"
"printer <name> or <number>"
"\n"
" Set default printer."
"\n"
"printers"
"\n"
" List available printers."
"\n")
def notice():
return ('Copyright 2024 Schimon Jehudah Zackary'
'\n\n'
'Permission is hereby granted, free of charge, to any person '
'obtaining a copy of this software and associated '
'documentation files (the “Software”), to deal in the '
'Software without restriction, including without limitation '
'the rights to use, copy, modify, merge, publish, distribute, '
'sublicense, and/or sell copies of the Software, and to '
'permit persons to whom the Software is furnished to do so, '
'subject to the following conditions:'
'\n\n'
'The above copyright notice and this permission notice shall '
'be included in all copies or substantial portions of the '
'Software.'
'\n\n'
'THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY '
'KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE '
'WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR '
'PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR '
'COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER '
'LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR '
'OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE '
'SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.')
class Slixprint(slixmpp.ClientXMPP):
"""
Slixprint - Printer bot for Jabber/XMPP.
Slixprint is a printer bot based on slixmpp.
"""
def __init__(self, jid, password):
slixmpp.ClientXMPP.__init__(self, jid, password)
# The session_start event will be triggered when
# the bot establishes its connection with the server
# and the XML streams are ready for use. We want to
# listen for this event so that we we can initialize
# our roster.
self.add_event_handler("session_start", self.start)
# The message event is triggered whenever a message
# stanza is received. Be aware that that includes
# MUC messages and error messages.
self.add_event_handler("message", self.process_message)
self.add_event_handler("disco_info", self.discovery)
async def start(self, event):
"""
Process the session_start event.
Typical actions for the session_start event are
requesting the roster and broadcasting an initial
presence stanza.
Arguments:
event -- An empty dictionary. The session_start
event does not provide any additional
data.
"""
self.adhoc_commands()
self.send_presence()
await self['xep_0115'].update_caps()
vcard = self.plugin['xep_0054'].make_vcard()
vcard['FN'] = 'Slixfeed'
vcard['NICKNAME'] = 'Slixfeed'
vcard['ROLE'] = 'Syndication News Bot'
vcard['ORG'] = 'RSS Task Force'
vcard['URL'] = 'https://gitgud.io/sjehuda/slixfeed'
vcard['NOTE'] = 'Printer bot made for XMPP.'
vcard['BDAY'] = '20 March 2024'
await self.plugin['xep_0054'].publish_vcard(vcard)
self['xep_0030'].add_identity(category='client',
itype='printer',
name='slixprint',
node=None,
jid=self.boundjid.full)
async def discovery(self, DiscoInfo):
jid = DiscoInfo['from']
await self['xep_0115'].update_caps(jid=jid)
async def process_message(self, message):
"""
Process incoming message stanzas. Be aware that this also
includes MUC messages and error messages. It is usually
a good idea to check the messages's type before processing
or sending replies.
Arguments:
message -- The received message stanza. See the documentation
for stanza objects and the Message stanza to see
how it may be used.
"""
if message['type'] in ('chat', 'normal'):
message_text = " ".join(message['body'].split())
if message_text.lower().startswith('http'):
url = message_text
content = Actions.download_document(url)
mimetype = Actions.determine_mimetype(content)
extension = mimetype.split('/')[0]
jid_bare = message['from'].bare
filename = ('{jid}_{timestamp}.{extension}'
.format(jid=jid_bare,
timestamp=time.time(),
extension=extension))
Actions.write_document(content, filename)
error = Actions.print_document(filename)
if error:
logging.error(error)
Actions.export_to_pdf(filename)
url_upload = await self['xep_0363'].upload_file(
filename + '.pdf')
message_body = 'Error: {text}'.format(text=error)
# message_body = ('Error: {text} {url}'
# .format(text=error, url=url_upload))
url_upload = saxutils.escape(url_upload)
# html = (f'<body xmlns="http://www.w3.org/1999/xhtml">'
# f'<a href="{url}">{url}</a></body>')
message_file = self.make_message(mto=message['from'],
mfrom=self.boundjid.full,
mbody=url_upload,
# mhtml=html,
mtype='chat')
message_file['oob']['url'] = url_upload
message_file.send()
else:
message_body = ('Please send a document or a URL.')
message.reply(message_body).send()
#message.reply("Thanks for sending\n%(body)s" % message).send()
def adhoc_commands(self):
self['xep_0050'].add_command(node='printers',
name='🖨️ Printers',
handler=self._handle_printers)
self['xep_0050'].add_command(node='help',
name='📔️ Help',
handler=self._handle_help)
self['xep_0050'].add_command(node='about',
name='📜️ About Slixprint',
handler=self._handle_about)
def _handle_printers(self, iq, session):
form = self['xep_0004'].make_form('form', 'Add')
form['instructions'] = 'Add a new bookmark'
printers = Actions.get_printers()
options = form.add_field(desc='Select a printer.',
ftype='list-single',
label='Printers',
required=True,
var='printer')
for printer in printers:
options.addOption(printer, printers[printer]["device-uri"])
session['has_next'] = True
session['next'] = self._handle_printer
session['payload'] = form
return session
def _handle_cancel(self, payload, session):
text_note = 'Operation has been cancelled.'
session['notes'] = [['info', text_note]]
return session
def _handle_printer(self, payload, session):
text_note = 'This form is not available yet.'
session['notes'] = [['info', text_note]]
return session
def _handle_help(self, iq, session):
text_note = Documentation.commands()
session['notes'] = [['info', text_note]]
return session
def _handle_about(self, iq, session):
text_note = Documentation.about()
session['notes'] = [['info', text_note]]
return session
if __name__ == '__main__':
# Setup the command line arguments.
parser = ArgumentParser(description=Slixprint.__doc__)
# Output verbosity options.
parser.add_argument("-q", "--quiet", help="set logging to ERROR",
action="store_const", dest="loglevel",
const=logging.ERROR, default=logging.INFO)
parser.add_argument("-d", "--debug", help="set logging to DEBUG",
action="store_const", dest="loglevel",
const=logging.DEBUG, default=logging.INFO)
# JID and password options.
parser.add_argument("-j", "--jid", dest="jid",
help="JID to use")
parser.add_argument("-p", "--password", dest="password",
help="password to use")
args = parser.parse_args()
# Setup logging.
logging.basicConfig(level=args.loglevel,
format='%(levelname)-8s %(message)s')
if args.jid is None:
args.jid = input("Username: ")
if args.password is None:
args.password = getpass("Password: ")
# Setup the bot and register plugins. Note that while plugins may
# have interdependencies, the order in which you register them does
# not matter.
xmpp = Slixprint(args.jid, args.password)
xmpp.register_plugin('xep_0030') # Service Discovery
xmpp.register_plugin('xep_0004') # Data Forms
xmpp.register_plugin('xep_0050') # Ad-Hoc Commands
xmpp.register_plugin('xep_0054') # vcard-temp
xmpp.register_plugin('xep_0060') # Publish-Subscribe
xmpp.register_plugin('xep_0066') # Out of Band Data
xmpp.register_plugin('xep_0115') # Entity Capabilities
xmpp.register_plugin('xep_0122') # Data Forms Validation
xmpp.register_plugin('xep_0199') # XMPP Ping
# xmpp.register_plugin('xep_0363') # HTTP File Upload
# Connect to the XMPP server and start processing XMPP stanzas.
xmpp.connect()
xmpp.process()