moviewyrm/bookwyrm/activitystreams.py

552 lines
19 KiB
Python
Raw Permalink Normal View History

2021-03-23 01:39:16 +00:00
""" access the activity streams stored in redis """
2021-09-11 18:37:10 +00:00
from datetime import timedelta
2021-03-23 01:39:16 +00:00
from django.dispatch import receiver
2021-08-05 02:54:47 +00:00
from django.db import transaction
from django.db.models import signals, Q
2021-09-11 18:37:10 +00:00
from django.utils import timezone
2021-03-23 01:39:16 +00:00
from bookwyrm import models
from bookwyrm.redis_store import RedisStore, r
from bookwyrm.tasks import app, LOW, MEDIUM, HIGH
2021-03-23 01:39:16 +00:00
class ActivityStream(RedisStore):
2021-08-05 02:19:24 +00:00
"""a category of activity stream (like home, local, books)"""
2021-03-23 01:39:16 +00:00
def stream_id(self, user):
2021-04-26 16:15:42 +00:00
"""the redis key for this user's instance of this stream"""
2021-09-18 04:39:18 +00:00
return f"{user.id}-{self.key}"
2021-03-23 01:39:16 +00:00
def unread_id(self, user):
2021-04-26 16:15:42 +00:00
"""the redis key for this user's unread count for this stream"""
2021-09-20 23:44:59 +00:00
stream_id = self.stream_id(user)
return f"{stream_id}-unread"
2021-03-23 01:39:16 +00:00
2021-11-24 18:00:30 +00:00
def unread_by_status_type_id(self, user):
"""the redis key for this user's unread count for this stream"""
stream_id = self.stream_id(user)
return f"{stream_id}-unread-by-type"
def get_rank(self, obj): # pylint: disable=no-self-use
2021-04-26 16:15:42 +00:00
"""statuses are sorted by date published"""
return obj.published_date.timestamp()
2021-04-02 17:44:30 +00:00
def add_status(self, status, increment_unread=False):
2021-04-26 16:15:42 +00:00
"""add a status to users' feeds"""
# the pipeline contains all the add-to-stream activities
pipeline = self.add_object_to_related_stores(status, execute=False)
if increment_unread:
for user in self.get_audience(status):
# add to the unread status count
pipeline.incr(self.unread_id(user))
2021-11-24 18:00:30 +00:00
# add to the unread status count for status type
pipeline.hincrby(
self.unread_by_status_type_id(user), get_status_type(status), 1
)
2021-03-23 01:39:16 +00:00
# and go!
pipeline.execute()
def add_user_statuses(self, viewer, user):
2021-04-26 16:15:42 +00:00
"""add a user's statuses to another user's feed"""
# only add the statuses that the viewer should be able to see (ie, not dms)
2021-10-06 17:37:09 +00:00
statuses = models.Status.privacy_filter(viewer).filter(user=user)
self.bulk_add_objects_to_store(statuses, self.stream_id(viewer))
def remove_user_statuses(self, viewer, user):
2021-04-26 16:15:42 +00:00
"""remove a user's status from another user's feed"""
# remove all so that followers only statuses are removed
statuses = user.status_set.all()
self.bulk_remove_objects_from_store(statuses, self.stream_id(viewer))
2021-03-23 01:39:16 +00:00
def get_activity_stream(self, user):
2021-04-26 16:15:42 +00:00
"""load the statuses to be displayed"""
2021-03-23 01:39:16 +00:00
# clear unreads for this feed
r.set(self.unread_id(user), 0)
2021-11-24 18:00:30 +00:00
r.delete(self.unread_by_status_type_id(user))
2021-03-23 01:39:16 +00:00
statuses = self.get_store(self.stream_id(user))
2021-03-23 01:54:17 +00:00
return (
models.Status.objects.select_subclasses()
.filter(id__in=statuses)
.select_related(
"user",
"reply_parent",
"comment__book",
"review__book",
"quotation__book",
)
.prefetch_related("mention_books", "mention_users")
2021-03-23 01:54:17 +00:00
.order_by("-published_date")
)
2021-03-23 01:39:16 +00:00
2021-03-23 19:52:38 +00:00
def get_unread_count(self, user):
2021-04-26 16:15:42 +00:00
"""get the unread status count for this user's feed"""
return int(r.get(self.unread_id(user)) or 0)
2021-03-23 19:52:38 +00:00
2021-11-24 18:00:30 +00:00
def get_unread_count_by_status_type(self, user):
"""get the unread status count for this user's feed's status types"""
status_types = r.hgetall(self.unread_by_status_type_id(user))
return {
str(key.decode("utf-8")): int(value) or 0
for key, value in status_types.items()
}
2021-04-05 20:13:56 +00:00
def populate_streams(self, user):
2021-04-26 16:15:42 +00:00
"""go from zero to a timeline"""
self.populate_store(self.stream_id(user))
2021-04-02 17:44:30 +00:00
def get_audience(self, status): # pylint: disable=no-self-use
2021-04-26 16:15:42 +00:00
"""given a status, what users should see it"""
2021-03-23 02:17:46 +00:00
# direct messages don't appeard in feeds, direct comments/reviews/etc do
2021-03-23 02:19:21 +00:00
if status.privacy == "direct" and status.status_type == "Note":
2021-03-26 19:09:37 +00:00
return []
2021-03-23 01:39:16 +00:00
# everybody who could plausibly see this status
audience = models.User.objects.filter(
is_active=True,
2021-03-23 01:54:17 +00:00
local=True, # we only create feeds for users of this instance
2021-03-23 01:39:16 +00:00
).exclude(
Q(id__in=status.user.blocks.all()) | Q(blocks=status.user) # not blocked
)
2021-03-23 21:59:51 +00:00
# only visible to the poster and mentioned users
if status.privacy == "direct":
audience = audience.filter(
Q(id=status.user.id) # if the user is the post's author
| Q(id__in=status.mention_users.all()) # if the user is mentioned
)
2021-03-23 01:39:16 +00:00
# only visible to the poster's followers and tagged users
2021-03-23 21:59:51 +00:00
elif status.privacy == "followers":
2021-03-23 01:39:16 +00:00
audience = audience.filter(
Q(id=status.user.id) # if the user is the post's author
2021-03-23 01:54:17 +00:00
| Q(following=status.user) # if the user is following the author
2021-03-23 01:39:16 +00:00
)
return audience.distinct()
2021-03-23 01:39:16 +00:00
def get_stores_for_object(self, obj):
return [self.stream_id(u) for u in self.get_audience(obj)]
def get_statuses_for_user(self, user): # pylint: disable=no-self-use
2021-04-26 16:15:42 +00:00
"""given a user, what statuses should they see on this stream"""
2021-10-06 17:37:09 +00:00
return models.Status.privacy_filter(
2021-03-23 01:39:16 +00:00
user,
2021-03-23 01:54:17 +00:00
privacy_levels=["public", "unlisted", "followers"],
2021-03-23 01:39:16 +00:00
)
def get_objects_for_store(self, store):
2021-04-05 18:10:26 +00:00
user = models.User.objects.get(id=store.split("-")[0])
return self.get_statuses_for_user(user)
2021-03-23 01:39:16 +00:00
class HomeStream(ActivityStream):
2021-04-26 16:15:42 +00:00
"""users you follow"""
2021-03-23 01:54:17 +00:00
key = "home"
2021-03-23 01:39:16 +00:00
def get_audience(self, status):
audience = super().get_audience(status)
2021-03-29 20:07:22 +00:00
if not audience:
return []
2021-03-23 01:39:16 +00:00
return audience.filter(
Q(id=status.user.id) # if the user is the post's author
2021-03-23 01:54:17 +00:00
| Q(following=status.user) # if the user is following the author
).distinct()
2021-03-23 01:39:16 +00:00
def get_statuses_for_user(self, user):
2021-10-06 17:37:09 +00:00
return models.Status.privacy_filter(
2021-03-23 01:39:16 +00:00
user,
2021-03-23 01:54:17 +00:00
privacy_levels=["public", "unlisted", "followers"],
).exclude(
~Q( # remove everything except
Q(user__followers=user) # user following
| Q(user=user) # is self
| Q(mention_users=user) # mentions user
),
2021-03-23 01:39:16 +00:00
)
class LocalStream(ActivityStream):
2021-04-26 16:15:42 +00:00
"""users you follow"""
2021-03-23 01:54:17 +00:00
key = "local"
2021-03-23 01:39:16 +00:00
def get_audience(self, status):
2021-03-23 01:39:16 +00:00
# this stream wants no part in non-public statuses
2021-03-23 21:59:51 +00:00
if status.privacy != "public" or not status.user.local:
2021-03-26 19:09:37 +00:00
return []
return super().get_audience(status)
2021-03-23 01:39:16 +00:00
def get_statuses_for_user(self, user):
2021-03-23 01:39:16 +00:00
# all public statuses by a local user
2021-10-06 17:37:09 +00:00
return models.Status.privacy_filter(
2021-03-23 01:39:16 +00:00
user,
privacy_levels=["public"],
2021-10-06 17:37:09 +00:00
).filter(user__local=True)
2021-03-23 01:39:16 +00:00
2021-08-04 23:56:08 +00:00
class BooksStream(ActivityStream):
"""books on your shelves"""
2021-03-23 01:54:17 +00:00
2021-08-04 23:56:08 +00:00
key = "books"
2021-03-23 01:39:16 +00:00
def get_audience(self, status):
2021-08-04 23:56:08 +00:00
"""anyone with the mentioned book on their shelves"""
# only show public statuses on the books feed,
# and only statuses that mention books
2021-08-05 00:25:31 +00:00
if status.privacy != "public" or not (
status.mention_books.exists() or hasattr(status, "book")
):
2021-03-26 19:09:37 +00:00
return []
2021-08-04 23:56:08 +00:00
2021-08-05 00:25:31 +00:00
work = (
status.book.parent_work
if hasattr(status, "book")
else status.mention_books.first().parent_work
)
2021-08-04 23:56:08 +00:00
audience = super().get_audience(status)
if not audience:
return []
2021-08-05 00:25:31 +00:00
return audience.filter(shelfbook__book__parent_work=work).distinct()
2021-03-23 01:39:16 +00:00
def get_statuses_for_user(self, user):
2021-08-05 02:09:00 +00:00
"""any public status that mentions the user's books"""
2021-08-05 00:25:31 +00:00
books = user.shelfbook_set.values_list(
"book__parent_work__id", flat=True
).distinct()
2021-10-06 17:37:09 +00:00
return (
models.Status.privacy_filter(
user,
privacy_levels=["public"],
)
2021-08-05 00:25:31 +00:00
.filter(
Q(comment__book__parent_work__id__in=books)
| Q(quotation__book__parent_work__id__in=books)
| Q(review__book__parent_work__id__in=books)
| Q(mention_books__parent_work__id__in=books)
)
2021-10-06 17:37:09 +00:00
.distinct()
2021-03-23 01:39:16 +00:00
)
2021-08-06 02:28:05 +00:00
def add_book_statuses(self, user, book):
"""add statuses about a book to a user's feed"""
work = book.parent_work
2021-10-06 17:37:09 +00:00
statuses = (
models.Status.privacy_filter(
user,
privacy_levels=["public"],
)
2021-08-06 02:28:05 +00:00
.filter(
Q(comment__book__parent_work=work)
| Q(quotation__book__parent_work=work)
| Q(review__book__parent_work=work)
| Q(mention_books__parent_work=work)
)
2021-10-06 17:37:09 +00:00
.distinct()
2021-08-06 02:28:05 +00:00
)
self.bulk_add_objects_to_store(statuses, self.stream_id(user))
def remove_book_statuses(self, user, book):
"""add statuses about a book to a user's feed"""
work = book.parent_work
2021-10-06 17:37:09 +00:00
statuses = (
models.Status.privacy_filter(
user,
privacy_levels=["public"],
)
2021-08-06 02:28:05 +00:00
.filter(
Q(comment__book__parent_work=work)
| Q(quotation__book__parent_work=work)
| Q(review__book__parent_work=work)
| Q(mention_books__parent_work=work)
)
2021-10-06 17:37:09 +00:00
.distinct()
2021-08-06 02:28:05 +00:00
)
self.bulk_remove_objects_from_store(statuses, self.stream_id(user))
2021-03-23 01:39:16 +00:00
2021-08-05 02:09:00 +00:00
# determine which streams are enabled in settings.py
2021-03-23 01:39:16 +00:00
streams = {
"home": HomeStream(),
"local": LocalStream(),
"books": BooksStream(),
2021-03-23 01:39:16 +00:00
}
2021-03-23 01:54:17 +00:00
2021-03-23 01:39:16 +00:00
@receiver(signals.post_save)
# pylint: disable=unused-argument
def add_status_on_create(sender, instance, created, *args, **kwargs):
2021-04-26 16:15:42 +00:00
"""add newly created statuses to activity feeds"""
# we're only interested in new statuses
2021-03-23 15:13:57 +00:00
if not issubclass(sender, models.Status):
return
if instance.deleted:
2021-08-05 02:54:47 +00:00
remove_status_task.delay(instance.id)
2021-03-23 01:39:16 +00:00
return
2021-08-05 02:54:47 +00:00
# when creating new things, gotta wait on the transaction
transaction.on_commit(
lambda: add_status_on_create_command(sender, instance, created)
)
2021-08-05 02:54:47 +00:00
def add_status_on_create_command(sender, instance, created):
2021-08-05 02:54:47 +00:00
"""runs this code only after the database commit completes"""
2021-11-12 16:55:47 +00:00
priority = HIGH
# check if this is an old status, de-prioritize if so
# (this will happen if federation is very slow, or, more expectedly, on csv import)
if instance.published_date < timezone.now() - timedelta(
days=1
) or instance.created_date < instance.published_date - timedelta(days=1):
2021-11-12 16:55:47 +00:00
priority = LOW
add_status_task.apply_async(
args=(instance.id,),
kwargs={"increment_unread": created},
queue=priority,
)
2021-09-06 22:39:32 +00:00
if sender == models.Boost:
handle_boost_task.delay(instance.id)
@receiver(signals.post_delete, sender=models.Boost)
# pylint: disable=unused-argument
def remove_boost_on_delete(sender, instance, *args, **kwargs):
2021-04-26 16:15:42 +00:00
"""boosts are deleted"""
2021-08-05 02:54:47 +00:00
# remove the boost
remove_status_task.delay(instance.id)
# re-add the original status
add_status_task.delay(instance.boosted_status.id)
@receiver(signals.post_save, sender=models.UserFollows)
# pylint: disable=unused-argument
def add_statuses_on_follow(sender, instance, created, *args, **kwargs):
2021-04-26 16:15:42 +00:00
"""add a newly followed user's statuses to feeds"""
if not created or not instance.user_subject.local:
return
2021-08-05 02:54:47 +00:00
add_user_statuses_task.delay(
instance.user_subject.id, instance.user_object.id, stream_list=["home"]
)
@receiver(signals.post_delete, sender=models.UserFollows)
# pylint: disable=unused-argument
2021-04-05 19:11:49 +00:00
def remove_statuses_on_unfollow(sender, instance, *args, **kwargs):
2021-04-26 16:15:42 +00:00
"""remove statuses from a feed on unfollow"""
if not instance.user_subject.local:
return
2021-08-05 02:54:47 +00:00
remove_user_statuses_task.delay(
instance.user_subject.id, instance.user_object.id, stream_list=["home"]
)
@receiver(signals.post_save, sender=models.UserBlocks)
# pylint: disable=unused-argument
2021-04-05 19:11:49 +00:00
def remove_statuses_on_block(sender, instance, *args, **kwargs):
2021-04-26 16:15:42 +00:00
"""remove statuses from all feeds on block"""
# blocks apply ot all feeds
if instance.user_subject.local:
2021-08-05 02:54:47 +00:00
remove_user_statuses_task.delay(
instance.user_subject.id, instance.user_object.id
)
# and in both directions
if instance.user_object.local:
2021-08-05 02:54:47 +00:00
remove_user_statuses_task.delay(
instance.user_object.id, instance.user_subject.id
)
@receiver(signals.post_delete, sender=models.UserBlocks)
# pylint: disable=unused-argument
def add_statuses_on_unblock(sender, instance, *args, **kwargs):
2021-10-03 18:46:26 +00:00
"""add statuses back to all feeds on unblock"""
# make sure there isn't a block in the other direction
if models.UserBlocks.objects.filter(
user_subject=instance.user_object,
user_object=instance.user_subject,
).exists():
return
public_streams = [k for (k, v) in streams.items() if k != "home"]
# add statuses back to streams with statuses from anyone
if instance.user_subject.local:
2021-08-05 02:54:47 +00:00
add_user_statuses_task.delay(
instance.user_subject.id,
instance.user_object.id,
stream_list=public_streams,
2021-08-05 02:54:47 +00:00
)
# add statuses back to streams with statuses from anyone
if instance.user_object.local:
2021-08-05 02:54:47 +00:00
add_user_statuses_task.delay(
instance.user_object.id,
instance.user_subject.id,
stream_list=public_streams,
2021-08-05 02:54:47 +00:00
)
2021-03-23 14:01:49 +00:00
@receiver(signals.post_save, sender=models.User)
# pylint: disable=unused-argument
2021-04-03 17:56:53 +00:00
def populate_streams_on_account_create(sender, instance, created, *args, **kwargs):
2021-04-26 16:15:42 +00:00
"""build a user's feeds when they join"""
2021-03-23 14:01:49 +00:00
if not created or not instance.local:
return
2022-01-04 22:17:14 +00:00
transaction.on_commit(
lambda: populate_streams_on_account_create_command(instance.id)
)
2021-03-23 14:01:49 +00:00
2022-01-04 22:17:14 +00:00
def populate_streams_on_account_create_command(instance_id):
"""wait for the transaction to complete"""
2021-09-07 16:39:42 +00:00
for stream in streams:
2022-01-04 22:17:14 +00:00
populate_stream_task.delay(stream, instance_id)
2021-08-05 02:54:47 +00:00
2021-08-06 02:28:05 +00:00
@receiver(signals.pre_save, sender=models.ShelfBook)
# pylint: disable=unused-argument
2021-08-06 00:13:47 +00:00
def add_statuses_on_shelve(sender, instance, *args, **kwargs):
2021-08-06 02:28:05 +00:00
"""update books stream when user shelves a book"""
2021-08-06 00:13:47 +00:00
if not instance.user.local:
2021-08-06 02:28:05 +00:00
return
2021-08-30 21:06:29 +00:00
book = instance.book
2021-08-06 02:28:05 +00:00
# check if the book is already on the user's shelves
editions = book.parent_work.editions.all()
if models.ShelfBook.objects.filter(user=instance.user, book__in=editions).exists():
2021-08-06 02:28:05 +00:00
return
2021-08-30 21:06:29 +00:00
add_book_statuses_task.delay(instance.user.id, book.id)
2021-08-06 02:28:05 +00:00
2021-08-05 23:39:09 +00:00
@receiver(signals.post_delete, sender=models.ShelfBook)
# pylint: disable=unused-argument
def remove_statuses_on_unshelve(sender, instance, *args, **kwargs):
2021-08-05 23:39:09 +00:00
"""update books stream when user unshelves a book"""
if not instance.user.local:
return
2021-08-30 21:06:29 +00:00
book = instance.book
2021-08-05 23:39:09 +00:00
# check if the book is actually unshelved, not just moved
editions = book.parent_work.editions.all()
if models.ShelfBook.objects.filter(user=instance.user, book__in=editions).exists():
2021-08-05 23:39:09 +00:00
return
2021-08-30 21:06:29 +00:00
remove_book_statuses_task.delay(instance.user.id, book.id)
2021-08-08 00:38:07 +00:00
2021-08-05 02:54:47 +00:00
# ---- TASKS
@app.task(queue=LOW)
2021-08-30 21:06:29 +00:00
def add_book_statuses_task(user_id, book_id):
"""add statuses related to a book on shelve"""
user = models.User.objects.get(id=user_id)
book = models.Edition.objects.get(id=book_id)
BooksStream().add_book_statuses(user, book)
@app.task(queue=LOW)
2021-08-30 21:06:29 +00:00
def remove_book_statuses_task(user_id, book_id):
"""remove statuses about a book from a user's books feed"""
user = models.User.objects.get(id=user_id)
book = models.Edition.objects.get(id=book_id)
BooksStream().remove_book_statuses(user, book)
@app.task(queue=MEDIUM)
2021-08-08 00:38:07 +00:00
def populate_stream_task(stream, user_id):
"""background task for populating an empty activitystream"""
user = models.User.objects.get(id=user_id)
stream = streams[stream]
stream.populate_streams(user)
@app.task(queue=MEDIUM)
2021-08-05 02:54:47 +00:00
def remove_status_task(status_ids):
"""remove a status from any stream it might be in"""
# this can take an id or a list of ids
if not isinstance(status_ids, list):
status_ids = [status_ids]
statuses = models.Status.objects.filter(id__in=status_ids)
for stream in streams.values():
for status in statuses:
stream.remove_object_from_related_stores(status)
@app.task(queue=HIGH)
def add_status_task(status_id, increment_unread=False):
"""add a status to any stream it should be in"""
2021-11-24 18:00:30 +00:00
status = models.Status.objects.select_subclasses().get(id=status_id)
# we don't want to tick the unread count for csv import statuses, idk how better
# to check than just to see if the states is more than a few days old
2021-09-11 18:37:10 +00:00
if status.created_date < timezone.now() - timedelta(days=2):
increment_unread = False
2021-03-23 14:01:49 +00:00
for stream in streams.values():
stream.add_status(status, increment_unread=increment_unread)
2021-08-05 02:54:47 +00:00
@app.task(queue=MEDIUM)
2021-08-05 02:54:47 +00:00
def remove_user_statuses_task(viewer_id, user_id, stream_list=None):
"""remove all statuses by a user from a viewer's stream"""
stream_list = [streams[s] for s in stream_list] if stream_list else streams.values()
viewer = models.User.objects.get(id=viewer_id)
user = models.User.objects.get(id=user_id)
for stream in stream_list:
stream.remove_user_statuses(viewer, user)
@app.task(queue=MEDIUM)
2021-08-05 02:54:47 +00:00
def add_user_statuses_task(viewer_id, user_id, stream_list=None):
2021-09-07 23:33:43 +00:00
"""add all statuses by a user to a viewer's stream"""
2021-08-05 02:54:47 +00:00
stream_list = [streams[s] for s in stream_list] if stream_list else streams.values()
viewer = models.User.objects.get(id=viewer_id)
user = models.User.objects.get(id=user_id)
for stream in stream_list:
2021-08-30 20:44:19 +00:00
stream.add_user_statuses(viewer, user)
2021-09-06 22:39:32 +00:00
@app.task(queue=MEDIUM)
2021-09-06 22:39:32 +00:00
def handle_boost_task(boost_id):
"""remove the original post and other, earlier boosts"""
instance = models.Status.objects.get(id=boost_id)
2021-09-06 23:16:45 +00:00
boosted = instance.boost.boosted_status
2021-09-06 22:39:32 +00:00
2021-10-04 16:47:33 +00:00
# previous boosts of this status
2021-09-06 22:39:32 +00:00
old_versions = models.Boost.objects.filter(
boosted_status__id=boosted.id,
created_date__lt=instance.created_date,
2021-09-08 16:21:15 +00:00
)
2021-09-06 22:39:32 +00:00
for stream in streams.values():
2021-10-04 16:47:33 +00:00
# people who should see the boost (not people who see the original status)
2021-09-06 22:39:32 +00:00
audience = stream.get_stores_for_object(instance)
stream.remove_object_from_related_stores(boosted, stores=audience)
for status in old_versions:
stream.remove_object_from_related_stores(status, stores=audience)
2021-11-24 18:00:30 +00:00
def get_status_type(status):
"""return status type even for boosted statuses"""
status_type = status.status_type.lower()
# Check if current status is a boost
if hasattr(status, "boost"):
# Act in accordance of your findings
if hasattr(status.boost.boosted_status, "review"):
status_type = "review"
if hasattr(status.boost.boosted_status, "comment"):
status_type = "comment"
if hasattr(status.boost.boosted_status, "quotation"):
status_type = "quotation"
return status_type