thecoffemaker/Slixfeed
1
0
Fork 0
forked from sch/Slixfeed
Code Pull requests Activity

Improve handling of configuration directories by allocating

Browse source 
a handler to each directory, instead of processing a function
to detect them.
This commit is contained in:
Schimon Jehudah, Adv. 2024-11-19 15:55:29 +02:00
parent 09bd838ad2
commit 79b80b1698
10 changed files with 109 additions and 45 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

1
slixfeed/assets/selector.toml
View file

@ -18,6 +18,7 @@ blacklist = [
"lana@redice.tv",
"larken@larkenrose.com",
"lee@oraclebroadcasting.com",
"mark@enclosedworld.com",
"mark@marksargent.com",
"nick@nightnationreview.com",
"oliver@postmarketos.org",

64
slixfeed/config.py
View file

@ -45,12 +45,76 @@ except:
logger = Logger(__name__)
class Cache:
def get_default_cache_directory():
"""
Determine the directory path where dbfile will be stored.
* If $XDG_DATA_HOME is defined, use it;
* else if $HOME exists, use it;
* else if the platform is Windows, use %APPDATA%;
* else use the current directory.
Returns
-------
str
Path to cache directory.
"""
# data_home = xdg.BaseDirectory.xdg_data_home
data_home = os.environ.get('XDG_CACHE_HOME')
if data_home is None:
if os.environ.get('HOME') is None:
if sys.platform == 'win32':
data_home = os.environ.get('APPDATA')
if data_home is None:
return os.path.abspath('.slixfeed/cache')
else:
return os.path.abspath('.slixfeed/cache')
else:
data_home = os.path.join(
os.environ.get('HOME'), '.cache'
)
return os.path.join(data_home, 'slixfeed')
# TODO Consider a class ConfigDefault for default values to be initiate at most
# basic level possible and a class ConfigJID for each JID (i.e. db_file) to be
# also initiated at same level or at least at event call, then check whether
# setting_jid.setting_key has value, otherwise resort to setting_default.setting_key.
class Config:
# TODO Write a similar function for file.
# NOTE the is a function of directory, noot file.
def get_default_config_directory():
"""
Determine the directory path where configuration will be stored.
* If $XDG_CONFIG_HOME is defined, use it;
* else if $HOME exists, use it;
* else if the platform is Windows, use %APPDATA%;
* else use the current directory.
Returns
-------
str
Path to configuration directory.
"""
# config_home = xdg.BaseDirectory.xdg_config_home
config_home = os.environ.get('XDG_CONFIG_HOME')
if config_home is None:
if os.environ.get('HOME') is None:
if sys.platform == 'win32':
config_home = os.environ.get('APPDATA')
if config_home is None:
return os.path.abspath('.')
else:
return os.path.abspath('.')
else:
config_home = os.path.join(
os.environ.get('HOME'), '.config'
)
return os.path.join(config_home, 'slixfeed')
def update_toml_file(filename, data):
with open(filename, 'w') as new_file:
content = tomli_w.dumps(data)

13
slixfeed/syndication.py
View file

@ -44,16 +44,15 @@ logger = Logger(__name__)
class Feed:
# NOTE Consider removal of MD (and any other option HTML and XBEL)
def export_feeds(jid_bare, ext):
def export_feeds(dir_cache, jid_bare, ext):
function_name = sys._getframe().f_code.co_name
logger.debug('{}: jid_bare: {}: ext: {}'.format(function_name, jid_bare, ext))
cache_dir = config.get_default_cache_directory()
if not os.path.isdir(cache_dir):
os.mkdir(cache_dir)
if not os.path.isdir(cache_dir + '/' + ext):
os.mkdir(cache_dir + '/' + ext)
if not os.path.isdir(dir_cache):
os.mkdir(dir_cache)
if not os.path.isdir(dir_cache + '/' + ext):
os.mkdir(dir_cache + '/' + ext)
filename = os.path.join(
cache_dir, ext, 'slixfeed_' + DateAndTime.timestamp() + '.' + ext)
dir_cache, ext, 'slixfeed_' + DateAndTime.timestamp() + '.' + ext)
db_file = config.get_pathname_to_database(jid_bare)
results = sqlite.get_feeds(db_file)
match ext:

4
slixfeed/version.py
View file

@ -1,2 +1,2 @@
__version__ = '0.1.100'
__version_info__ = (0, 1, 100)
__version__ = '0.1.101'
__version_info__ = (0, 1, 101)

9
slixfeed/xmpp/chat.py
View file

