doesnm/JabberCard
1
0
Fork 0
Code Issues 15 Pull requests Projects Releases Packages Wiki Activity Actions

Add file PyProject;

Browse source 
Support display of a single pubsub node item;
Update document README;
Modularize code;
This commit is contained in:
Schimon Jehudah, Adv. 2024-11-17 17:30:38 +02:00
parent 37aa7e8f40
commit 5e495579c2
32 changed files with 2431 additions and 2059 deletions
Download patch file Download diff file Expand all files Collapse all files

22
jabbercard/xmpp/instance.py Normal file
View file

@ -0,0 +1,22 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
from slixmpp import ClientXMPP
class XmppInstance(ClientXMPP):
def __init__(self, jid, password, jid_bare):
super().__init__(jid, password)
self.jid_bare = jid_bare
self.register_plugin('xep_0030') # XEP-0030: Service Discovery
self.register_plugin('xep_0045') # XEP-0045: Multi-User Chat
self.register_plugin('xep_0054') # XEP-0054: vcard-temp
self.register_plugin('xep_0060') # XEP-0060: Publish-Subscribe
self.register_plugin('xep_0369') # XEP-0369: Mediated Information eXchange (MIX)
self.add_event_handler("session_start", self.on_session_start)
async def on_session_start(self, event):
self.send_presence()
#self.disconnect()

467
jabbercard/xmpp/utilities.py Normal file
View file

