moviewyrm/bookwyrm/views/helpers.py

224 lines
6.7 KiB
Python
Raw Normal View History

2021-03-08 16:49:10 +00:00
""" helper functions used in various views """
2021-01-12 22:43:59 +00:00
import re
from datetime import datetime, timedelta
import dateutil.parser
import dateutil.tz
from dateutil.parser import ParserError
2021-01-12 22:43:59 +00:00
from requests import HTTPError
2021-11-24 12:37:09 +00:00
from django.db.models import Q
from django.conf import settings as django_settings
2022-03-02 09:12:32 +00:00
from django.shortcuts import redirect
from django.http import Http404
from django.utils import translation
2021-01-12 22:43:59 +00:00
from bookwyrm import activitypub, models, settings
2021-01-12 22:43:59 +00:00
from bookwyrm.connectors import ConnectorException, get_data
2021-01-13 19:45:08 +00:00
from bookwyrm.status import create_generated_note
2021-01-12 22:43:59 +00:00
from bookwyrm.utils import regex
2021-01-12 18:44:17 +00:00
2021-12-06 06:02:47 +00:00
# pylint: disable=unnecessary-pass
class WebFingerError(Exception):
2021-12-06 05:47:04 +00:00
"""empty error class for problems finding user information with webfinger"""
2021-12-06 05:59:51 +00:00
pass
2021-12-06 06:02:47 +00:00
2021-02-23 20:41:37 +00:00
def get_user_from_username(viewer, username):
2021-04-26 16:15:42 +00:00
"""helper function to resolve a localname or a username to a user"""
2021-05-23 04:33:56 +00:00
if viewer.is_authenticated and viewer.localname == username:
# that's yourself, fool
return viewer
# raises 404 if the user isn't found
2021-01-12 20:05:30 +00:00
try:
2021-02-23 21:05:43 +00:00
return models.User.viewer_aware_objects(viewer).get(localname=username)
2021-01-12 20:05:30 +00:00
except models.User.DoesNotExist:
pass
# if the localname didn't match, try the username
try:
2021-02-23 20:41:37 +00:00
return models.User.viewer_aware_objects(viewer).get(username=username)
except models.User.DoesNotExist:
raise Http404()
2021-01-12 20:05:30 +00:00
def is_api_request(request):
2021-04-26 16:15:42 +00:00
"""check whether a request is asking for html or data"""
return "json" in request.headers.get("Accept", "") or re.match(
r".*\.json/?$", request.path
)
2021-01-12 20:05:30 +00:00
2021-02-23 19:34:15 +00:00
def is_bookwyrm_request(request):
2021-04-26 16:15:42 +00:00
"""check if the request is coming from another bookwyrm instance"""
2021-03-08 16:49:10 +00:00
user_agent = request.headers.get("User-Agent")
2021-06-18 21:12:56 +00:00
if user_agent is None or re.search(regex.BOOKWYRM_USER_AGENT, user_agent) is None:
2021-01-12 21:47:00 +00:00
return False
return True
def handle_remote_webfinger(query):
2021-04-26 16:15:42 +00:00
"""webfingerin' other servers"""
2021-01-12 21:47:00 +00:00
user = None
# usernames could be @user@domain or user@domain
if not query:
return None
2021-03-08 16:49:10 +00:00
if query[0] == "@":
2021-01-12 21:47:00 +00:00
query = query[1:]
try:
2021-03-08 16:49:10 +00:00
domain = query.split("@")[1]
2021-01-12 21:47:00 +00:00
except IndexError:
return None
try:
2021-04-08 16:59:21 +00:00
user = models.User.objects.get(username__iexact=query)
2021-01-12 21:47:00 +00:00
except models.User.DoesNotExist:
2021-09-18 18:32:00 +00:00
url = f"https://{domain}/.well-known/webfinger?resource=acct:{query}"
2021-01-12 21:47:00 +00:00
try:
data = get_data(url)
except (ConnectorException, HTTPError):
return None
2021-03-08 16:49:10 +00:00
for link in data.get("links"):
if link.get("rel") == "self":
2021-01-12 21:47:00 +00:00
try:
user = activitypub.resolve_remote_id(
2021-03-08 16:49:10 +00:00
link["href"], model=models.User
2021-01-12 21:47:00 +00:00
)
except (KeyError, activitypub.ActivitySerializerError):
2021-01-12 21:47:00 +00:00
return None
return user
2021-11-28 09:09:29 +00:00
def subscribe_remote_webfinger(query):
"""get subscribe template from other servers"""
template = None
# usernames could be @user@domain or user@domain
if not query:
return WebFingerError("invalid_username")
if query[0] == "@":
query = query[1:]
2021-11-28 09:09:29 +00:00
try:
domain = query.split("@")[1]
except IndexError:
return WebFingerError("invalid_username")
2021-11-28 09:09:29 +00:00
url = f"https://{domain}/.well-known/webfinger?resource=acct:{query}"
2021-11-28 09:09:29 +00:00
try:
data = get_data(url)
except (ConnectorException, HTTPError):
return WebFingerError("user_not_found")
2021-11-28 09:09:29 +00:00
for link in data.get("links"):
if link.get("rel") == "http://ostatus.org/schema/1.0/subscribe":
template = link["template"]
return template
2021-11-28 10:38:28 +00:00
def get_edition(book_id):
2021-04-26 16:15:42 +00:00
"""look up a book in the db and return an edition"""
book = models.Book.objects.select_subclasses().get(id=book_id)
if isinstance(book, models.Work):
book = book.default_edition
return book
2021-01-13 19:45:08 +00:00
def handle_reading_status(user, shelf, book, privacy):
2021-04-26 16:15:42 +00:00
"""post about a user reading a book"""
2021-01-13 19:45:08 +00:00
# tell the world about this cool thing that happened
try:
message = {
2021-03-08 16:49:10 +00:00
"to-read": "wants to read",
"reading": "started reading",
"read": "finished reading",
2022-02-28 19:56:59 +00:00
"stopped-reading": "stopped reading",
2021-01-13 19:45:08 +00:00
}[shelf.identifier]
except KeyError:
# it's a non-standard shelf, don't worry about it
return
2021-03-08 16:49:10 +00:00
status = create_generated_note(user, message, mention_books=[book], privacy=privacy)
2021-01-13 19:45:08 +00:00
status.save()
2021-01-26 16:31:55 +00:00
def is_blocked(viewer, user):
2021-04-26 16:15:42 +00:00
"""is this viewer blocked by the user?"""
2021-01-26 16:31:55 +00:00
if viewer.is_authenticated and viewer in user.blocks.all():
return True
return False
2021-03-21 02:14:41 +00:00
def load_date_in_user_tz_as_utc(date_str: str, user: models.User) -> datetime:
"""ensures that data is stored consistently in the UTC timezone"""
if not date_str:
return None
user_tz = dateutil.tz.gettz(user.preferred_timezone)
date = dateutil.parser.parse(date_str, ignoretz=True)
try:
return date.replace(tzinfo=user_tz).astimezone(dateutil.tz.UTC)
except ParserError:
return None
def set_language(user, response):
"""Updates a user's language"""
if user.preferred_language:
translation.activate(user.preferred_language)
response.set_cookie(
settings.LANGUAGE_COOKIE_NAME,
user.preferred_language,
expires=datetime.now() + timedelta(seconds=django_settings.SESSION_COOKIE_AGE),
)
return response
2021-11-24 12:37:09 +00:00
def filter_stream_by_status_type(activities, allowed_types=None):
"""filter out activities based on types"""
if not allowed_types:
allowed_types = []
if "review" not in allowed_types:
activities = activities.filter(
Q(review__isnull=True), Q(boost__boosted_status__review__isnull=True)
)
if "comment" not in allowed_types:
activities = activities.filter(
Q(comment__isnull=True), Q(boost__boosted_status__comment__isnull=True)
)
if "quotation" not in allowed_types:
activities = activities.filter(
Q(quotation__isnull=True), Q(boost__boosted_status__quotation__isnull=True)
)
if "everything" not in allowed_types:
activities = activities.filter(
Q(generatednote__isnull=True),
Q(boost__boosted_status__generatednote__isnull=True),
)
return activities
2022-03-02 09:12:32 +00:00
2022-03-02 09:47:08 +00:00
2022-03-02 09:12:32 +00:00
def maybe_redirect_local_path(request, model):
"""
2022-03-12 04:14:45 +00:00
if the request had an invalid path, return a permanent redirect response to the
correct one, including a slug if any.
2022-03-02 09:12:32 +00:00
if path is valid, returns False.
"""
# don't redirect empty path for unit tests which currently have this
2022-03-12 04:14:45 +00:00
if request.path in ("/", model.local_path):
2022-03-02 09:12:32 +00:00
return False
2022-03-02 09:47:08 +00:00
new_path = model.local_path
2022-03-02 09:12:32 +00:00
if len(request.GET) > 0:
new_path = f"{model.local_path}?{request.GET.urlencode()}"
return redirect(new_path, permanent=True)