594 lines
22 KiB
Python
594 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
|
#
|
|
# combo - content management system
|
|
# Copyright (C) 2014-2015 Entr'ouvert
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify it
|
|
# under the terms of the GNU Affero General Public License as published
|
|
# by the Free Software Foundation, either version 3 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU Affero General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU Affero General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
|
|
import copy
|
|
import logging
|
|
|
|
from django import template
|
|
from django.conf import settings
|
|
from django.db import models
|
|
from django.forms import models as model_forms
|
|
from django.forms import Select
|
|
from django.utils.translation import ugettext_lazy as _
|
|
|
|
from jsonfield import JSONField
|
|
|
|
from combo.data.models import CellBase
|
|
from combo.data.library import register_cell_class
|
|
from combo.utils import requests
|
|
|
|
from .utils import get_wcs_json, is_wcs_enabled, get_wcs_services
|
|
|
|
@register_cell_class
|
|
class WcsFormCell(CellBase):
|
|
template_name = 'combo/wcs/form.html'
|
|
|
|
formdef_reference = models.CharField(_('Form'), max_length=150)
|
|
|
|
cached_title = models.CharField(_('Title'), max_length=150)
|
|
cached_url = models.URLField(_('URL'))
|
|
cached_json = JSONField(blank=True)
|
|
|
|
is_enabled = classmethod(is_wcs_enabled)
|
|
|
|
class Meta:
|
|
verbose_name = _('Form Link')
|
|
|
|
def get_default_form_class(self):
|
|
from .forms import WcsFormCellForm
|
|
return WcsFormCellForm
|
|
|
|
def save(self, *args, **kwargs):
|
|
if self.formdef_reference:
|
|
wcs_key, form_slug = self.formdef_reference.split(':')
|
|
wcs_site = get_wcs_services().get(wcs_key)
|
|
forms_response_json = get_wcs_json(wcs_site, 'api/formdefs/')
|
|
if isinstance(forms_response_json, dict):
|
|
# forward compability with future w.c.s. API
|
|
forms_response_json = forms_response_json.get('data')
|
|
for form in forms_response_json:
|
|
slug = form.get('slug')
|
|
if slug == form_slug:
|
|
self.cached_title = form.get('title')
|
|
self.cached_url = form.get('url')
|
|
self.cached_json = form
|
|
return super(WcsFormCell, self).save(*args, **kwargs)
|
|
|
|
def get_cell_extra_context(self, context):
|
|
context = super(WcsFormCell, self).get_cell_extra_context(context)
|
|
context['slug'] = self.formdef_reference.split(':')[-1]
|
|
context['title'] = self.cached_title
|
|
context['url'] = self.cached_url
|
|
if self.cached_json:
|
|
for attribute in self.cached_json:
|
|
if not attribute in context:
|
|
context[attribute] = self.cached_json.get(attribute)
|
|
return context
|
|
|
|
def get_additional_label(self):
|
|
if not self.cached_title:
|
|
return
|
|
return self.cached_title
|
|
|
|
def render_for_search(self):
|
|
return ''
|
|
|
|
def get_external_links_data(self):
|
|
if not (self.cached_url and self.cached_title):
|
|
return []
|
|
text = ''
|
|
if self.cached_json:
|
|
text = ' '.join([self.cached_json.get('description', ''),
|
|
' '.join(self.cached_json.get('keywords', []))]).strip()
|
|
return [{
|
|
'url': self.cached_url,
|
|
'title': self.cached_title,
|
|
'text': text,
|
|
}]
|
|
|
|
def get_asset_slots(self):
|
|
slots = {}
|
|
for slot_template_key, slot_template_data in settings.WCS_FORM_ASSET_SLOTS.items():
|
|
suffix = ''
|
|
if slot_template_data.get('suffix'):
|
|
suffix = ' (%s)' % slot_template_data['suffix']
|
|
slot_key = 'wcs:form:%s:%s' % (slot_template_key, self.formdef_reference)
|
|
slots[slot_key] = {
|
|
'label': u'%(prefix)s — %(label)s%(suffix)s' % {
|
|
'prefix': slot_template_data['prefix'],
|
|
'label': self.cached_title,
|
|
'suffix': suffix}}
|
|
slots[slot_key].update(slot_template_data)
|
|
return slots
|
|
|
|
|
|
class WcsCommonCategoryCell(CellBase):
|
|
is_enabled = classmethod(is_wcs_enabled)
|
|
category_reference = models.CharField(_('Category'), max_length=150)
|
|
|
|
cached_title = models.CharField(_('Title'), max_length=150)
|
|
cached_description = models.TextField(_('Description'), blank=True)
|
|
cached_url = models.URLField(_('Cached URL'))
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def save(self, *args, **kwargs):
|
|
if self.category_reference:
|
|
wcs_key, category_slug = self.category_reference.split(':')
|
|
wcs_site = get_wcs_services().get(wcs_key)
|
|
categories_response_json = get_wcs_json(wcs_site, 'api/categories/')
|
|
for category in categories_response_json.get('data'):
|
|
slug = category.get('slug')
|
|
if slug == category_slug:
|
|
self.cached_title = category.get('title')
|
|
self.cached_description = category.get('description') or ''
|
|
self.cached_url = category.get('url')
|
|
return super(WcsCommonCategoryCell, self).save(*args, **kwargs)
|
|
|
|
def get_additional_label(self):
|
|
if not self.cached_title:
|
|
return
|
|
return self.cached_title
|
|
|
|
def get_asset_slots(self):
|
|
slots = {}
|
|
for slot_template_key, slot_template_data in settings.WCS_CATEGORY_ASSET_SLOTS.items():
|
|
suffix = ''
|
|
if slot_template_data.get('suffix'):
|
|
suffix = ' (%s)' % slot_template_data['suffix']
|
|
slot_key = 'wcs:category:%s:%s' % (slot_template_key, self.category_reference)
|
|
slots[slot_key] = {
|
|
'label': u'%(prefix)s — %(label)s%(suffix)s' % {
|
|
'prefix': slot_template_data['prefix'],
|
|
'label': self.cached_title,
|
|
'suffix': suffix}}
|
|
slots[slot_key].update(slot_template_data)
|
|
return slots
|
|
|
|
|
|
@register_cell_class
|
|
class WcsCategoryCell(WcsCommonCategoryCell):
|
|
template_name = 'combo/wcs/category.html'
|
|
|
|
class Meta:
|
|
verbose_name = _('Category Link')
|
|
|
|
def get_default_form_class(self):
|
|
from .forms import WcsCategoryCellForm
|
|
return WcsCategoryCellForm
|
|
|
|
def get_cell_extra_context(self, context):
|
|
context = super(WcsCategoryCell, self).get_cell_extra_context(context)
|
|
if not self.category_reference:
|
|
return context
|
|
context['slug'] = self.category_reference.split(':')[-1]
|
|
context['title'] = self.cached_title
|
|
context['description'] = self.cached_description
|
|
context['url'] = self.cached_url
|
|
return context
|
|
|
|
|
|
class WcsBlurpMixin(object):
|
|
is_enabled = classmethod(is_wcs_enabled)
|
|
cache_duration = 5
|
|
api_url = None
|
|
warn_on_404 = True
|
|
|
|
def get_api_url(self, context):
|
|
return self.api_url
|
|
|
|
def get_data(self, context):
|
|
if context.get('placeholder_search_mode'):
|
|
# don't call webservices when we're just looking for placeholders
|
|
return {}
|
|
if self.wcs_site:
|
|
try:
|
|
wcs_sites = {self.wcs_site: get_wcs_services()[self.wcs_site]}
|
|
except KeyError:
|
|
# in case of the site disappeared from settings
|
|
return {}
|
|
else:
|
|
wcs_sites = get_wcs_services()
|
|
|
|
wcs_sites = copy.deepcopy(wcs_sites)
|
|
returns = set([])
|
|
api_url = self.get_api_url(context)
|
|
for wcs_slug, wcs_site in wcs_sites.items():
|
|
url = wcs_site.get('url')
|
|
if not url.endswith('/'):
|
|
url += '/'
|
|
wcs_site['base_url'] = url
|
|
wcs_site['slug'] = wcs_slug
|
|
|
|
response = requests.get(
|
|
api_url,
|
|
remote_service=wcs_site,
|
|
user=getattr(context.get('request'), 'user', None),
|
|
cache_duration=self.cache_duration,
|
|
raise_if_not_cached=not(context.get('synchronous')),
|
|
log_errors=False)
|
|
returns.add(response.status_code)
|
|
if response.status_code == 200:
|
|
json_response = response.json()
|
|
if isinstance(json_response, list):
|
|
# backward compat with older w.c.s.
|
|
wcs_site['data'] = json_response
|
|
elif json_response.get('err', 0) == 0:
|
|
wcs_site['data'] = json_response['data']
|
|
else:
|
|
# skip errors
|
|
continue
|
|
# and mark all items with the site info
|
|
for item in wcs_site['data']:
|
|
item['site_slug'] = wcs_slug
|
|
|
|
if not 200 in returns: # not a single valid answer
|
|
if returns != set([404]) or self.warn_on_404:
|
|
logging.error('failed to get data from any %s (%r)', api_url, returns)
|
|
|
|
return wcs_sites
|
|
|
|
def get_cell_extra_context(self, context):
|
|
return {self.variable_name: self.get_data(context)}
|
|
|
|
|
|
class WcsDataBaseCell(CellBase, WcsBlurpMixin):
|
|
is_enabled = classmethod(is_wcs_enabled)
|
|
wcs_site = models.CharField(_('Site'), max_length=50, blank=True)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def get_additional_label(self):
|
|
wcs_sites = dict([(x, y.get('title')) for x, y in get_wcs_services().items()])
|
|
if len(wcs_sites.keys()) < 2:
|
|
return ''
|
|
if self.wcs_site in wcs_sites:
|
|
return wcs_sites[self.wcs_site]
|
|
return _('All Sites')
|
|
|
|
def get_form_fields(self):
|
|
if len(get_wcs_services()) == 1:
|
|
return []
|
|
return ['wcs_site']
|
|
|
|
def get_form_widgets(self):
|
|
if len(get_wcs_services()) == 1:
|
|
return {}
|
|
combo_wcs_sites = [('', _('All'))]
|
|
wcs_sites = [(x, y.get('title')) for x, y in get_wcs_services().items()]
|
|
wcs_sites.sort(key=lambda x: x[1])
|
|
combo_wcs_sites.extend(wcs_sites)
|
|
return {'wcs_site': Select(choices=combo_wcs_sites)}
|
|
|
|
def get_default_form_class(self):
|
|
return model_forms.modelform_factory(self.__class__,
|
|
fields=self.get_form_fields(),
|
|
widgets=self.get_form_widgets())
|
|
|
|
def get_cell_extra_context(self, context):
|
|
extra_context = super(WcsDataBaseCell, self).get_cell_extra_context(context)
|
|
extra_context.update(WcsBlurpMixin.get_cell_extra_context(self, context))
|
|
return extra_context
|
|
|
|
|
|
class WcsUserDataBaseCell(WcsDataBaseCell):
|
|
warn_on_404 = False
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def is_visible(self, user=None):
|
|
if not user or user.is_anonymous:
|
|
return False
|
|
return super(WcsUserDataBaseCell, self).is_visible(user)
|
|
|
|
|
|
@register_cell_class
|
|
class WcsCurrentFormsCell(WcsUserDataBaseCell):
|
|
variable_name = 'user_forms'
|
|
|
|
categories = JSONField(_('Categories'), blank=True)
|
|
current_forms = models.BooleanField(_('Current Forms'), default=True)
|
|
done_forms = models.BooleanField(_('Done Forms'), default=False)
|
|
|
|
class Meta:
|
|
verbose_name = _('User Forms')
|
|
|
|
def get_default_form_class(self):
|
|
from .forms import WcsCurrentFormsCellForm
|
|
return WcsCurrentFormsCellForm
|
|
|
|
def get_api_url(self, context):
|
|
user = self.get_concerned_user(context)
|
|
if user:
|
|
user_name_id = user.get_name_id()
|
|
if user_name_id:
|
|
return '/api/users/%s/forms?limit=100&sort=desc' % user_name_id
|
|
return '/api/user/forms?limit=100&sort=desc'
|
|
|
|
@property
|
|
def template_name(self):
|
|
if self.current_forms and self.done_forms:
|
|
return 'combo/wcs/user_all_forms.html'
|
|
if self.done_forms:
|
|
return 'combo/wcs/user_done_forms.html'
|
|
return 'combo/wcs/current_forms.html'
|
|
|
|
def get_additional_label(self):
|
|
initial_label = super(WcsCurrentFormsCell, self).get_additional_label()
|
|
if self.current_forms and self.done_forms:
|
|
label = _('All Forms')
|
|
elif self.done_forms:
|
|
label = _('Done Forms')
|
|
else:
|
|
label = _('Current Forms')
|
|
if initial_label:
|
|
return '%s - %s' % (initial_label, label)
|
|
return label
|
|
|
|
def get_cell_extra_context(self, context):
|
|
context = super(WcsCurrentFormsCell, self).get_cell_extra_context(context)
|
|
|
|
categories_filter = {}
|
|
if self.categories:
|
|
for category in self.categories.get('data', []):
|
|
categories_filter[tuple(category.split(':'))] = True
|
|
|
|
for wcs_site in context['user_forms']:
|
|
if not context['user_forms'].get(wcs_site):
|
|
continue
|
|
if not context['user_forms'][wcs_site].get('data'):
|
|
continue
|
|
forms = context['user_forms'][wcs_site]['data']
|
|
if categories_filter:
|
|
forms = [x for x in forms if (wcs_site, x.get('category_slug')) in categories_filter]
|
|
if self.current_forms and self.done_forms:
|
|
pass # keep them all
|
|
elif self.current_forms:
|
|
forms = [x for x in forms if not x.get('form_status_is_endpoint')]
|
|
elif self.done_forms:
|
|
forms = [x for x in forms if x.get('form_status_is_endpoint')]
|
|
else:
|
|
forms = [] # nothing left
|
|
context['user_forms'][wcs_site]['data'] = forms # put it back
|
|
|
|
context['current_forms'] = context['user_forms'] # legacy
|
|
|
|
# regroup all forms in a flat list
|
|
context['forms'] = []
|
|
for wcs_site in context['user_forms']:
|
|
if not context['user_forms'].get(wcs_site):
|
|
continue
|
|
if not context['user_forms'][wcs_site].get('data'):
|
|
continue
|
|
context['forms'].extend(context['user_forms'][wcs_site]['data'])
|
|
|
|
return context
|
|
|
|
@register_cell_class
|
|
class WcsCurrentDraftsCell(WcsUserDataBaseCell):
|
|
variable_name = 'current_drafts'
|
|
template_name = 'combo/wcs/current_drafts.html'
|
|
|
|
categories = JSONField(_('Categories'), blank=True)
|
|
|
|
class Meta:
|
|
verbose_name = _('Current Drafts')
|
|
|
|
def get_default_form_class(self):
|
|
from .forms import WcsCurrentDraftsCellForm
|
|
return WcsCurrentDraftsCellForm
|
|
|
|
def get_api_url(self, context):
|
|
user = self.get_concerned_user(context)
|
|
if user:
|
|
user_name_id = user.get_name_id()
|
|
if user_name_id:
|
|
return '/api/users/%s/drafts' % user_name_id
|
|
return '/api/user/drafts'
|
|
|
|
def get_cell_extra_context(self, context):
|
|
context = super(WcsCurrentDraftsCell, self).get_cell_extra_context(context)
|
|
|
|
categories_filter = {}
|
|
if self.categories:
|
|
for category in self.categories.get('data', []):
|
|
categories_filter[tuple(category.split(':'))] = True
|
|
|
|
for wcs_site in context['current_drafts']:
|
|
if not context['current_drafts'].get(wcs_site):
|
|
continue
|
|
if not context['current_drafts'][wcs_site].get('data'):
|
|
continue
|
|
forms = context['current_drafts'][wcs_site]['data']
|
|
if categories_filter:
|
|
forms = [x for x in forms if (wcs_site, x.get('category_slug')) in categories_filter]
|
|
context['current_drafts'][wcs_site]['data'] = forms # put it back
|
|
|
|
# regroup all forms in a flat list
|
|
context['drafts'] = []
|
|
for wcs_site in context['current_drafts']:
|
|
if not context['current_drafts'].get(wcs_site):
|
|
continue
|
|
if not context['current_drafts'][wcs_site].get('data'):
|
|
continue
|
|
context['drafts'].extend(context['current_drafts'][wcs_site]['data'])
|
|
|
|
return context
|
|
|
|
|
|
@register_cell_class
|
|
class WcsFormsOfCategoryCell(WcsCommonCategoryCell, WcsBlurpMixin):
|
|
ordering = models.CharField(_('Order'), max_length=20,
|
|
default='alpha', blank=False,
|
|
choices=[('alpha', _('Default (alphabetical)')),
|
|
('popularity', _('Popularity')),
|
|
('manual', _('Manual'))])
|
|
manual_order = JSONField(blank=True,
|
|
verbose_name=_('Manual Order'),
|
|
help_text=_('Use drag and drop to reorder forms'))
|
|
limit = models.PositiveSmallIntegerField(_('Limit'),
|
|
null=True, blank=True)
|
|
|
|
class Meta:
|
|
verbose_name = _('Forms of Category')
|
|
|
|
variable_name = 'forms'
|
|
template_name = 'combo/wcs/forms_of_category.html'
|
|
cache_duration = 600
|
|
|
|
def get_default_form_class(self):
|
|
from .forms import WcsFormsOfCategoryCellForm
|
|
return WcsFormsOfCategoryCellForm
|
|
|
|
@property
|
|
def wcs_site(self):
|
|
return self.category_reference.split(':')[0]
|
|
|
|
def get_api_url(self, context):
|
|
api_url = '/api/categories/%s/formdefs/' % self.category_reference.split(':')[1]
|
|
if self.ordering == 'popularity':
|
|
api_url += '?include-count=on'
|
|
return api_url
|
|
|
|
def is_relevant(self, context):
|
|
return bool(self.category_reference)
|
|
|
|
def get_cell_extra_context(self, context):
|
|
extra_context = super(WcsFormsOfCategoryCell, self).get_cell_extra_context(context)
|
|
extra_context.update(WcsBlurpMixin.get_cell_extra_context(self, context))
|
|
if not self.category_reference:
|
|
return extra_context
|
|
extra_context['slug'] = self.category_reference.split(':')[-1]
|
|
extra_context['title'] = self.cached_title
|
|
extra_context['description'] = self.cached_description
|
|
try:
|
|
extra_context['forms'] = list(extra_context['forms'][self.wcs_site]['data'])
|
|
except (KeyError, TypeError) as e:
|
|
# an error occured when getting the data
|
|
extra_context['forms'] = []
|
|
|
|
# default sort is alphabetical, it's always done as this will serve as
|
|
# secondary sort key (thanks to Python stable sort)
|
|
extra_context['forms'] = sorted(extra_context['forms'], key=lambda x: x.get('title'))
|
|
extra_context['more_forms'] = []
|
|
|
|
if self.ordering == 'popularity':
|
|
extra_context['forms'] = sorted(extra_context['forms'], key=lambda x: x.get('count'), reverse=True)
|
|
elif self.ordering == 'manual':
|
|
if self.manual_order:
|
|
manual_order = self.manual_order.get('data')
|
|
for form in extra_context['forms']:
|
|
form_reference = '%s:%s' % (self.category_reference, form['slug'])
|
|
try:
|
|
form['order'] = manual_order.index(form_reference)
|
|
except ValueError:
|
|
form['order'] = 9999
|
|
extra_context['forms'] = sorted(extra_context['forms'], key=lambda x: x.get('order'))
|
|
|
|
if self.limit:
|
|
if len(extra_context['forms']) > self.limit:
|
|
extra_context['more_forms'] = extra_context['forms'][self.limit:]
|
|
extra_context['forms'] = extra_context['forms'][:self.limit]
|
|
|
|
return extra_context
|
|
|
|
def render_for_search(self):
|
|
return ''
|
|
|
|
def get_external_links_data(self):
|
|
if not self.category_reference:
|
|
return
|
|
formdefs = self.get_data({'synchronous': True})
|
|
for site in formdefs.values():
|
|
for formdef in site.get('data', []):
|
|
text = ' '.join([formdef.get('description', '')] + formdef.get('keywords', []))
|
|
yield {
|
|
'url': formdef['url'],
|
|
'title': formdef['title'],
|
|
'text': text}
|
|
|
|
|
|
@register_cell_class
|
|
class WcsCareFormsCell(WcsDataBaseCell):
|
|
api_url = '/api/forms/?limit=10'
|
|
variable_name = 'care_forms'
|
|
template_name = 'combo/wcs/care_forms.html'
|
|
cache_duration = 600
|
|
|
|
class Meta:
|
|
verbose_name = _('Forms to process')
|
|
|
|
|
|
@register_cell_class
|
|
class CategoriesCell(WcsDataBaseCell):
|
|
api_url = '/api/categories/?full=on'
|
|
variable_name = 'form_categories'
|
|
template_name = 'combo/wcs/form_categories.html'
|
|
cache_duration = 600
|
|
|
|
class Meta:
|
|
verbose_name = _('Form Categories')
|
|
|
|
|
|
@register_cell_class
|
|
class TrackingCodeInputCell(CellBase):
|
|
is_enabled = classmethod(is_wcs_enabled)
|
|
wcs_site = models.CharField(_('Site'), max_length=50, blank=True)
|
|
template_name = 'combo/wcs/tracking_code_input.html'
|
|
|
|
class Meta:
|
|
verbose_name = _('Tracking Code Input')
|
|
|
|
def get_default_form_class(self):
|
|
if len(get_wcs_services()) == 1:
|
|
return None
|
|
combo_wcs_sites = [('', _('All Sites'))] + [(x, y.get('title')) for x, y in get_wcs_services().items()]
|
|
return model_forms.modelform_factory(self.__class__,
|
|
fields=['wcs_site'],
|
|
widgets={'wcs_site': Select(choices=combo_wcs_sites)})
|
|
|
|
def get_cell_extra_context(self, context):
|
|
extra_context = super(TrackingCodeInputCell, self).get_cell_extra_context(context)
|
|
if not self.wcs_site:
|
|
self.wcs_site = list(get_wcs_services().keys())[0]
|
|
extra_context['url'] = get_wcs_services().get(self.wcs_site).get('url')
|
|
return extra_context
|
|
|
|
|
|
@register_cell_class
|
|
class BackofficeSubmissionCell(WcsDataBaseCell):
|
|
api_url = '/api/formdefs/?backoffice-submission=on'
|
|
variable_name = 'all_formdefs'
|
|
template_name = 'combo/wcs/backoffice_submission.html'
|
|
cache_duration = 600
|
|
|
|
class Meta:
|
|
verbose_name = _('Backoffice Submission')
|
|
|
|
def get_cell_extra_context(self, context):
|
|
context = super(BackofficeSubmissionCell, self).get_cell_extra_context(context)
|
|
# add a fake category where it's missing
|
|
for site_formdefs in context['all_formdefs'].values():
|
|
for formdef in site_formdefs['data']:
|
|
if not 'category' in formdef:
|
|
formdef['category'] = _('Misc')
|
|
return context
|