microblog.pub/app/lookup.py

41 lines
1.2 KiB
Python
Raw Normal View History

2022-06-22 20:11:22 +02:00
import mf2py # type: ignore
from app import activitypub as ap
from app import webfinger
from app.actor import Actor
from app.actor import fetch_actor
from app.ap_object import RemoteObject
2022-06-29 20:43:17 +02:00
from app.database import AsyncSession
2022-06-22 20:11:22 +02:00
2022-06-29 20:43:17 +02:00
async def lookup(db_session: AsyncSession, query: str) -> Actor | RemoteObject:
2022-06-22 20:11:22 +02:00
if query.startswith("@"):
query = webfinger.get_actor_url(query) # type: ignore # None check below
if not query:
raise ap.NotAnObjectError(query)
try:
ap_obj = ap.fetch(query)
except ap.NotAnObjectError as not_an_object_error:
resp = not_an_object_error.resp
if not resp:
raise ap.NotAnObjectError(query)
alternate_obj = None
if resp.headers.get("content-type", "").startswith("text/html"):
for alternate in mf2py.parse(doc=resp.text).get("alternates", []):
if alternate.get("type") == "application/activity+json":
alternate_obj = ap.fetch(alternate["url"])
if alternate_obj:
ap_obj = alternate_obj
else:
raise
if ap_obj["type"] in ap.ACTOR_TYPES:
2022-06-29 20:43:17 +02:00
actor = await fetch_actor(db_session, ap_obj["id"])
2022-06-22 20:11:22 +02:00
return actor
else:
return RemoteObject(ap_obj)