microblog.pub/app/ap_object.py

206 lines
5.7 KiB
Python
Raw Normal View History

2022-06-23 19:07:20 +00:00
import base64
2022-06-22 18:11:22 +00:00
import hashlib
from datetime import datetime
from typing import Any
import pydantic
from dateutil.parser import isoparse
from markdown import markdown
from app import activitypub as ap
from app.actor import LOCAL_ACTOR
from app.actor import Actor
from app.actor import RemoteActor
2022-06-22 19:15:07 +00:00
from app.utils import opengraph
2022-06-22 18:11:22 +00:00
class Object:
@property
def is_from_db(self) -> bool:
return False
@property
def ap_type(self) -> str:
return self.ap_object["type"]
@property
def ap_object(self) -> ap.RawObject:
raise NotImplementedError
@property
def ap_id(self) -> str:
return ap.get_id(self.ap_object["id"])
@property
def ap_actor_id(self) -> str:
return ap.get_actor_id(self.ap_object)
@property
def ap_published_at(self) -> datetime | None:
# TODO: default to None? or now()?
if "published" in self.ap_object:
return isoparse(self.ap_object["published"])
elif "created" in self.ap_object:
return isoparse(self.ap_object["created"])
return None
@property
def actor(self) -> Actor:
raise NotImplementedError()
@property
def visibility(self) -> ap.VisibilityEnum:
return ap.object_visibility(self.ap_object)
@property
def context(self) -> str | None:
2022-06-24 09:33:05 +00:00
return self.ap_object.get("context") or self.ap_object.get("conversation")
2022-06-22 18:11:22 +00:00
@property
def sensitive(self) -> bool:
return self.ap_object.get("sensitive", False)
@property
2022-06-24 09:33:05 +00:00
def attachments(self) -> list["Attachment"]:
2022-06-23 19:07:20 +00:00
attachments = []
for obj in self.ap_object.get("attachment", []):
proxied_url = _proxied_url(obj["url"])
attachments.append(
Attachment.parse_obj(
{
"proxiedUrl": proxied_url,
"resizedUrl": proxied_url + "/740"
if obj["mediaType"].startswith("image")
else None,
**obj,
}
)
)
2022-06-22 18:11:22 +00:00
# Also add any video Link (for PeerTube compat)
if self.ap_type == "Video":
for link in ap.as_list(self.ap_object.get("url", [])):
if (isinstance(link, dict)) and link.get("type") == "Link":
if link.get("mediaType", "").startswith("video"):
2022-06-23 19:07:20 +00:00
proxied_url = _proxied_url(link["href"])
2022-06-22 18:11:22 +00:00
attachments.append(
Attachment(
type="Video",
mediaType=link["mediaType"],
url=link["href"],
2022-06-23 19:07:20 +00:00
proxiedUrl=proxied_url,
2022-06-22 18:11:22 +00:00
)
)
break
return attachments
@property
def url(self) -> str | None:
obj_url = self.ap_object.get("url")
if isinstance(obj_url, str):
return obj_url
elif obj_url:
for u in ap.as_list(obj_url):
if u["mediaType"] == "text/html":
return u["href"]
return None
@property
def content(self) -> str | None:
content = self.ap_object.get("content")
if not content:
return None
# PeerTube returns the content as markdown
if self.ap_object.get("mediaType") == "text/markdown":
return markdown(content, extensions=["mdx_linkify"])
return content
@property
def permalink_id(self) -> str:
return (
"permalink-"
+ hashlib.md5(
self.ap_id.encode(),
usedforsecurity=False,
).hexdigest()
)
@property
def activity_object_ap_id(self) -> str | None:
if "object" in self.ap_object:
return ap.get_id(self.ap_object["object"])
return None
@property
def in_reply_to(self) -> str | None:
return self.ap_object.get("inReplyTo")
def _to_camel(string: str) -> str:
cased = "".join(word.capitalize() for word in string.split("_"))
return cased[0:1].lower() + cased[1:]
class BaseModel(pydantic.BaseModel):
class Config:
alias_generator = _to_camel
2022-06-23 19:07:20 +00:00
def _proxied_url(url: str) -> str:
return "/proxy/media/" + base64.urlsafe_b64encode(url.encode()).decode()
2022-06-22 18:11:22 +00:00
class Attachment(BaseModel):
type: str
media_type: str
name: str | None
url: str
2022-06-23 19:07:20 +00:00
# Extra fields for the templates
proxied_url: str
resized_url: str | None = None
2022-06-22 18:11:22 +00:00
class RemoteObject(Object):
def __init__(self, raw_object: ap.RawObject, actor: Actor | None = None):
self._raw_object = raw_object
self._actor: Actor
# Pre-fetch the actor
actor_id = ap.get_actor_id(raw_object)
if actor_id == LOCAL_ACTOR.ap_id:
self._actor = LOCAL_ACTOR
elif actor:
if actor.ap_id != actor_id:
raise ValueError(
f"Invalid actor, got {actor.ap_id}, " f"expected {actor_id}"
)
self._actor = actor
else:
self._actor = RemoteActor(
ap_actor=ap.fetch(ap.get_actor_id(raw_object)),
)
self._og_meta = None
if self.ap_type == "Note":
self._og_meta = opengraph.og_meta_from_note(self._raw_object)
@property
def og_meta(self) -> list[dict[str, Any]] | None:
if self._og_meta:
return [og_meta.dict() for og_meta in self._og_meta]
return None
@property
def ap_object(self) -> ap.RawObject:
return self._raw_object
@property
def actor(self) -> Actor:
return self._actor