Skip to content

muck_out.process

activity_stub 

activity_stub(
    data: dict, actor_id: str | None = None
) -> ActivityStub

Builds the activity stub

Source code in muck_out/process/activity.py 
15
16
17
18
19
20
21
22
23
24
25
def activity_stub(data: dict, actor_id: str | None = None) -> ActivityStub:
    """Builds the activity stub"""
    stub = ActivityStub.model_validate(data)

    if stub.to == [] and actor_id:
        stub.to = [actor_id]

    if stub.type == "EmojiReact":
        stub.type = "Like"

    return stub

actor_stub 

actor_stub(data: dict[str, Any]) -> ActorStub

Returns the stub actor

Source code in muck_out/process/actor.py 
24
25
26
27
28
29
30
31
def actor_stub(data: dict[str, Any]) -> ActorStub:
    """Returns the stub actor"""
    stub = ActorStub.model_validate(data)

    if data.get("identifiers") is None:
        stub.identifiers = calculate_identifiers(stub)

    return stub

collection_stub 

collection_stub(data: dict[str, Any]) -> CollectionStub

Returns a normalized version of a collection, which possible null values

Source code in muck_out/process/collection.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
def collection_stub(data: dict[str, Any]) -> CollectionStub:
    """Returns a normalized version of a collection, which possible null values"""
    result = CollectionStub.model_validate(data)

    if not result.items and data.get("orderedItems"):
        result.items = transform_to_list_of_uris(data.get("orderedItems"))

    if isinstance(data.get("first"), dict):
        first = CollectionStub.model_validate(data.get("first"))
        result.next = first.next
        result.items += first.items
        result.first = None

    return result

normalize_activity 

normalize_activity(
    activity: dict, actor_id: str | None = None
) -> Activity

Normalizes activities.

Parameters:

Name Type Description Default
activity dict

The activity being normalized

 required
actor_id str | None

Actor receiving this activity

 None

Returns:

Type Description
Activity
Source code in muck_out/process/activity.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def normalize_activity(activity: dict, actor_id: str | None = None) -> Activity:
    """
    Normalizes activities.

    :param activity: The activity being normalized
    :param actor_id: Actor receiving this activity
    :returns:
    """
    try:
        obj = activity.get("object")
        if isinstance(obj, dict):
            try:
                obj = normalize_object(obj)
            except Exception:
                if isinstance(obj, dict):
                    obj = obj.get("id")

        stub = activity_stub(activity, actor_id=actor_id)
        dumped = stub.model_dump(by_alias=True)
        dumped["object"] = obj

        return Activity.model_validate(dumped)

    except Exception as e:
        raise e

normalize_actor 

normalize_actor(data: dict[str, Any]) -> Actor | None

Normalizes an ActivityPub actor

Source code in muck_out/process/actor.py 
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
def normalize_actor(data: dict[str, Any]) -> Actor | None:
    """Normalizes an ActivityPub actor"""

    try:
        stub = actor_stub(data)

        if stub.inbox is None:
            return None

        if stub.identifiers is None or len(stub.identifiers) == 0:
            if stub.id is None:
                return
            stub.identifiers = [stub.id]

        return Actor.model_validate(stub.model_dump(by_alias=True))

    except Exception as e:
        logger.info(e)
        return None

normalize_collection 

normalize_collection(
    collection: dict[str, Any],
) -> Collection | None

Normalizes a collection

Source code in muck_out/process/collection.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
def normalize_collection(collection: dict[str, Any]) -> Collection | None:
    """Normalizes a collection"""

    stub = collection_stub(collection)

    if stub.type is None or stub.type not in (
        "Collection",
        "OrderedCollection",
        "CollectionPage",
        "OrderedCollectionPage",
    ):
        return None

    return Collection.model_validate(stub.model_dump(by_alias=True))

normalize_object 

normalize_object(obj: dict[str, Any]) -> Object

Normalizes an object

Parameters:

Name Type Description Default
obj dict[str, Any]

The object to be normalized

 required

Returns:

Type Description
Object
Source code in muck_out/process/object.py 
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
def normalize_object(obj: dict[str, Any]) -> Object:
    """Normalizes an object

    :params obj: The object to be normalized
    :returns:
    """

    stub = object_stub(obj)
    data = stub.model_dump(by_alias=True)
    if "content" in data:
        data["contentPlain"] = data["content"]
    result = Object.model_validate(data)

    if not any([result.content, result.summary, result.name]):
        raise ValueError(
            "At least one of the properties content, summary, or name should be set"
        )

    return result

object_stub 

object_stub(data: dict[str, Any]) -> ObjectStub

Constructs a stub from data

This function is not a direct equivalent to ObjectStub.model_validate as functionality happens that is not field to field for

  • content filled from contentMap
Source code in muck_out/process/object.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def object_stub(data: dict[str, Any]) -> ObjectStub:
    """Constructs a stub from data

    This function is not a direct equivalent to ObjectStub.model_validate
    as functionality happens that is not field to field for

    - `content` filled from `contentMap`

    """

    data = update_for_media_type(data)

    stub = ObjectStub.model_validate(data)

    if stub.content is None:
        content_map = data.get("contentMap")
        if isinstance(content_map, dict):
            values = content_map.values()
            if len(values) > 0:
                stub.content = sanitize_html(list(content_map.values())[0])

    return stub