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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
from flask import current_app
from little_boxes.activitypub import ActivityType
from little_boxes import activitypub as ap
from typing import Dict, Any
from .vars import DEFAULT_CTX
ObjectType = Dict[str, Any]
def ap_url(klass, username):
if klass == "url":
return f"https://{current_app.config['AP_DOMAIN']}/user/{username}"
elif klass == "shared_inbox":
return f"https://{current_app.config['AP_DOMAIN']}/inbox"
elif klass == "inbox":
return f"https://{current_app.config['AP_DOMAIN']}" f"/user/{username}/inbox"
elif klass == "outbox":
return f"https://{current_app.config['AP_DOMAIN']}" f"/user/{username}/outbox"
elif klass == "followings":
return f"https://{current_app.config['AP_DOMAIN']}" f"/user/{username}/followings"
elif klass == "followers":
return f"https://{current_app.config['AP_DOMAIN']}" f"/user/{username}/followers"
else:
return None
def full_url(path):
if path.startswith("http://") or path.startswith("https://"):
return path
root = current_app.config["AP_DOMAIN"]
if path.startswith("/"):
return root + path[1:]
elif path.startswith("/"):
return root + "/" + path
else:
return root + path
def embed_collection(total_items, first_page_id):
"""Helper creating a root OrderedCollection
with a link to the first page."""
return {
"type": ap.ActivityType.ORDERED_COLLECTION.value,
"totalItems": total_items,
"first": f"{first_page_id}?page=first",
"id": first_page_id,
}
def add_extra_collection(item: Dict[str, Any]) -> Dict[str, Any]:
if item["type"] != ActivityType.CREATE.value:
return item
item["object"]["replies"] = embed_collection(
item.get("meta", {}).get("count_direct_reply", 0), f'{item["id"]}/replies'
)
item["object"]["likes"] = embed_collection(item.get("meta", {}).get("count_like", 0), f'{item["id"]}/likes')
item["object"]["shares"] = embed_collection(item.get("meta", {}).get("count_boost", 0), f'{item["id"]}/shares')
return item
def remove_context(activity: Dict[str, Any]) -> Dict[str, Any]:
if "@context" in activity:
del activity["@context"]
return activity
def activity_from_doc(item: Dict[str, Any], embed: bool = False) -> Dict[str, Any]:
item = add_extra_collection(item)
activity = clean_activity(item)
if embed:
return remove_context(activity)
return activity
def clean_activity(activity: ObjectType) -> Dict[str, Any]:
"""Clean the activity before rendering it.
- Remove the hidden bco and bcc field
"""
for field in ["bto", "bcc", "source"]:
if field in activity:
del activity[field]
if activity["type"] == "Create" and field in activity["object"]:
del activity["object"][field]
return activity
def build_ordered_collection(items, actor_id, page, limit=50, switch_side=False):
try:
total_items = len(items)
except TypeError:
total_items = items.count()
if total_items <= 0:
return {
"@context": DEFAULT_CTX,
"id": f"{actor_id}/followers",
"totalItems": 0,
"type": ap.ActivityType.ORDERED_COLLECTION.value,
"orderedItems": [],
}
if not page:
resp = {
"@context": DEFAULT_CTX,
"id": f"{actor_id}/followers",
"totalItems": total_items,
"type": ap.ActivityType.ORDERED_COLLECTION.value,
"first": {
"id": f"{actor_id}/followers?page=0",
"orderedItems": [(item.target.url if switch_side else item.actor.url) for item in items],
"partOf": f"{actor_id}/followers",
"totalItems": total_items,
"type": ap.ActivityType.ORDERED_COLLECTION_PAGE.value,
},
}
if total_items == limit:
resp["first"]["next"] = f"{actor_id}/followers?page=1"
return resp
# return resp
|
