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

Snake case

merge-requests/30/head
Bob Mottram 2022-01-02 15:16:47 +00:00
parent 7da2381aa0
commit 27c91490b0
1 changed files with 179 additions and 177 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

356
happening.py
View File

@ -20,33 +20,34 @@ from utils import has_object_dict
from utils import acct_dir from utils import acct_dir
def _valid_uuid(testUuid: str, version: int): def _valid_uuid(test_uuid: str, version: int):
"""Check if uuid_to_test is a valid UUID """Check if uuid_to_test is a valid UUID
""" """
try: try:
uuid_obj = UUID(testUuid, version=version) uuid_obj = UUID(test_uuid, version=version)
except ValueError: except ValueError:
return False return False
return str(uuid_obj) == testUuid return str(uuid_obj) == test_uuid
def _remove_event_from_timeline(eventId: str, tlEventsFilename: str) -> None: def _remove_event_from_timeline(event_id: str,
tl_events_filename: str) -> None:
"""Removes the given event Id from the timeline """Removes the given event Id from the timeline
""" """
if eventId + '\n' not in open(tlEventsFilename).read(): if event_id + '\n' not in open(tl_events_filename).read():
return return
with open(tlEventsFilename, 'r') as fp: with open(tl_events_filename, 'r') as fp_tl:
eventsTimeline = fp.read().replace(eventId + '\n', '') events_timeline = fp_tl.read().replace(event_id + '\n', '')
try: try:
with open(tlEventsFilename, 'w+') as fp2: with open(tl_events_filename, 'w+') as fp2:
fp2.write(eventsTimeline) fp2.write(events_timeline)
except OSError: except OSError:
print('EX: ERROR: unable to save events timeline') print('EX: ERROR: unable to save events timeline')
def save_event_post(base_dir: str, handle: str, post_id: str, def save_event_post(base_dir: str, handle: str, post_id: str,
eventJson: {}) -> bool: event_json: {}) -> bool:
"""Saves an event to the calendar and/or the events timeline """Saves an event to the calendar and/or the events timeline
If an event has extra fields, as per Mobilizon, If an event has extra fields, as per Mobilizon,
Then it is saved as a separate entity and added to the Then it is saved as a separate entity and added to the
@ -57,96 +58,96 @@ def save_event_post(base_dir: str, handle: str, post_id: str,
if not os.path.isdir(base_dir + '/accounts/' + handle): if not os.path.isdir(base_dir + '/accounts/' + handle):
print('WARN: Account does not exist at ' + print('WARN: Account does not exist at ' +
base_dir + '/accounts/' + handle) base_dir + '/accounts/' + handle)
calendarPath = base_dir + '/accounts/' + handle + '/calendar' calendar_path = base_dir + '/accounts/' + handle + '/calendar'
if not os.path.isdir(calendarPath): if not os.path.isdir(calendar_path):
os.mkdir(calendarPath) os.mkdir(calendar_path)
# get the year, month and day from the event # get the year, month and day from the event
eventTime = datetime.strptime(eventJson['startTime'], event_time = datetime.strptime(event_json['startTime'],
"%Y-%m-%dT%H:%M:%S%z") "%Y-%m-%dT%H:%M:%S%z")
eventYear = int(eventTime.strftime("%Y")) event_year = int(event_time.strftime("%Y"))
if eventYear < 2020 or eventYear >= 2100: if event_year < 2020 or event_year >= 2100:
return False return False
eventMonthNumber = int(eventTime.strftime("%m")) event_month_number = int(event_time.strftime("%m"))
if eventMonthNumber < 1 or eventMonthNumber > 12: if event_month_number < 1 or event_month_number > 12:
return False return False
eventDayOfMonth = int(eventTime.strftime("%d")) event_day_of_month = int(event_time.strftime("%d"))
if eventDayOfMonth < 1 or eventDayOfMonth > 31: if event_day_of_month < 1 or event_day_of_month > 31:
return False return False
if eventJson.get('name') and eventJson.get('actor') and \ if event_json.get('name') and event_json.get('actor') and \
eventJson.get('uuid') and eventJson.get('content'): event_json.get('uuid') and event_json.get('content'):
if not _valid_uuid(eventJson['uuid'], 4): if not _valid_uuid(event_json['uuid'], 4):
return False return False
print('Mobilizon type event') print('Mobilizon type event')
# if this is a full description of an event then save it # if this is a full description of an event then save it
# as a separate json file # as a separate json file
eventsPath = base_dir + '/accounts/' + handle + '/events' events_path = base_dir + '/accounts/' + handle + '/events'
if not os.path.isdir(eventsPath): if not os.path.isdir(events_path):
os.mkdir(eventsPath) os.mkdir(events_path)
eventsYearPath = \ events_year_path = \
base_dir + '/accounts/' + handle + '/events/' + str(eventYear) base_dir + '/accounts/' + handle + '/events/' + str(event_year)
if not os.path.isdir(eventsYearPath): if not os.path.isdir(events_year_path):
os.mkdir(eventsYearPath) os.mkdir(events_year_path)
eventId = str(eventYear) + '-' + eventTime.strftime("%m") + '-' + \ event_id = str(event_year) + '-' + event_time.strftime("%m") + '-' + \
eventTime.strftime("%d") + '_' + eventJson['uuid'] event_time.strftime("%d") + '_' + event_json['uuid']
eventFilename = eventsYearPath + '/' + eventId + '.json' event_filename = events_year_path + '/' + event_id + '.json'
save_json(eventJson, eventFilename) save_json(event_json, event_filename)
# save to the events timeline # save to the events timeline
tlEventsFilename = base_dir + '/accounts/' + handle + '/events.txt' tl_events_filename = base_dir + '/accounts/' + handle + '/events.txt'
if os.path.isfile(tlEventsFilename): if os.path.isfile(tl_events_filename):
_remove_event_from_timeline(eventId, tlEventsFilename) _remove_event_from_timeline(event_id, tl_events_filename)
try: try:
with open(tlEventsFilename, 'r+') as tlEventsFile: with open(tl_events_filename, 'r+') as tl_events_file:
content = tlEventsFile.read() content = tl_events_file.read()
if eventId + '\n' not in content: if event_id + '\n' not in content:
tlEventsFile.seek(0, 0) tl_events_file.seek(0, 0)
tlEventsFile.write(eventId + '\n' + content) tl_events_file.write(event_id + '\n' + content)
except OSError as ex: except OSError as ex:
print('EX: Failed to write entry to events file ' + print('EX: Failed to write entry to events file ' +
tlEventsFilename + ' ' + str(ex)) tl_events_filename + ' ' + str(ex))
return False return False
else: else:
try: try:
with open(tlEventsFilename, 'w+') as tlEventsFile: with open(tl_events_filename, 'w+') as tl_events_file:
tlEventsFile.write(eventId + '\n') tl_events_file.write(event_id + '\n')
except OSError: except OSError:
print('EX: unable to write ' + tlEventsFilename) print('EX: unable to write ' + tl_events_filename)
# create a directory for the calendar year # create a directory for the calendar year
if not os.path.isdir(calendarPath + '/' + str(eventYear)): if not os.path.isdir(calendar_path + '/' + str(event_year)):
os.mkdir(calendarPath + '/' + str(eventYear)) os.mkdir(calendar_path + '/' + str(event_year))
# calendar month file containing event post Ids # calendar month file containing event post Ids
calendarFilename = calendarPath + '/' + str(eventYear) + \ calendar_filename = calendar_path + '/' + str(event_year) + \
'/' + str(eventMonthNumber) + '.txt' '/' + str(event_month_number) + '.txt'
# Does this event post already exist within the calendar month? # Does this event post already exist within the calendar month?
if os.path.isfile(calendarFilename): if os.path.isfile(calendar_filename):
if post_id in open(calendarFilename).read(): if post_id in open(calendar_filename).read():
# Event post already exists # Event post already exists
return False return False
# append the post Id to the file for the calendar month # append the post Id to the file for the calendar month
try: try:
with open(calendarFilename, 'a+') as calendarFile: with open(calendar_filename, 'a+') as calendar_file:
calendarFile.write(post_id + '\n') calendar_file.write(post_id + '\n')
except OSError: except OSError:
print('EX: unable to append ' + calendarFilename) print('EX: unable to append ' + calendar_filename)
# create a file which will trigger a notification that # create a file which will trigger a notification that
# a new event has been added # a new event has been added
calNotifyFilename = base_dir + '/accounts/' + handle + '/.newCalendar' cal_notify_filename = base_dir + '/accounts/' + handle + '/.newCalendar'
notifyStr = \ notify_str = \
'/calendar?year=' + str(eventYear) + '?month=' + \ '/calendar?year=' + str(event_year) + '?month=' + \
str(eventMonthNumber) + '?day=' + str(eventDayOfMonth) str(event_month_number) + '?day=' + str(event_day_of_month)
try: try:
with open(calNotifyFilename, 'w+') as calendarNotificationFile: with open(cal_notify_filename, 'w+') as cal_file:
calendarNotificationFile.write(notifyStr) cal_file.write(notify_str)
except OSError: except OSError:
print('EX: unable to write ' + calNotifyFilename) print('EX: unable to write ' + cal_notify_filename)
return False return False
return True return True
@ -185,39 +186,39 @@ def get_todays_events(base_dir: str, nickname: str, domain: str,
else: else:
year = currYear year = currYear
if not currMonthNumber: if not currMonthNumber:
monthNumber = now.month month_number = now.month
else: else:
monthNumber = currMonthNumber month_number = currMonthNumber
if not currDayOfMonth: if not currDayOfMonth:
dayNumber = now.day day_number = now.day
else: else:
dayNumber = currDayOfMonth day_number = currDayOfMonth
calendarFilename = \ calendar_filename = \
acct_dir(base_dir, nickname, domain) + \ acct_dir(base_dir, nickname, domain) + \
'/calendar/' + str(year) + '/' + str(monthNumber) + '.txt' '/calendar/' + str(year) + '/' + str(month_number) + '.txt'
events = {} events = {}
if not os.path.isfile(calendarFilename): if not os.path.isfile(calendar_filename):
return events return events
calendarPostIds = [] calendar_post_ids = []
recreateEventsFile = False recreate_events_file = False
with open(calendarFilename, 'r') as eventsFile: with open(calendar_filename, 'r') as events_file:
for post_id in eventsFile: for post_id in events_file:
post_id = post_id.replace('\n', '').replace('\r', '') post_id = post_id.replace('\n', '').replace('\r', '')
post_filename = locate_post(base_dir, nickname, domain, post_id) post_filename = locate_post(base_dir, nickname, domain, post_id)
if not post_filename: if not post_filename:
recreateEventsFile = True recreate_events_file = True
continue continue
post_json_object = load_json(post_filename) post_json_object = load_json(post_filename)
if not _is_happening_post(post_json_object): if not _is_happening_post(post_json_object):
continue continue
publicEvent = is_public_post(post_json_object) public_event = is_public_post(post_json_object)
postEvent = [] post_event = []
dayOfMonth = None day_of_month = None
for tag in post_json_object['object']['tag']: for tag in post_json_object['object']['tag']:
if not _is_happening_event(tag): if not _is_happening_event(tag):
continue continue
@ -226,38 +227,38 @@ def get_todays_events(base_dir: str, nickname: str, domain: str,
# tag is an event # tag is an event
if not tag.get('startTime'): if not tag.get('startTime'):
continue continue
eventTime = \ event_time = \
datetime.strptime(tag['startTime'], datetime.strptime(tag['startTime'],
"%Y-%m-%dT%H:%M:%S%z") "%Y-%m-%dT%H:%M:%S%z")
if int(eventTime.strftime("%Y")) == year and \ if int(event_time.strftime("%Y")) == year and \
int(eventTime.strftime("%m")) == monthNumber and \ int(event_time.strftime("%m")) == month_number and \
int(eventTime.strftime("%d")) == dayNumber: int(event_time.strftime("%d")) == day_number:
dayOfMonth = str(int(eventTime.strftime("%d"))) day_of_month = str(int(event_time.strftime("%d")))
if '#statuses#' in post_id: if '#statuses#' in post_id:
# link to the id so that the event can be # link to the id so that the event can be
# easily deleted # easily deleted
tag['post_id'] = post_id.split('#statuses#')[1] tag['post_id'] = post_id.split('#statuses#')[1]
tag['sender'] = post_id.split('#statuses#')[0] tag['sender'] = post_id.split('#statuses#')[0]
tag['sender'] = tag['sender'].replace('#', '/') tag['sender'] = tag['sender'].replace('#', '/')
tag['public'] = publicEvent tag['public'] = public_event
postEvent.append(tag) post_event.append(tag)
else: else:
# tag is a place # tag is a place
postEvent.append(tag) post_event.append(tag)
if postEvent and dayOfMonth: if post_event and day_of_month:
calendarPostIds.append(post_id) calendar_post_ids.append(post_id)
if not events.get(dayOfMonth): if not events.get(day_of_month):
events[dayOfMonth] = [] events[day_of_month] = []
events[dayOfMonth].append(postEvent) events[day_of_month].append(post_event)
# if some posts have been deleted then regenerate the calendar file # if some posts have been deleted then regenerate the calendar file
if recreateEventsFile: if recreate_events_file:
try: try:
with open(calendarFilename, 'w+') as calendarFile: with open(calendar_filename, 'w+') as calendar_file:
for post_id in calendarPostIds: for post_id in calendar_post_ids:
calendarFile.write(post_id + '\n') calendar_file.write(post_id + '\n')
except OSError: except OSError:
print('EX: unable to write ' + calendarFilename) print('EX: unable to write ' + calendar_filename)
return events return events
@ -267,18 +268,18 @@ def day_events_check(base_dir: str, nickname: str, domain: str,
"""Are there calendar events for the given date? """Are there calendar events for the given date?
""" """
year = currDate.year year = currDate.year
monthNumber = currDate.month month_number = currDate.month
dayNumber = currDate.day day_number = currDate.day
calendarFilename = \ calendar_filename = \
acct_dir(base_dir, nickname, domain) + \ acct_dir(base_dir, nickname, domain) + \
'/calendar/' + str(year) + '/' + str(monthNumber) + '.txt' '/calendar/' + str(year) + '/' + str(month_number) + '.txt'
if not os.path.isfile(calendarFilename): if not os.path.isfile(calendar_filename):
return False return False
eventsExist = False events_exist = False
with open(calendarFilename, 'r') as eventsFile: with open(calendar_filename, 'r') as events_file:
for post_id in eventsFile: for post_id in events_file:
post_id = post_id.replace('\n', '').replace('\r', '') post_id = post_id.replace('\n', '').replace('\r', '')
post_filename = locate_post(base_dir, nickname, domain, post_id) post_filename = locate_post(base_dir, nickname, domain, post_id)
if not post_filename: if not post_filename:
@ -297,19 +298,19 @@ def day_events_check(base_dir: str, nickname: str, domain: str,
# tag is an event # tag is an event
if not tag.get('startTime'): if not tag.get('startTime'):
continue continue
eventTime = \ event_time = \
datetime.strptime(tag['startTime'], datetime.strptime(tag['startTime'],
"%Y-%m-%dT%H:%M:%S%z") "%Y-%m-%dT%H:%M:%S%z")
if int(eventTime.strftime("%d")) != dayNumber: if int(event_time.strftime("%d")) != day_number:
continue continue
if int(eventTime.strftime("%m")) != monthNumber: if int(event_time.strftime("%m")) != month_number:
continue continue
if int(eventTime.strftime("%Y")) != year: if int(event_time.strftime("%Y")) != year:
continue continue
eventsExist = True events_exist = True
break break
return eventsExist return events_exist
def get_this_weeks_events(base_dir: str, nickname: str, domain: str) -> {}: def get_this_weeks_events(base_dir: str, nickname: str, domain: str) -> {}:
@ -319,34 +320,34 @@ def get_this_weeks_events(base_dir: str, nickname: str, domain: str) -> {}:
Note: currently not used but could be with a weekly calendar screen Note: currently not used but could be with a weekly calendar screen
""" """
now = datetime.now() now = datetime.now()
endOfWeek = now + timedelta(7) end_of_week = now + timedelta(7)
year = now.year year = now.year
monthNumber = now.month month_number = now.month
calendarFilename = \ calendar_filename = \
acct_dir(base_dir, nickname, domain) + \ acct_dir(base_dir, nickname, domain) + \
'/calendar/' + str(year) + '/' + str(monthNumber) + '.txt' '/calendar/' + str(year) + '/' + str(month_number) + '.txt'
events = {} events = {}
if not os.path.isfile(calendarFilename): if not os.path.isfile(calendar_filename):
return events return events
calendarPostIds = [] calendar_post_ids = []
recreateEventsFile = False recreate_events_file = False
with open(calendarFilename, 'r') as eventsFile: with open(calendar_filename, 'r') as events_file:
for post_id in eventsFile: for post_id in events_file:
post_id = post_id.replace('\n', '').replace('\r', '') post_id = post_id.replace('\n', '').replace('\r', '')
post_filename = locate_post(base_dir, nickname, domain, post_id) post_filename = locate_post(base_dir, nickname, domain, post_id)
if not post_filename: if not post_filename:
recreateEventsFile = True recreate_events_file = True
continue continue
post_json_object = load_json(post_filename) post_json_object = load_json(post_filename)
if not _is_happening_post(post_json_object): if not _is_happening_post(post_json_object):
continue continue
postEvent = [] post_event = []
weekDayIndex = None week_day_index = None
for tag in post_json_object['object']['tag']: for tag in post_json_object['object']['tag']:
if not _is_happening_event(tag): if not _is_happening_event(tag):
continue continue
@ -355,63 +356,63 @@ def get_this_weeks_events(base_dir: str, nickname: str, domain: str) -> {}:
# tag is an event # tag is an event
if not tag.get('startTime'): if not tag.get('startTime'):
continue continue
eventTime = \ event_time = \
datetime.strptime(tag['startTime'], datetime.strptime(tag['startTime'],
"%Y-%m-%dT%H:%M:%S%z") "%Y-%m-%dT%H:%M:%S%z")
if eventTime >= now and eventTime <= endOfWeek: if event_time >= now and event_time <= end_of_week:
weekDayIndex = (eventTime - now).days() week_day_index = (event_time - now).days()
postEvent.append(tag) post_event.append(tag)
else: else:
# tag is a place # tag is a place
postEvent.append(tag) post_event.append(tag)
if postEvent and weekDayIndex: if post_event and week_day_index:
calendarPostIds.append(post_id) calendar_post_ids.append(post_id)
if not events.get(weekDayIndex): if not events.get(week_day_index):
events[weekDayIndex] = [] events[week_day_index] = []
events[weekDayIndex].append(postEvent) events[week_day_index].append(post_event)
# if some posts have been deleted then regenerate the calendar file # if some posts have been deleted then regenerate the calendar file
if recreateEventsFile: if recreate_events_file:
try: try:
with open(calendarFilename, 'w+') as calendarFile: with open(calendar_filename, 'w+') as calendar_file:
for post_id in calendarPostIds: for post_id in calendar_post_ids:
calendarFile.write(post_id + '\n') calendar_file.write(post_id + '\n')
except OSError: except OSError:
print('EX: unable to write ' + calendarFilename) print('EX: unable to write ' + calendar_filename)
return events return events
def get_calendar_events(base_dir: str, nickname: str, domain: str, def get_calendar_events(base_dir: str, nickname: str, domain: str,
year: int, monthNumber: int) -> {}: year: int, month_number: int) -> {}:
"""Retrieves calendar events """Retrieves calendar events
Returns a dictionary indexed by day number of lists containing Returns a dictionary indexed by day number of lists containing
Event and Place activities Event and Place activities
""" """
calendarFilename = \ calendar_filename = \
acct_dir(base_dir, nickname, domain) + \ acct_dir(base_dir, nickname, domain) + \
'/calendar/' + str(year) + '/' + str(monthNumber) + '.txt' '/calendar/' + str(year) + '/' + str(month_number) + '.txt'
events = {} events = {}
if not os.path.isfile(calendarFilename): if not os.path.isfile(calendar_filename):
return events return events
calendarPostIds = [] calendar_post_ids = []
recreateEventsFile = False recreate_events_file = False
with open(calendarFilename, 'r') as eventsFile: with open(calendar_filename, 'r') as events_file:
for post_id in eventsFile: for post_id in events_file:
post_id = post_id.replace('\n', '').replace('\r', '') post_id = post_id.replace('\n', '').replace('\r', '')
post_filename = locate_post(base_dir, nickname, domain, post_id) post_filename = locate_post(base_dir, nickname, domain, post_id)
if not post_filename: if not post_filename:
recreateEventsFile = True recreate_events_file = True
continue continue
post_json_object = load_json(post_filename) post_json_object = load_json(post_filename)
if not _is_happening_post(post_json_object): if not _is_happening_post(post_json_object):
continue continue
postEvent = [] post_event = []
dayOfMonth = None day_of_month = None
for tag in post_json_object['object']['tag']: for tag in post_json_object['object']['tag']:
if not _is_happening_event(tag): if not _is_happening_event(tag):
continue continue
@ -420,57 +421,58 @@ def get_calendar_events(base_dir: str, nickname: str, domain: str,
# tag is an event # tag is an event
if not tag.get('startTime'): if not tag.get('startTime'):
continue continue
eventTime = \ event_time = \
datetime.strptime(tag['startTime'], datetime.strptime(tag['startTime'],
"%Y-%m-%dT%H:%M:%S%z") "%Y-%m-%dT%H:%M:%S%z")
if int(eventTime.strftime("%Y")) == year and \ if int(event_time.strftime("%Y")) == year and \
int(eventTime.strftime("%m")) == monthNumber: int(event_time.strftime("%m")) == month_number:
dayOfMonth = str(int(eventTime.strftime("%d"))) day_of_month = str(int(event_time.strftime("%d")))
postEvent.append(tag) post_event.append(tag)
else: else:
# tag is a place # tag is a place
postEvent.append(tag) post_event.append(tag)
if postEvent and dayOfMonth: if post_event and day_of_month:
calendarPostIds.append(post_id) calendar_post_ids.append(post_id)
if not events.get(dayOfMonth): if not events.get(day_of_month):
events[dayOfMonth] = [] events[day_of_month] = []
events[dayOfMonth].append(postEvent) events[day_of_month].append(post_event)
# if some posts have been deleted then regenerate the calendar file # if some posts have been deleted then regenerate the calendar file
if recreateEventsFile: if recreate_events_file:
try: try:
with open(calendarFilename, 'w+') as calendarFile: with open(calendar_filename, 'w+') as calendar_file:
for post_id in calendarPostIds: for post_id in calendar_post_ids:
calendarFile.write(post_id + '\n') calendar_file.write(post_id + '\n')
except OSError: except OSError:
print('EX: unable to write ' + calendarFilename) print('EX: unable to write ' + calendar_filename)
return events return events
def remove_calendar_event(base_dir: str, nickname: str, domain: str, def remove_calendar_event(base_dir: str, nickname: str, domain: str,
year: int, monthNumber: int, messageId: str) -> None: year: int, month_number: int,
message_id: str) -> None:
"""Removes a calendar event """Removes a calendar event
""" """
calendarFilename = \ calendar_filename = \
acct_dir(base_dir, nickname, domain) + \ acct_dir(base_dir, nickname, domain) + \
'/calendar/' + str(year) + '/' + str(monthNumber) + '.txt' '/calendar/' + str(year) + '/' + str(month_number) + '.txt'
if not os.path.isfile(calendarFilename): if not os.path.isfile(calendar_filename):
return return
if '/' in messageId: if '/' in message_id:
messageId = messageId.replace('/', '#') message_id = message_id.replace('/', '#')
if messageId not in open(calendarFilename).read(): if message_id not in open(calendar_filename).read():
return return
lines = None lines = None
with open(calendarFilename, 'r') as f: with open(calendar_filename, 'r') as fp_cal:
lines = f.readlines() lines = fp_cal.readlines()
if not lines: if not lines:
return return
try: try:
with open(calendarFilename, 'w+') as f: with open(calendar_filename, 'w+') as fp_cal:
for line in lines: for line in lines:
if messageId not in line: if message_id not in line:
f.write(line) fp_cal.write(line)
except OSError: except OSError:
print('EX: unable to write ' + calendarFilename) print('EX: unable to write ' + calendar_filename)