Display webmentions on notes

app/webmentions.py

@@ -1,17 +1,63 @@
+from dataclasses import asdict
+from dataclasses import dataclass
+from typing import Any
+from typing import Optional
+
 from bs4 import BeautifulSoup  # type: ignore
 from fastapi import APIRouter
+from fastapi import Depends
 from fastapi import HTTPException
 from fastapi import Request
 from fastapi.responses import JSONResponse
 from loguru import logger
 
+from app.boxes import get_outbox_object_by_ap_id
+from app.database import AsyncSession
+from app.database import get_db_session
+from app.database import now
 from app.utils import microformats
 from app.utils.url import check_url
 from app.utils.url import is_url_valid
+from app.utils.url import make_abs
 
 router = APIRouter()
 
 
+@dataclass
+class Webmention:
+    actor_icon_url: str
+    actor_name: str
+    url: str
+    received_at: str
+
+    @classmethod
+    def from_microformats(
+        cls, items: list[dict[str, Any]], url: str
+    ) -> Optional["Webmention"]:
+        for item in items:
+            if item["type"][0] == "h-card":
+                return cls(
+                    actor_icon_url=make_abs(
+                        item["properties"]["photo"][0], url
+                    ),  # type: ignore
+                    actor_name=item["properties"]["name"][0],
+                    url=url,
+                    received_at=now().isoformat(),
+                )
+            if item["type"][0] == "h-entry":
+                author = item["properties"]["author"][0]
+                return cls(
+                    actor_icon_url=make_abs(
+                        author["properties"]["photo"][0], url
+                    ),  # type: ignore
+                    actor_name=author["properties"]["name"][0],
+                    url=url,
+                    received_at=now().isoformat(),
+                )
+
+        return None
+
+
 def is_source_containing_target(source_html: str, target_url: str) -> bool:
     soup = BeautifulSoup(source_html, "html5lib")
     for link in soup.find_all("a"):
@@ -28,6 +74,7 @@ def is_source_containing_target(source_html: str, target_url: str) -> bool:
 @router.post("/webmentions")
 async def webmention_endpoint(
     request: Request,
+    db_session: AsyncSession = Depends(get_db_session),
 ) -> JSONResponse:
     form_data = await request.form()
     try:
@@ -45,7 +92,11 @@ async def webmention_endpoint(
 
     logger.info(f"Received webmention {source=} {target=}")
 
-    # TODO: get outbox via ap_id (URL is the same as ap_id)
+    mentioned_object = await get_outbox_object_by_ap_id(db_session, target)
+    if not mentioned_object:
+        logger.info(f"Invalid target {target=}")
+        raise HTTPException(status_code=400, detail="Invalid target")
+
     maybe_data_and_html = await microformats.fetch_and_parse(source)
     if not maybe_data_and_html:
         logger.info("failed to fetch source")
@@ -57,6 +108,25 @@ async def webmention_endpoint(
         logger.warning("target not found in source")
         raise HTTPException(status_code=400, detail="target not found in source")
 
-    logger.info(f"{data=}")
+    try:
+        webmention = Webmention.from_microformats(data["items"], source)
+        if not webmention:
+            raise ValueError("Failed to fetch target data")
+    except Exception:
+        logger.warning("Failed build Webmention for {source=} with {data=}")
+        return JSONResponse(content={}, status_code=200)
+
+    logger.info(f"{webmention=}")
+
+    if mentioned_object.webmentions is None:
+        mentioned_object.webmentions = [asdict(webmention)]
+    else:
+        mentioned_object.webmentions = [asdict(webmention)] + [
+            wm  # type: ignore
+            for wm in mentioned_object.webmentions  # type: ignore
+            if wm["url"] != source  # type: ignore
+        ]
+
+    await db_session.commit()
 
     return JSONResponse(content={}, status_code=200)