Source code for org.models.ticket

from functools import cached_property
from markupsafe import Markup
from onegov.chat.collections import ChatCollection
from onegov.core.templates import render_macro
from onegov.directory import Directory, DirectoryEntry
from onegov.event import EventCollection
from onegov.form import FormSubmissionCollection
from onegov.org import _
from onegov.org.layout import DefaultLayout, EventLayout
from onegov.chat import Message
from onegov.core.elements import Link, LinkGroup, Confirm, Intercooler, Trait
from onegov.org.views.utils import show_tags, show_filters
from onegov.reservation import Allocation, Resource, Reservation
from onegov.ticket import Ticket, Handler, handlers
from onegov.search.utils import extract_hashtags
from purl import URL
from sqlalchemy import desc
from sqlalchemy import func
from sqlalchemy.orm import object_session


from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
    from onegov.chat.models import Chat
    from onegov.event import Event
    from onegov.form import Form, FormSubmission
    from onegov.org.request import OrgRequest
    from onegov.pay import Payment
    from onegov.ticket.handler import _Q
    from sqlalchemy import Column
    from sqlalchemy.orm import Query, Session
    from uuid import UUID


[docs] def ticket_submitter(ticket: Ticket) -> str | None: handler = ticket.handler mail = handler.deleted and ticket.snapshot.get('email') or handler.email # case of EventSubmissionHandler for imported events if handler.data.get('source'): mail = handler.data.get('user', mail) return mail
[docs] class OrgTicketMixin: """ Adds additional methods to the ticket, needed by the organisations implementation of it. Strictly limited to things that do not belong into onegov.ticket. """ if TYPE_CHECKING:
[docs] number: Column[str]
group: Column[str]
[docs] def reference(self, request: 'OrgRequest') -> str: """ Returns the reference which should be used wherever we talk about a ticket, but do not show it (say in an e-mail subject). This reference should not be changed so it stays consistent. If you want, you can override the content of the reference group, shown in brackets (see :meth:`reference_group`). """ return f'{self.number} / {self.reference_group(request)}'
[docs] def reference_group(self, request: 'OrgRequest') -> str: return request.translate(self.group)
@cached_property
[docs] def extra_localized_text(self) -> str: # extracts of attachments are currently not searchable - if they were # we would add this here - probably in a raw SQL query that # concatenates all the text # # for now I've decided against it as it would lower the hit-rate # for notes (which should be very easy to find), just to be able # to search through files which are mostly going to be irrelevant # for what the user wants to find # # if the user wants to have a ticket findable through the file content # we should advise them to enter a meaningful note with the file # instead. # q = object_session(self).query(Message) q = q.filter_by(channel_id=self.number) q = q.filter(Message.type.in_(('ticket_note', 'ticket_chat'))) q = q.with_entities(Message.text) return ' '.join(n.text for n in q if n.text)
@property
[docs] def es_tags(self) -> list[str] | None: if self.extra_localized_text: return [ tag.lstrip('#') for tag in extract_hashtags( self.extra_localized_text ) ] return None
[docs] class FormSubmissionTicket(OrgTicketMixin, Ticket):
[docs] __mapper_args__ = {'polymorphic_identity': 'FRM'} # type:ignore
[docs] es_type_name = 'submission_tickets'
[docs] class ReservationTicket(OrgTicketMixin, Ticket):
[docs] __mapper_args__ = {'polymorphic_identity': 'RSV'} # type:ignore
[docs] es_type_name = 'reservation_tickets'
[docs] class EventSubmissionTicket(OrgTicketMixin, Ticket):
[docs] __mapper_args__ = {'polymorphic_identity': 'EVN'} # type:ignore
[docs] es_type_name = 'event_tickets'
if TYPE_CHECKING:
[docs] handler: 'EventSubmissionHandler'
[docs] def reference_group(self, request: 'OrgRequest') -> str: return self.title
[docs] class DirectoryEntryTicket(OrgTicketMixin, Ticket):
[docs] __mapper_args__ = {'polymorphic_identity': 'DIR'} # type:ignore
[docs] es_type_name = 'directory_tickets'
@handlers.registered_handler('FRM')
[docs] class FormSubmissionHandler(Handler):
[docs] id: 'UUID'
[docs] handler_title = _('Form Submissions')
[docs] code_title = _('Forms')
@cached_property
[docs] def collection(self) -> FormSubmissionCollection: return FormSubmissionCollection(self.session)
@cached_property
[docs] def submission(self) -> 'FormSubmission | None': return self.collection.by_id(self.id)
@cached_property
[docs] def form(self) -> 'Form': assert self.submission is not None return self.submission.form_class(data=self.submission.data)
@property
[docs] def deleted(self) -> bool: return self.submission is None
# FIXME: We should be a little bit more careful about data from # properties that can be None @property
[docs] def email(self) -> str: return ( self.submission.email or '' if self.submission is not None else '' )
@property
[docs] def title(self) -> str: return ( self.submission.title or '' if self.submission is not None else '' )
@property
[docs] def subtitle(self) -> None: return None
@property
[docs] def group(self) -> str: return ( self.submission.form.title # type:ignore[union-attr] if self.submission is not None else '' )
@property
[docs] def payment(self) -> 'Payment | None': return self.submission.payment if self.submission is not None else None
@property
[docs] def extra_data(self) -> list[str]: return [ v for v in self.submission.data.values() if isinstance(v, str) ] if self.submission is not None else []
@property
[docs] def undecided(self) -> bool: if self.deleted: return False assert self.submission is not None # submissions without registration window do not present a decision if not self.submission.registration_window: return False if self.submission.claimed is None: return True return False
[docs] def get_summary( self, request: 'OrgRequest' # type:ignore[override] ) -> Markup: layout = DefaultLayout(self.submission, request) if self.submission is not None: return render_macro(layout.macros['display_form'], request, { 'form': self.form, 'layout': layout }) return Markup('')
@handlers.registered_handler('RSV')
[docs] class ReservationHandler(Handler):
[docs] id: 'UUID'
[docs] handler_title = _('Reservations')
[docs] code_title = _('Reservations')
@cached_property
[docs] def resource(self) -> Resource | None: if self.deleted: return None query = self.session.query(Resource) query = query.filter(Resource.id == self.reservations[0].resource) return query.one()
[docs] def reservations_query(self) -> 'Query[Reservation]': # libres allows for multiple reservations with a single request (token) # for now we don't really have that case in onegov.org, but we # try to be aware of it as much as possible query = self.session.query(Reservation) query = query.filter(Reservation.token == self.id) query = query.order_by(Reservation.start) return query
@cached_property
[docs] def reservations(self) -> tuple[Reservation, ...]: return tuple(self.reservations_query())
@cached_property
[docs] def has_future_reservation(self) -> bool: exists = self.reservations_query().filter( Reservation.start > func.now() ).exists() return self.session.query(exists).scalar()
@cached_property
[docs] def most_future_reservation(self) -> Reservation | None: return ( self.session.query(Reservation) .order_by(desc(Reservation.start)) .first() )
@cached_property
[docs] def submission(self) -> 'FormSubmission | None': return FormSubmissionCollection(self.session).by_id(self.id)
@property
[docs] def payment(self) -> 'Payment | None': return self.reservations and self.reservations[0].payment or None
@property
[docs] def deleted(self) -> bool: return not self.reservations
@property
[docs] def extra_data(self) -> list[str]: return self.submission and [ v for v in self.submission.data.values() if isinstance(v, str) ] or []
@property
[docs] def email(self) -> str: # the e-mail is the same over all reservations if self.deleted: return self.ticket.snapshot.get('email') # type:ignore return self.reservations[0].email
@property
[docs] def undecided(self) -> bool: # if there is no reservation with an 'accept' marker, the user # has not yet made a decision if self.deleted: return False for r in self.reservations: if r.data and r.data.get('accepted'): return False return True
[docs] def prepare_delete_ticket(self) -> None: for reservation in self.reservations or (): self.session.delete(reservation)
@cached_property
[docs] def ticket_deletable(self) -> bool: return not self.has_future_reservation and super().ticket_deletable
@property
[docs] def title(self) -> str: parts = [] for ix, reservation in enumerate(self.reservations): parts.append(self.get_reservation_title(reservation)) if ix == 4: parts.append('…') break return ', '.join(parts)
[docs] def get_reservation_title(self, reservation: Reservation) -> str: assert self.resource and hasattr(self.resource, 'reservation_title') return self.resource.reservation_title(reservation)
@property
[docs] def subtitle(self) -> str | None: if self.submission: return ', '.join( p for p in (self.email, self.submission.title) if p) elif self.reservations: return self.email else: return None
@property
[docs] def group(self) -> str | None: return self.resource.title if self.resource else None
@classmethod
[docs] def handle_extra_parameters( cls, session: 'Session', query: '_Q', extra_parameters: dict[str, Any] ) -> '_Q': if 'allocation_id' in extra_parameters: allocations = session.query(Allocation.group) allocations = allocations.filter( Allocation.id == int(extra_parameters['allocation_id'])) tokens = session.query(Reservation.token) tokens = tokens.filter( Reservation.target.in_(allocations.subquery())) handler_ids = tuple(t[0].hex for t in tokens) if handler_ids: query = query.filter(Ticket.handler_id.in_(handler_ids)) else: query = query.filter(False) return query
[docs] def get_summary( self, request: 'OrgRequest' # type:ignore[override] ) -> Markup: layout = DefaultLayout(self.resource, request) parts = [] parts.append( render_macro(layout.macros['reservations'], request, { 'reservations': self.reservations, 'layout': layout }) ) if self.submission: form = self.submission.form_class(data=self.submission.data) parts.append( render_macro(layout.macros['display_form'], request, { 'form': form, 'layout': layout }) ) return Markup('').join(parts)
@handlers.registered_handler('EVN')
[docs] class EventSubmissionHandler(Handler):
[docs] id: 'UUID'
[docs] handler_title = _('Events')
[docs] code_title = _('Events')
@cached_property
[docs] def collection(self) -> EventCollection: return EventCollection(self.session)
@cached_property
[docs] def event(self) -> 'Event | None': return self.collection.by_id(self.id)
@property
[docs] def deleted(self) -> bool: return self.event is None
@cached_property
[docs] def source(self) -> str | None: # values stored only when importing with cli return self.data.get('source')
@cached_property
[docs] def import_user(self) -> str | None: # values stored only when importing with cli return self.data.get('user')
@cached_property
[docs] def email(self) -> str | None: return self.event.meta.get('submitter_email') if self.event else None
@property
[docs] def title(self) -> str: return self.event.title if self.event else ''
@property
[docs] def subtitle(self) -> str | None: if self.deleted: return None assert self.event is not None parts = ( self.event.meta.get('submitter_email'), '{:%d.%m.%Y %H:%M}'.format(self.event.localized_start) ) return ', '.join(p for p in parts if p)
@property
[docs] def extra_data(self) -> list[str]: assert self.event is not None return [ self.event.description or '', self.event.title, self.event.location or '' ]
@property
[docs] def undecided(self) -> bool: return self.event and self.event.state == 'submitted' or False
@property
[docs] def ticket_deletable(self) -> bool: # We don't want to delete the event. So we will redact the ticket # instead. return False
@cached_property
[docs] def group(self) -> str: return _('Event')
[docs] def get_summary( self, request: 'OrgRequest' # type:ignore[override] ) -> Markup: assert self.event is not None layout = EventLayout(self.event, request) return render_macro(layout.macros['display_event'], request, { 'event': self.event, 'layout': layout, 'show_tags': show_tags(request), 'show_filters': show_filters(request), })
@handlers.registered_handler('DIR')
[docs] class DirectoryEntryHandler(Handler):
[docs] id: 'UUID'
[docs] handler_title = _('Directory Entry Submissions')
[docs] code_title = _('Directory Entry Submissions')
@cached_property
[docs] def collection(self) -> FormSubmissionCollection: return FormSubmissionCollection(self.session)
@cached_property
[docs] def submission(self) -> 'FormSubmission | None': return self.collection.by_id(self.id)
@cached_property
[docs] def form(self) -> 'Form | None': return ( self.submission.form_class(data=self.submission.data) if self.submission is not None else None )
# FIXME: This should probably query ExtendedDirectory, since we rely # on a method that only exists on ExtendedDirectory @cached_property
[docs] def directory(self) -> Directory | None: if self.submission: directory_id = self.submission.meta['directory'] else: directory_id = self.ticket.handler_data['directory'] return self.session.query(Directory).filter_by(id=directory_id).first()
@cached_property
[docs] def entry(self) -> DirectoryEntry | None: if self.submission: id = self.submission.meta.get('directory_entry') else: id = self.ticket.handler_data.get('directory_entry') return self.session.query(DirectoryEntry).filter_by(id=id).first()
@property
[docs] def deleted(self) -> bool: if not self.directory: return True if self.kind == 'change-request': if self.submission: data = self.submission.meta else: data = self.ticket.handler_data entry = ( self.session.query(DirectoryEntry) .filter_by(id=data['directory_entry']) .first() ) if not entry: return True if self.state == 'adopted': name = self.ticket.handler_data.get('entry_name') if name is None: return True return not self.directory.entry_with_name_exists(name) return False
@property
[docs] def email(self) -> str: return ( # we don't allow directory entry submissions without an email self.submission.email # type:ignore[return-value] if self.submission is not None else '' )
@property
[docs] def submitter_name(self) -> str | None: submitter_name: str | None = ( self.ticket.snapshot.get('submitter_name') if self.deleted else None ) if submitter_name is None and self.submission is not None: submitter_name = self.submission.submitter_name return submitter_name
@property
[docs] def submitter_phone(self) -> str | None: submitter_phone: str | None = ( self.ticket.snapshot.get('submitter_phone') if self.deleted else None ) if submitter_phone is None and self.submission is not None: submitter_phone = self.submission.submitter_phone return submitter_phone
@property
[docs] def submitter_address(self) -> str | None: submitter_address: str | None = ( self.ticket.snapshot.get('submitter_address') if self.deleted else None ) if submitter_address is None and self.submission is not None: submitter_address = self.submission.submitter_address return submitter_address
@property
[docs] def title(self) -> str: return ( self.submission.title or '' if self.submission is not None else '' )
@property
[docs] def subtitle(self) -> None: return None
@property
[docs] def group(self) -> str: if self.directory: return self.directory.title elif self.ticket.group: return self.ticket.group return '-'
@property
[docs] def payment(self) -> 'Payment | None': return self.submission.payment if self.submission else None
@property
[docs] def state(self) -> str | None: return self.ticket.handler_data.get('state')
@property
[docs] def extra_data(self) -> list[str]: return self.submission and [ v for v in self.submission.data.values() if isinstance(v, str) ] or []
@property
[docs] def undecided(self) -> bool: if not self.directory or self.deleted: return False return self.state is None
@property
[docs] def kind(self) -> str: if self.submission: data = self.submission.meta else: data = self.ticket.handler_data if 'change-request' in data.get('extensions', ()): return 'change-request' else: return 'new-entry'
[docs] def get_summary( self, request: 'OrgRequest' # type:ignore[override] ) -> Markup: assert self.form is not None layout = DefaultLayout(self.submission, request) # XXX this is a poor man's request.get_form self.form.request = request self.form.model = self.submission macro = layout.macros['directory_entry_submission'] return render_macro(macro, request, { 'form': self.form, 'layout': layout, 'handler': self, })
[docs] class ChatTicket(OrgTicketMixin, Ticket):
[docs] __mapper_args__ = {'polymorphic_identity': 'CHT'} # type:ignore
[docs] es_type_name = 'chat_tickets'
[docs] def reference_group(self, request: 'OrgRequest') -> str: return self.handler.title
@handlers.registered_handler('CHT')
[docs] class ChatHandler(Handler):
[docs] handler_title = _('Chats')
[docs] code_title = _('Chats')
@cached_property
[docs] def collection(self) -> ChatCollection: return ChatCollection(self.session)
@cached_property
[docs] def chat(self) -> 'Chat | None': return self.collection.by_id(self.id)
@property
[docs] def deleted(self) -> bool: return self.chat is None
@property
[docs] def title(self) -> str: if self.chat is not None: return f'Chat - {self.chat.customer_name}' else: return ''
@property
[docs] def subtitle(self) -> None: return None
@property
[docs] def group(self) -> str | None: return self.chat.topic if self.chat is not None else None
@property
[docs] def email(self) -> str: return self.chat.email if self.chat is not None else ''
[docs] def get_summary( self, request: 'OrgRequest' # type: ignore[override] ) -> Markup: layout = DefaultLayout(self.collection, request) if self.chat is not None: return render_macro(layout.macros['display_chat'], request, { 'chat': self.chat, 'layout': layout }) return Markup('')