@ -0,0 +1,467 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
from jabbercard.config import Cache
from jabbercard.utilities.graphics import Graphics
from jabbercard.utilities.toml import Toml
from jabbercard.xmpp.instance import XmppInstance
from jabbercard.xmpp.xep_0030 import XmppXep0030
from jabbercard.xmpp.xep_0045 import XmppXep0045
from jabbercard.xmpp.xep_0054 import XmppXep0054
from jabbercard.xmpp.xep_0060 import XmppXep0060
from jabbercard.xmpp.xep_0369 import XmppXep0369
import os
from dateutil import parser
from slixmpp import stanza
import xml.etree.ElementTree as ET
class XmppUtilities:
async def cache_vcard_data(
jabber_id, password, jid_bare, node_name_vcard4, item_id_vcard4):
# Start an XMPP instance and retrieve information
xmpp_instance = XmppInstance(jabber_id, password, jid_bare)
xmpp_instance.connect()
vcard4_data = await XmppXep0060.get_node_items(
xmpp_instance, jid_bare, node_name_vcard4, item_ids=[item_id_vcard4])
xmpp_instance.disconnect()
if vcard4_data:
directory_cache = Cache.get_directory()
supdirectory = os.path.join(directory_cache, 'xep_0060', jid_bare)
if not os.path.exists(supdirectory): os.mkdir(supdirectory)
directory = os.path.join(directory_cache, 'xep_0060', jid_bare, node_name_vcard4)
if not os.path.exists(directory): os.mkdir(directory)
if isinstance(vcard4_data['iq'], stanza.iq.Iq):
iq = vcard4_data['iq']
for item in iq['pubsub']['items']:
filename = directory + item_id_vcard4 + '.xml'
xml_item_as_string = str(item)
Toml.save_to_file(filename, xml_item_as_string)
#item_payload = item['payload']
#vcard4_info = Syndication.extract_vcard4_items(item_payload)
async def cache_node_data(
jabber_id, password, jid_bare, node_name):
# Start an XMPP instance and retrieve information
xmpp_instance = XmppInstance(jabber_id, password, jid_bare)
xmpp_instance.connect()
node_items = await XmppXep0060.get_node_items(
xmpp_instance, jid_bare, node_name)
xmpp_instance.disconnect()
if node_items:
supdirectory = 'xep_0060/{}/'.format(jid_bare)
if not os.path.exists(supdirectory): os.mkdir(supdirectory)
directory = os.path.join(directory_cache, 'xep_0060', jid_bare, node_name)
if not os.path.exists(directory): os.mkdir(directory)
if isinstance(node_items['iq'], stanza.iq.Iq):
iq = node_items['iq']
namespace = '{http://www.w3.org/2005/Atom}'
for item in iq['pubsub']['items']:
item_payload = item['payload']
date_element = item_payload.find(namespace + 'updated')
if not date_element: date_element = item_payload.find(namespace + 'published')
if isinstance(date_element, ET.Element):
date = date_element.text
modification_time = parser.parse(date).timestamp()
filename = directory + item['id'] + '.xml'
xml_item_as_string = str(item)
Toml.save_to_file(filename, xml_item_as_string)
if isinstance(date_element, ET.Element):
file_statistics = os.stat(filename)
access_time = file_statistics.st_atime
os.utime(filename, (access_time, modification_time))
#item_payload = item['payload']
#entry = Syndication.extract_atom_items(item_payload)
async def cache_jid_data(
jabber_id, password, jid_bare, node_name=None, item_id=None, alias=None):
iq_disco_items_list = iq_disco_items_items_list = node_note = node_title = title = ''
jid_vcard = {
'name' : '',
'note' : '',
'type' : '',
'bin' : ''}
#filename = 'details/{}.toml'.format(jid_bare)
#if os.path.exists(filename): jid_details = Toml.open_file_toml(filename)
# Start an XMPP instance and retrieve information
xmpp_instance = XmppInstance(jabber_id, password, jid_bare)
xmpp_instance.connect()
# JID kind
print('JID kind')
instance = message = node_id = None
jid_info = await XmppXep0030.get_jid_info(xmpp_instance, jid_bare)
jid_info_iq = jid_info['iq']
jid_kind = jid_info['kind']
# Set node name to 'urn:xmpp:microblog:0' if JID is an account
if jid_kind == 'account' and not node_name: node_name = 'urn:xmpp:microblog:0'
# vCard4 data
node_name_vcard4 = 'urn:xmpp:vcard4'
item_id_vcard4 = 'current'
vcard4_data = await XmppXep0060.get_node_items(
xmpp_instance, jid_bare, node_name_vcard4, item_ids=[item_id_vcard4])
if vcard4_data:
directory_cache = Cache.get_directory()
supdirectory = os.path.join(directory_cache, 'xep_0060', jid_bare)
if not os.path.exists(supdirectory): os.mkdir(supdirectory)
directory = os.path.join(directory_cache, 'xep_0060', jid_bare, node_name_vcard4)
if not os.path.exists(directory): os.mkdir(directory)
if isinstance(vcard4_data['iq'], stanza.iq.Iq):
iq = vcard4_data['iq']
for item in iq['pubsub']['items']:
filename = directory + item_id_vcard4 + '.xml'
xml_item_as_string = str(item)
Toml.save_to_file(filename, xml_item_as_string)
#item_payload = item['payload']
#vcard4_info = Syndication.extract_vcard4_items(item_payload)
# JID info
print('JID info')
# NOTE Group chat of Psi+ Project at jabber.ru has a note in its vCard.
vcard_data = await XmppXep0054.get_vcard_data(xmpp_instance, jid_bare)
if not vcard_data['error']:
conference_title = None
if jid_kind in ('mix', 'muc'):
for identity in jid_info_iq['disco_info']['identities']:
if identity[3]:
conference_title = identity[3]
break
vcard_temp = vcard_data['iq']['vcard_temp']
jid_vcard = {
'name' : vcard_temp['FN'] or conference_title or '',
'note' : vcard_temp['notes'] or node_id or '',
'type' : vcard_temp['PHOTO']['TYPE'] or '',
'bin' : vcard_temp['PHOTO']['BINVAL'] or ''}
# TODO /d/pubsub.nicoco.fr/blog/urn-uuid-53e43061-1962-3112-bb8a-1473dca61719
count = ''
jid_items = await XmppXep0030.get_jid_items(xmpp_instance, jid_bare)
if isinstance(jid_items['iq'], stanza.iq.Iq):
iq = jid_items['iq']
iq_disco_items = iq['disco_items']
iq_disco_items_items = iq_disco_items['items']
#iq_disco_items_set = {''}
iq_disco_items_list = []
iq_disco_items_items_list = []
for item in iq_disco_items_items:
if jid_kind == 'muc':
#iq_disco_items_set.update([item[2]])
iq_disco_items_list.append(item[2])
else:
#iq_disco_items_set.update([item[1]])
iq_disco_items_list.append(item[1])
iq_disco_items_items_list.append(
[item[0] or '', item[1] or '', item[2] or ''])
#count = len(iq_disco_items_set)
count = len(iq_disco_items_list)
# Title
print('Title')
if (jid_kind not in ('conference', 'mix', 'muc') and
'@' in jid_bare and
not node_name):
# NOTE Variables node_name and node_title do not appear to be utilized.
node_name = 'urn:xmpp:microblog:0'
node_title = 'Journal'
elif jid_kind == 'pubsub':
category = 'unsorted'
for item in iq_disco_items_items:
if item[2] and item[1] == node_name:
node_title = item[2]
break
else:
jid_items = None
if jid_kind == 'server':
if jid_info_iq:
for identity in jid_info_iq['disco_info']['identities']:
if jid_kind == identity[0] and identity[1] == 'im' and identity[3]:
title = identity[3]
print(jid_bare)
print(identity)
print(jid_info)
# String 'undefined' is sourced from JID discuss@conference.conversejs.org
if not title:
if jid_vcard['name'] and not 'undefined' in jid_vcard['name']:
title = jid_vcard['name']
else:
title = jid_bare.split('@')[0]
# JID item count
#count = await XmppUtilities.count_jid_items(xmpp_instance, jid_bare, node_name, jid_kind)
#if jid_kind in ('mix', 'muc', 'conference', 'server'):
# jid_items = await XmppXep0030.get_jid_items(xmpp_instance, jid_bare)
# if isinstance(jid_items['iq'], stanza.iq.Iq):
# iq = jid_items['iq']
# count = len(iq['disco_items']['items'])
#elif jid_kind in ('account', 'pubsub'):
# node_item_ids = await XmppXep0060.get_node_item_ids(xmpp_instance, jid_bare, node_name)
# if isinstance(node_item_ids, stanza.iq.Iq):
# count = len(node_item_ids['disco_items']['items'])
# Group chat messages
print('Group chat messages')
messages = []
subject = ''
if jid_kind == 'muc':
#action = 'Join'
# TODO Create configurations for group chat preview
room_info_muc = await XmppXep0045.get_room_information(
xmpp_instance, jid_bare, alias, maxstanzas=50)
# NOTE Do not mix error messages with node titles and descriptions etc.
if isinstance(room_info_muc['iq'], tuple):
iq = room_info_muc['iq']
for message in iq[3]:
messages.append({
'id' : message['id'],
'alias' : message['mucnick'],
'body' : message['body'],
'timestamp' : message['delay']['stamp'].__str__()})
messages.reverse()
subject = iq[1]['subject']
#title = title or node_name
if not node_title: node_title = node_name
node_note = jid_bare
else:
message = '{}: {} (XEP-0045)'.format(room_info_muc['condition'], room_info_muc['text'])
elif jid_kind == 'mix':
room_info_muc = await XmppXep0369.get_room_information(
xmpp_instance, jid_bare, alias)
if isinstance(room_info_muc['iq'], tuple):
iq = room_info_muc['iq']
for message in iq[3]:
messages.append({
'id' : message['id'],
'alias' : message['mucnick'],
'body' : message['body'],
'timestamp' : message['delay']['stamp'].__str__()})
messages.reverse()
subject = iq[1]['subject']
#title = title or node_name
if not node_title: node_title = node_name
node_note = jid_bare
else:
message = '{}: {} (XEP-0369)'.format(room_info_muc['condition'], room_info_muc['text'])
# Node items
print('Node items')
nodes = {}
#if node_name and node_name in iq_disco_items_set:
if iq_disco_items_list and node_name and node_name in iq_disco_items_list:
#action = 'Browse'
node_item_ids = await XmppXep0060.get_node_item_ids(xmpp_instance, jid_bare, node_name)
if isinstance(node_item_ids['iq'], stanza.iq.Iq):
iq = node_item_ids['iq']
nodes[node_name] = {}
nodes[node_name]['title'] = node_title
nodes[node_name]['count'] = len(iq['disco_items']['items'])
nodes[node_name]['item_ids'] = []
for item_id in iq['disco_items']['items']:
nodes[node_name]['item_ids'].append(
[item_id[0] or '', item_id[1] or '', item_id[2] or ''])
item_ids = []
for item in nodes[node_name]['item_ids']:
item_ids.append(item[2])
node_items = await XmppXep0060.get_node_items(
xmpp_instance, jid_bare, node_name)
if node_items:
directory_cache = Cache.get_directory()
supdirectory = os.path.join(directory_cache, 'xep_0060', jid_bare)
if not os.path.exists(supdirectory): os.mkdir(supdirectory)
directory = os.path.join(directory_cache, 'xep_0060', jid_bare, node_name)
if not os.path.exists(directory): os.mkdir(directory)
if isinstance(node_items['iq'], stanza.iq.Iq):
iq = node_items['iq']
namespace = '{http://www.w3.org/2005/Atom}'
for item in iq['pubsub']['items']:
item_payload = item['payload']
date_element = item_payload.find(namespace + 'updated')
if not date_element: date_element = item_payload.find(namespace + 'published')
if isinstance(date_element, ET.Element):
date = date_element.text
modification_time = parser.parse(date).timestamp()
filename = directory + item['id'] + '.xml'
xml_item_as_string = str(item)
Toml.save_to_file(filename, xml_item_as_string)
if isinstance(date_element, ET.Element):
file_statistics = os.stat(filename)
access_time = file_statistics.st_atime
os.utime(filename, (access_time, modification_time))
#item_payload = item['payload']
#entry = Syndication.extract_atom_items(item_payload)
xmpp_instance.disconnect()
# Notes
print('Notes')
jid_vcard_note = jid_vcard['note']
if isinstance(jid_vcard_note, list) and len(jid_vcard_note):
note = jid_vcard_note[0]['NOTE']
else:
note = jid_vcard_note
#if not note and jid_vcard['name'] and not 'undefined' in jid_vcard['name'] and title != jid_vcard['name']:
# note = jid_vcard['name']
jid_details = {
'count' : count or '',
'error' : jid_info['error'],
'error_text' : jid_info['text'] or '',
'error_condition' : jid_info['condition'] or '',
'image_type' : jid_vcard['type'],
'items' : iq_disco_items_items_list,
'kind' : jid_kind or '',
'messages' : messages or '',
'name' : title,
'nodes' : nodes,
'note' : note or '',
'subject' : subject or ''}
print(jid_details)
# Query URI href
link_href = XmppUtilities.get_link_href(jid_bare, jid_kind, node_name)
Graphics.handle_photo(jid_bare, jid_vcard, link_href)
directory_cache = Cache.get_directory()
filename = os.path.join(directory_cache, 'details', jid_bare + '.toml')
Toml.save_to_toml(filename, jid_details)
return jid_details
def set_action_instance_type(jid_kind, node_name=None):
if jid_kind in ('conference', 'server'):
action = 'Discover'
if jid_kind == 'conference':
instance = 'conferences'
elif jid_kind == 'server':
instance = 'services'
elif jid_kind in ('mix', 'muc'):
action = 'Join'
instance = 'occupants'
elif jid_kind == 'pubsub':
if node_name:
action = 'Subscribe'
instance = 'articles'
else:
action = 'Browse'
instance = 'nodes'
elif jid_kind == 'account':
action = 'Message'
instance = 'articles'
else: # jid_info['error']
action = 'Contact'
return action, instance
def get_link_href(jid_bare, jid_kind, node_name=None):
if jid_kind in ('conference', 'server'):
link_href = 'xmpp:{}?disco;type=get;request=items'.format(jid_bare)
elif jid_kind in ('mix', 'muc'):
link_href = 'xmpp:{}?join'.format(jid_bare)
elif jid_kind == 'pubsub':
if node_name:
link_href = 'xmpp:{}?pubsub;node={};action=subscribe'.format(jid_bare, node_name)
else:
link_href = 'xmpp:{}?disco;type=get;request=items'.format(jid_bare)
elif jid_kind == 'account':
link_href = 'xmpp:{}?message'.format(jid_bare)
else: # jid_info['error']
link_href = 'xmpp:{}'.format(jid_bare)
return link_href
def get_view_href(jid_bare, jid_kind, node_name=None):
links = []
view_href = None
if jid_kind in ('conference', 'server'):
view_href = '/d/' + jid_bare
elif jid_kind in ('mix', 'muc'):
view_href = '/v/' + jid_bare
elif jid_kind == 'pubsub':
if node_name:
view_href = '/d/{}/{}'.format(jid_bare, node_name)
else:
view_href = '/d/' + jid_bare
elif jid_kind == 'account':
view_href = '/d/{}/{}'.format(jid_bare, node_name)
return view_href
def get_xmpp_uri(jid_bare, jid_kind, node_name=None):
links = []
view_href = None
xmpp_uri = jid_bare
if jid_kind in ('conference', 'server'):
xmpp_uri = jid_bare
elif jid_kind in ('mix', 'muc'):
xmpp_uri = jid_bare
elif jid_kind == 'pubsub':
if node_name:
xmpp_uri = '{}?;node={}'.format(jid_bare, node_name)
else:
xmpp_uri = jid_bare
elif jid_kind == 'account':
xmpp_uri = jid_bare
return xmpp_uri
def get_query_uri_links(jid_bare, jid_kind, node_name=None, item_id=None):
links = []
if jid_kind in ('conference', 'server'):
links.append({'name' : 'Discover',
'href' : 'xmpp:{}?disco;type=get;request=items'.format(jid_bare),
'iden' : 'discover'})
xmpp_uri = jid_bare
elif jid_kind in ('mix', 'muc'):
links.append({'name' : 'Join',
'href' : 'xmpp:{}?join'.format(jid_bare),
'iden' : 'join'})
elif jid_kind == 'pubsub':
links.append({'name' : 'Browse',
'href' : 'xmpp:{}?disco;type=get;request=items'.format(jid_bare),
'iden' : 'browse'})
elif jid_kind == 'account':
links.append({'name' : 'Message',
'href' : 'xmpp:{}?message'.format(jid_bare),
'iden' : 'message'})
links.append({'name' : 'Add',
'href' : 'xmpp:{}?roster'.format(jid_bare),
'iden' : 'add'})
else: # jid_info['error']
links.append({'name' : 'Connect',
'href' : 'xmpp:{}'.format(jid_bare),
'iden' : 'connect'})
links.append({'name' : 'Add',
'href' : 'xmpp:{}?roster'.format(jid_bare),
'iden' : 'add'})
if item_id:
links.append({'name' : 'Subscribe',
'href' : 'xmpp:{}?pubsub;node={};item={};action=subscribe'.format(jid_bare, node_name, item_id),
'iden' : 'subscribe'})
links.append({'name' : 'View',
'href' : 'xmpp:{}?pubsub;node={};item={}'.format(jid_bare, node_name, item_id),
'iden' : 'view'})
elif node_name:
links.append({'name' : 'Subscribe',
'href' : 'xmpp:{}?pubsub;node={};action=subscribe'.format(jid_bare, node_name),
'iden' : 'subscribe'})
links.append({'name' : 'View',
'href' : 'xmpp:{}?pubsub;node={}'.format(jid_bare, node_name),
'iden' : 'view'})
links.append({'name' : 'vCard',
'href' : 'xmpp:{}?vcard'.format(jid_bare),
'iden' : 'vcard'})
return links

