Select Git revision
base.py 34.53 KiB
import logging
from collections import namedtuple
from enum import Enum
from functools import reduce
from urllib.parse import quote
from django.apps import apps
from django.core.paginator import Paginator
from django.db import models
from django.db.models import Q
from django.db.models.expressions import F, Subquery, Value
from django.utils import timezone
from modelcluster.fields import ParentalKey, ParentalManyToManyField
from taggit.models import ItemBase, Tag, TagBase
from wagtail.admin.panels import FieldPanel, MultiFieldPanel, PublishingPanel
from wagtail.fields import StreamField
from wagtail.models import Page
from wagtail.search import index
from shared.blocks import (
DEFAULT_CONTENT_BLOCKS,
FooterLinksBlock,
MenuItemBlock,
MenuParentBlock,
)
logger = logging.getLogger(__name__)
class SubpageMixin:
"""Must be used in class definition before MetadataPageMixin!"""
@property
def root_page(self):
if not hasattr(self, "_root_page"):
# vypada to hackove ale lze takto pouzit: dle dokumentace get_ancestors
# vraci stranky v poradi od rootu, tedy domovska stranka je druha v poradi
self._root_page = self.get_ancestors().specific()[1]
return self._root_page
def get_meta_image(self):
return self.search_image or self.root_page.get_meta_image()
class ArticleMixin(models.Model):
"""
Common fields for articles.
Must be used in class definition before MetadataPageMixin!
If you want to tag articles, add tags as `tags` field in article page model.
"""
### FIELDS
content = StreamField(
DEFAULT_CONTENT_BLOCKS,
verbose_name="Článek",
blank=True,
use_json_field=True,
)
date = models.DateField("datum", default=timezone.now)
perex = models.TextField("perex")
author = models.CharField("autor", max_length=250, blank=True, null=True)
image = models.ForeignKey(
"wagtailimages.Image",
on_delete=models.PROTECT,
blank=True,
null=True,
verbose_name="obrázek",
)
"""
Hidden field describing the source of shared articles, can be of values "district", "uniweb", "elections"
or "main", depending on from which type of site this article was shared from
"""
shared_type = models.TextField(null=True, blank=True)
"""
Hidden field which links to a Page model of ArticlesMixin page to which this article was shared.
Example: DistrictArticlesPage has shared tag "main", which this article shares as well -> shared_from will contain a reference to DistrictArticlesPage
"""
shared_from = models.ForeignKey(
Page,
null=True,
blank=True,
related_name="+",
on_delete=models.PROTECT,
)
search_fields = Page.search_fields + [
index.SearchField("title"),
index.SearchField("author"),
index.SearchField("perex"),
index.SearchField("content"),
]
### PANELS
content_panels = Page.content_panels + [
FieldPanel("date"),
FieldPanel("perex"),
FieldPanel("content"),
FieldPanel("author"),
FieldPanel("image"),
]
settings_panels = [PublishingPanel()]
class Meta:
abstract = True
@property
def get_original_url(self):
return self.full_url
@property
def get_no_index(self):
"""
Indicates that a link to self should contain rel="noindex"
"""
return self.shared_from is not None
@property
def get_rel(self):
"""
Returns "rel" property for a link to this article
"""
return "rel=noindex" if self.get_no_index else ""
def get_url(self, request=None):
# 'request' kwarg for Wagtail compatibility
if self.shared_from is not None:
return f"{self.shared_from.url}sdilene?sdilene={quote(self.slug)}"
return self.url
def get_full_url(self, request=None):
# 'request' kwarg for Wagtail compatibility
if self.shared_from is not None:
return f"{self.shared_from.full_url}sdilene?sdilene={quote(self.slug)}"
return self.full_url
@property
def articles_page(self):
"""
Returns articles page on which this article is displayed
"""
return (
self.shared_from.get_specific()
if self.shared_from
else self.get_parent().get_specific()
)
@property
def root_page(self):
"""
Returns root page of article, or a root page of Articles page to which this article was shared
"""
if self.shared_from is None:
return self.get_parent().get_ancestors().specific().live().last()
return self.shared_from.get_specific().root_page
@property
def get_tags(self):
"""
Returns all tags, including tags of shared articles from another site
"""
if self.shared_from is not None:
return self.articles_page.search_tags_by_unioned_id_query([self])
return self.tags.all()
@classmethod
def has_tags(cls):
try:
cls._meta.get_field("tags")
except models.FieldDoesNotExist:
return False
return True
def get_meta_image(self):
if hasattr(self, "search_image") and self.search_image:
return self.search_image
return self.image
def get_meta_description(self):
if hasattr(self, "search_description") and self.search_description:
return self.search_description
return self.perex
class MenuMixin(Page):
menu = StreamField(
[("menu_item", MenuItemBlock()), ("menu_parent", MenuParentBlock())],
verbose_name="Menu",
blank=True,
use_json_field=True,
)
menu_panels = [
MultiFieldPanel(
[
FieldPanel("menu"),
],
heading="Nastavení menu",
),
]
class Meta:
abstract = True
class ExtendedMetadataHomePageMixin(models.Model):
"""Use for site home page to define metadata title suffix.
Must be used in class definition before MetadataPageMixin!
"""
title_suffix = models.CharField(
"Přípona titulku stránky",
max_length=100,
blank=True,
null=True,
help_text="Umožňuje přidat příponu k základnímu titulku stránky. Pokud "
"je např. titulek stránky pojmenovaný 'Kontakt' a do přípony vyplníte "
"'MS Pardubice | Piráti', výsledný titulek bude "
"'Kontakt | MS Pardubice | Piráti'. Pokud příponu nevyplníte, použije "
"se název webu.",
)
class Meta:
abstract = True
def get_meta_title_suffix(self):
if self.title_suffix:
return self.title_suffix
if hasattr(super(), "get_meta_title"):
return super().get_meta_title()
return self.get_site().site_name
def get_meta_title(self):
title = super().get_meta_title()
suffix = self.get_meta_title_suffix()
# Covers scenario when title_suffix is not set and evaluates to super().get_meta_title() value.
# Rather than having MS Pardubice | MS Pardubice, just use MS Pardubice alone.
if title != suffix:
return f"{super().get_meta_title()} | {self.get_meta_title_suffix()}"
return title
class ExtendedMetadataPageMixin(models.Model):
"""Use for pages except for home page to use shared metadata title suffix.
There are few rules on how to use this:
- Do not forget to list ExtendedMetadataHomePageMixin among ancestors of the related HomePage class.
- Must be used in class definition before MetadataPageMixin.
- Expects SubpageMixin or equivalent exposing `root_page` property to be used for the page too.
"""
class Meta:
abstract = True
def get_meta_title_suffix(self):
if not hasattr(self, "root_page"):
logger.warning(
"Using `ExtendedMetadataPageMixin` without `SubpageMixin` for %s",
repr(self),
)
return None
if not hasattr(self.root_page, "get_meta_title_suffix"):
logger.warning(
"Using `ExtendedMetadataPageMixin` without `ExtendedMetadataHomePageMixin` on the root page for %s",
repr(self),
)
return None
return self.root_page.get_meta_title_suffix()
def get_meta_title(self):
suffix = self.get_meta_title_suffix()
if not suffix:
return super().get_meta_title()
return f"{super().get_meta_title()} | {self.get_meta_title_suffix()}"
class FooterMixin(models.Model):
footer_links = StreamField(
[
("footer_links", FooterLinksBlock()),
],
verbose_name="Odkazy v zápatí webu",
blank=True,
max_num=1,
use_json_field=True,
)
class Meta:
abstract = True
class SharedTag(TagBase):
class Meta:
verbose_name = "sdílený tag"
verbose_name_plural = "sdílené tagy"
class SharedTaggedDistrictArticle(ItemBase):
tag = models.ForeignKey(
SharedTag, related_name="shared_district_tags", on_delete=models.CASCADE
)
content_object = ParentalKey(
to="district.DistrictArticlePage",
on_delete=models.CASCADE,
related_name="shared_district_articles",
)
class SharedTaggedUniwebArticle(ItemBase):
tag = models.ForeignKey(
SharedTag, related_name="shared_uniweb_tags", on_delete=models.CASCADE
)
content_object = ParentalKey(
to="uniweb.UniwebArticlePage",
on_delete=models.CASCADE,
related_name="shared_uniweb_articles",
)
class SharedTaggedMainArticle(ItemBase):
tag = models.ForeignKey(
SharedTag, related_name="shared_main_tags", on_delete=models.CASCADE
)
content_object = ParentalKey(
to="main.MainArticlePage",
on_delete=models.CASCADE,
related_name="shared_main_articles",
)
class SharedTaggedElectionsArticle(ItemBase):
tag = models.ForeignKey(
SharedTag, related_name="shared_elections_tags", on_delete=models.CASCADE
)
content_object = ParentalKey(
to="elections.ElectionsArticlePage",
on_delete=models.CASCADE,
related_name="shared_elections_articles",
)
class SharedArticlesPageType(Enum):
DISTRICT = "district"
UNIWEB = "uniweb"
MAIN = "main"
ELECTIONS = "elections"
class ArticlesMixin:
def get_shared_tags(self):
"""
Relies on property articles_page being present within the child page, returns shared tags field
"""
return (
self.articles_page.shared_tags
if self.articles_page is not None
else SharedTag.objects.none()
)
def merge_dict(self, aDict: dict, bDict: dict):
"""
Utility for efficiently merging dict objects in lambda queries
"""
aDict.update(bDict)
return aDict
def determine_page_type(self):
"""
Determines which article type to use based on the module from which this method is run from
"""
if self._meta.app_label == "district":
return SharedArticlesPageType.DISTRICT
elif self._meta.app_label == "uniweb":
return SharedArticlesPageType.UNIWEB
elif self._meta.app_label == "main":
return SharedArticlesPageType.MAIN
elif self._meta.app_label == "elections":
return SharedArticlesPageType.ELECTIONS
def evaluate_page_query(self, results):
"""
Utility for merging and materializing articles query to prevent duplicities.
Prefers original articles as opposed to shared ones (if we share an article to the same web that it originates from)
"""
return list(
reduce(
lambda unique, item: unique
if item["union_page_ptr_id"] in unique
and "union_shared_from_id" in item
and item["union_shared_from_id"] is not None
else self.merge_dict(unique, {item["union_page_ptr_id"]: item}),
list(results),
{},
).values()
)
def unique_articles_by_id(self, results):
"""
Utility creating an unique results list with preference for non-shared articles
Prefers original articles as opposed to shared ones (if we share an article to the same web that it originates from)
"""
return list(
reduce(
lambda unique, item: unique
if item.page_ptr.id in unique and item.shared_from is not None
else self.merge_dict(unique, {item.page_ptr.id: item}),
results,
{},
).values()
)
def create_base_shared_query(self, query, original_query):
"""
Returns a query filtered by shared tags,
Filters out page ids that would be duplicates of original query (shared articles dispayed on the same page)
"""
filtered_query = (
query.filter(
~Q(page_ptr_id__in=Subquery(original_query.values("page_ptr_id"))),
shared_tags__slug__in=self.get_shared_tags().values_list(
"slug", flat=True
),
)
if isinstance(original_query, models.QuerySet)
else (
query.filter(
~Q(
page_ptr_id__in=list(
map(lambda article: article.pk, original_query)
)
),
shared_tags__slug__in=self.get_shared_tags().values_list(
"slug", flat=True
),
)
if original_query is not None
else query.filter(
shared_tags__slug__in=self.get_shared_tags().values_list(
"slug", flat=True
),
)
)
)
return filtered_query.live().specific()
def append_all_shared_articles_query(
self,
previous_query: models.QuerySet | None = None,
custom_article_query=None,
):
"""
Creates articles query with shared articles as well as articles pre-selected by previous_query parameter
Returns an unionized query with .values() being applied on it. Unionized queries cannot be annotated or filtered.
If you wish to run annotation or additional filters, use custom_article_query param. This parameter accepts lambdas with
two parameters: shared article query (before unionizing) and shared articles enum, denoting the origin of shared articles
"""
# To prevent circular deps, we get class models during runtime
DistrictArticlePage = apps.get_model(app_label="district.DistrictArticlePage")
UniwebArticlePage = apps.get_model(app_label="uniweb.UniwebArticlePage")
MainArticlePage = apps.get_model(app_label="main.MainArticlePage")
ElectionsArticlePage = apps.get_model(
app_label="elections.ElectionsArticlePage"
)
page_type = self.determine_page_type()
# In order to balance union() requirements for tables with same-fields only, we are adding null fields using values(). These values must be in correct order
main_meta_fields = MainArticlePage._meta.fields
elections_meta_fields = ElectionsArticlePage._meta.fields
district_meta_fields = DistrictArticlePage._meta.fields
uniweb_meta_fields = UniwebArticlePage._meta.fields
# dictionary of class types of fields, assumes that a field cannot have a different type in each web
fields_dict = reduce(
lambda class_dict, field: self.merge_dict(
class_dict, {f"union_{field.column}": field}
),
[
*main_meta_fields,
*elections_meta_fields,
*district_meta_fields,
*uniweb_meta_fields,
],
{},
)
fields_reducer = (
lambda assigned, field: assigned
if field.column == "shared_from_id" or field.column == "shared_type"
else self.merge_dict(assigned, {f"union_{field.column}": F(field.column)})
)
setup_fields_order = lambda orderBy, orderFrom: reduce(
lambda orderTo, field: self.merge_dict(orderTo, {field: orderFrom[field]}),
orderBy.keys(),
{},
)
district_only_fields = reduce(fields_reducer, district_meta_fields, {})
uniweb_only_fields = reduce(fields_reducer, uniweb_meta_fields, {})
main_only_fields = reduce(fields_reducer, main_meta_fields, {})
elections_only_fields = reduce(fields_reducer, elections_meta_fields, {})
key_fields_default_values = {
"union_thumb_image_id": F("search_image_id"),
"union_is_black": Value(False, models.BooleanField()),
"union_article_type": Value(2, models.PositiveSmallIntegerField()),
}
create_complementary_field = lambda key: (
key_fields_default_values[key]
if key in key_fields_default_values
else (
# any type will suffice in the foreign key
Value(
None,
models.ForeignKey(
DistrictArticlePage, blank=True, on_delete=models.SET_NULL
),
)
if isinstance(fields_dict[key], models.ForeignKey)
else Value(None, fields_dict[key].__class__())
)
)
reduce_complementary_fields = lambda complementary, item, notIn: (
self.merge_dict(
complementary, {item[0]: create_complementary_field(item[0])}
)
if item[0] not in notIn
else complementary
)
create_complementary_fields = lambda notIn, fromFields: reduce(
lambda complementary, item: reduce_complementary_fields(
complementary, item, dict(notIn)
),
reduce(
lambda sources, source: sources + list(source.items()), fromFields, []
),
{},
)
district_complementary_fields = create_complementary_fields(
district_only_fields,
[uniweb_only_fields, main_only_fields, elections_only_fields],
)
uniweb_complementary_fields = create_complementary_fields(
uniweb_only_fields,
[district_only_fields, main_only_fields, elections_only_fields],
)
main_complementary_fields = create_complementary_fields(
main_only_fields,
[uniweb_only_fields, district_only_fields, elections_only_fields],
)
elections_complementary_fields = create_complementary_fields(
elections_only_fields,
[uniweb_only_fields, district_only_fields, main_only_fields],
)
main_fields = main_only_fields | main_complementary_fields
elections_fields = setup_fields_order(
main_fields,
elections_only_fields | elections_complementary_fields,
)
district_fields = setup_fields_order(
main_fields,
district_only_fields | district_complementary_fields,
)
uniweb_fields = setup_fields_order(
main_fields,
uniweb_only_fields | uniweb_complementary_fields,
)
district_article_query: models.QuerySet = DistrictArticlePage.objects
uniweb_article_query: models.QuerySet = UniwebArticlePage.objects
main_article_query: models.QuerySet = MainArticlePage.objects
elections_article_query: models.QuerySet = ElectionsArticlePage.objects
apply_additional_filter = (
lambda query: custom_article_query(query)
if custom_article_query is not None
else query
)
create_query_by_slug = lambda query: apply_additional_filter(
self.create_base_shared_query(query, previous_query)
)
district_by_slug = create_query_by_slug(district_article_query)
uniweb_by_slug = create_query_by_slug(uniweb_article_query)
main_by_slug = create_query_by_slug(main_article_query)
elections_by_slug = create_query_by_slug(elections_article_query)
shared_field = Value(
self.page_ptr.id if hasattr(self, "page_ptr") else None, # preview fix
output_field=models.ForeignKey(
Page, null=True, blank=True, related_name="+", on_delete=models.PROTECT
),
)
main_by_values = main_by_slug.values(
**main_fields,
union_shared_from_id=shared_field,
union_shared_type=Value(
SharedArticlesPageType.MAIN.value, output_field=models.TextField()
),
)
uniweb_by_values = uniweb_by_slug.values(
**uniweb_fields,
union_shared_from_id=shared_field,
union_shared_type=Value(
SharedArticlesPageType.UNIWEB.value, output_field=models.TextField()
),
)
district_by_values = district_by_slug.values(
**district_fields,
union_shared_from_id=shared_field,
union_shared_type=Value(
SharedArticlesPageType.DISTRICT.value, output_field=models.TextField()
),
)
elections_by_values = elections_by_slug.values(
**elections_fields,
union_shared_from_id=shared_field,
union_shared_type=Value(
SharedArticlesPageType.ELECTIONS.value, output_field=models.TextField()
),
)
empty_shared_field = Value(
None,
output_field=models.ForeignKey(
Page, null=True, blank=True, related_name="+", on_delete=models.PROTECT
),
)
empty_shared_type = Value(None, output_field=models.TextField())
if previous_query is not None:
prepared_query = previous_query.live().specific()
if page_type == SharedArticlesPageType.DISTRICT:
prepared_query = prepared_query.values(
**district_fields,
union_shared_from_id=empty_shared_field,
union_shared_type=empty_shared_type,
)
elif page_type == SharedArticlesPageType.UNIWEB:
prepared_query = prepared_query.values(
**uniweb_fields,
union_shared_from_id=empty_shared_field,
union_shared_type=empty_shared_type,
)
elif page_type == SharedArticlesPageType.MAIN:
prepared_query = prepared_query.values(
**main_fields,
union_shared_from_id=empty_shared_field,
union_shared_type=empty_shared_type,
)
elif page_type == SharedArticlesPageType.ELECTIONS:
prepared_query = prepared_query.values(
**elections_fields,
union_shared_from_id=empty_shared_field,
union_shared_type=empty_shared_type,
)
if self.get_shared_tags().count() == 0:
return prepared_query.order_by("-union_date")
return (
prepared_query.union(main_by_values)
.union(uniweb_by_values)
.union(district_by_values)
.union(elections_by_values)
.order_by("-union_date")
)
else:
return (
main_by_values.union(uniweb_by_values)
.union(district_by_values)
.union(elections_by_values)
.order_by("-union_date")
)
def materialize_shared_articles_query(self, results):
"""
Corresponding method to append_all_shared_articles_query.
Materializes article query as article type corresponding to the module from which
this function is run. Put query from append_all_shared_articles_query as results parameter.
"""
# To prevent circular deps, we get class models during runtime
page_type = self.determine_page_type()
DistrictArticlePage = apps.get_model(app_label="district.DistrictArticlePage")
UniwebArticlePage = apps.get_model(app_label="uniweb.UniwebArticlePage")
MainArticlePage = apps.get_model(app_label="main.MainArticlePage")
ElectionsArticlePage = apps.get_model(
app_label="elections.ElectionsArticlePage"
)
main_meta_fields = MainArticlePage._meta.fields
elections_meta_fields = ElectionsArticlePage._meta.fields
district_meta_fields = DistrictArticlePage._meta.fields
uniweb_meta_fields = UniwebArticlePage._meta.fields
assign_to_model = lambda unioned: lambda assignment, field: self.merge_dict(
assignment, {field.column: unioned[f"union_{field.column}"]}
)
evaluated = self.evaluate_page_query(
results
) # We MUST eval here since we can't turn values() into concrete class instances in QuerySet after union
if page_type == SharedArticlesPageType.DISTRICT:
return list(
map(
lambda unioned: DistrictArticlePage(
**reduce(assign_to_model(unioned), district_meta_fields, {})
),
evaluated,
)
)
if page_type == SharedArticlesPageType.UNIWEB:
return list(
map(
lambda unioned: UniwebArticlePage(
**reduce(assign_to_model(unioned), uniweb_meta_fields, {})
),
evaluated,
)
)
if page_type == SharedArticlesPageType.MAIN:
return list(
map(
lambda unioned: MainArticlePage(
**reduce(assign_to_model(unioned), main_meta_fields, {})
),
evaluated,
)
)
if page_type == SharedArticlesPageType.ELECTIONS:
return list(
map(
lambda unioned: ElectionsArticlePage(
**reduce(assign_to_model(unioned), elections_meta_fields, {})
),
evaluated,
)
)
def get_page_with_shared_articles(
self, query: models.QuerySet, page_size: int, page: int
):
"""
Returns Page object whose property object_list has been materialized, uses Paginator internally
"""
paginator = Paginator(
query,
page_size,
)
paginator_page = paginator.get_page(page)
paginator_page.object_list = self.materialize_shared_articles_query(
paginator_page.object_list
)
return paginator_page
def get_article_page_by_slug(self, slug: str):
"""
Filters articles + shared articles based on "tag" field,
returns first result sorted by date
"""
articles = self.append_all_shared_articles_query(
custom_article_query=lambda query: query.filter(slug=slug)
)[:1]
return self.materialize_shared_articles_query(articles)[0]
def setup_article_page_context(self, request):
"""
Use this method to setup page context for shared article at /sdilene
"""
slug = request.GET.get("sdilene", "")
return self.get_article_page_by_slug(slug).serve(request)
def materialize_articles_as_id_only(self, articles):
"""
Returns a temporary article class with pk, shared and date as the only properties.
Useful when optimizing large article queries
"""
TmpArticle = namedtuple("TemporaryArticle", field_names=["page_ptr"])
TmpPrimaryKey = namedtuple("TemporaryPk", field_names=["id"])
return list(
map(
lambda unioned: TmpArticle(
page_ptr=TmpPrimaryKey(id=unioned["union_page_ptr_id"]),
),
articles.values("union_page_ptr_id", "union_date"),
)
)
def search_tags_with_count(self, articles: list):
"""
Returns a list of tags based on article ids with each count
"""
if isinstance(articles, models.QuerySet):
articles = self.materialize_articles_as_id_only(articles)
article_ids = list(map(lambda article: article.page_ptr.id, articles))
tag_list = list(
Tag.objects.filter(
district_districtarticletag_items__content_object_id__in=article_ids
)
.union(
Tag.objects.filter(
main_mainarticletag_items__content_object_id__in=article_ids
),
Tag.objects.filter(
uniweb_uniwebarticletag_items__content_object_id__in=article_ids
),
Tag.objects.filter(
elections_electionsarticletag_items__content_object_id__in=article_ids
),
all=True,
)
.order_by("slug")
.values()
)
tag_aggregate = reduce(
lambda aggregate, tag: self.merge_dict(
aggregate,
{tag["slug"]: aggregate[tag["slug"]] + 1}
if tag["slug"] in aggregate
else {tag["slug"]: 1},
),
tag_list,
{},
)
already_present = {}
unique_tags = []
for tag in tag_list:
tag["count"] = tag_aggregate[tag["slug"]]
if tag["slug"] not in already_present:
unique_tags.append(tag)
already_present[tag["slug"]] = True
return unique_tags
def search_tags_by_unioned_id_query(
self,
articles: list,
):
"""
Search tags based on article query or list of articles.
Returns a list of Tag objects, unique and sorted by slug
"""
if isinstance(articles, models.QuerySet):
articles = self.materialize_articles_as_id_only(articles)
article_ids = list(map(lambda article: article.page_ptr.id, articles))
tag_query = Tag.objects.filter(
Q(district_districtarticletag_items__content_object_id__in=article_ids)
| Q(main_mainarticletag_items__content_object_id__in=article_ids)
| Q(uniweb_uniwebarticletag_items__content_object_id__in=article_ids)
| Q(elections_electionsarticletag_items__content_object_id__in=article_ids)
)
return tag_query.order_by("slug").distinct("slug")
def search_articles(
self,
query: str,
page_size: int,
page: int,
previous_query: models.QuerySet | None = None,
):
"""
Uses wagtail search to lookup articles based on a phrase. Accepts a string phrase query + a previous_query param, which can be any articles query that you want to filter by.
Returns a list of articles with models based on from which module is this method run.
To optimize search results we use paginator internally
"""
DistrictArticlePage = apps.get_model(app_label="district.DistrictArticlePage")
UniwebArticlePage = apps.get_model(app_label="uniweb.UniwebArticlePage")
MainArticlePage = apps.get_model(app_label="main.MainArticlePage")
ElectionsArticlePage = apps.get_model(
app_label="elections.ElectionsArticlePage"
)
# .search() runs annotate, so its impossible to search after .union()
# .search() also returns an object that cannot be broken down by .values()
# therefore, shared search has to happen here
search_factory = lambda search_query: list(
search_query.search(query).annotate_score("score")
)
current_query = search_factory(previous_query)
shared_district_search = search_factory(
self.create_base_shared_query(DistrictArticlePage.objects, current_query)
)
shared_uniweb_search = search_factory(
self.create_base_shared_query(UniwebArticlePage.objects, current_query)
)
shared_main_search = search_factory(
self.create_base_shared_query(MainArticlePage.objects, current_query)
)
shared_elections_search = search_factory(
self.create_base_shared_query(ElectionsArticlePage.objects, current_query)
)
# .search is not lazy either, making this the best optimized query possible AFAIK
sorted = self.unique_articles_by_id(
current_query
+ shared_district_search
+ shared_uniweb_search
+ shared_main_search
+ shared_elections_search
)
sorted.sort(key=lambda item: item.score)
sorted = Paginator(sorted, page_size).get_page(page)
sorted_ids = list(map(lambda article: article.pk, sorted))
converted_query = self.materialize_shared_articles_query(
self.append_all_shared_articles_query(
previous_query.filter(page_ptr_id__in=sorted_ids),
custom_article_query=lambda query: query.filter(
page_ptr_id__in=sorted_ids
),
)
)
converted_query_map = reduce(
lambda map, article: self.merge_dict(map, {article.pk: article}),
converted_query,
{},
)
sorted_final_result = []
for sorted_result in sorted:
sorted_final_result.append(converted_query_map[sorted_result.pk])
return sorted_final_result
def filter_by_tag_name(self, tag: str):
"""
Returns a dict which can be used to filter articles based on tag name
"""
return {
"tags__name": tag,
}
class Meta:
abstract = True
class ArticlesPageMixin(ArticlesMixin, models.Model):
def get_shared_tags(self):
"""
Overrides get_shared_tags from ArticlesMixin, returns shared tags
"""
return self.shared_tags
shared_tags = ParentalManyToManyField(
"shared.SharedTag",
verbose_name="Výběr tagů pro články sdílené mezi sítěmi",
help_text="Pro výběr jednoho tagu klikněte na tag a uložte nebo publikujte stránku. Pro výběr více tagů využijte podržte Ctrl a vyberte příslušné tagy.",
blank=True,
)
content_panels = Page.content_panels + [FieldPanel("shared_tags")]
class Meta:
abstract = True