Something something like soundcloud but not like soundcloud.
Log in, upload records, done.
Simple, easy, KISS.
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

utils.py 3.9KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121
  1. from flask import current_app
  2. from little_boxes.activitypub import ActivityType
  3. from little_boxes import activitypub as ap
  4. from typing import Dict, Any
  5. ObjectType = Dict[str, Any]
  6. def ap_url(klass, username):
  7. if klass == "url":
  8. return f"https://{current_app.config['AP_DOMAIN']}/user/{username}"
  9. elif klass == "shared_inbox":
  10. return f"https://{current_app.config['AP_DOMAIN']}/inbox"
  11. elif klass == "inbox":
  12. return f"https://{current_app.config['AP_DOMAIN']}" f"/user/{username}/inbox"
  13. elif klass == "outbox":
  14. return f"https://{current_app.config['AP_DOMAIN']}" f"/user/{username}/outbox"
  15. elif klass == "followings":
  16. return f"https://{current_app.config['AP_DOMAIN']}" f"/user/{username}/followings"
  17. elif klass == "followers":
  18. return f"https://{current_app.config['AP_DOMAIN']}" f"/user/{username}/followers"
  19. else:
  20. return None
  21. def full_url(path):
  22. if path.startswith("http://") or path.startswith("https://"):
  23. return path
  24. root = current_app.config["AP_DOMAIN"]
  25. if path.startswith("/"):
  26. return root + path[1:]
  27. elif path.startswith("/"):
  28. return root + "/" + path
  29. else:
  30. return root + path
  31. def embed_collection(total_items, first_page_id):
  32. """Helper creating a root OrderedCollection
  33. with a link to the first page."""
  34. return {
  35. "type": ap.ActivityType.ORDERED_COLLECTION.value,
  36. "totalItems": total_items,
  37. "first": f"{first_page_id}?page=first",
  38. "id": first_page_id,
  39. }
  40. def add_extra_collection(item: Dict[str, Any]) -> Dict[str, Any]:
  41. if item["type"] != ActivityType.CREATE.value:
  42. return item
  43. item["object"]["replies"] = embed_collection(
  44. item.get("meta", {}).get("count_direct_reply", 0), f'{item["id"]}/replies'
  45. )
  46. item["object"]["likes"] = embed_collection(item.get("meta", {}).get("count_like", 0), f'{item["id"]}/likes')
  47. item["object"]["shares"] = embed_collection(item.get("meta", {}).get("count_boost", 0), f'{item["id"]}/shares')
  48. return item
  49. def remove_context(activity: Dict[str, Any]) -> Dict[str, Any]:
  50. if "@context" in activity:
  51. del activity["@context"]
  52. return activity
  53. def activity_from_doc(item: Dict[str, Any], embed: bool = False) -> Dict[str, Any]:
  54. item = add_extra_collection(item)
  55. activity = clean_activity(item)
  56. if embed:
  57. return remove_context(activity)
  58. return activity
  59. def clean_activity(activity: ObjectType) -> Dict[str, Any]:
  60. """Clean the activity before rendering it.
  61. - Remove the hidden bco and bcc field
  62. """
  63. for field in ["bto", "bcc", "source"]:
  64. if field in activity:
  65. del activity[field]
  66. if activity["type"] == "Create" and field in activity["object"]:
  67. del activity["object"][field]
  68. return activity
  69. def build_ordered_collection(items, actor_id, page, limit=50, switch_side=False):
  70. total_items = len(items)
  71. if total_items <= 0:
  72. return {
  73. "@context": ap.COLLECTION_CTX,
  74. "id": f"{actor_id}/followers",
  75. "totalItems": 0,
  76. "type": ap.ActivityType.ORDERED_COLLECTION.value,
  77. "orderedItems": [],
  78. }
  79. if not page:
  80. resp = {
  81. "@context": ap.COLLECTION_CTX,
  82. "id": f"{actor_id}/followers",
  83. "totalItems": total_items,
  84. "type": ap.ActivityType.ORDERED_COLLECTION.value,
  85. "first": {
  86. "id": f"{actor_id}/followers?page=0",
  87. "orderedItems": [(item.target.url if switch_side else item.actor.url) for item in items],
  88. "partOf": f"{actor_id}/followers",
  89. "totalItems": total_items,
  90. "type": ap.ActivityType.ORDERED_COLLECTION_PAGE.value,
  91. },
  92. }
  93. if len(items) == limit:
  94. resp["first"]["next"] = f"{actor_id}/followers?page=1"
  95. return resp
  96. # return resp