1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
|
import argparse
import enum
import errno
import logging
import time
import requests
import sys
import stat
from typing import Optional
from threading import Thread, Lock, Event
from queue import Queue
import urllib.parse
from fuse import FUSE, fuse_exit
from de_uvok.activitypub_fuse.status_fuse import StatusFileSystem
from de_uvok.activitypub_fuse.status_provider import (
ActivityPubStatusProvider,
MastodonStatusProvider,
StatusProvider,
)
logger = logging.getLogger(__name__)
class APIChoice(enum.Enum):
ACTIVITYPUB = "ActivityPub"
MASTODON = "Mastodon"
def parse_arguments():
parser = argparse.ArgumentParser(
description="Mount a read-only FUSE filesystem for ActivityPub or Mastodon"
)
parser.add_argument(
"mountpoint", help="The directory where the filesystem will be mounted"
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument(
"-a", "--activitypub", action="store_true", help="Use ActivityPub API"
)
group.add_argument("-m", "--mastodon", action="store_true", help="Use Mastodon API")
parser.add_argument("-s", "--server", required=True, help="The server/host URL")
parser.add_argument(
"-u", "--username", required=True, help="The username to fetch statuses for"
)
args = parser.parse_args()
if args.activitypub:
args.api_choice = APIChoice.ACTIVITYPUB
elif args.mastodon:
args.api_choice = APIChoice.MASTODON
else:
parser.error("Must choose either ActivityPub or Mastodon API")
return args
# todo: make this a loop supporting paging
def status_fetcher(fs: StatusFileSystem, sp: StatusProvider, sig_quit: Event):
max_id = ""
while (max_id is not None) and not (sig_quit.wait(5)):
logger.debug("Fetch statuses.")
st_rep = sp.load_statuses(max_id)
logger.debug("Add statuses to FS.")
fs.add_statuses(st_rep[0])
max_id = st_rep[1]
logger.debug("Waiting to fetch statuses.")
logger.debug("Done.")
def main(args):
quit_evt = Event()
t = None
try:
if args.api_choice == APIChoice.MASTODON:
status_provider = MastodonStatusProvider(args.server, args.username)
else:
status_provider = ActivityPubStatusProvider(args.server, args.username)
myfs = StatusFileSystem()
t = Thread(target=status_fetcher, args=(myfs, status_provider, quit_evt))
t.start()
FUSE(myfs, args.mountpoint, nothreads=True, foreground=True)
except:
fuse_exit()
raise
finally:
quit_evt.set()
if t:
t.join()
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
logger.setLevel(logging.DEBUG)
args = parse_arguments()
main(args)
|
