moviewyrm/bookwyrm/views/helpers.py

187 lines
5.5 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
import dateutil.parser
import dateutil.tz
from dateutil.parser import ParserError
2021-01-12 22:43:59 +00:00
from requests import HTTPError
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
class WebFingerError(Exception):
2021-12-06 05:47:04 +00:00
"""empty error class for problems finding user information with webfinger"""
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
if query[0] == "@":
query = query[1:]
try:
domain = query.split("@")[1]
except IndexError:
2021-01-12 21:47:00 +00:00
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 10:38:28 +00:00
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",
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
2021-08-07 18:15:02 +00:00
def get_landing_books():
"""list of books for the landing page"""
return list(
set(
models.Edition.objects.filter(
review__published_date__isnull=False,
review__deleted=False,
review__user__local=True,
review__privacy__in=["public", "unlisted"],
)
.exclude(cover__exact="")
.distinct()
.order_by("-review__published_date")[:6]
2021-03-21 02:14:41 +00:00
)
2021-05-22 16:55:38 +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)
return response