146 lines
4.5 KiB
Python
146 lines
4.5 KiB
Python
# lingo - payment and billing system
|
|
# Copyright (C) 2022 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 json
|
|
|
|
from django.conf import settings
|
|
from django.utils.translation import ugettext_lazy as _
|
|
from requests.exceptions import RequestException
|
|
|
|
from lingo.agendas.models import Agenda
|
|
from lingo.utils import requests
|
|
|
|
|
|
class ChronoError(Exception):
|
|
pass
|
|
|
|
|
|
def is_chrono_enabled():
|
|
return hasattr(settings, 'KNOWN_SERVICES') and settings.KNOWN_SERVICES.get('chrono')
|
|
|
|
|
|
def get_chrono_service():
|
|
if not is_chrono_enabled():
|
|
return {}
|
|
return list(settings.KNOWN_SERVICES.get('chrono').values())[0]
|
|
|
|
|
|
def get_chrono_json(path, params=None, log_errors=True):
|
|
chrono_site = get_chrono_service()
|
|
if chrono_site is None:
|
|
return
|
|
try:
|
|
response = requests.get(
|
|
path,
|
|
params=params or {},
|
|
remote_service=chrono_site,
|
|
without_user=True,
|
|
headers={'accept': 'application/json'},
|
|
log_errors=log_errors,
|
|
)
|
|
response.raise_for_status()
|
|
except RequestException as e:
|
|
if e.response is not None:
|
|
try:
|
|
# return json if available (on 404 responses by example)
|
|
return e.response.json()
|
|
except json.JSONDecodeError:
|
|
pass
|
|
return
|
|
return response.json()
|
|
|
|
|
|
def collect_agenda_data():
|
|
result = get_chrono_json('api/agenda/')
|
|
if result is None:
|
|
return
|
|
if result.get('data') is None:
|
|
return
|
|
|
|
agenda_data = []
|
|
for agenda in result['data']:
|
|
if agenda['kind'] != 'events':
|
|
continue
|
|
agenda_data.append(
|
|
{
|
|
'slug': agenda['slug'],
|
|
'label': agenda['text'],
|
|
'category_slug': agenda['category'],
|
|
'category_label': agenda['category_label'],
|
|
}
|
|
)
|
|
return agenda_data
|
|
|
|
|
|
def refresh_agendas():
|
|
result = collect_agenda_data()
|
|
if result is None:
|
|
return
|
|
|
|
# fetch existing agendas
|
|
existing_agendas = {a.slug: a for a in Agenda.objects.all()}
|
|
seen_agendas = []
|
|
|
|
# build agendas from chrono
|
|
for agenda_data in result:
|
|
slug = agenda_data['slug']
|
|
agenda = existing_agendas.get(slug) or Agenda(slug=slug)
|
|
for key, value in agenda_data.items():
|
|
if key == 'slug':
|
|
continue
|
|
setattr(agenda, key, value)
|
|
agenda.save()
|
|
seen_agendas.append(agenda.slug)
|
|
|
|
# now check outdated agendas
|
|
for slug, agenda in existing_agendas.items():
|
|
if slug not in seen_agendas:
|
|
agenda.delete()
|
|
|
|
|
|
def get_event(event_slug):
|
|
return get_events(
|
|
[event_slug],
|
|
error_message=_('Unable to get event details'),
|
|
error_message_with_details=_('Unable to get event details (%s)'),
|
|
)[0]
|
|
|
|
|
|
def get_events(event_slugs, error_message=None, error_message_with_details=None):
|
|
error_message = error_message or _('Unable to get events details')
|
|
error_message_with_details = error_message_with_details or _('Unable to get events details (%s)')
|
|
result = get_chrono_json('api/agendas/events/', params={'slots': event_slugs})
|
|
if not result:
|
|
raise ChronoError(error_message)
|
|
if result.get('err'):
|
|
raise ChronoError(error_message_with_details % result['err_desc'])
|
|
if not result.get('data'):
|
|
raise ChronoError(error_message)
|
|
return result['data']
|
|
|
|
|
|
def get_subscriptions(agenda_slug, user_external_id):
|
|
result = get_chrono_json(
|
|
'api/agenda/%s/subscription/?user_external_id=%s' % (agenda_slug, user_external_id)
|
|
)
|
|
if not result or not result.get('data'):
|
|
raise ChronoError(_('Unable to get subscription details'))
|
|
if result.get('err'):
|
|
raise ChronoError(_('Unable to get subscription details (%s)') % result['err_desc'])
|
|
if not result.get('data'):
|
|
raise ChronoError(_('Unable to get subscription details'))
|
|
return result['data']
|