305 lines
9.7 KiB
Python
305 lines
9.7 KiB
Python
import logging
|
|
from typing import Dict, Optional
|
|
from uuid import uuid4
|
|
|
|
import pendulum
|
|
import requests
|
|
import trafilatura
|
|
from bs4 import BeautifulSoup
|
|
from django.apps import apps
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.files.base import ContentFile
|
|
from django.db import models
|
|
from django.urls import reverse
|
|
from django_extensions.db.models import TimeStampedModel
|
|
from htmldate import find_date
|
|
from imagekit.models import ImageSpecField
|
|
from imagekit.processors import ResizeToFit
|
|
from scrobbles.mixins import ScrobblableConstants, ScrobblableMixin
|
|
from taggit.managers import TaggableManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
BNULL = {"blank": True, "null": True}
|
|
User = get_user_model()
|
|
|
|
|
|
class Domain(TimeStampedModel):
|
|
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
|
|
root = models.CharField(max_length=255)
|
|
name = models.CharField(max_length=255, **BNULL)
|
|
|
|
tags = TaggableManager(blank=True)
|
|
|
|
def __str__(self) -> str:
|
|
if self.name:
|
|
return str(self.name)
|
|
return str(self.root)
|
|
|
|
def scrobbles_for_user(self, user_id):
|
|
from scrobbles.models import Scrobble
|
|
|
|
return Scrobble.objects.filter(
|
|
web_page__domain=self, user_id=user_id
|
|
).order_by("-timestamp")
|
|
|
|
|
|
class WebPage(ScrobblableMixin):
|
|
COMPLETION_PERCENT = getattr(settings, "WEBSITE_COMPLETION_PERCENT", 100)
|
|
|
|
uuid = models.UUIDField(default=uuid4, editable=False, **BNULL)
|
|
url = models.URLField(max_length=500)
|
|
date = models.DateField(**BNULL)
|
|
domain = models.ForeignKey(Domain, on_delete=models.DO_NOTHING, **BNULL)
|
|
extract = models.TextField(**BNULL)
|
|
|
|
image = models.ImageField(upload_to="webpage/", **BNULL)
|
|
image_small = ImageSpecField(
|
|
source="image",
|
|
processors=[ResizeToFit(100, 100)],
|
|
format="JPEG",
|
|
options={"quality": 60},
|
|
)
|
|
image_medium = ImageSpecField(
|
|
source="image",
|
|
processors=[ResizeToFit(300, 300)],
|
|
format="JPEG",
|
|
options={"quality": 75},
|
|
)
|
|
|
|
def __str__(self) -> str:
|
|
if self.title:
|
|
return "{} ({})".format(self.title, self.domain)
|
|
if self.domain:
|
|
return "Unknown ({})".format(self.domain)
|
|
return str(self.uuid)
|
|
|
|
def _raw_domain(self):
|
|
self.url.split("//")[-1].split("/")[0]
|
|
|
|
def _update_extract_from_web(self, raw_text: str = "", force=True):
|
|
if not raw_text:
|
|
raw_text = requests.get(self.url, headers=headers).text
|
|
if not self.extract or force:
|
|
self.extract = trafilatura.extract(raw_text)
|
|
self.save(update_fields=["extract"])
|
|
|
|
def get_absolute_url(self):
|
|
return reverse("webpages:webpage_detail", kwargs={"slug": self.uuid})
|
|
|
|
def get_read_url(self):
|
|
return reverse("webpages:webpage_read", kwargs={"slug": self.uuid})
|
|
|
|
@property
|
|
def estimated_time_to_read_in_seconds(self):
|
|
if not self.extract:
|
|
return 600
|
|
|
|
words_per_minute = getattr(settings, "READING_WORDS_PER_MINUTE", 200)
|
|
words = len(self.extract.split(" "))
|
|
return int(words / words_per_minute) * 60
|
|
|
|
@property
|
|
def estimated_time_to_read_in_minutes(self):
|
|
return int(self.estimated_time_to_read_in_seconds / 60)
|
|
|
|
@property
|
|
def strings(self) -> ScrobblableConstants:
|
|
return ScrobblableConstants(verb="Browsing", tags="earth_americas")
|
|
|
|
@property
|
|
def subtitle(self):
|
|
return self.domain
|
|
|
|
@property
|
|
def primary_image_url(self) -> str:
|
|
if self.image:
|
|
return self.image.url
|
|
return ""
|
|
|
|
def scrobble_for_user(self, user_id):
|
|
Scrobble = apps.get_model("scrobbles", "Scrobble")
|
|
scrobble_data = self.basic_scrobble_data(user_id)
|
|
logger.info(
|
|
"[scrobble_for_user] called for webpage",
|
|
extra={
|
|
"webpage_id": self.id,
|
|
"user_id": user_id,
|
|
"scrobble_data": scrobble_data,
|
|
"media_type": Scrobble.MediaType.WEBPAGE,
|
|
},
|
|
)
|
|
scrobble = Scrobble.create_or_update(self, user_id, scrobble_data)
|
|
# TODO Possibly make this async?
|
|
scrobble.push_to_archivebox()
|
|
return scrobble
|
|
|
|
def scrobbles(self, user):
|
|
Scrobble = apps.get_model("scrobbles", "Scrobble")
|
|
return Scrobble.objects.filter(user=user, web_page=self).order_by(
|
|
"-timestamp"
|
|
)
|
|
|
|
def clean_title(self, title: str, save=True):
|
|
if len(title.split("|")) > 1:
|
|
title = title.split("|")[0]
|
|
if len(title.split("–")) > 1:
|
|
title = title.split("–")[0]
|
|
if len(title.split(" - ")) > 1:
|
|
title = title.split(" - ")[0]
|
|
self.title = title.strip()
|
|
|
|
if save:
|
|
self.save(update_fields=["title"])
|
|
|
|
def _update_domain_from_url(self, save=False):
|
|
domain = self.url.split("//")[-1].split("/")[0].split("www.")[-1]
|
|
self.domain, created = Domain.objects.get_or_create(root=domain)
|
|
|
|
if save:
|
|
self.save(update_fields=["domain"])
|
|
|
|
def _update_title_from_web(self, raw_text: str, save=False):
|
|
if not raw_text:
|
|
return
|
|
|
|
soup = BeautifulSoup(raw_text, "html.parser")
|
|
title_tag = soup.find("title")
|
|
if title_tag and title_tag.string:
|
|
self.title = title_tag.string
|
|
elif not self.title and self.extract:
|
|
first_line = self.extract.split("\n")[0]
|
|
self.title = first_line[:254]
|
|
|
|
if save:
|
|
self.save(update_fields=["title"])
|
|
|
|
def _update_date_from_web(self, save=False):
|
|
try:
|
|
date_str = find_date(str(self.url))
|
|
except ValueError:
|
|
date_str = ""
|
|
if date_str:
|
|
self.date = pendulum.parse(date_str).date()
|
|
|
|
if save:
|
|
self.save(update_fields=["date"])
|
|
|
|
def push_to_archivebox(self, url: str, username: str, password: str):
|
|
login_url = requests.compat.urljoin(url, "admin/login/")
|
|
session = requests.Session()
|
|
response = session.get(login_url)
|
|
csrf_token = response.cookies.get_dict().get("csrftoken")
|
|
response = session.post(
|
|
login_url,
|
|
data={
|
|
"username": username,
|
|
"password": password,
|
|
"csrfmiddlewaretoken": csrf_token,
|
|
},
|
|
)
|
|
try:
|
|
response = session.post(
|
|
requests.compat.urljoin(url, "add/"),
|
|
data={
|
|
"url": self.url + "\n",
|
|
"tags": "vrobbler",
|
|
"depth": "0",
|
|
"parser": "auto",
|
|
},
|
|
timeout=2,
|
|
)
|
|
except requests.exceptions.ReadTimeout:
|
|
return
|
|
|
|
if response.status_code == 200:
|
|
logger.info(
|
|
"Website already exists in archive", extra={"url": self.url}
|
|
)
|
|
else:
|
|
raise Exception(
|
|
f"Failed to push URL to archivebox (Response {response.status_code})"
|
|
)
|
|
|
|
def _extract_image_url_from_html(self, raw_html: str) -> Optional[str]:
|
|
"""Extract og:image or twitter:image from HTML."""
|
|
if not raw_html:
|
|
return None
|
|
|
|
soup = BeautifulSoup(raw_html, "html.parser")
|
|
|
|
og_image = soup.find("meta", property="og:image")
|
|
if og_image and og_image.get("content"):
|
|
return og_image["content"]
|
|
|
|
twitter_image = soup.find("meta", attrs={"name": "twitter:image"})
|
|
if twitter_image and twitter_image.get("content"):
|
|
return twitter_image["content"]
|
|
|
|
return None
|
|
|
|
def _download_and_save_image(self, image_url: str) -> bool:
|
|
"""Download and save the webpage image."""
|
|
if not image_url:
|
|
return False
|
|
|
|
try:
|
|
response = requests.get(image_url, timeout=10)
|
|
response.raise_for_status()
|
|
|
|
filename = image_url.split("/")[-1].split("?")[0]
|
|
if not filename.endswith((".jpg", ".jpeg", ".png", ".webp")):
|
|
filename = "webpage_image.jpg"
|
|
|
|
self.image.save(filename, ContentFile(response.content), save=True)
|
|
return True
|
|
except Exception:
|
|
return False
|
|
|
|
def fetch_data_from_web(self, save=True, force=True):
|
|
raw_text = trafilatura.fetch_url(self.url)
|
|
if not self.extract or force:
|
|
self.extract = trafilatura.extract(
|
|
raw_text,
|
|
include_links=False,
|
|
include_comments=False,
|
|
)
|
|
|
|
if not self.title or force:
|
|
self._update_title_from_web(raw_text)
|
|
|
|
if not self.date or force:
|
|
self._update_date_from_web()
|
|
|
|
if not self.domain or force:
|
|
self._update_domain_from_url()
|
|
|
|
if not self.base_run_time_seconds or force:
|
|
self.base_run_time_seconds = self.estimated_time_to_read_in_seconds
|
|
|
|
if (not self.image or force) and raw_text:
|
|
image_url = self._extract_image_url_from_html(raw_text)
|
|
if image_url:
|
|
self._download_and_save_image(image_url)
|
|
|
|
if save:
|
|
self.save()
|
|
|
|
@classmethod
|
|
def find_or_create(cls, data_dict: Dict) -> "WebPage":
|
|
"""Given a data dict from an manual URL scrobble, does the heavy lifting of looking up
|
|
the url, creating if if doesn't exist yet.
|
|
|
|
"""
|
|
# TODO Add constants for all these data keys
|
|
if "url" not in data_dict.keys():
|
|
logger.error("No url in data dict")
|
|
return
|
|
|
|
webpage = cls.objects.filter(url=data_dict.get("url")).first()
|
|
|
|
if not webpage:
|
|
webpage = cls(url=data_dict.get("url"))
|
|
webpage.fetch_data_from_web(save=True)
|
|
return webpage
|