epicyon/announce.py

84 lines
3.2 KiB
Python
Raw Normal View History

2019-07-02 09:25:29 +00:00
__filename__ = "announce.py"
__author__ = "Bob Mottram"
__license__ = "AGPL3+"
__version__ = "0.0.1"
__maintainer__ = "Bob Mottram"
__email__ = "bob@freedombone.net"
__status__ = "Production"
import json
import commentjson
from utils import getStatusNumber
from utils import createOutboxDir
2019-07-02 10:43:54 +00:00
from utils import urlPermitted
2019-07-02 09:25:29 +00:00
2019-07-02 20:54:22 +00:00
def createAnnounce(baseDir: str,federationList: [], \
2019-07-03 09:40:27 +00:00
nickname: str, domain: str, port: int, \
2019-07-03 19:00:03 +00:00
toUrl: str, ccUrl: str, httpPrefix: str, \
2019-07-02 20:54:22 +00:00
objectUrl: str, saveToFile: bool) -> {}:
2019-07-02 09:25:29 +00:00
"""Creates an announce message
Typically toUrl will be https://www.w3.org/ns/activitystreams#Public
and ccUrl might be a specific person favorited or repeated and the followers url
objectUrl is typically the url of the message, corresponding to url or atomUri in createPostBase
"""
2019-07-02 10:43:54 +00:00
if not urlPermitted(objectUrl,federationList):
return None
2019-07-02 09:25:29 +00:00
if port!=80 and port!=443:
domain=domain+':'+str(port)
statusNumber,published = getStatusNumber()
2019-07-03 19:00:03 +00:00
newAnnounceId=httpPrefix+'://'+domain+'/users/'+nickname+'/statuses/'+statusNumber
2019-07-02 09:25:29 +00:00
newAnnounce = {
2019-07-03 19:00:03 +00:00
'actor': httpPrefix+'://'+domain+'/users/'+nickname,
'atomUri': httpPrefix+'://'+domain+'/users/'+nickname+'/statuses/'+statusNumber,
2019-07-02 09:25:29 +00:00
'cc': [],
'id': newAnnounceId+'/activity',
'object': objectUrl,
'published': published,
'to': [toUrl],
'type': 'Announce'
}
if ccUrl:
if len(ccUrl)>0:
newAnnounce['cc']=ccUrl
if saveToFile:
if ':' in domain:
domain=domain.split(':')[0]
2019-07-03 09:40:27 +00:00
outboxDir = createOutboxDir(nickname,domain,baseDir)
2019-07-02 09:25:29 +00:00
filename=outboxDir+'/'+newAnnounceId.replace('/','#')+'.json'
with open(filename, 'w') as fp:
commentjson.dump(newAnnounce, fp, indent=4, sort_keys=False)
return newAnnounce
2019-07-02 20:54:22 +00:00
def announcePublic(baseDir: str,federationList: [], \
2019-07-03 19:00:03 +00:00
nickname: str, domain: str, port: int, httpPrefix: str, \
2019-07-02 20:54:22 +00:00
objectUrl: str, saveToFile: bool) -> {}:
2019-07-02 09:25:29 +00:00
"""Makes a public announcement
"""
fromDomain=domain
if port!=80 and port!=443:
fromDomain=fromDomain+':'+str(port)
toUrl = 'https://www.w3.org/ns/activitystreams#Public'
2019-07-03 19:00:03 +00:00
ccUrl = httpPrefix + '://'+fromDomain+'/users/'+nickname+'/followers'
2019-07-03 09:40:27 +00:00
return createAnnounce(baseDir,nickname, domain, port, \
2019-07-03 19:00:03 +00:00
toUrl, ccUrl, httpPrefix, objectUrl, saveToFile)
2019-07-02 09:25:29 +00:00
2019-07-02 20:54:22 +00:00
def repeatPost(baseDir: str,federationList: [], \
2019-07-03 19:00:03 +00:00
nickname: str, domain: str, port: int, httpPrefix: str, \
2019-07-03 09:40:27 +00:00
announceNickname: str, announceDomain: str, \
2019-07-03 19:00:03 +00:00
announcePort: int, announceHttpsPrefix: str, \
2019-07-02 20:54:22 +00:00
announceStatusNumber: int, saveToFile: bool) -> {}:
2019-07-02 14:02:58 +00:00
"""Repeats a given status post
2019-07-02 09:52:37 +00:00
"""
announcedDomain=announceDomain
if announcePort!=80 and announcePort!=443:
announcedDomain=announcedDomain+':'+str(announcePort)
2019-07-03 19:00:03 +00:00
objectUrl = announceHttpsPrefix + '://'+announcedDomain+'/users/'+ \
2019-07-03 09:40:27 +00:00
announceNickname+'/statuses/'+str(announceStatusNumber)
2019-07-02 09:52:37 +00:00
2019-07-03 19:00:03 +00:00
return announcePublic(baseDir,nickname, domain, port, httpPrefix, objectUrl, saveToFile)
2019-07-02 09:52:37 +00:00