microblog.pub/migrations.py

"""Migrations that will be run automatically at startup."""
from typing import Any
from typing import Dict
from urllib.parse import urlparse

from little_boxes import activitypub as ap

from utils.migrations import DB
from utils.migrations import Migration
from utils.migrations import logger
from utils.migrations import perform  # noqa:  just here for export


class _1_MetaMigrationt(Migration):
    """Add new metadata to simplify querying."""

    def __guess_visibility(self, data: Dict[str, Any]) -> ap.Visibility:
        to = data.get("to", [])
        cc = data.get("cc", [])
        if ap.AS_PUBLIC in to:
            return ap.Visibility.PUBLIC
        elif ap.AS_PUBLIC in cc:
            return ap.Visibility.UNLISTED
        else:
            # Uses a bit of heuristic here, it's too expensive to fetch the actor, so assume the followers
            # collection has "/collection" in it (which is true for most software), and at worst, we will
            # classify it as "DIRECT" which behave the same as "FOLLOWERS_ONLY" (i.e. no Announce)
            followers_only = False
            for item in to:
                if "/followers" in item:
                    followers_only = True
                    break
            if not followers_only:
                for item in cc:
                    if "/followers" in item:
                        followers_only = True
                        break
            if followers_only:
                return ap.Visibility.FOLLOWERS_ONLY

        return ap.Visibility.DIRECT

    def migrate(self) -> None:  # noqa: C901  # too complex
        for data in DB.activities.find():
            logger.info(f"before={data}")
            obj = data["activity"].get("object")
            set_meta: Dict[str, Any] = {}

            # Set `meta.object_id` (str)
            if not data["meta"].get("object_id"):
                set_meta["meta.object_id"] = None
                if obj:
                    if isinstance(obj, str):
                        set_meta["meta.object_id"] = data["activity"]["object"]
                    elif isinstance(obj, dict):
                        obj_id = obj.get("id")
                        if obj_id:
                            set_meta["meta.object_id"] = obj_id

            # Set `meta.object_visibility` (str)
            if not data["meta"].get("object_visibility"):
                set_meta["meta.object_visibility"] = None
                object_id = data["meta"].get("object_id") or set_meta.get(
                    "meta.object_id"
                )
                if object_id:
                    obj = data["meta"].get("object") or data["activity"].get("object")
                    if isinstance(obj, dict):
                        set_meta["meta.object_visibility"] = self.__guess_visibility(
                            obj
                        ).name

            # Set `meta.actor_id` (str)
            if not data["meta"].get("actor_id"):
                set_meta["meta.actor_id"] = None
                actor = data["activity"].get("actor")
                if actor:
                    if isinstance(actor, str):
                        set_meta["meta.actor_id"] = data["activity"]["actor"]
                    elif isinstance(actor, dict):
                        actor_id = actor.get("id")
                        if actor_id:
                            set_meta["meta.actor_id"] = actor_id

            # Set `meta.poll_answer` (bool)
            if not data["meta"].get("poll_answer"):
                set_meta["meta.poll_answer"] = False
                if obj:
                    if isinstance(obj, dict):
                        if (
                            obj.get("name")
                            and not obj.get("content")
                            and obj.get("inReplyTo")
                        ):
                            set_meta["meta.poll_answer"] = True

            # Set `meta.visibility` (str)
            if not data["meta"].get("visibility"):
                set_meta["meta.visibility"] = self.__guess_visibility(
                    data["activity"]
                ).name

            if not data["meta"].get("server"):
                set_meta["meta.server"] = urlparse(data["remote_id"]).netloc

            logger.info(f"meta={set_meta}\n")
            DB.activities.update_one({"_id": data["_id"]}, {"$set": set_meta})