109
jabbercard/xmpp/xep_0030.py Normal file
View file

@ -0,0 +1,109 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
from asyncio import TimeoutError
from slixmpp.exceptions import IqError, IqTimeout, PresenceError
class XmppXep0030:
async def get_jid_items(self, jid_bare):
try:
condition = text = ''
error = False
iq = await self['xep_0030'].get_items(jid=jid_bare)
except (IqError, IqTimeout) as e:
#logger.warning('Chat type could not be determined for {}'.format(jid_bare))
#logger.error(e)
iq = ''
error = True
condition = e.iq['error']['condition']
text = e.iq['error']['text'] or 'Error'
#if not text:
# # NOTE We might want to set a specific photo for condition remote-server-not-found
# if condition:
# text = 'Could not determine JID type'
# else:
# text = 'Unknown Error'
result = {
'condition' : condition,
'error' : error,
'iq' : iq,
'text' : text}
return result
# NOTE
# Feature "urn:xmpp:mucsub:0" is present in both, MUC local and MUC hostname
# Feature "urn:xmpp:serverinfo:0" is present in both, MUC hostname and main hostname
async def get_jid_info(self, jid_bare):
jid_kind = None
try:
error = False
condition = text = ''
iq = await self['xep_0030'].get_info(jid=jid_bare)
iq_disco_info = iq['disco_info']
if iq_disco_info:
features = iq_disco_info['features']
if 'http://jabber.org/protocol/muc#unique' in features:
jid_kind = 'conference'
elif 'urn:xmpp:mix:core:1' in features:
jid_kind = 'mix'
elif ('muc_moderated' in features or
'muc_open' in features or
'muc_persistent' in features or
'muc_public' in features or
'muc_semianonymous' in features or
'muc_unmoderated' in features or
'muc_unsecured' in features):
jid_kind = 'muc'
elif '@' in jid_bare:
for identity in iq_disco_info['identities']:
#if identity[0] == 'pubsub' and identity[1] == 'pep':
if identity[0] == 'account':
#if 'urn:xmpp:bookmarks:1#compat-pep' in features:
#if 'urn:xmpp:bookmarks:1#compat' in features:
#if 'urn:xmpp:push:0' in features:
#if 'urn:xmpp:pep-vcard-conversion:0' in features:
#if 'urn:xmpp:sid:0' in features:
# Also in MIX
#if 'urn:xmpp:mam:2' in features:
#if 'urn:xmpp:mam:2#extended' in features:
jid_kind = 'account'
break
if identity[0] == 'client' and identity[1] == 'bot':
jid_kind = 'bot'
else:
for identity in iq_disco_info['identities']:
if identity[0] == 'pubsub' and identity[1] == 'service':
#if 'http://jabber.org/protocol/pubsub' in features:
#if 'http://jabber.org/protocol/pubsub#access-authorize' in features:
#if 'http://jabber.org/protocol/rsm' in features:
jid_kind = 'pubsub'
break
if identity[0] == 'server' and identity[1] == 'im':
jid_kind = 'server'
break
#logger.info('Jabber ID: {}\n'
# 'Chat Type: {}'.format(jid_bare, result))
else:
iq = condition = text = ''
except (IqError, IqTimeout) as e:
#logger.warning('Chat type could not be determined for {}'.format(jid_bare))
#logger.error(e)
iq = ''
error = True
condition = e.iq['error']['condition']
text = e.iq['error']['text'] or 'Error'
#if not text:
# # NOTE We might want to set a specific photo for condition remote-server-not-found
# if condition:
# text = 'Could not determine JID type'
# else:
# text = 'Unknown Error'
result = {
'condition' : condition,
'error' : error,
'iq' : iq,
'text' : text,
'kind' : jid_kind}
return result

