diasporadiaries

utils.py (14385B)

---

 # Copyright (c) 2019 Ivan Jelincic <parazyd@dyne.org>
 #
 # This file is part of diasporadiaries
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License as published by
 # the Free Software Foundation, either version 3 of the License, or
 # (at your option) any later version.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU Affero General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 """
 Utility functions.
 """
 import json
 from hashlib import sha256
 from os import makedirs, listdir, remove
 from os.path import join, isfile
 from random import SystemRandom, shuffle
 from shutil import rmtree
 from string import ascii_lowercase, digits
 from time import gmtime, strftime, time
 
 from bcrypt import gensalt, hashpw
 from flask import Markup
 
 from config import sem
 from db import (sql_select_col_where, sql_select_col, sql_insert,
                 sql_delete_row_where)
 
 
 COUNTRYMAP = {}
 with open('world.json') as worldfile:
     JSON_DATA = json.load(worldfile)
 for cname in JSON_DATA:
     COUNTRYMAP[cname['alpha2']] = cname['name']
 COUNTRYMAP['xk'] = 'Kosovo'
 del JSON_DATA
 
 
 def randomstring(length):
     """
     Returns a random lowercase+digit string of given length.
     """
     return ''.join(SystemRandom().choice(ascii_lowercase+digits)
                    for _ in range(length))
 
 
 def getcountryname(ccode):
     """
     Returns a country name matching the given country code.
     """
     return COUNTRYMAP.get(ccode, 'Unknown countries')
 
 
 def fill_story(row):
     """
     Returns a story dict filled with the given row from the database.
     """
     return {
         'id': row[0],
         'name': row[1],
         'embark': row[2],
         'embarkname': getcountryname(row[2]),
         'disembark': row[3],
         'disembarkname': getcountryname(row[3]),
         'email': row[4],
         'city': row[5],
         'about': row[6],
         'story': row[7],
         'time': row[8],
         'visible': row[9],
         'deletekey': row[10],
         'abstract': row[11],
     }
 
 
 def get_story(story_id):
     """
     Returns a specific story matching the story_id.
     """
     if not story_id:
         return None
     res = sql_select_col_where('*', 'id', story_id)
     if not res:
         return None
     return fill_story(res[0])
 
 
 def get_recent_stories():
     """
     Returns a list of the 3 most recent stories in the database.
     """
     rows = sql_select_col_where('*', 'visible', 1)
     if len(rows) > 3:
         rows = rows[-3:]
 
     stories = []
     for i in rows:
         j = fill_story(i)
         stories.append(j)
 
     return stories[::-1]
 
 
 def get_multiple_stories(col, val, reverse=True):
     """
     Returns a list of stories where col=val.
     """
     rows = sql_select_col_where('*', col, val)
 
     stories = []
     for i in rows:
         j = fill_story(i)
         stories.append(j)
 
     if reverse:
         return stories[::-1]
     return stories
 
 
 def get_multiple_stories_filtered(col, val, param, reverse=True):
     """
     Returns a filtered list of stories where col=val.
 
     param is a tuple, where [0] will be compared to [1] for filtering.
     """
     stories = get_multiple_stories(col, val, reverse=reverse)
     # NOTE: Possible improvement, check when stories is empty and
     # act accordingly in country.html
 
     filtered_stories = []
     for i in stories:
         if i[param[0]] == param[1]:
             filtered_stories.append(i)
 
     return filtered_stories
 
 
 def get_multiple_users(col, val, reverse=False):
     """
     Returns a list of users where col=val.
     """
     if col and val:
         rows = sql_select_col_where('*', col, val, table='users')
     else:
         rows = sql_select_col('*', table='users')
 
     users = []
     for i in rows:
         j = fill_user_dict(i)
         users.append(j)
 
     if reverse:
         return users[::-1]
     return users
 
 
 def makenav(randomize=False):
     """
     Queries the disembark column for filling up the by-country dropdown
     navigation.
     """
     rows = sql_select_col_where('disembark', 'visible', 1)
 
     havec = [i[0] for i in rows]
     havec = list(set(havec))
     navlist = [(i, getcountryname(i)) for i in havec]
 
     if randomize:
         shuffle(navlist)
     return navlist
 
 
 def fill_user_dict(row):
     """
     Function to fill a dict with the given user's row from the database.
     """
     return {
         'id': row[0],
         'email': row[1],
         'name': row[2],
         'password': row[3],
         'cap': row[4],
         'first_seen': row[5],
         'is_active': row[6],
     }
 
 
 def find_user_by_id(user_id):
     """
     Queries the database for a specific user id and returns the user dict.
     """
     row = sql_select_col_where('*', 'id', user_id, table='users')
     if not row:
         return None
 
     return fill_user_dict(row[0])
 
 
 def find_user_by_email(email):
     """
     Queries the database for a user's email (username) and returns the user
     dict.
     """
     row = sql_select_col_where('*', 'email', email, table='users')
     if not row:
         return None
 
     return fill_user_dict(row[0])
 
 
 def validate_user(user, password):
     """
     Validates a user login.
     """
     if hashpw(password.encode(), user['password']) == user['password']:
         return True
 
     return False
 
 
 def make_profile(name, email):
     """
     Helper function to generate and insert a profile into the database.
     """
     if sql_select_col_where('email', 'email', email, table='users'):
         return None
 
     plain_pw = randomstring(24)
 
     # If it's the first user, make them an admin.
     cap = 2
     if sql_select_col('id', table='users'):
         cap = 0
 
     # hashed = bcrypt.hashpw(password, bcrypt.gensalt())
     # bcrypt.hashpw(plaintext, hashed) == hashed
     password = hashpw(plain_pw.encode(), gensalt())
 
     userargs = [
         None,
         email,
         name,
         password,
         cap,
         int(time()),
         0,
     ]
     sql_insert(userargs)
 
     makedirs('follows', exist_ok=True)
     with open(join('follows', email), 'w') as follow_file:
         json.dump([], follow_file)
 
     msgpath = join('messages', email)
     makedirs(msgpath, exist_ok=True)
 
     initial_user = find_user_by_id(1)
     if initial_user:
         welcome_msg = "Welcome to Diaspora Diaries! Enjoy your stay :)"
         with open(join(msgpath, initial_user['email']), 'w') as msgfile:
             json.dump([{'from': 'Diaspora Diaries', 'message': welcome_msg,
                         'time': int(time()), 'unread': 1}], msgfile)
 
     return plain_pw
 
 
 def get_latest_messages(user_id):
     """
     Gets the latest messages of a user to list them in a table.
     """
     msgpath = join('messages', user_id)
     ppl = listdir(msgpath)
     latest = []
     for i in ppl:
         if isfile(join(msgpath, i)):
             with open(join(msgpath, i)) as msgfile:
                 data = json.load(msgfile)
                 user = find_user_by_email(i)
                 if not user:
                     continue
                 data[-1]['message'] = data[-1]['message'][:64] + "..."
                 latest.append([user['name'], user['id'], data[-1]])
 
     return sorted(latest, key=lambda x: x[2]['time'], reverse=True)
 
 
 def get_messages(user_id, id_from):
     """
     Gets all messages in a single conversation.
     """
     email = sql_select_col_where('email', 'id', id_from, table='users')
     if email:
         email = email[0][0]
     else:
         return []
 
     msgpath = join('messages', user_id, email)
 
     data = []
     if isfile(msgpath):
         with open(msgpath) as msgfile:
             data = json.load(msgfile)
     else:
         with open(msgpath, 'w') as msgfile:
             json.dump(data, msgfile)
         return []
 
     messages = []
     for i in data:
         i['unread'] = 0
         messages.append(i)
 
     sem.acquire()
     with open(msgpath, 'w') as msgfile:
         json.dump(messages, msgfile)
     sem.release()
 
     return messages
 
 
 def send_message(id_to, msg, id_us):
     """
     Function for sending/recieving a message.
     """
     if not id_to or not msg or not id_us:
         return
     ours = find_user_by_id(id_us)
     them = find_user_by_id(id_to)
     if not ours or not them:
         return
 
     makedirs(join('messages', ours['email']), exist_ok=True)
     makedirs(join('messages', them['email']), exist_ok=True)
     our_msgpath = join('messages', ours['email'], them['email'])
     their_msgpath = join('messages', them['email'], ours['email'])
 
     msgdict = {
         'from': ours['name'],
         'message': Markup(msg).striptags(),
         'time': int(time()),
         'unread': 1,
     }
 
     ourdata = []
     theirdata = []
 
     if isfile(our_msgpath):
         with open(our_msgpath) as ourmsgs:
             ourdata = json.load(ourmsgs)
 
     sem.acquire()
     ourdata.append(msgdict)
     with open(our_msgpath, 'w') as ourmsgs:
         json.dump(ourdata, ourmsgs)
     sem.release()
 
     if our_msgpath == their_msgpath:
         return
 
     if isfile(their_msgpath):
         with open(their_msgpath) as theirmsgs:
             theirdata = json.load(theirmsgs)
 
     sem.acquire()
     theirdata.append(msgdict)
     with open(their_msgpath, 'w') as theirmsgs:
         json.dump(theirdata, theirmsgs)
     sem.release()
 
 
 def send_shout(id_to, msg, id_us):
     """
     Sends a shout to a user.
     """
     if not id_to or not msg or not id_us:
         return
     ours = find_user_by_id(id_us)
     them = find_user_by_id(id_to)
     if not ours or not them:
         return
 
     makedirs('shouts', exist_ok=True)
     shoutpath = (join('shouts', them['email']))
 
     m = sha256()
     m.update(Markup(msg).striptags().encode())
     msgdict = {
         'from': ours['name'],
         'email': ours['email'],
         'message': Markup(msg).striptags(),
         'time': int(time()),
         'id': m.hexdigest(),
     }
 
     theirdata = []
 
     if isfile(shoutpath):
         with open(shoutpath) as shouts_file:
             theirdata = json.load(shouts_file)
 
     sem.acquire()
     theirdata.append(msgdict)
     with open(shoutpath, 'w') as shouts_file:
         json.dump(theirdata, shouts_file)
     sem.release()
 
 
 def delete_shout(id_to, id_shout):
     """
     Deletes a specific shout.
     """
     if not id_to or not id_shout:
         return
     ours = find_user_by_id(id_to)
     if not ours:
         return
 
     shoutpath = join('shouts', ours['email'])
     if isfile(shoutpath):
         with open(shoutpath) as shouts_file:
             shouts = json.load(shouts_file)
 
         for i in shouts:
             if i['id'] == id_shout:
                 shouts.remove(i)
 
         sem.acquire()
         with open(shoutpath, 'w') as shouts_file:
             json.dump(shouts, shouts_file)
         sem.release()
 
 
 def get_shouts(email):
     """
     Gets all shouts for a user.
     """
     if not isfile(join('shouts', email)):
         return []
 
     shouts = []
     with open(join('shouts', email)) as shouts:
         shouts = json.load(shouts)
 
     return shouts[::-1]
 
 
 def follow_user(id_us, id_them):
     """
     Follows a user.
     """
     if not id_us or not id_them:
         return
     ours = find_user_by_id(id_us)
     them = find_user_by_id(id_them)
     if not ours or not them:
         return
 
     our_follow = join('follows', ours['email'])
     with open(our_follow) as follow_file:
         ourdata = json.load(follow_file)
 
     sem.acquire()
     ourdata.append(them['email'])
     with open(our_follow, 'w') as follow_file:
         json.dump(ourdata, follow_file)
     sem.release()
 
 
 def unfollow_user(id_us, id_them):
     """
     Unfollows a user.
     """
     if not id_us or not id_them:
         return
     ours = find_user_by_id(id_us)
     them = find_user_by_id(id_them)
     if not ours or not them:
         return
 
     our_follow = join('follows', ours['email'])
     with open(our_follow) as follow_file:
         ourdata = json.load(follow_file)
 
     if ourdata:
         while them['email'] in ourdata:
             ourdata.remove(them['email'])
 
     sem.acquire()
     with open(our_follow, 'w') as follow_file:
         json.dump(ourdata, follow_file)
     sem.release()
 
 
 def is_following(email_ours, email_them):
     """
     Function to check if we're following a specific user.
     """
     with open(join('follows', email_ours)) as follow_file:
         ourdata = json.load(follow_file)
 
     if email_them in ourdata:
         return True
 
     return False
 
 
 def get_following(email):
     """
     Gets all the users we are following.
     """
     with open(join('follows', email)) as follow_file:
         return json.load(follow_file)
 
 
 def delete_user(user_id):
     """
     Deletes a user and their messages directory.
     """
     user = find_user_by_id(user_id)
     rmtree(join('messages', user['email']), ignore_errors=True)
     remove(join('follows', user['email']))
     sql_delete_row_where('id', user_id, table='users')
 
 
 def get_profiles_from_stories(stories):
     """
     Map emails to ids from given stories.
     """
     profiles = {}
     for i in stories:
         if i['email']:
             uid = sql_select_col_where('id', 'email', i['email'],
                                        table='users')
             if uid:
                 profiles[i['email']] = uid[0][0]
 
     return profiles
 
 
 def parsetime(then):
     """
     Parse epoch to return human-readable time
     """
     ds = int(time()) - then
     dm = int(ds / 60)
 
     def f(x):
         if int(x) != 1:
             return 's'
         return ''
 
     if ds < 60:
         return '%d second%s ago' % (ds, f(ds))
     elif dm < 60:
         return '%d minute%s ago' % (dm, f(dm))
     elif dm < (24 * 60):
         return '%d hour%s ago' % (int(dm / 60), f(dm/60))
     elif dm < (24 * 60 * 7):
         return '%d day%s ago' % (int(dm / (60*24)), f(dm/(60*24)))
     elif dm < (24 * 60 * 31):
         return '%d week%s ago' % (int(dm / (60*24*7)), f(dm/(60*24*7)))
     elif dm < (24 * 60 * 365.25):
         return '%d month%s ago' % (int(dm / (60*24*30)), f(dm/(60*24*30)))
 
     return '%d year%s ago' % (int(dm / (60*24*365)), f(dm/(60*24*365)))
     # return strftime('%d.%m.%Y. %H:%M UTC', gmtime(then))