epicyon/utils.py

307 lines
11 KiB
Python
Raw Normal View History

2019-07-02 09:25:29 +00:00
__filename__ = "utils.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "0.0.1"
__maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
import os
import datetime
2019-07-14 17:02:41 +00:00
import commentjson
2019-07-02 09:25:29 +00:00
def getStatusNumber() -> (str,str):
"""Returns the status number and published date
"""
currTime=datetime.datetime.utcnow()
daysSinceEpoch=(currTime - datetime.datetime(1970,1,1)).days
# status is the number of seconds since epoch
statusNumber=str(((daysSinceEpoch*24*60*60) + (currTime.hour*60*60) + (currTime.minute*60) + currTime.second)*1000000 + currTime.microsecond)
published=currTime.strftime("%Y-%m-%dT%H:%M:%SZ")
return statusNumber,published
2019-07-04 10:02:56 +00:00
def createPersonDir(nickname: str,domain: str,baseDir: str,dirname: str) -> str:
"""Create a directory for a person
2019-07-02 09:25:29 +00:00
"""
2019-07-28 13:30:19 +00:00
handle=nickname+'@'+domain
2019-07-02 09:25:29 +00:00
if not os.path.isdir(baseDir+'/accounts/'+handle):
os.mkdir(baseDir+'/accounts/'+handle)
2019-07-04 10:02:56 +00:00
boxDir=baseDir+'/accounts/'+handle+'/'+dirname
if not os.path.isdir(boxDir):
os.mkdir(boxDir)
return boxDir
def createOutboxDir(nickname: str,domain: str,baseDir: str) -> str:
"""Create an outbox for a person
"""
return createPersonDir(nickname,domain,baseDir,'outbox')
def createInboxQueueDir(nickname: str,domain: str,baseDir: str) -> str:
"""Create an inbox queue and returns the feed filename and directory
"""
return createPersonDir(nickname,domain,baseDir,'queue')
2019-07-02 10:39:55 +00:00
def domainPermitted(domain: str, federationList: []):
if len(federationList)==0:
return True
2019-07-11 12:29:31 +00:00
if ':' in domain:
domain=domain.split(':')[0]
2019-07-02 10:39:55 +00:00
if domain in federationList:
return True
return False
2019-07-09 14:20:23 +00:00
def urlPermitted(url: str, federationList: [],capability: str):
2019-07-27 20:30:58 +00:00
if url.endswith('gab.com') or url.endswith('gabfed.com'):
2019-07-12 20:56:26 +00:00
return False
2019-07-02 10:39:55 +00:00
if len(federationList)==0:
return True
for domain in federationList:
if domain in url:
return True
return False
2019-07-06 15:17:21 +00:00
def getDisplayName(actor: str,personCache: {}) -> str:
"""Returns the display name for the given actor
2019-08-22 12:41:16 +00:00
"""
if '/statuses/' in actor:
2019-08-22 13:21:16 +00:00
actor=actor.split('/statuses/')[0]
2019-08-22 13:29:57 +00:00
if not personCache.get(actor):
return None
2019-08-22 12:56:33 +00:00
if personCache[actor].get('actor'):
if personCache[actor]['actor'].get('name'):
return personCache[actor]['actor']['name']
2019-08-22 12:41:16 +00:00
return None
2019-07-06 15:17:21 +00:00
def getNicknameFromActor(actor: str) -> str:
"""Returns the nickname from an actor url
"""
if '/users/' not in actor:
2019-08-21 16:23:06 +00:00
# https://domain/@nick
if '/@' in actor:
nickStr=actor.split('/@')[1]
if '/' in nickStr:
nickStr=nickStr.split('/')[0]
return nickStr
2019-07-06 15:17:21 +00:00
return None
2019-07-10 09:47:07 +00:00
nickStr=actor.split('/users/')[1].replace('@','')
if '/' not in nickStr:
return nickStr
else:
return nickStr.split('/')[0]
2019-07-06 15:17:21 +00:00
def getDomainFromActor(actor: str) -> (str,int):
"""Returns the domain name from an actor url
"""
port=None
if '/users/' not in actor:
domain = actor.replace('https://','').replace('http://','').replace('dat://','')
2019-08-21 16:23:06 +00:00
if '/' in actor:
domain=domain.split('/')[0]
2019-07-06 15:17:21 +00:00
else:
domain = actor.split('/users/')[0].replace('https://','').replace('http://','').replace('dat://','')
if ':' in domain:
port=int(domain.split(':')[1])
domain=domain.split(':')[0]
return domain,port
2019-07-06 19:24:52 +00:00
def followPerson(baseDir: str,nickname: str, domain: str, \
followNickname: str, followDomain: str, \
federationList: [],debug: bool, \
followFile='following.txt') -> bool:
"""Adds a person to the follow list
"""
if not domainPermitted(followDomain.lower().replace('\n',''), \
federationList):
if debug:
print('DEBUG: follow of domain '+followDomain+' not permitted')
return False
2019-07-11 12:29:31 +00:00
if debug:
print('DEBUG: follow of domain '+followDomain)
2019-07-16 22:57:45 +00:00
if ':' in domain:
handle=nickname+'@'+domain.split(':')[0].lower()
else:
handle=nickname+'@'+domain.lower()
if ':' in followDomain:
handleToFollow=followNickname+'@'+followDomain.split(':')[0].lower()
else:
handleToFollow=followNickname+'@'+followDomain.lower()
2019-07-06 19:24:52 +00:00
if not os.path.isdir(baseDir+'/accounts'):
os.mkdir(baseDir+'/accounts')
if not os.path.isdir(baseDir+'/accounts/'+handle):
os.mkdir(baseDir+'/accounts/'+handle)
filename=baseDir+'/accounts/'+handle+'/'+followFile
if os.path.isfile(filename):
if handleToFollow in open(filename).read():
2019-07-11 12:29:31 +00:00
if debug:
print('DEBUG: follow already exists')
2019-07-06 19:24:52 +00:00
return True
with open(filename, "a") as followfile:
2019-07-16 22:57:45 +00:00
followfile.write(followNickname+'@'+followDomain+'\n')
2019-07-11 12:29:31 +00:00
if debug:
print('DEBUG: follow added')
2019-07-06 19:24:52 +00:00
return True
2019-07-11 12:29:31 +00:00
if debug:
print('DEBUG: creating new following file')
2019-07-06 19:24:52 +00:00
with open(filename, "w") as followfile:
2019-07-16 22:57:45 +00:00
followfile.write(followNickname+'@'+followDomain+'\n')
2019-07-06 19:24:52 +00:00
return True
2019-07-11 12:29:31 +00:00
2019-07-13 19:28:14 +00:00
def locatePost(baseDir: str,nickname: str,domain: str,postUrl: str,replies=False) -> str:
2019-07-11 12:29:31 +00:00
"""Returns the filename for the given status post url
"""
2019-07-13 19:28:14 +00:00
if not replies:
extension='json'
else:
extension='replies'
2019-07-11 19:31:02 +00:00
# if this post in the shared inbox?
handle='inbox@'+domain
boxName='inbox'
2019-07-13 19:28:14 +00:00
postFilename=baseDir+'/accounts/'+nickname+'@'+domain+'/'+boxName+'/'+postUrl.replace('/','#')+'.'+extension
2019-07-11 12:29:31 +00:00
if not os.path.isfile(postFilename):
2019-07-11 19:31:02 +00:00
boxName='outbox'
2019-07-13 19:28:14 +00:00
postFilename=baseDir+'/accounts/'+nickname+'@'+domain+'/'+boxName+'/'+postUrl.replace('/','#')+'.'+extension
2019-07-11 12:29:31 +00:00
if not os.path.isfile(postFilename):
2019-07-11 19:31:02 +00:00
# if this post in the inbox of the person?
boxName='inbox'
2019-07-13 19:28:14 +00:00
postFilename=baseDir+'/accounts/'+nickname+'@'+domain+'/'+boxName+'/'+postUrl.replace('/','#')+'.'+extension
2019-07-11 12:29:31 +00:00
if not os.path.isfile(postFilename):
postFilename=None
return postFilename
2019-07-14 16:37:01 +00:00
2019-07-14 16:57:06 +00:00
def removeAttachment(baseDir: str,httpPrefix: str,domain: str,postJson: {}):
if not postJson.get('attachment'):
return
if not postJson['attachment'][0].get('url'):
return
if port:
if port!=80 and port!=443:
if ':' not in domain:
domain=domain+':'+str(port)
2019-07-14 16:57:06 +00:00
attachmentUrl=postJson['attachment'][0]['url']
if not attachmentUrl:
return
mediaFilename=baseDir+'/'+attachmentUrl.replace(httpPrefix+'://'+domain+'/','')
if os.path.isfile(mediaFilename):
os.remove(mediaFilename)
postJson['attachment']=[]
2019-08-12 18:02:29 +00:00
def removeModerationPostFromIndex(baseDir: str,postUrl: str,debug: bool) -> None:
"""Removes a url from the moderation index
"""
moderationIndexFile=baseDir+'/accounts/moderation.txt'
if not os.path.isfile(moderationIndexFile):
return
postId=postUrl.replace('/activity','')
if postId in open(moderationIndexFile).read():
with open(moderationIndexFile, "r") as f:
lines = f.readlines()
with open(moderationIndexFile, "w+") as f:
for line in lines:
if line.strip("\n") != postId:
f.write(line)
else:
if debug:
print('DEBUG: removed '+postId+' from moderation index')
2019-07-14 17:02:41 +00:00
def deletePost(baseDir: str,httpPrefix: str,nickname: str,domain: str,postFilename: str,debug: bool):
2019-07-14 16:37:01 +00:00
"""Recursively deletes a post and its replies and attachments
"""
2019-07-14 16:57:06 +00:00
with open(postFilename, 'r') as fp:
postJsonObject=commentjson.load(fp)
# remove any attachment
2019-07-14 17:02:41 +00:00
removeAttachment(baseDir,httpPrefix,domain,postJsonObject)
2019-08-12 13:22:17 +00:00
# remove from moderation index file
if postJsonObject.get('moderationStatus'):
2019-08-12 18:02:29 +00:00
if postJsonObject.get('object'):
if isinstance(postJsonObject['object'], dict):
if postJsonObject['object'].get('id'):
postId=postJsonObject['object']['id'].replace('/activity','')
removeModerationPostFromIndex(baseDir,postId,debug)
2019-07-14 17:02:41 +00:00
# remove any hashtags index entries
removeHashtagIndex=False
if postJsonObject.get('object'):
if isinstance(postJsonObject['object'], dict):
if postJsonObject['object'].get('content'):
if '#' in postJsonObject['object']['content']:
removeHashtagIndex=True
if removeHashtagIndex:
if postJsonObject['object'].get('id') and postJsonObject['object'].get('tag'):
# get the id of the post
postId=postJsonObject['object']['id'].replace('/activity','')
for tag in postJsonObject['object']['tag']:
if tag['type']!='Hashtag':
continue
# find the index file for this tag
tagIndexFilename=baseDir+'/tags/'+tag['name'][1:]+'.txt'
if not os.path.isfile(tagIndexFilename):
continue
# remove postId from the tag index file
with open(tagIndexFilename, "r") as f:
lines = f.readlines()
with open(tagIndexFilename, "w+") as f:
for line in lines:
if line.strip("\n") != postId:
f.write(line)
2019-07-14 17:02:41 +00:00
# remove any replies
2019-07-14 16:37:01 +00:00
repliesFilename=postFilename.replace('.json','.replies')
if os.path.isfile(repliesFilename):
if debug:
print('DEBUG: removing replies to '+postFilename)
with open(repliesFilename,'r') as f:
for replyId in f:
replyFile=locatePost(baseDir,nickname,domain,replyId)
if replyFile:
if os.path.isfile(replyFile):
deletePost(baseDir,nickname,domain,replyFile,debug)
2019-07-14 17:02:41 +00:00
# remove the replies file
2019-07-14 16:37:01 +00:00
os.remove(repliesFilename)
2019-07-14 17:02:41 +00:00
# finally, remove the post itself
2019-07-14 16:37:01 +00:00
os.remove(postFilename)
2019-07-27 22:48:34 +00:00
2019-08-23 13:47:29 +00:00
def validNickname(domain: str,nickname: str) -> bool:
2019-07-27 22:48:34 +00:00
forbiddenChars=['.',' ','/','?',':',';','@']
for c in forbiddenChars:
if c in nickname:
return False
2019-08-23 13:47:29 +00:00
if nickname==domain:
return False
2019-07-27 22:48:34 +00:00
reservedNames=['inbox','outbox','following','followers','capabilities']
if nickname in reservedNames:
return False
return True
2019-08-08 11:24:26 +00:00
def noOfAccounts(baseDir: str) -> bool:
"""Returns the number of accounts on the system
"""
accountCtr=0
for subdir, dirs, files in os.walk(baseDir+'/accounts'):
for account in dirs:
if '@' in account:
if not account.startswith('inbox'):
accountCtr+=1
return accountCtr
2019-08-10 11:31:42 +00:00
def isPublicPost(postJsonObject: {}) -> bool:
"""Returns true if the given post is public
"""
if not postJsonObject.get('type'):
return False
if postJsonObject['type']!='Create':
return False
if not postJsonObject.get('object'):
return False
if not isinstance(postJsonObject['object'], dict):
return False
if not postJsonObject['object'].get('to'):
return False
for recipient in postJsonObject['object']['to']:
if recipient.endswith('#Public'):
return True
return False