58
jabbercard/xmpp/xep_0045.py Normal file
View file

@ -0,0 +1,58 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
from asyncio import TimeoutError
from datetime import datetime
from slixmpp.exceptions import IqError, IqTimeout, PresenceError
class XmppXep0045:
async def get_room_information(self, jid, alias, maxchars=None, maxstanzas=None, seconds=None):
#logger.info('Joining groupchat\nJID : {}\n'.format(jid))
#jid_from = str(self.boundjid) if self.is_component else None
if not maxchars: maxchars = 1000
if not maxstanzas: maxstanzas = 50
if not seconds: seconds = 864000
try:
error = False
condition = text = ''
#since = datetime.fromtimestamp(time.time()-seconds)
iq = await self['xep_0045'].join_muc_wait(
jid,
alias,
#maxchars=maxchars,
maxstanzas=maxstanzas,
#password=None,
#presence_options = {"pfrom" : jid_from},
#seconds=seconds,
#since=since,
timeout=10
)
except TimeoutError as e:
#raise HTTPException(status_code=504, detail='request-timeout-reached')
error = True
iq = e
condition = 'Request timeout reached'
text = str(e)
except (IqError, IqTimeout, PresenceError) as e:
error = True
iq = e
condition = e.iq['error']['condition']
text = e.iq['error']['text']
result = {
'error' : error,
'condition' : condition,
'text' : text,
'iq' : iq}
return result
async def get_room_data(self, jid_bare):
return await self['xep_0045'].get_room_config(jid_bare)
async def get_room_participants(self, jid_bare):
return await self['xep_0045'].get_roster(jid_bare)
# NOTE: "Item not found", yet is a group chat
# That is, JID has no vcard
# messaging-off@conference.movim.eu

