indymedia
/
epicyon
mirror of https://gitlab.com/bashrc2/epicyon
7
4
Fork 1
Code Issues 14 Releases Wiki Activity

Replace file operations with functions

main
bashrc 2026-04-25 22:36:46 +01:00
parent 9ae07571a7
commit 583a1ae83f
11 changed files with 124 additions and 234 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
conversation.py
View File

@ -21,6 +21,8 @@ from utils import resembles_url
from keys import get_instance_actor_key
from session import get_json
from session import get_json_valid
from data import save_string
from data import append_string
def _get_conversation_filename(base_dir: str, nickname: str, domain: str,
@ -63,23 +65,17 @@ def update_conversation(base_dir: str, nickname: str, domain: str,
return False
post_id = remove_id_ending(post_json_object['object']['id'])
if not os.path.isfile(conversation_filename):
try:
with open(conversation_filename, 'w+',
encoding='utf-8') as fp_conv:
fp_conv.write(post_id + '\n')
return True
except OSError:
print('EX: update_conversation ' +
'unable to write to ' + conversation_filename)
if save_string(post_id + '\n', conversation_filename,
'EX: update_conversation ' +
'unable to write to ' +
conversation_filename):
return True
elif not text_in_file(post_id + '\n', conversation_filename):
try:
with open(conversation_filename, 'a+',
encoding='utf-8') as fp_conv:
fp_conv.write(post_id + '\n')
return True
except OSError:
print('EX: update_conversation 2 ' +
'unable to write to ' + conversation_filename)
if append_string(post_id + '\n', conversation_filename,
'EX: update_conversation 2 ' +
'unable to write to ' +
conversation_filename):
return True
return False
@ -97,12 +93,8 @@ def mute_conversation(base_dir: str, nickname: str, domain: str,
return
if os.path.isfile(conversation_filename + '.muted'):
return
try:
with open(conversation_filename + '.muted', 'w+',
encoding='utf-8') as fp_conv:
fp_conv.write('\n')
except OSError:
print('EX: unable to write mute ' + conversation_filename)
save_string('\n', conversation_filename + '.muted',
'EX: unable to write mute ' + conversation_filename)
def unmute_conversation(base_dir: str, nickname: str, domain: str,

21
crawlers.py
View File

@ -19,6 +19,8 @@ from blocking import get_bsky_domains_list
from blocking import get_nostr_domains_list
from blocking import update_blocked_cache
from blocking import is_blocked_domain
from data import load_string
from data import save_string
default_user_agent_blocks = [
'fedilist', 'ncsc scan', 'fedifetcher'
@ -64,13 +66,9 @@ def load_known_web_bots(base_dir: str) -> []:
known_bots_filename = data_dir(base_dir) + '/knownBots.txt'
if not os.path.isfile(known_bots_filename):
return []
crawlers_str = None
try:
with open(known_bots_filename, 'r', encoding='utf-8') as fp_crawlers:
crawlers_str = fp_crawlers.read()
except OSError:
print('EX: unable to load web bots from ' +
known_bots_filename)
crawlers_str = load_string(known_bots_filename,
'EX: unable to load web bots from ' +
known_bots_filename)
if not crawlers_str:
return []
known_bots: list[str] = []
@ -93,12 +91,9 @@ def _save_known_web_bots(base_dir: str, known_bots: []) -> bool:
known_bots_str: str = ''
for crawler in known_bots:
known_bots_str += crawler.strip() + '\n'
try:
with open(known_bots_filename, 'w+', encoding='utf-8') as fp_crawlers:
fp_crawlers.write(known_bots_str)
except OSError:
print("EX: unable to save known web bots to " +
known_bots_filename)
if not save_string(known_bots_str, known_bots_filename,
"EX: unable to save known web bots to " +
known_bots_filename):
return False
return True

22
daemon.py
View File

@ -105,6 +105,7 @@ from daemon_utils import has_accept
from daemon_utils import is_authorized
from poison import load_dictionary
from poison import load_2grams
from data import load_string
class PubServer(BaseHTTPRequestHandler):
@ -583,14 +584,9 @@ def load_tokens(base_dir: str, tokens_dict: {}, tokens_lookup: {}) -> None:
if not os.path.isfile(token_filename):
continue
nickname = handle.split('@')[0]
token = None
try:
with open(token_filename, 'r',
encoding='utf-8') as fp_tok:
token = fp_tok.read()
except OSError as ex:
print('WARN: Unable to read token for ' +
nickname + ' ' + str(ex))
token = load_string(token_filename,
'WARN: Unable to read token for ' +
nickname + ' [ex]')
if not token:
continue
tokens_dict[nickname] = token
@ -753,13 +749,9 @@ def run_daemon(accounts_data_dir: str,
robots_txt_filename = data_dir(base_dir) + '/robots.txt'
httpd.robots_txt = None
if os.path.isfile(robots_txt_filename):
new_robots_txt = ''
try:
with open(robots_txt_filename, 'r',
encoding='utf-8') as fp_robots:
new_robots_txt = fp_robots.read()
except OSError:
print('EX: error reading 1 ' + robots_txt_filename)
new_robots_txt = \
load_string(robots_txt_filename,
'EX: error reading 1 ' + robots_txt_filename)
if new_robots_txt:
httpd.robots_txt = new_robots_txt

20
daemon_get.py
View File

@ -220,6 +220,7 @@ from daemon_get_links import edit_links2
from daemon_get_login import redirect_to_login_screen
from daemon_get_login import show_login_screen
from poison import html_poisoned
from data import load_string
# Blogs can be longer, so don't show many per page
MAX_POSTS_IN_BLOGS_FEED = 4
@ -6255,12 +6256,8 @@ def daemon_http_get(self) -> None:
return
if os.path.isfile(filename):
content = None
try:
with open(filename, 'r', encoding='utf-8') as fp_rfile:
content = fp_rfile.read()
except OSError:
print('EX: unable to read file ' + filename)
content = load_string(filename,
'EX: unable to read file ' + filename)
if content:
try:
content_json = json.loads(content)
@ -6500,13 +6497,10 @@ def _get_ontology(self, calling_domain: str,
if ontology_str.endswith('.json'):
ontology_file_type = 'application/ld+json'
if os.path.isfile(ontology_filename):
ontology_file = None
try:
with open(ontology_filename, 'r',
encoding='utf-8') as fp_ont:
ontology_file = fp_ont.read()
except OSError:
print('EX: unable to read ontology ' + ontology_filename)
ontology_file = \
load_string(ontology_filename,
'EX: unable to read ontology ' +
ontology_filename)
if ontology_file:
ontology_file = \
ontology_file.replace('static.datafoodconsortium.org',

15
daemon_get_images.py
View File

@ -29,6 +29,7 @@ from daemon_utils import etag_exists
from fitnessFunctions import fitness_performance
from person import save_person_qrcode
from lxmf import save_lxmf_qrcode
from data import load_string
def show_avatar_or_banner(self, referer_domain: str, path: str,
@ -442,16 +443,12 @@ def show_media(self, path: str, base_dir: str,
last_modified_time.strftime('%a, %d %b %Y %H:%M:%S GMT')
if media_filename.endswith('.vtt'):
media_transcript = None
try:
with open(media_filename, 'r',
encoding='utf-8') as fp_vtt:
media_transcript = fp_vtt.read()
media_file_type = 'text/vtt; charset=utf-8'
except OSError:
print('EX: unable to read media binary ' +
media_filename)
media_transcript = \
load_string(media_filename,
'EX: unable to read media binary ' +
media_filename)
if media_transcript:
media_file_type = 'text/vtt; charset=utf-8'
media_transcript = media_transcript.encode('utf-8')
set_headers_etag(self, media_filename, media_file_type,
media_transcript, None,

19
daemon_get_post.py
View File

@ -49,6 +49,7 @@ from fitnessFunctions import fitness_performance
from securemode import secure_mode
from context import get_individual_post_context
from conversation import convthread_id_to_conversation_tag
from data import load_string
def _show_post_from_file(self, post_filename: str, liked_by: str,
@ -308,12 +309,9 @@ def show_individual_post(self, ssml_getreq: bool, authorized: bool,
if not os.path.isfile(ssml_filename):
http_404(self, 74)
return True
ssml_str = None
try:
with open(ssml_filename, 'r', encoding='utf-8') as fp_ssml:
ssml_str = fp_ssml.read()
except OSError:
print('EX: unable to read ssml file ' + ssml_filename)
ssml_str = load_string(ssml_filename,
'EX: unable to read ssml file ' +
ssml_filename)
if ssml_str:
msg = ssml_str.encode('utf-8')
msglen = len(msg)
@ -661,12 +659,9 @@ def show_individual_at_post(self, ssml_getreq: bool, authorized: bool,
if not os.path.isfile(ssml_filename):
http_404(self, 67)
return True
ssml_str = None
try:
with open(ssml_filename, 'r', encoding='utf-8') as fp_ssml:
ssml_str = fp_ssml.read()
except OSError:
print('EX: unable to read ssml file 2 ' + ssml_filename)
ssml_str = load_string(ssml_filename,
'EX: unable to read ssml file 2 ' +
ssml_filename)
if ssml_str:
msg = ssml_str.encode('utf-8')
msglen = len(msg)

24
daemon_head.py
View File

@ -23,6 +23,8 @@ from media import path_is_video
from media import path_is_audio
from daemon_utils import get_user_agent
from daemon_utils import log_epicyon_instances
from data import load_string
from data import save_string
def daemon_http_head(self) -> None:
@ -118,13 +120,11 @@ def daemon_http_head(self) -> None:
last_modified_time.strftime(time_format_str)
media_tag_filename = media_filename + '.etag'
if os.path.isfile(media_tag_filename):
try:
with open(media_tag_filename, 'r',
encoding='utf-8') as fp_efile:
etag = fp_efile.read()
except OSError:
print('EX: do_HEAD unable to read ' +
media_tag_filename)
etag_str = load_string(media_tag_filename,
'EX: do_HEAD unable to read ' +
media_tag_filename)
if etag_str:
etag = etag_str
else:
media_binary = None
try:
@ -135,13 +135,9 @@ def daemon_http_head(self) -> None:
media_filename)
if media_binary:
etag = md5(media_binary).hexdigest() # nosec
try:
with open(media_tag_filename, 'w+',
encoding='utf-8') as fp_efile:
fp_efile.write(etag)
except OSError:
print('EX: do_HEAD unable to write ' +
media_tag_filename)
save_string(etag, media_tag_filename,
'EX: do_HEAD unable to write ' +
media_tag_filename)
else:
http_404(self, 151)
return

48
daemon_post_links.py
View File

@ -18,6 +18,7 @@ from utils import get_nickname_from_actor
from utils import get_config_param
from httpheaders import redirect_headers
from content import extract_text_fields_in_post
from data import save_string
def _links_update_edited(fields: {}, links_filename: str) -> None:
@ -30,24 +31,16 @@ def _links_update_edited(fields: {}, links_filename: str) -> None:
if not links_str.endswith('\n'):
links_str += '\n'
links_str += fields['newColLink'] + '\n'
try:
with open(links_filename, 'w+',
encoding='utf-8') as fp_links:
fp_links.write(links_str)
except OSError:
print('EX: _links_update unable to write ' +
links_filename)
save_string(links_str, links_filename,
'EX: _links_update unable to write ' +
links_filename)
else:
if fields.get('newColLink'):
# the text area is empty but there is a new link added
links_str = fields['newColLink'] + '\n'
try:
with open(links_filename, 'w+',
encoding='utf-8') as fp_links:
fp_links.write(links_str)
except OSError:
print('EX: _links_update unable to write ' +
links_filename)
save_string(links_str, links_filename,
'EX: _links_update unable to write ' +
links_filename)
else:
if os.path.isfile(links_filename):
try:
@ -65,13 +58,9 @@ def _links_update_about(fields: {}, allow_local_network_access: bool,
about_str = fields['editedAbout']
if not dangerous_markup(about_str,
allow_local_network_access, []):
try:
with open(about_filename, 'w+',
encoding='utf-8') as fp_about:
fp_about.write(about_str)
except OSError:
print('EX: unable to write about ' +
about_filename)
save_string(about_str, about_filename,
'EX: unable to write about ' +
about_filename)
else:
if os.path.isfile(about_filename):
try:
@ -89,11 +78,8 @@ def _links_update_tos(fields: {}, allow_local_network_access: bool,
tos_str = fields['editedTOS']
if not dangerous_markup(tos_str,
allow_local_network_access, []):
try:
with open(tos_filename, 'w+', encoding='utf-8') as fp_tos:
fp_tos.write(tos_str)
except OSError:
print('EX: unable to write TOS ' + tos_filename)
save_string(tos_str, tos_filename,
'EX: unable to write TOS ' + tos_filename)
else:
if os.path.isfile(tos_filename):
try:
@ -109,13 +95,9 @@ def _links_update_sepcification(fields: {},
"""
if fields.get('editedSpecification'):
specification_str = fields['editedSpecification']
try:
with open(specification_filename, 'w+',
encoding='utf-8') as fp_specification:
fp_specification.write(specification_str)
except OSError:
print('EX: unable to write specification ' +
specification_filename)
save_string(specification_str, specification_filename,
'EX: unable to write specification ' +
specification_filename)
else:
if os.path.isfile(specification_filename):
try:

34
daemon_post_login.py
View File

@ -36,6 +36,8 @@ from flags import is_system_account
from person import person_upgrade_actor
from person import activate_account2
from person import register_account
from data import load_string
from data import save_string
def post_login_screen(self, calling_domain: str, cookie: str,
@ -180,21 +182,15 @@ def post_login_screen(self, calling_domain: str, cookie: str,
acct_dir(base_dir, login_nickname, domain) + '/.salt'
salt = create_password(32)
if os.path.isfile(salt_filename):
try:
with open(salt_filename, 'r',
encoding='utf-8') as fp_salt:
salt = fp_salt.read()
except OSError as ex:
print('EX: Unable to read salt for ' +
login_nickname + ' ' + str(ex))
salt_str = load_string(salt_filename,
'EX: Unable to read salt for ' +
login_nickname + ' [ex]')
if salt_str:
salt = salt_str
else:
try:
with open(salt_filename, 'w+',
encoding='utf-8') as fp_salt:
fp_salt.write(salt)
except OSError as ex:
print('EX: Unable to save salt for ' +
login_nickname + ' ' + str(ex))
save_string(salt, salt_filename,
'EX: Unable to save salt for ' +
login_nickname + ' [ex]')
token_text = login_nickname + login_password + salt
token = sha256(token_text.encode('utf-8')).hexdigest()
@ -202,13 +198,9 @@ def post_login_screen(self, calling_domain: str, cookie: str,
login_handle = login_nickname + '@' + domain
token_filename = \
data_dir(base_dir) + '/' + login_handle + '/.token'
try:
with open(token_filename, 'w+',
encoding='utf-8') as fp_tok:
fp_tok.write(token)
except OSError as ex:
print('EX: Unable to save token for ' +
login_nickname + ' ' + str(ex))
save_string(token, token_filename,
'EX: Unable to save token for ' +
login_nickname + ' [ex]')
dir_str = data_dir(base_dir)
person_upgrade_actor(base_dir, None,

80
daemon_post_newswire.py
View File

@ -26,6 +26,7 @@ from timeFunctions import date_from_string_format
from httpheaders import redirect_headers
from content import extract_text_fields_in_post
from content import load_dogwhistles
from data import save_string
def newswire_update(self, calling_domain: str, cookie: str,
@ -116,22 +117,14 @@ def newswire_update(self, calling_domain: str, cookie: str,
if not newswire_str.endswith('\n'):
newswire_str += '\n'
newswire_str += fields['newNewswireFeed'] + '\n'
try:
with open(newswire_filename, 'w+',
encoding='utf-8') as fp_news:
fp_news.write(newswire_str)
except OSError:
print('EX: unable to write ' + newswire_filename)
save_string(newswire_str, newswire_filename,
'EX: unable to write ' + newswire_filename)
else:
if fields.get('newNewswireFeed'):
# the text area is empty but there is a new feed added
newswire_str = fields['newNewswireFeed'] + '\n'
try:
with open(newswire_filename, 'w+',
encoding='utf-8') as fp_news:
fp_news.write(newswire_str)
except OSError:
print('EX: unable to write ' + newswire_filename)
save_string(newswire_str, newswire_filename,
'EX: unable to write ' + newswire_filename)
else:
# text area has been cleared and there is no new feed
if os.path.isfile(newswire_filename):
@ -145,13 +138,10 @@ def newswire_update(self, calling_domain: str, cookie: str,
filter_newswire_filename = \
data_dir(base_dir) + '/' + 'news@' + domain + '/filters.txt'
if fields.get('filteredWordsNewswire'):
try:
with open(filter_newswire_filename, 'w+',
encoding='utf-8') as fp_filter:
fp_filter.write(fields['filteredWordsNewswire'])
except OSError:
print('EX: newswire_update unable to write ' +
filter_newswire_filename)
save_string(fields['filteredWordsNewswire'],
filter_newswire_filename,
'EX: newswire_update unable to write ' +
filter_newswire_filename)
else:
if os.path.isfile(filter_newswire_filename):
try:
@ -163,36 +153,26 @@ def newswire_update(self, calling_domain: str, cookie: str,
# save dogwhistle words list
dogwhistles_filename = data_dir(base_dir) + '/dogwhistles.txt'
if fields.get('dogwhistleWords'):
try:
with open(dogwhistles_filename, 'w+',
encoding='utf-8') as fp_dogwhistles:
fp_dogwhistles.write(fields['dogwhistleWords'])
except OSError:
print('EX: newswire_update unable to write 2 ' +
dogwhistles_filename)
save_string(fields['dogwhistleWords'],
dogwhistles_filename,
'EX: newswire_update unable to write 2 ' +
dogwhistles_filename)
self.server.dogwhistles = \
load_dogwhistles(dogwhistles_filename)
else:
# save an empty file
try:
with open(dogwhistles_filename, 'w+',
encoding='utf-8') as fp_dogwhistles:
fp_dogwhistles.write('')
except OSError:
print('EX: newswire_update unable unable to write 3 ' +
dogwhistles_filename)
save_string('', dogwhistles_filename,
'EX: newswire_update unable unable to write 3 ' +
dogwhistles_filename)
self.server.dogwhistles = {}
# save news tagging rules
hashtag_rules_filename = data_dir(base_dir) + '/hashtagrules.txt'
if fields.get('hashtagRulesList'):
try:
with open(hashtag_rules_filename, 'w+',
encoding='utf-8') as fp_rules:
fp_rules.write(fields['hashtagRulesList'])
except OSError:
print('EX: newswire_update unable to write 4 ' +
hashtag_rules_filename)
save_string(fields['hashtagRulesList'],
hashtag_rules_filename,
'EX: newswire_update unable to write 4 ' +
hashtag_rules_filename)
else:
if os.path.isfile(hashtag_rules_filename):
try:
@ -206,13 +186,9 @@ def newswire_update(self, calling_domain: str, cookie: str,
newswire_trusted = fields['trustedNewswire']
if not newswire_trusted.endswith('\n'):
newswire_trusted += '\n'
try:
with open(newswire_tusted_filename, 'w+',
encoding='utf-8') as fp_trust:
fp_trust.write(newswire_trusted)
except OSError:
print('EX: newswire_update unable to write 5 ' +
newswire_tusted_filename)
save_string(newswire_trusted, newswire_tusted_filename,
'EX: newswire_update unable to write 5 ' +
newswire_tusted_filename)
else:
if os.path.isfile(newswire_tusted_filename):
try:
@ -315,13 +291,9 @@ def citations_update(self, calling_domain: str, cookie: str,
citations_str += citation_date + '\n'
# save citations dates, so that they can be added when
# reloading the newblog screen
try:
with open(citations_filename, 'w+',
encoding='utf-8') as fp_cit:
fp_cit.write(citations_str)
except OSError:
print('EX: citations_update unable to write ' +
citations_filename)
save_string(citations_str, citations_filename,
'EX: citations_update unable to write ' +
citations_filename)
# redirect back to the default timeline
redirect_headers(self, actor_str + '/newblog',

39
daemon_post_person_options.py
View File

@ -49,6 +49,7 @@ from blocking import blocked_quote_toots_remove
from notifyOnPost import add_notify_on_post
from notifyOnPost import remove_notify_on_post
from flags import is_moderator
from data import save_string
def _person_options_page_number(options_confirm_params: str) -> int:
@ -611,16 +612,9 @@ def _person_options_post_to_news(self, options_confirm_params: str,
else:
if os.path.isdir(account_dir):
nw_filename = newswire_blocked_filename
nw_written = False
try:
with open(nw_filename, 'w+',
encoding='utf-8') as fp_no:
fp_no.write('\n')
nw_written = True
except OSError as ex:
print('EX: _person_options_post_to_news unable ' +
'to write ' + nw_filename + ' ' + str(ex))
if nw_written:
if save_string('\n', nw_filename,
'EX: _person_options_post_to_news unable ' +
'to write ' + nw_filename + ' [ex]'):
refresh_newswire(base_dir)
users_path_str = \
users_path + '/' + default_timeline + \
@ -671,17 +665,10 @@ def _person_options_post_to_features(self, options_confirm_params: str,
else:
if os.path.isdir(account_dir):
feat_filename = features_blocked_filename
feat_written = False
try:
with open(feat_filename, 'w+',
encoding='utf-8') as fp_no:
fp_no.write('\n')
feat_written = True
except OSError as ex:
print('EX: _person_options_post_to_features ' +
'unable to write ' + feat_filename +
' ' + str(ex))
if feat_written:
if save_string('\n', feat_filename,
'EX: _person_options_post_to_features ' +
'unable to write ' + feat_filename +
' [ex]'):
refresh_newswire(base_dir)
users_path_str = \
users_path + '/' + default_timeline + \
@ -731,13 +718,9 @@ def _person_options_mod_news(self, options_confirm_params: str,
else:
if os.path.isdir(account_dir):
nw_filename = newswire_mod_filename
try:
with open(nw_filename, 'w+',
encoding='utf-8') as fp_mod:
fp_mod.write('\n')
except OSError:
print('EX: _person_options_mod_news ' +
'unable to write ' + nw_filename)
save_string('\n', nw_filename,
'EX: _person_options_mod_news ' +
'unable to write ' + nw_filename)
users_path_str = \
users_path + '/' + default_timeline + \
'?page=' + str(page_number)