@ -225,14 +225,14 @@ class XmppChat:
response = ('Invalid. Enter command key '
'or command key & name')
case 'info':
entries = XmppCommands.print_info_list()
entries = XmppCommands.print_info_list(self)
response = ('Available command options:\n'
'```\n{}\n```\n'
'Usage: `info <option>`'
.format(entries))
case _ if command_lowercase.startswith('info'):
entry = command[5:].lower()
response = XmppCommands.print_info_specific(entry)
response = XmppCommands.print_info_specific(self, entry)
case _ if command_lowercase in ['greetings', 'hallo', 'hello',
'hey', 'hi', 'hola', 'holla',
'hollo']:
@ -355,7 +355,7 @@ class XmppChat:
XmppPresence.send(self, jid_bare, status_message,
status_type=status_type)
pathname, response = XmppCommands.export_feeds(
jid_bare, ext)
self.dir_cache, jid_bare, ext)
encrypt_omemo = Config.get_setting_value(self, jid_bare, 'omemo')
encrypted = True if encrypt_omemo else False
url = await XmppUpload.start(self, jid_bare, Path(pathname), encrypted=encrypted)
@ -788,11 +788,10 @@ class XmppChatAction:
news_digest = ''
# Send media
if self.omemo_present and encrypt_omemo:
cache_dir = config.get_default_cache_directory()
# if not media_url.startswith('data:'):
filename = media_url.split('/').pop().split('?')[0]
if not filename: breakpoint()
pathname = os.path.join(cache_dir, filename)
pathname = os.path.join(self.dir_cache, filename)
# http_response = await Http.response(media_url)
http_headers = await Http.fetch_headers(media_url)
if ('Content-Length' in http_headers and

28
slixfeed/xmpp/client.py
View file

@ -47,7 +47,7 @@ import slixmpp
# from lxml import etree
import slixfeed.config as config
from slixfeed.config import Config, Data
from slixfeed.config import Cache, Config, Data
import slixfeed.fetch as fetch
from slixfeed.log import Logger
import slixfeed.sqlite as sqlite
@ -117,6 +117,11 @@ class XmppClient(slixmpp.ClientXMPP):
# Handler for ping
self.task_ping_instance = {}
# Handlers for directories
self.dir_config = Config.get_default_config_directory()
self.dir_cache = Cache.get_default_cache_directory()
self.dir_data = Data.get_default_data_directory()
# Handler for default configuration
self.defaults = config.get_values('settings.toml')
# Handler for configuration
@ -2151,8 +2156,7 @@ class XmppClient(slixmpp.ClientXMPP):
.format(function_name, jid_full))
values = payload['values']
search_type = values['search_type']
config_dir = config.get_default_config_directory()
db_file = config_dir + '/feeds.sqlite'
db_file = os.path.join(self.dir_config, 'feeds.sqlite')
if os.path.isfile(db_file):
form = self['xep_0004'].make_form('form', 'Discover & Search')
match search_type:
@ -2207,8 +2211,7 @@ class XmppClient(slixmpp.ClientXMPP):
.format(function_name, jid_full))
values = payload['values']
category = values['category']
config_dir = config.get_default_config_directory()
db_file = config_dir + '/feeds.sqlite'
db_file = os.path.join(self.dir_config, 'feeds.sqlite')
form = self['xep_0004'].make_form('form', 'Discover & Search')
form['instructions'] = 'Browsing category "{}"'.format(category)
options = form.add_field(desc='Select a subscription to add.',
@ -2792,8 +2795,8 @@ class XmppClient(slixmpp.ClientXMPP):
label='About',
required=True,
value='about')
config_dir = config.get_default_config_directory()
with open(config_dir + '/' + 'about.toml', mode="rb") as information:
file_about = os.path.join(self.dir_config, 'about.toml')
with open(file_about, mode="rb") as information:
entries = tomllib.load(information)
for entry in entries:
label = entries[entry][0]['title']
@ -2810,8 +2813,8 @@ class XmppClient(slixmpp.ClientXMPP):
function_name = sys._getframe().f_code.co_name
logger.debug('{}: jid_full: {}'
.format(function_name, jid_full))
config_dir = config.get_default_config_directory()
with open(config_dir + '/' + 'about.toml', mode="rb") as information:
file_about = os.path.join(self.dir_config, 'about.toml')
with open(file_about, mode="rb") as information:
entries = tomllib.load(information)
entry_key = payload['values']['option']
# case 'terms':
@ -2887,9 +2890,8 @@ class XmppClient(slixmpp.ClientXMPP):
function_name = sys._getframe().f_code.co_name
logger.debug('{}: jid_full: {}'
.format(function_name, jid_full))
config_dir = config.get_default_config_directory()
with open(config_dir + '/' + 'commands.toml', mode="rb") as commands:
file_cmds = os.path.join(self.dir_config, 'commands.toml')
with open(file_cmds, mode="rb") as commands:
cmds = tomllib.load(commands)
form = self['xep_0004'].make_form('result', 'Manual')
@ -2980,7 +2982,7 @@ class XmppClient(slixmpp.ClientXMPP):
# form['instructions'] = ('✅️ Feeds have been exported')
exts = values['filetype']
for ext in exts:
filename = Feed.export_feeds(jid_bare, ext)
filename = Feed.export_feeds(self.dir_cache, jid_bare, ext)
encrypt_omemo = Config.get_setting_value(self, jid_bare, 'omemo')
encrypted = True if encrypt_omemo else False
url = await XmppUpload.start(

22
slixfeed/xmpp/commands.py
View file

@ -77,17 +77,17 @@ class XmppCommands:
return message
def print_info_list():
config_dir = config.get_default_config_directory()
with open(config_dir + '/' + 'information.toml', mode="rb") as information:
def print_info_list(self):
file_info = os.path.join(self.dir_config, 'information.toml')
with open(file_info, mode="rb") as information:
result = tomllib.load(information)
message = '\n'.join(result)
return message
def print_info_specific(entry):
config_dir = config.get_default_config_directory()
with open(config_dir + '/' + 'information.toml', mode="rb") as information:
def print_info_specific(self, entry):
file_info = os.path.join(self.dir_config, 'information.toml')
with open(file_info, mode="rb") as information:
entries = tomllib.load(information)
if entry in entries:
# command_list = '\n'.join(command_list)
@ -331,8 +331,8 @@ class XmppCommands:
await sqlite.set_filter_value(db_file, ['deny', val])
def export_feeds(jid_bare, ext):
pathname = Feed.export_feeds(jid_bare, ext)
def export_feeds(dir_cache, jid_bare, ext):
pathname = Feed.export_feeds(dir_cache, jid_bare, ext)
message = 'Feeds successfuly exported to {}.'.format(ext)
return pathname, message
@ -1091,8 +1091,7 @@ class XmppCommands:
def add_jid_to_selector(self, jid, list_type):
config_dir = config.get_default_config_directory()
filename = os.path.join(config_dir, 'selector.toml')
filename = os.path.join(self.dir_config, 'selector.toml')
match list_type:
case 'blacklist':
list_type_list = self.blacklist
@ -1108,8 +1107,7 @@ class XmppCommands:
def del_jid_from_selector(self, jid, list_type):
config_dir = config.get_default_config_directory()
filename = os.path.join(config_dir, 'selector.toml')
filename = os.path.join(self.dir_config, 'selector.toml')
match list_type:
case 'blacklist':
list_type_list = self.blacklist

6
slixfeed/xmpp/ipc.py
View file

@ -177,7 +177,7 @@ class XmppIpcServer:
ext = command[7:]
if ext in ('md', 'opml'):
filename, result = XmppCommands.export_feeds(
self, jid_bare, ext)
self.dir_cache, jid_bare, ext)
response = result + ' : ' + filename
else:
response = 'Unsupported filetype. Try: md or opml'
@ -204,10 +204,10 @@ class XmppIpcServer:
response = await XmppCommands.import_opml(
self, db_file, jid_bare, command)
case 'info':
response = XmppCommands.print_info_list()
response = XmppCommands.print_info_list(self)
case _ if command.startswith('info'):
entry = command[5:].lower()
response = XmppCommands.print_info_specific(entry)
response = XmppCommands.print_info_specific(self, entry)
case 'pubsub list':
response = await XmppCommands.pubsub_list(
self, jid_bare)

3
slixfeed/xmpp/muc.py
View file

@ -81,6 +81,9 @@ class XmppMuc:
e.presence['error']['code'] == '403'):
logger.warning('{} is banned from {}'.format(self.alias, jid))
result = 'ban'
elif e.condition == 'conflict':
logger.warning(e.presence['error']['text'])
result = 'conflict'
else:
result = 'error'
return result

4
slixfeed/xmpp/profile.py
View file

@ -57,9 +57,7 @@ async def update(self):
async def set_avatar(self):
config_dir = config.get_default_config_directory()
if not os.path.isdir(config_dir):
config_dir = '/usr/share/slixfeed/'
config_dir = self.dir_config
filename = glob.glob(config_dir + '/image.*')
if not filename and os.path.isdir('/usr/share/slixfeed/'):
# filename = '/usr/share/slixfeed/image.svg'