29
jabbercard/xmpp/xep_0054.py Normal file
View file

@ -0,0 +1,29 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
from asyncio import TimeoutError
from slixmpp.exceptions import IqError, IqTimeout, PresenceError
class XmppXep0054:
async def get_vcard_data(self, jid_bare):
try:
error = False
condition = text = ''
iq = await self['xep_0054'].get_vcard(jid_bare)
except (IqError, IqTimeout) as e:
error = True
condition = e.iq['error']['condition']
text = e.iq['error']['text']
if not text:
if condition:
text = 'Could not retrieve vCard'
else:
text = 'Unknown Error'
iq = ''
result = {
'error' : error,
'condition' : condition,
'text' : text,
'iq' : iq}
return result

69
jabbercard/xmpp/xep_0060.py Normal file
View file

@ -0,0 +1,69 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
from asyncio import TimeoutError
from slixmpp.exceptions import IqError, IqTimeout, PresenceError
class XmppXep0060:
async def get_node_items(self, jid_bare, node_name, item_ids=None, max_items=None):
try:
error = False
condition = text = ''
if max_items:
iq = await self['xep_0060'].get_items(
jid_bare, node_name, timeout=5)
it = self['xep_0060'].get_items(
jid_bare, node_name, timeout=5, max_items=max_items, iterator=True)
q = rsm.Iq()
q['to'] = jid_bare
q['disco_items']['node'] = node_name
async for item in rsm.ResultIterator(q, 'disco_items', '10'):
print(item['disco_items']['items'])
else:
iq = await self['xep_0060'].get_items(
jid_bare, node_name, timeout=5, item_ids=item_ids)
result = iq
except (IqError, IqTimeout) as e:
error = True
iq = ''
condition = e.iq['error']['condition']
text = e.iq['error']['text']
if not text:
if condition:
text = 'Could not retrieve node items'
else:
text = 'Unknown Error'
result = {
'error' : error,
'condition' : condition,
'text' : text,
'iq' : iq}
return result
async def get_node_item_ids(self, jid_bare, node_name):
try:
error = False
condition = text = ''
iq = await self['xep_0030'].get_items(
jid_bare, node_name)
# Broken. See https://codeberg.org/poezio/slixmpp/issues/3548
#iq = await self['xep_0060'].get_item_ids(
# jid_bare, node_name, timeout=5)
except (IqError, IqTimeout) as e:
error = True
iq = ''
condition = e.iq['error']['condition']
text = e.iq['error']['text']
if not text:
if condition:
text = 'Could not retrieve node item IDs'
else:
text = 'Unknown Error'
result = {
'error' : error,
'condition' : condition,
'text' : text,
'iq' : iq}
return result

34
jabbercard/xmpp/xep_0369.py Normal file
View file

@ -0,0 +1,34 @@
#!/usr/bin/python
# -*- coding: utf-8 -*-
from asyncio import TimeoutError
from slixmpp.exceptions import IqError, IqTimeout, PresenceError
class XmppXep0369:
async def get_room_information(self, jid, alias):
iq = await self['xep_0369'].join_channel(jid, alias)
breakpoint()
print('GOOD DAY! Please contact the developer!')
try:
error = False
condition = text = ''
#iq = await self['xep_0369'].get_channel_info(jid)
iq = await self['xep_0369'].join_channel(jid, alias)
except TimeoutError as e:
#raise HTTPException(status_code=504, detail='request-timeout-reached')
error = True
iq = e
condition = 'Request timeout reached'
text = str(e)
except (IqError, IqTimeout, PresenceError) as e:
error = True
iq = e
condition = e.iq['error']['condition']
text = e.iq['error']['text']
result = {
'error' : error,
'condition' : condition,
'text' : text,
'iq' : iq}
return result