From 003cd12be734b05d7281579d3d377f79425531fa Mon Sep 17 00:00:00 2001
From: Simeon Simeonov
Date: Sat, 10 Sep 2022 10:27:18 +0200
Subject: Remove obsolete code and use black alike code styling
---
beinc_generic_client.py | 86 +++++++++------
beinc_pull.py | 110 +++++++++++--------
beinc_server.py | 156 +++++++++++++++-----------
beinc_weechat.py | 286 +++++++++++++++++++++++++++---------------------
4 files changed, 376 insertions(+), 262 deletions(-)
diff --git a/beinc_generic_client.py b/beinc_generic_client.py
index ca3847a..af1e74f 100755
--- a/beinc_generic_client.py
+++ b/beinc_generic_client.py
@@ -2,7 +2,7 @@
# -*- coding: utf-8 -*-
# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0
-# Copyright (C) 2013-2020 Simeon Simeonov
+# Copyright (C) 2013-2022 Simeon Simeonov
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -29,9 +29,8 @@ import sys
import urllib.parse
import urllib.request
-
__author__ = 'Simeon Simeonov'
-__version__ = '4.1'
+__version__ = '4.2'
__license__ = 'GPL3'
@@ -58,7 +57,7 @@ def fetch_password(args_password):
sys.exit(errno.EACCES)
elif os.path.isfile(args_password):
try:
- with io.open(args_password, 'r') as fp:
+ with io.open(args_password, 'r', encoding='utf-8') as fp:
passwd = fp.readline()
if passwd.strip():
return passwd.strip()
@@ -86,10 +85,12 @@ def pull_notifications(ssl_context, args):
data=urllib.parse.urlencode(
(
('resource_name', args.rname),
- ('password', args.password)
- )).encode('utf-8'),
+ ('password', args.password),
+ )
+ ).encode('utf-8'),
timeout=args.socket_timeout,
- context=ssl_context)
+ context=ssl_context,
+ )
response_dict = json.loads(response.read().decode('utf-8'))
if response.code != 200:
raise socket.error(response_dict.get('message', ''))
@@ -113,10 +114,12 @@ def push_notification(ssl_context, args):
('resource_name', args.rname),
('password', args.password),
('title', args.title),
- ('message', args.message)
- )).encode('utf-8'),
+ ('message', args.message),
+ )
+ ).encode('utf-8'),
timeout=args.socket_timeout,
- context=ssl_context)
+ context=ssl_context,
+ )
response_dict = json.loads(response.read().decode('utf-8'))
if response.code != 200:
raise socket.error(response_dict.get('message', ''))
@@ -125,91 +128,112 @@ def push_notification(ssl_context, args):
def main(inargs=None):
"""main entry"""
parser = argparse.ArgumentParser(
- description='The following options are available')
+ description='The following options are available'
+ )
parser.add_argument(
'url',
metavar='URL',
type=str,
- help='BEINC server destination URL')
+ help='BEINC server destination URL',
+ )
parser.add_argument(
- '-c', '--cert-file',
+ '-c',
+ '--cert-file',
metavar='FILE',
type=str,
dest='cert',
default='',
help='CA-cert to check the server-cert against'
- '(default: Check disabled)')
+ '(default: Check disabled)',
+ )
parser.add_argument(
'--ciphers',
metavar='CIPHERS',
type=str,
dest='ciphers',
default='',
- help='Preferred ciphers list (default: auto)')
+ help='Preferred ciphers list (default: auto)',
+ )
parser.add_argument(
'--disable-hostname-check',
action='store_true',
dest='disable_hostname_check',
default=False,
- help='Do not check whether server cert matches server hostname')
+ help='Do not check whether server cert matches server hostname',
+ )
parser.add_argument(
- '-m', '--message',
+ '-m',
+ '--message',
metavar='MESSAGE',
type=str,
dest='message',
default='BEINC message',
- help='BEINC message (default: "BEINC message")')
+ help='BEINC message (default: "BEINC message")',
+ )
parser.add_argument(
- '-n', '--resource-name',
+ '-n',
+ '--resource-name',
metavar='NAME',
type=str,
dest='rname',
required=True,
- help='The name of the BEINC-resource on the remote server')
+ help='The name of the BEINC-resource on the remote server',
+ )
parser.add_argument(
- '-p', '--password',
+ '-p',
+ '--password',
metavar='PASSWORD[FILE]',
type=str,
dest='password',
default='',
help='BEINC taget-password / text-file containing the target password'
- ' (default & recommended: prompt for passwd)')
+ ' (default & recommended: prompt for passwd)',
+ )
parser.add_argument(
'--pull',
action='store_true',
dest='pull',
default=False,
- help='Perform a pull operation (default: push)')
+ help='Perform a pull operation (default: push)',
+ )
parser.add_argument(
- '-T', '--socket-timeout',
+ '-T',
+ '--socket-timeout',
metavar='SECONDS',
type=int,
dest='socket_timeout',
default=3,
- help='Socket timeout in seconds (0=Python default) (default: 3)')
+ help='Socket timeout in seconds (0=Python default) (default: 3)',
+ )
parser.add_argument(
- '-t', '--title',
+ '-t',
+ '--title',
metavar='TITLE',
type=str,
dest='title',
default='BEINC title',
- help='BEINC title (default: "BEINC title"')
+ help='BEINC title (default: "BEINC title"',
+ )
parser.add_argument(
- '-v', '--version',
+ '-v',
+ '--version',
action='version',
version=f'%(prog)s {__version__}',
- help='display program-version and exit')
+ help='display program-version and exit',
+ )
args = parser.parse_args(inargs)
if args.socket_timeout:
socket.setdefaulttimeout(args.socket_timeout)
args.password = fetch_password(args.password)
try:
- context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
- context.verify_mode = ssl.CERT_NONE
+ context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
if args.cert:
context.verify_mode = ssl.CERT_REQUIRED
context.load_verify_locations(cafile=os.path.expanduser(args.cert))
context.check_hostname = bool(not args.disable_hostname_check)
+ else:
+ context.check_hostname = False
+ context.verify_mode = ssl.CERT_NONE
if args.ciphers:
context.set_ciphers(args.ciphers)
if args.pull:
diff --git a/beinc_pull.py b/beinc_pull.py
index c99cde8..67b6e63 100755
--- a/beinc_pull.py
+++ b/beinc_pull.py
@@ -2,7 +2,7 @@
# -*- coding: utf-8 -*-
# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0
-# Copyright (C) 2013-2020 Simeon Simeonov
+# Copyright (C) 2013-2022 Simeon Simeonov
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -38,7 +38,7 @@ except ImportError:
__author__ = 'Simeon Simeonov'
-__version__ = '4.1'
+__version__ = '4.2'
__license__ = 'GPL3'
@@ -65,7 +65,7 @@ def fetch_password(args_password):
sys.exit(errno.EACCES)
elif os.path.isfile(args_password):
try:
- with io.open(args_password, 'r') as fp:
+ with io.open(args_password, 'r', encoding='utf-8') as fp:
passwd = fp.readline()
if passwd.strip():
return passwd.strip()
@@ -93,11 +93,14 @@ def display_notification(args, title, message):
raise Exception(
'Could not load "pynotify".'
'Please install "pynotify" or use a different osd-system!'
- 'Terminating...')
+ 'Terminating...'
+ )
if not pynotify.init('BEINC Notify'):
raise Exception('There was a problem with libnotify')
- notification_obj = pynotify.Notification(summary=title,
- message=message)
+ notification_obj = pynotify.Notification(
+ summary=title,
+ message=message,
+ )
notification_obj.timeout = 1000 * args.osd_timeout
notification_obj.set_category('im.received')
notification_obj.show()
@@ -116,35 +119,37 @@ def pull_notifications(scheduler, args):
:type args: argparse.Namespace
"""
try:
- context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
- context.verify_mode = ssl.CERT_NONE
+ context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
if args.cert:
context.verify_mode = ssl.CERT_REQUIRED
context.load_verify_locations(cafile=os.path.expanduser(args.cert))
context.check_hostname = bool(not args.disable_hostname_check)
+ else:
+ context.check_hostname = False
+ context.verify_mode = ssl.CERT_NONE
if args.ciphers:
context.set_ciphers(args.ciphers)
response = urllib.request.urlopen(
args.url,
data=urllib.parse.urlencode(
- (
- ('resource_name', args.rname),
- ('password', args.password)
- )).encode('utf-8'),
+ (('resource_name', args.rname), ('password', args.password))
+ ).encode('utf-8'),
timeout=args.socket_timeout,
- context=context)
+ context=context,
+ )
response_dict = json.loads(response.read().decode('utf-8'))
if response.code != 200:
raise socket.error(response_dict.get('message', ''))
for entry in response_dict['data']['messages']:
- display_notification(args,
- entry.get('title', ''),
- entry.get('message', ''))
+ display_notification(
+ args,
+ entry.get('title', ''),
+ entry.get('message', ''),
+ )
response.close()
- scheduler.enter(args.frequency,
- 1,
- pull_notifications,
- (scheduler, args))
+ scheduler.enter(
+ args.frequency, 1, pull_notifications, (scheduler, args)
+ )
except ssl.SSLError as e:
eprint(f'BEINC SSL/TLS error: {e}')
sys.exit(errno.EPERM)
@@ -159,88 +164,105 @@ def pull_notifications(scheduler, args):
def main(inargs=None):
"""main entry"""
parser = argparse.ArgumentParser(
- description='The following options are available')
+ description='The following options are available'
+ )
parser.add_argument(
'url',
metavar='URL',
type=str,
- help='BEINC server destination URL')
+ help='BEINC server destination URL',
+ )
parser.add_argument(
- '-c', '--cert-file',
+ '-c',
+ '--cert-file',
metavar='FILE',
type=str,
dest='cert',
default='',
help='CA-cert to check the server-cert against '
- '(default: Check disabled)')
+ '(default: Check disabled)',
+ )
parser.add_argument(
'--ciphers',
metavar='CIPHERS',
type=str,
dest='ciphers',
default='',
- help='Preferred ciphers list (default: auto)')
+ help='Preferred ciphers list (default: auto)',
+ )
parser.add_argument(
'--disable-hostname-check',
action='store_true',
dest='disable_hostname_check',
default=False,
- help='Do not check whether server cert matches server hostname')
+ help='Do not check whether server cert matches server hostname',
+ )
parser.add_argument(
- '-f', '--frequency',
+ '-f',
+ '--frequency',
metavar='SECONDS',
type=int,
dest='frequency',
default=10,
- help='Pulling frequency in seconds (default: 10)')
+ help='Pulling frequency in seconds (default: 10)',
+ )
parser.add_argument(
- '-n', '--resource-name',
+ '-n',
+ '--resource-name',
metavar='NAME',
type=str,
dest='rname',
required=True,
- help='The name of the BEINC-resource on the remote server')
+ help='The name of the BEINC-resource on the remote server',
+ )
parser.add_argument(
- '-o', '--osd-system',
+ '-o',
+ '--osd-system',
metavar='SYSTEM',
type=str,
dest='osd_sys',
default='pynotify',
- help='BEINC osd-system: "pynotify" (default)')
+ help='BEINC osd-system: "pynotify" (default)',
+ )
parser.add_argument(
- '-p', '--password',
+ '-p',
+ '--password',
metavar='PASSWORD[FILE]',
type=str,
dest='password',
default='',
help='BEINC taget-password / text-file containing the target password'
- ' (default & recommended: prompt for passwd)')
+ ' (default & recommended: prompt for passwd)',
+ )
parser.add_argument(
- '-T', '--socket-timeout',
+ '-T',
+ '--socket-timeout',
metavar='SECONDS',
type=int,
dest='socket_timeout',
default=3,
- help='Socket timeout in seconds (0=Python default) (default: 3)')
+ help='Socket timeout in seconds (0=Python default) (default: 3)',
+ )
parser.add_argument(
- '-t', '--osd-timeout',
+ '-t',
+ '--osd-timeout',
metavar='SECONDS',
type=int,
dest='osd_timeout',
default=5,
- help='OSD timeout (default: 5)')
+ help='OSD timeout (default: 5)',
+ )
parser.add_argument(
- '-v', '--version',
+ '-v',
+ '--version',
action='version',
version=f'%(prog)s {__version__}',
- help='display program-version and exit')
+ help='display program-version and exit',
+ )
args = parser.parse_args(inargs)
args.password = fetch_password(args.password)
scheduler = sched.scheduler(time.time, time.sleep)
- scheduler.enter(args.frequency,
- 1,
- pull_notifications,
- (scheduler, args))
+ scheduler.enter(args.frequency, 1, pull_notifications, (scheduler, args))
scheduler.run()
diff --git a/beinc_server.py b/beinc_server.py
index 8957e73..77aa35e 100755
--- a/beinc_server.py
+++ b/beinc_server.py
@@ -2,7 +2,7 @@
# -*- coding: utf-8 -*-
# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0
-# Copyright (C) 2013-2020 Simeon Simeonov
+# Copyright (C) 2013-2022 Simeon Simeonov
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -21,15 +21,15 @@
import argparse
import cgi
import errno
+import io
import json
import logging
import os
import ssl
import sys
-
from functools import wraps
-from logging.config import fileConfig
from http.server import BaseHTTPRequestHandler, HTTPServer
+from logging.config import fileConfig
try:
import notify2 as pynotify
@@ -38,7 +38,7 @@ except ImportError:
__author__ = 'Simeon Simeonov'
-__version__ = '4.1'
+__version__ = '4.2'
__license__ = 'GPL3'
@@ -71,6 +71,7 @@ def eprint(*arg, **kwargs):
def beinc_login_required(method):
"""Decorator for checking login credentials"""
+
@wraps(method)
def wrapper(self, data, *arg, **kwargs):
if data.get('resource_name') is None:
@@ -84,6 +85,7 @@ def beinc_login_required(method):
if not instance.password_match(data.get('password')):
raise BEINCError401('Wrong instance or password')
return method(self, data, *arg, **kwargs)
+
return wrapper
@@ -105,19 +107,24 @@ class BEINCInstance:
self._queue_size = 0 # disable queueing
if pynotify is None:
eprint('This server does not possess pynotify capability')
- eprint(f'Remove the instance {self._name} or define it with '
- f'"osd_system": "none" or other '
- f'available backend')
+ eprint(
+ f'Remove the instance {self._name} or define it with '
+ f'"osd_system": "none" or other '
+ f'available backend'
+ )
sys.exit(errno.EPERM)
try:
self._osd_notification = pynotify.Notification(' ')
self._osd_notification.timeout = 1000 * int(
- instance_dict.get('osd_timeout', 5))
+ instance_dict.get('osd_timeout', 5)
+ )
self._osd_notification.set_category('im.received')
self._osd_type = BEINC_OSD_TYPE_PYNOTIFY
except Exception as e:
- eprint(f'Unable to set up a pynotify notification object '
- f'for "{self._name}" ({e})')
+ eprint(
+ f'Unable to set up a pynotify notification object '
+ f'for "{self._name}" ({e})'
+ )
sys.exit(errno.EPERM)
@property
@@ -195,6 +202,7 @@ class BEINCInstance:
class BEINCCustomHandler(BaseHTTPRequestHandler):
"""Custom handler"""
+
def do_POST(self):
"""Handle POST requests"""
if self.path.strip('/') not in ('beinc/push', 'beinc/pull'):
@@ -203,14 +211,18 @@ class BEINCCustomHandler(BaseHTTPRequestHandler):
form = cgi.FieldStorage(
fp=self.rfile,
headers=self.headers,
- environ={'REQUEST_METHOD': 'POST',
- 'CONTENT_TYPE': self.headers['Content-Type']})
+ environ={
+ 'REQUEST_METHOD': 'POST',
+ 'CONTENT_TYPE': self.headers['Content-Type'],
+ },
+ )
# extract all known fields
POST_data = dict(
resource_name=form.getvalue('resource_name'),
password=form.getvalue('password'),
title=form.getvalue('title', ''),
- message=form.getvalue('message', ''))
+ message=form.getvalue('message', ''),
+ )
try:
result = {}
if self.path.strip('/') == 'beinc/push':
@@ -249,10 +261,11 @@ class BEINCCustomHandler(BaseHTTPRequestHandler):
instance = self.server.instances[data.get('resource_name')]
try:
if not instance.queueable:
- raise BEINCError405(
- 'This instance does not support queuing')
- return {'message': 'OK. Fetched.',
- 'data': {'messages': instance.get_queue()}}
+ raise BEINCError405('This instance does not support queuing')
+ return {
+ 'message': 'OK. Fetched.',
+ 'data': {'messages': instance.get_queue()},
+ }
except Exception as e:
self._generate_json_error(500, str(e))
@@ -270,9 +283,9 @@ class BEINCCustomHandler(BaseHTTPRequestHandler):
self.send_header('Content-type', 'application/json; charset=utf-8')
self.end_headers()
msg = {'code': code, 'message': message, 'data': {}}
- self.wfile.write(json.dumps(msg,
- sort_keys=True,
- indent=4).encode('utf-8'))
+ self.wfile.write(
+ json.dumps(msg, sort_keys=True, indent=4).encode('utf-8')
+ )
def _render_to_JSON_response(self, context):
"""
@@ -284,13 +297,14 @@ class BEINCCustomHandler(BaseHTTPRequestHandler):
self.send_response(200)
self.send_header('Content-type', 'application/json; charset=utf-8')
self.end_headers()
- self.wfile.write(json.dumps(response,
- sort_keys=True,
- indent=4).encode('utf-8'))
+ self.wfile.write(
+ json.dumps(response, sort_keys=True, indent=4).encode('utf-8')
+ )
class BEINCNotifyServer(HTTPServer):
"""BEINCNotifyServer class"""
+
def __init__(self, *arg, **kwargs):
"""Default constructor"""
super().__init__(*arg, **kwargs)
@@ -319,9 +333,7 @@ class BEINCNotifyServer(HTTPServer):
self._instances[instance['name']] = BEINCInstance(instance)
logger.info('Instance %s added', instance['name'])
except Exception as e:
- eprint('Unable to create instance "{0}": {1}'.format(
- instance['name'],
- e))
+ eprint(f"Unable to create instance \"{instance['name']}\": {e}")
sys.exit(1)
@property
@@ -334,51 +346,66 @@ class BEINCNotifyServer(HTTPServer):
if __name__ == '__main__':
parser = argparse.ArgumentParser(
- description='The following options are available')
+ description='The following options are available'
+ )
parser.add_argument(
- '-H', '--hostname',
+ '-H',
+ '--hostname',
metavar='HOSTNAME',
type=str,
dest='hostname',
default='127.0.0.1',
- help='BEINC server IP / hostname (default: 127.0.0.1)')
+ help='BEINC server IP / hostname (default: 127.0.0.1)',
+ )
parser.add_argument(
- '-L', '--logger-name',
+ '-L',
+ '--logger-name',
metavar='NAME',
type=str,
dest='logger_name',
default='',
- help="BEINC logger name (default: 'beinc')")
+ help="BEINC logger name (default: 'beinc')",
+ )
parser.add_argument(
- '-l', '--logger-config',
+ '-l',
+ '--logger-config',
metavar='CONFIG',
type=str,
dest='logger_config',
default=os.path.expanduser('~/.beinc_server_logger.ini'),
- help=('BEINC logger config (.ini) '
- '(default: ~/.beinc_server_logger.ini)'))
+ help=(
+ 'BEINC logger config (.ini) '
+ '(default: ~/.beinc_server_logger.ini)'
+ ),
+ )
parser.add_argument(
- '-p', '--port',
+ '-p',
+ '--port',
metavar='PORT',
type=int,
dest='port',
default=9998,
- help='BEINC server port (default: 9998)')
+ help='BEINC server port (default: 9998)',
+ )
parser.add_argument(
- '-f', '--config-file',
+ '-f',
+ '--config-file',
metavar='FILE',
type=str,
default=os.path.expanduser('~/.beinc_server.json'),
dest='config_file',
- help='BEINC config file (default: ~/.beinc_server.json)')
+ help='BEINC config file (default: ~/.beinc_server.json)',
+ )
parser.add_argument(
- '-v', '--version',
+ '-v',
+ '--version',
action='version',
version=f'%(prog)s {__version__}',
- help='Display program-version and exit')
+ help='Display program-version and exit',
+ )
args = parser.parse_args()
try:
- with open(args.config_file, 'r') as fp:
+ with io.open(args.config_file, 'r', encoding='utf-8') as fp:
config_dict = json.load(fp)
except Exception as e:
eprint(f'Unable to parse {args.config_file}: {e}')
@@ -390,35 +417,42 @@ if __name__ == '__main__':
else:
logging.basicConfig(
format='%(asctime)s - %(levelname)s - %(message)s',
- level=logging.DEBUG)
+ level=logging.DEBUG,
+ )
logger = logging.getLogger('beinc')
logger.info('BEINC starting. Loading config...')
if config_dict.get('config_version') != BEINC_CURRENT_CONFIG_VERSION:
eprint(
- 'WARNING: The version of the config-file: {0} ({1}) '
- 'does not correspond to the latest version supported '
- 'by this program ({2})\nCheck beinc_config_sample.json '
- 'for the newest features!'.format(
- args.config_file,
- config_dict.get('config_version', 'Not set'),
- BEINC_CURRENT_CONFIG_VERSION))
+ f'WARNING: The version of the config-file: {args.config_file} '
+ f'({config_dict.get("config_version", "Not set")}) does not '
+ 'correspond to the latest version supported by this program '
+ f'({BEINC_CURRENT_CONFIG_VERSION})\n'
+ 'Check beinc_config_sample.json for the newest features!'
+ )
ssl_certificate = config_dict['server']['general'].get(
- 'ssl_certificate')
+ 'ssl_certificate'
+ )
ssl_private_key = config_dict['server']['general'].get(
- 'ssl_private_key')
+ 'ssl_private_key'
+ )
ssl_acceptable_ciphers_str = config_dict['server']['general'].get(
- 'ssl_ciphers')
- beinc_server = BEINCNotifyServer((args.hostname, args.port),
- BEINCCustomHandler)
+ 'ssl_ciphers'
+ )
+ beinc_server = BEINCNotifyServer(
+ (args.hostname, args.port),
+ BEINCCustomHandler,
+ )
beinc_server.set_config(config_dict)
if ssl_certificate and ssl_private_key:
- beinc_server.socket = ssl.wrap_socket(
- beinc_server.socket,
- keyfile=ssl_private_key,
- certfile=ssl_certificate,
- server_side=True,
- ssl_version=ssl.PROTOCOL_TLSv1_2,
- ciphers=ssl_acceptable_ciphers_str)
+ context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+ context.load_cert_chain(
+ certfile=ssl_certificate, keyfile=ssl_private_key
+ )
+ if ssl_acceptable_ciphers_str is not None:
+ context.set_ciphers(ssl_acceptable_ciphers_str)
+ beinc_server.socket = context.wrap_socket(
+ beinc_server.socket, server_side=True
+ )
logger.info('Done!')
beinc_server.serve_forever()
except KeyboardInterrupt:
diff --git a/beinc_weechat.py b/beinc_weechat.py
index 75af86a..2288b74 100644
--- a/beinc_weechat.py
+++ b/beinc_weechat.py
@@ -1,7 +1,7 @@
# -*- coding: utf-8 -*-
# Blackmore's Enhanced IRC-Notification Collection (BEINC) v4.0
-# Copyright (C) 2013-2020 Simeon Simeonov
+# Copyright (C) 2013-2022 Simeon Simeonov
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
@@ -17,6 +17,7 @@
# along with this program. If not, see .
"""BEINC client for Weechat"""
import datetime
+import io
import json
import os
import socket
@@ -26,9 +27,8 @@ import urllib.request
import weechat
-
__author__ = 'Simeon Simeonov'
-__version__ = '4.1'
+__version__ = '4.2'
__license__ = 'GPL3'
@@ -61,38 +61,46 @@ class WeechatTarget:
if self._url == '':
raise Exception('"target_url" not defined for target')
self._password = target_dict.get('target_password', '')
- self._pm_title_template = target_dict.get('pm_title_template',
- '%s @ %S')
- self._pm_message_template = target_dict.get('pm_message_template',
- '%m')
- self._cm_title_template = target_dict.get('cm_title_template',
- '%c @ %S')
- self._cm_message_template = target_dict.get('cm_message_template',
- '%s -> %m')
- self._nm_title_template = target_dict.get('nm_title_template',
- '%c @ %S')
- self._nm_message_template = target_dict.get('nm_message_template',
- '%s -> %m')
+ self._pm_title_template = target_dict.get(
+ 'pm_title_template', '%s @ %S'
+ )
+ self._pm_message_template = target_dict.get(
+ 'pm_message_template', '%m'
+ )
+ self._cm_title_template = target_dict.get(
+ 'cm_title_template', '%c @ %S'
+ )
+ self._cm_message_template = target_dict.get(
+ 'cm_message_template', '%s -> %m'
+ )
+ self._nm_title_template = target_dict.get(
+ 'nm_title_template', '%c @ %S'
+ )
+ self._nm_message_template = target_dict.get(
+ 'nm_message_template', '%s -> %m'
+ )
self._chans = set(target_dict.get('channel_list', []))
self._nicks = set(target_dict.get('nick_list', []))
- self._chan_messages_policy = int(target_dict.get(
- 'channel_messages_policy',
- BEINC_POLICY_LIST_ONLY))
- self._priv_messages_policy = int(target_dict.get(
- 'private_messages_policy',
- BEINC_POLICY_ALL))
- self._notifications_policy = int(target_dict.get(
- 'notifications_policy',
- BEINC_POLICY_ALL))
+ self._chan_messages_policy = int(
+ target_dict.get('channel_messages_policy', BEINC_POLICY_LIST_ONLY)
+ )
+ self._priv_messages_policy = int(
+ target_dict.get('private_messages_policy', BEINC_POLICY_ALL)
+ )
+ self._notifications_policy = int(
+ target_dict.get('notifications_policy', BEINC_POLICY_ALL)
+ )
self._cert_file = target_dict.get('target_cert_file')
- self._timestamp_format = target_dict.get('target_timestamp_format',
- '%H:%M:%S')
+ self._timestamp_format = target_dict.get(
+ 'target_timestamp_format', '%H:%M:%S'
+ )
self._debug = bool(target_dict.get('debug', False))
self._enabled = bool(target_dict.get('enabled', True))
self._socket_timeout = int(target_dict.get('socket_timeout', 3))
self._ssl_ciphers = target_dict.get('ssl_ciphers', '')
self._disable_hostname_check = bool(
- target_dict.get('disable-hostname-check', False))
+ target_dict.get('disable-hostname-check', False)
+ )
self._ssl_version = target_dict.get('ssl_version', 'auto')
self._last_message = None # datetime.datetime instance
self._context = None
@@ -143,26 +151,22 @@ class WeechatTarget:
last_message = 'never'
if self._last_message is not None:
last_message = self._last_message.strftime('%Y-%m-%d %H:%M:%S')
- return ('name: {0}\nurl: {1}\nenabled: {2}\nchannel_list: {3}\n'
- 'nick_list: {4}\nchannel_messages_policy: {5}\n'
- 'private_messages_policy: {6}\nnotifications_policy: {7}\n'
- 'last message: {8}\nsocket timeout: {9}\nssl-version: {10}\n'
- 'ciphers: {11}\ndisable hostname check: {12}\n'
- 'debug: {13}\n\n'.format(
- self._name,
- self._url,
- 'yes' if self._enabled else 'no',
- ', '.join(self._chans),
- ', '.join(self._nicks),
- self._chan_messages_policy,
- self._priv_messages_policy,
- self._notifications_policy,
- last_message,
- self._socket_timeout,
- self._ssl_version,
- self._ssl_ciphers or 'auto',
- 'yes' if self._disable_hostname_check else 'no',
- 'yes' if self._debug else 'no'))
+ return (
+ f'name: {self._name}\nurl: {self._url}\n'
+ f"enabled: {'yes' if self._enabled else 'no'}\n"
+ f"channel_list: {', '.join(self._chans)}\n"
+ f"nick_list: {', '.join(self._nicks)}\n"
+ f'channel_messages_policy: {self._chan_messages_policy}\n'
+ f'private_messages_policy: {self._priv_messages_policy}\n'
+ f'notifications_policy: {self._notifications_policy}\n'
+ f'last message: {last_message}\n'
+ f'socket timeout: {self._socket_timeout}\n'
+ f'ssl-version: {self._ssl_version}\n'
+ f"ciphers: {self._ssl_ciphers or 'auto'}\n"
+ "disable hostname check: "
+ f"{'yes' if self._disable_hostname_check else 'no'}\n"
+ f"debug: {'yes' if self._debug else 'no'}\n\n"
+ )
def send_private_message_notification(self, values):
"""
@@ -172,20 +176,22 @@ class WeechatTarget:
:type value: dict
"""
try:
- title = self._fetch_formatted_str(self._pm_title_template,
- values)
+ title = self._fetch_formatted_str(self._pm_title_template, values)
message = self._fetch_formatted_str(
self._pm_message_template,
- values)
+ values,
+ )
if not self._send_beinc_message(title, message) and self._debug:
beinc_prnt(
f'BEINC DEBUG: send_private_message_notification-ERROR '
- f'for "{self._name}": _send_beinc_message -> False')
+ f'for "{self._name}": _send_beinc_message -> False'
+ )
except Exception as e:
if self._debug:
beinc_prnt(
f'BEINC DEBUG: send_private_message_notification-ERROR '
- f'for "{self._name}": {e}')
+ f'for "{self._name}": {e}'
+ )
def send_channel_message_notification(self, values):
"""
@@ -195,19 +201,21 @@ class WeechatTarget:
:type value: dict
"""
try:
- title = self._fetch_formatted_str(self._cm_title_template,
- values)
- message = self._fetch_formatted_str(self._cm_message_template,
- values)
+ title = self._fetch_formatted_str(self._cm_title_template, values)
+ message = self._fetch_formatted_str(
+ self._cm_message_template, values
+ )
if not self._send_beinc_message(title, message) and self._debug:
beinc_prnt(
f'BEINC DEBUG: send_channel_message_notification-ERROR '
- f'for "{self._name}": _send_beinc_message -> False')
+ f'for "{self._name}": _send_beinc_message -> False'
+ )
except Exception as e:
if self._debug:
beinc_prnt(
f'BEINC DEBUG: send_channel_message_notification-ERROR '
- f'for "{self._name}": {e}')
+ f'for "{self._name}": {e}'
+ )
def send_notify_message_notification(self, values):
"""
@@ -217,19 +225,21 @@ class WeechatTarget:
:type value: dict
"""
try:
- title = self._fetch_formatted_str(self._nm_title_template,
- values)
- message = self._fetch_formatted_str(self._nm_message_template,
- values)
+ title = self._fetch_formatted_str(self._nm_title_template, values)
+ message = self._fetch_formatted_str(
+ self._nm_message_template, values
+ )
if not self._send_beinc_message(title, message) and self._debug:
beinc_prnt(
f'BEINC DEBUG: send_notify_message_notification-ERROR '
- f'for "{self._name}": _send_beinc_message -> False')
+ f'for "{self._name}": _send_beinc_message -> False'
+ )
except Exception as e:
if self._debug:
beinc_prnt(
f'BEINC DEBUG: send_notify_message_notification-ERROR '
- f'for "{self._name}": {e}')
+ f'for "{self._name}": {e}'
+ )
def send_broadcast_notification(self, message):
"""
@@ -244,26 +254,30 @@ class WeechatTarget:
if not self._send_beinc_message(title, message) and self._debug:
beinc_prnt(
f'BEINC DEBUG: send_broadcast_notification-ERROR '
- f'for "{self._name}": _send_beinc_message -> False')
+ f'for "{self._name}": _send_beinc_message -> False'
+ )
except Exception as e:
if self._debug:
beinc_prnt(
f'BEINC DEBUG: send_broadcast_notification-ERROR '
- f'for "{self._name}": {e}')
+ f'for "{self._name}": {e}'
+ )
def _context_setup(self):
"""Sets up the SSL context"""
if self._context is not None:
return True
try:
- context = ssl.SSLContext(ssl.PROTOCOL_TLSv1_2)
- context.verify_mode = ssl.CERT_NONE
+ context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
if self._cert_file:
context.verify_mode = ssl.CERT_REQUIRED
- context.load_verify_locations(cafile=os.path.expanduser(
- self._cert_file))
- context.check_hostname = bool(
- not self._disable_hostname_check)
+ context.load_verify_locations(
+ cafile=os.path.expanduser(self._cert_file)
+ )
+ context.check_hostname = bool(not self._disable_hostname_check)
+ else:
+ context.check_hostname = False
+ context.verify_mode = ssl.CERT_NONE
if self._ssl_ciphers and self._ssl_ciphers != 'auto':
context.set_ciphers(self._ssl_ciphers)
self._context = context
@@ -292,13 +306,15 @@ class WeechatTarget:
:rtype: str
"""
timestamp = datetime.datetime.now().strftime(self._timestamp_format)
- replacements = {'%S': values['server'],
- '%s': values['source_nick'],
- '%c': values['channel'],
- '%m': values['message'],
- '%t': timestamp,
- '%p': 'BEINC',
- '%n': values['own_nick']}
+ replacements = {
+ '%S': values['server'],
+ '%s': values['source_nick'],
+ '%c': values['channel'],
+ '%m': values['message'],
+ '%t': timestamp,
+ '%p': 'BEINC',
+ '%n': values['own_nick'],
+ }
for key, value in replacements.items():
template = template.replace(key, value)
return template
@@ -327,16 +343,20 @@ class WeechatTarget:
('resource_name', self._name),
('password', self._password),
('title', title),
- ('message', message)
- )).encode('utf-8'),
+ ('message', message),
+ )
+ ).encode('utf-8'),
timeout=self._socket_timeout,
- context=self._context)
+ context=self._context,
+ )
response_dict = json.loads(response.read().decode('utf-8'))
if response.code != 200:
raise socket.error(response_dict.get('message', ''))
if self._debug:
- beinc_prnt('BEINC DEBUG: Server responded: {0}'.format(
- response_dict.get('message')))
+ beinc_prnt(
+ "BEINC DEBUG: Server responded: "
+ f"{response_dict.get('message')}"
+ )
self._last_message = datetime.datetime.now()
return True
except ssl.SSLError as e:
@@ -378,7 +398,7 @@ def beinc_cmd_target_handler(cmd_tokens):
if cmd_tokens[0] == 'list':
beinc_prnt('--- Globals ---')
for key, value in global_values.items():
- beinc_prnt('{key} -> {value}'.format(key=key, value=str(value)))
+ beinc_prnt(f'{key} -> {str(value)}')
beinc_prnt('--- Targets ---')
for target in target_list:
beinc_prnt(str(target))
@@ -430,8 +450,10 @@ def beinc_command(data, buffer_obj, args):
elif cmd_tokens[0] == 'target':
return beinc_cmd_target_handler(cmd_tokens[1:])
else:
- beinc_prnt('syntax: /beinc < on | off | reload |'
- ' broadcast | target >')
+ beinc_prnt(
+ 'syntax: /beinc < on | off | reload |'
+ ' broadcast | target >'
+ )
return weechat.WEECHAT_RC_OK
@@ -439,8 +461,9 @@ def beinc_privmsg_handler(data, signal, signal_data):
"""Callback function the *PRIVMSG* IRC messages hooked by Weechat"""
if not enabled:
return weechat.WEECHAT_RC_OK
- prvmsg_dict = weechat.info_get_hashtable('irc_message_parse',
- {'message': signal_data})
+ prvmsg_dict = weechat.info_get_hashtable(
+ 'irc_message_parse', {'message': signal_data}
+ )
# packing the privmsg handler values
ph_values = {}
ph_values['server'] = signal.split(',')[0]
@@ -448,7 +471,8 @@ def beinc_privmsg_handler(data, signal, signal_data):
ph_values['channel'] = prvmsg_dict['arguments'].split(':')[0].strip()
ph_values['source_nick'] = prvmsg_dict['nick']
ph_values['message'] = ':'.join(
- prvmsg_dict['arguments'].split(':')[1:]).strip()
+ prvmsg_dict['arguments'].split(':')[1:]
+ ).strip()
if ph_values['channel'] == ph_values['own_nick']:
# priv messages are handled here
if not global_values['global_private_messages_policy']:
@@ -458,10 +482,10 @@ def beinc_privmsg_handler(data, signal, signal_data):
continue
p_messages_policy = target.private_messages_policy
if p_messages_policy == BEINC_POLICY_ALL or (
- p_messages_policy == BEINC_POLICY_LIST_ONLY and
- '{0}.{1}'.format(
- ph_values['server'],
- ph_values['source_nick'].lower()) in target.nicks):
+ p_messages_policy == BEINC_POLICY_LIST_ONLY
+ and f"{ph_values['server']}.{ph_values['source_nick'].lower()}"
+ in target.nicks
+ ):
target.send_private_message_notification(ph_values)
elif ph_values['own_nick'].lower() in ph_values['message'].lower():
# notify messages are handled here
@@ -471,10 +495,9 @@ def beinc_privmsg_handler(data, signal, signal_data):
if not target.enabled:
continue
if target.notifications_policy == BEINC_POLICY_ALL or (
- target.notifications_policy == BEINC_POLICY_LIST_ONLY and
- '{0}.{1}'.format(
- ph_values['server'],
- ph_values['channel'].lower()) in target.chans
+ target.notifications_policy == BEINC_POLICY_LIST_ONLY
+ and f"{ph_values['server']}.{ph_values['channel'].lower()}"
+ in target.chans
):
target.send_notify_message_notification(ph_values)
elif global_values['global_channel_messages_policy']:
@@ -486,10 +509,10 @@ def beinc_privmsg_handler(data, signal, signal_data):
continue
c_messages_policy = target.channel_messages_policy
if c_messages_policy == BEINC_POLICY_ALL or (
- c_messages_policy == BEINC_POLICY_LIST_ONLY and
- '{0}.{1}'.format(
- ph_values['server'],
- ph_values['channel'].lower()) in target.chans):
+ c_messages_policy == BEINC_POLICY_LIST_ONLY
+ and f"{ph_values['server']}.{ph_values['channel'].lower()}"
+ in target.chans
+ ):
target.send_channel_message_notification(ph_values)
return weechat.WEECHAT_RC_OK
@@ -517,24 +540,28 @@ def beinc_init():
try:
beinc_config_file_str = os.path.join(
weechat.info_get('weechat_dir', ''),
- 'beinc_weechat.json')
+ 'beinc_weechat.json',
+ )
beinc_prnt(f'Parsing {beinc_config_file_str}...')
custom_error = 'load error'
- with open(beinc_config_file_str, 'r') as fp:
+ with io.open(beinc_config_file_str, 'r', encoding='utf-8') as fp:
config_dict = json.load(fp)
custom_error = 'target parse error'
global_values['use_current_buffer'] = bool(
- config_dict['irc_client'].get(
- 'use_current_buffer', False))
- if config_dict.get('config_version',
- 0) != BEINC_CURRENT_CONFIG_VERSION:
- beinc_prnt('WARNING: The version of the config-file: {0} ({1}) '
- 'does not correspond to the latest version supported '
- 'by this program ({2})\nCheck beinc_config_sample.json '
- 'for the newest features!'.format(
- beinc_config_file_str,
- config_dict.get('config_version', 0),
- BEINC_CURRENT_CONFIG_VERSION))
+ config_dict['irc_client'].get('use_current_buffer', False)
+ )
+ if (
+ config_dict.get('config_version', 0)
+ != BEINC_CURRENT_CONFIG_VERSION
+ ):
+ beinc_prnt(
+ "WARNING: The version of the config-file: "
+ f"{beinc_config_file_str} "
+ f"({config_dict.get('config_version', 0)}) "
+ "does not correspond to the latest version supported "
+ f"by this program ({BEINC_CURRENT_CONFIG_VERSION})\n"
+ "Check beinc_config_sample.json for the newest features!"
+ )
for target in config_dict['irc_client']['targets']:
try:
new_target = WeechatTarget(target)
@@ -551,8 +578,10 @@ def beinc_init():
beinc_prnt(f'BEINC target "{new_target.name}" added')
beinc_prnt('Done!')
except Exception as e:
- beinc_prnt(f'ERROR: unable to parse {beinc_config_file_str}: '
- f'{custom_error} - {e}\nBEINC is now disabled')
+ beinc_prnt(
+ f'ERROR: unable to parse {beinc_config_file_str}: '
+ f'{custom_error} - {e}\nBEINC is now disabled'
+ )
enabled = False
# do not return error / exit the script
# in order to give a smoother opportunity to fix a 'broken' config
@@ -567,19 +596,24 @@ weechat.register(
__license__,
'Blackmore\'s Extended IRC Notification Collection (Weechat Client)',
'',
- '')
+ '',
+)
version = weechat.info_get('version_number', '') or 0
if int(version) < 0x00040000:
weechat.prnt('', 'WeeChat version >= 0.4.0 is required to run beinc')
else:
- weechat.hook_command('beinc',
- 'BEINC command', ('< broadcast | on | off |'
- ' reload | target >'),
- ('Available target actions:\n'
- 'disable \nenable \nlist'),
- 'None',
- 'beinc_command',
- '')
+ weechat.hook_command(
+ 'beinc',
+ 'BEINC command',
+ '< broadcast | on | off | reload | target >',
+ (
+ 'Available target actions:\n'
+ 'disable \nenable \nlist'
+ ),
+ 'None',
+ 'beinc_command',
+ '',
+ )
weechat.hook_signal('*,irc_in2_privmsg', 'beinc_privmsg_handler', '')
beinc_init()
weechat.prnt('', 'beinc initiated!')
--
cgit v1.3