Source code for pas.views.settlement_run

from __future__ import annotations

from io import BytesIO
from webob import Response
from onegov.core.utils import module_path
from decimal import Decimal
from operator import itemgetter
from weasyprint import HTML, CSS  # type: ignore[import-untyped]
from weasyprint.text.fonts import (  # type: ignore[import-untyped]
    FontConfiguration)

from onegov.core.elements import Link
from onegov.core.security import Private
from onegov.core.utils import normalize_for_filename
from onegov.pas import _
from onegov.pas import PasApp
from onegov.pas.calculate_pay import calculate_rate
from onegov.pas.collections import (
    AttendenceCollection,
    SettlementRunCollection,
)
from onegov.pas.custom import get_current_rate_set
from onegov.pas.export_single_parliamentarian import (
    generate_parliamentarian_settlement_pdf
)
from onegov.pas.forms import SettlementRunForm
from onegov.pas.layouts import SettlementRunCollectionLayout
from onegov.pas.layouts import SettlementRunLayout
from onegov.pas.models import (
    Attendence,
    PASCommission,
    PASCommissionMembership,
    PASParliamentarian,
    Party,
    SettlementRun,
)
from onegov.pas.models.attendence import TYPES
from onegov.pas.path import SettlementRunExport, SettlementRunAllExport
from onegov.pas.utils import (
    format_swiss_number,
    get_commissions_with_memberships,
    get_parliamentarians_with_settlements,
    get_parties_with_settlements,
    is_commission_president,
)
from onegov.pas.views.abschlussliste import (
    generate_abschlussliste_xlsx,
    generate_buchungen_abrechnungslauf_xlsx
)
from onegov.pas.views.pas_excel_export_nr_3_lohnart_fibu import (
        generate_fibu_export_rows)


from typing import Any, Literal, TypeAlias, TYPE_CHECKING
if TYPE_CHECKING:
    from sqlalchemy.orm import Session
    from datetime import date
    from onegov.core.types import RenderData
    from onegov.town6.request import TownRequest

[docs] SettlementDataRow: TypeAlias = tuple[ 'date', PASParliamentarian, str, Decimal, Decimal, Decimal ]
TotalRow: TypeAlias = tuple[ str, Decimal, Decimal, Decimal, Decimal, Decimal ]
[docs] XLSX_MIMETYPE = ( 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet' )
[docs] def get_commission_closure_status( session: Session, settlement_run: SettlementRun, commissions: list[PASCommission] | None = None ) -> list[dict[str, Any]]: """ Get closure status organized by commission showing completion summary. Args: session: Database session settlement_run: The settlement run to check commissions: Optional pre-fetched list of commissions Returns: List of dicts with structure: [ { 'commission_name': str, 'total_members': int, 'completed_members': int, 'completion_ratio': str, 'incomplete_members': [ {'name': str, 'has_attendance': bool}, ... ] }, ... ] """ # Query commissions if not provided if commissions is None: commissions = get_commissions_with_memberships( session, settlement_run.start, settlement_run.end ) commission_status = [] for commission in commissions: # Get all members of this commission during settlement period memberships = session.query(PASCommissionMembership).filter( PASCommissionMembership.commission_id == commission.id, ( PASCommissionMembership.start.is_(None) | (PASCommissionMembership.start <= settlement_run.end) ), ( PASCommissionMembership.end.is_(None) | (PASCommissionMembership.end >= settlement_run.start) ) ).all() total_members = len(memberships) if total_members == 0: continue completed_members = 0 incomplete_members = [] complete_members = [] for membership in memberships: parliamentarian = membership.parliamentarian parl_name = ( f'{parliamentarian.first_name} ' f'{parliamentarian.last_name}' ) # Check if this parliamentarian has closed attendance for this # commission closed_attendance = session.query(Attendence).filter( Attendence.parliamentarian_id == parliamentarian.id, Attendence.commission_id == commission.id, Attendence.date >= settlement_run.start, Attendence.date <= settlement_run.end, Attendence.abschluss == True ).first() if closed_attendance: completed_members += 1 complete_members.append({ 'name': parl_name, }) else: # Check if they have any attendance at all for this commission has_attendance = session.query(Attendence).filter( Attendence.parliamentarian_id == parliamentarian.id, Attendence.commission_id == commission.id, Attendence.date >= settlement_run.start, Attendence.date <= settlement_run.end ).first() is not None incomplete_members.append({ 'name': parl_name, 'has_attendance': has_attendance }) completion_ratio = f'{completed_members}/{total_members}' commission_status.append({ 'commission_name': commission.name, 'total_members': total_members, 'completed_members': completed_members, 'completion_ratio': completion_ratio, 'complete_members': complete_members, 'incomplete_members': incomplete_members, 'is_complete': completed_members == total_members }) # Sort by completion status (incomplete first) then by name commission_status.sort( key=itemgetter('is_complete', 'commission_name') ) return commission_status
@PasApp.html( model=SettlementRunCollection, template='settlement_runs.pt', permission=Private )
[docs] def view_settlement_runs( self: SettlementRunCollection, request: TownRequest ) -> RenderData: layout = SettlementRunCollectionLayout(self, request) filters = {} filters['active'] = [ Link( text=request.translate(title), active=self.active == value, url=request.link(self.for_filter(active=value)) ) for title, value in ( (_('Active'), True), (_('Inactive'), False) ) ] return { 'add_link': request.link(self, name='new'), 'filters': filters, 'layout': layout, 'settlement_runs': self.query().all(), 'title': layout.title, }
@PasApp.form(model=SettlementRunCollection, name='new', template='form.pt', permission=Private, form=SettlementRunForm )
[docs] def add_settlement_run( self: SettlementRunCollection, request: TownRequest, form: SettlementRunForm ) -> RenderData | Response: if form.submitted(request): settlement_run = self.add(**form.get_useful_data()) request.success(_('Added a new settlement run')) return request.redirect(request.link(settlement_run)) layout = SettlementRunCollectionLayout(self, request) layout.breadcrumbs.append(Link(_('New'), '#')) return { 'layout': layout, 'title': _('New settlement run'), 'form': form, 'form_width': 'large' }
@PasApp.html( model=SettlementRun, template='settlement_run.pt', permission=Private )
[docs] def view_settlement_run( self: SettlementRun, request: TownRequest ) -> RenderData: """ A page where all exports are listed and grouped by category. """ layout = SettlementRunLayout(self, request) session = request.session # Get parties active during settlement run period parties = get_parties_with_settlements(session, self.start, self.end) # Get commissions active during settlement run period commissions = get_commissions_with_memberships( session, self.start, self.end ) # Get parliamentarians active during settlement run period with settlements parliamentarians = get_parliamentarians_with_settlements( session, self.start, self.end ) # Get commission closure status for the control list commission_closure_status = get_commission_closure_status( session, self, commissions ) pdf_categories = { 'party': { 'title': _('Settlements by Party'), 'links': [ Link( party.title + ' Total', request.link( SettlementRunExport( self, party, category='party' ), 'run-export' ), ) for party in parties ], }, 'all': { 'title': _('All Settlements'), 'links': [ Link( _('All Parties'), request.link( SettlementRunAllExport( settlement_run=self, category='all-parties' ), name='run-export' ), ), ], }, 'commissions': { 'title': _('Settlements by Commission'), 'links': [ Link( commission.title + ' Total', request.link( SettlementRunExport( self, commission, category='commission' ), 'run-export' ), ) for commission in commissions ], }, 'parliamentarians': { 'title': _('Settlements by Parliamentarian'), 'links': [ Link( f'{p.last_name} {p.first_name}', request.link( SettlementRunExport( self, p, category='parliamentarian' ), 'run-export' ), ) for p in parliamentarians ], }, } excel_categories = { 'abschlussliste_export': { 'title': _('Abschlussliste'), 'links': [ Link( _('Abschlussliste (XLSX)'), request.link( SettlementRunAllExport( settlement_run=self, category='abschlussliste-xlsx-export' ), name='run-export' ), ) ] }, 'salary_export': { 'title': _('Buchungen Abrechnungslauf'), 'links': [ Link( _('Buchungen Abrechnungslauf (Kontrollliste)'), request.link( SettlementRunAllExport( settlement_run=self, category='buchungen-abrechnungslauf-kontroll-xlsx-export' ), name='run-export' ), ), ] }, 'fibu_export': { 'title': _('KR-Entschädigungen (CSV)'), 'links': [ Link( _('KR-Entschädigungen (CSV)'), request.link( SettlementRunAllExport( settlement_run=self, category='fibu-csv-export' ), name='run-export' ), ) ] } } export_tabs_data = { 'pdf': { 'tab_title': _('PDF Exports'), 'panel_id': 'panel-pdf-exports', 'categories': pdf_categories }, 'excel': { 'tab_title': _('Excel Exports'), 'panel_id': 'panel-excel-exports', 'categories': excel_categories } } return { 'layout': layout, 'settlement_run': self, 'export_tabs_data': export_tabs_data, 'commission_closure_status': commission_closure_status, 'title': layout.title, }
[docs] def _get_commission_totals( settlement_run: SettlementRun, request: TownRequest, commission: PASCommission ) -> list[TotalRow]: """Get totals for a specific commission grouped by party.""" session = request.session rate_set = get_current_rate_set(session, settlement_run) if not rate_set: return [] # Get all attendences in period for this commission attendences = AttendenceCollection( session, date_from=settlement_run.start, date_to=settlement_run.end, commission_id=str(commission.id) ).query() cola_multiplier = Decimal( str(1 + (rate_set.cost_of_living_adjustment / 100)) ) # Initialize party totals party_totals: dict[str, dict[str, Decimal]] = {} for attendence in attendences: # Get parliamentarian's party current_party = None for role in attendence.parliamentarian.roles: if role.party and (role.end is None or role.end >= settlement_run.start): current_party = role.party break if not current_party: continue party_name = current_party.name if party_name not in party_totals: party_totals[party_name] = { 'Meetings': Decimal('0'), 'Expenses': Decimal('0'), 'Total': Decimal('0'), 'Cost-of-living Allowance': Decimal('0'), 'Final': Decimal('0') } is_president = is_commission_president( attendence.parliamentarian, commission.id, settlement_run ) base_rate = calculate_rate( rate_set=rate_set, attendence_type=attendence.type, duration_minutes=int(attendence.duration), is_president=is_president, commission_type=commission.type ) # Update totals party_totals[party_name]['Meetings'] += Decimal(str(base_rate)) # Note: Expenses will be implemented in the future base_total = party_totals[party_name]['Meetings'] expenses = party_totals[party_name]['Expenses'] cola_amount = base_total * (cola_multiplier - 1) party_totals[party_name]['Total'] = base_total + expenses party_totals[party_name]['Cost-of-living Allowance'] = cola_amount party_totals[party_name]['Final'] = base_total + expenses + cola_amount # Convert to sorted list of tuples result = [ ( name, data['Meetings'], data['Expenses'], data['Total'], data['Cost-of-living Allowance'], data['Final'] ) for name, data in sorted(party_totals.items()) ] # Add total row totals = ( sum(r[1] for r in result), # Sessions sum(r[2] for r in result), # Expenses sum(r[3] for r in result), # Total sum(r[4] for r in result), # COLA sum(r[5] for r in result), # Final ) result.append(( f'Total {commission.name}', Decimal(str(totals[0])), Decimal(str(totals[1])), Decimal(str(totals[2])), Decimal(str(totals[3])), Decimal(str(totals[4])) )) return result
[docs] def _get_party_totals_for_export_all( self: SettlementRun, request: TownRequest ) -> list[tuple[str, Decimal, Decimal, Decimal, Decimal, Decimal]]: """Get totals grouped by party.""" session = request.session rate_set = get_current_rate_set(session, self) if not rate_set: return [] cola_multiplier = Decimal( str(1 + (rate_set.cost_of_living_adjustment / 100)) ) # Get all parties with settlements during the period parties = get_parties_with_settlements(session, self.start, self.end) # Initialize party totals party_totals: dict[str, dict[str, Decimal]] = { party.name: { 'Meetings': Decimal('0'), 'Expenses': Decimal('0'), 'Total': Decimal('0'), 'Cost-of-living Allowance': Decimal('0'), 'Final': Decimal('0'), } for party in parties } # Process attendances for each party with proper temporal alignment for party in parties: attendances = ( AttendenceCollection(session) .by_party( party_id=str(party.id), start_date=self.start, end_date=self.end, ) .query() .all() ) for attendance in attendances: # Calculate base rate is_president = any( r.role == 'president' for r in attendance.parliamentarian.roles ) base_rate = calculate_rate( rate_set=rate_set, attendence_type=attendance.type, duration_minutes=int(attendance.duration), is_president=is_president, commission_type=( attendance.commission.type if attendance.commission else None ), ) # Update totals party_totals[party.name]['Meetings'] += Decimal(str(base_rate)) base_total = party_totals[party.name]['Meetings'] expenses = party_totals[party.name]['Expenses'] cola_amount = base_total * (cola_multiplier - 1) party_totals[party.name]['Total'] = base_total + expenses party_totals[party.name]['Cost-of-living Allowance'] = cola_amount party_totals[party.name]['Final'] = ( base_total + expenses + cola_amount ) # Convert to sorted list of tuples result = [ ( name, data['Meetings'], data['Expenses'], data['Total'], data['Cost-of-living Allowance'], data['Final'], ) for name, data in sorted(party_totals.items()) ] # Add total row totals = ( Decimal(sum(r[1] for r in result)), # Sessions Decimal(sum(r[2] for r in result)), # Expenses Decimal(sum(r[3] for r in result)), # Total Decimal(sum(r[4] for r in result)), # COLA Decimal(sum(r[5] for r in result)), # Final ) result.append(('Total Parteien', *totals)) return result
[docs] def generate_settlement_pdf( settlement_run: SettlementRun, request: TownRequest, entity_type: Literal['all', 'commission', 'party', 'parliamentarian'], entity: PASCommission | Party | PASParliamentarian | None = None, ) -> bytes: """ Entry point for almost all settlement PDF generations. """ font_config = FontConfiguration() css_path = module_path('onegov.pas', 'views/templates/settlement_pdf.css') with open(css_path) as f: css = CSS(string=f.read()) if entity_type == 'commission' and isinstance(entity, PASCommission): settlement_data = _get_commission_settlement_data( settlement_run, request, entity ) totals = _get_commission_totals(settlement_run, request, entity) elif entity_type == 'party' and isinstance(entity, Party): settlement_data = _get_party_settlement_data( settlement_run, request, entity ) totals = get_party_specific_totals(settlement_run, request, entity) elif entity_type == 'all': settlement_data = _get_data_export_all(settlement_run, request) totals = _get_party_totals_for_export_all(settlement_run, request) assert len(totals) > 0 else: raise ValueError(f'Unsupported entity type: {entity_type}') html = _generate_settlement_html( settlement_data=settlement_data, totals=totals, subtitle='Einträge Journal', ) return HTML(string=html).write_pdf( stylesheets=[css], font_config=font_config )
[docs] def _get_commission_settlement_data( settlement_run: SettlementRun, request: TownRequest, commission: PASCommission ) -> list[SettlementDataRow]: """Get settlement data for a specific commission.""" session = request.session rate_set = get_current_rate_set(request.session, settlement_run) if not rate_set: return [] attendences = AttendenceCollection( session, date_from=settlement_run.start, date_to=settlement_run.end ).query().filter( Attendence.commission_id == commission.id ) cola_multiplier = Decimal( str(1 + (rate_set.cost_of_living_adjustment / 100)) ) result = [] for attendence in attendences: is_president = is_commission_president( attendence.parliamentarian, commission.id, settlement_run ) base_rate = calculate_rate( rate_set=rate_set, attendence_type=attendence.type, duration_minutes=int(attendence.duration), is_president=is_president, commission_type=commission.type ) rate_with_cola = Decimal(str(base_rate)) * cola_multiplier result.append(( attendence.date, attendence.parliamentarian, request.translate(TYPES[attendence.type]), attendence.calculate_value(), Decimal(str(base_rate)), rate_with_cola )) return sorted(result, key=itemgetter(0))
[docs] def _generate_settlement_html( settlement_data: list[SettlementDataRow], totals: list[TotalRow], subtitle: str, ) -> str: """Generate HTML for settlement PDF.""" html = f""" <!DOCTYPE html> <html> <head><meta charset="utf-8"></head> <body> <table class="journal-table"> <thead> <tr> <th colspan="7">{subtitle}</th> </tr> <tr> <th>Datum</th> <th>Pers-Nr</th> <th>Person</th> <th>Typ</th> <th>Wert</th> <th>CHF</th> <th>CHF + TZ</th> </tr> </thead> <tbody> """ for settlement_row in settlement_data: name = f'{settlement_row[1].first_name} {settlement_row[1].last_name}' html += f""" <tr> <td>{settlement_row[0].strftime('%d.%m.%Y')}</td> <td>{settlement_row[1].personnel_number}</td> <td>{name}</td> <td>{settlement_row[2]}</td> <td class="numeric">{settlement_row[3]}</td> <td class="numeric">{settlement_row[4]:,.2f}</td> <td class="numeric">{settlement_row[5]:,.2f}</td> </tr> """ html += """ </tbody> </table> <table class="summary-table"> <thead> <tr> <th>Name</th> <th>Sitzungen</th> <th>Spesen</th> <th>Total Einträge</th> <th>Teuerungszulagen</th> <th>Auszahlung</th> </tr> </thead> <tbody> """ for total_row in totals[:-1]: # All but last html += f""" <tr> <td>{total_row[0]}</td> <td class="numeric">{format_swiss_number(total_row[1])}</td> <td class="numeric">{format_swiss_number(total_row[2])}</td> <td class="numeric">{format_swiss_number(total_row[3])}</td> <td class="numeric">{format_swiss_number(total_row[4])}</td> <td class="numeric">{format_swiss_number(total_row[5])}</td> </tr> """ # Handle last row separately with total-row class if totals: final_row = totals[-1] html += f""" <tr class="total-row"> <td>{final_row[0]}</td> <td class="numeric">{format_swiss_number(final_row[1])}</td> <td class="numeric">{format_swiss_number(final_row[2])}</td> <td class="numeric">{format_swiss_number(final_row[3])}</td> <td class="numeric">{format_swiss_number(final_row[4])}</td> <td class="numeric">{format_swiss_number(final_row[5])}</td> </tr> """ html += """ </tbody> </table> </body> </html> """ return html
[docs] def _get_data_export_all( self: SettlementRun, request: TownRequest ) -> list[SettlementDataRow]: session = request.session rate_set = get_current_rate_set(session, self) if not rate_set: return [] # Use collection to get all attendences in period attendences = AttendenceCollection( session, date_from=self.start, date_to=self.end ) cola_multiplier = Decimal( str(1 + (rate_set.cost_of_living_adjustment / 100)) ) settlement_data: list[SettlementDataRow] = [] # Add type hint here # Group by parliamentarian for summary parliamentarian_totals: dict[str, Decimal] = {} for attendence in attendences.query(): is_president = is_commission_president( attendence.parliamentarian, attendence, self ) # Calculate base rate base_rate = calculate_rate( rate_set=rate_set, attendence_type=attendence.type, duration_minutes=int(attendence.duration), is_president=is_president, commission_type=( attendence.commission.type if attendence.commission else None ), ) # Apply cost of living adjustment rate_with_cola = Decimal(base_rate * cola_multiplier) # Build description of the session/meeting translated_type = request.translate(TYPES[attendence.type]) if attendence.type == 'plenary': type_desc = translated_type elif attendence.type == 'commission' or attendence.type == 'study': commission_name = ( attendence.commission.name if attendence.commission else '' ) type_desc = f'{translated_type} - {commission_name}' else: # shortest type_desc = translated_type # Add row for this attendence settlement_data.append( ( attendence.date, attendence.parliamentarian, type_desc, attendence.calculate_value(), base_rate, rate_with_cola, ) ) # Update parliamentarian total parliamentarian_id = str(attendence.parliamentarian.id) if parliamentarian_id not in parliamentarian_totals: parliamentarian_totals[parliamentarian_id] = Decimal('0') parliamentarian_totals[parliamentarian_id] += rate_with_cola # Sort by date settlement_data.sort(key=itemgetter(0)) return settlement_data
[docs] def get_party_specific_totals( settlement_run: SettlementRun, request: TownRequest, party: Party ) -> list[TotalRow]: """Get totals for a specific party.""" session = request.session rate_set = get_current_rate_set(session, settlement_run) cola_multiplier = Decimal( str(1 + (rate_set.cost_of_living_adjustment / 100)) ) # Use AttendenceCollection with party filter attendences = ( AttendenceCollection( session, date_from=settlement_run.start, date_to=settlement_run.end, party_id=str(party.id), ) .query() .all() ) parliamentarian_totals: dict[str, dict[str, Decimal]] = {} for attendence in attendences: parliamentarian = attendence.parliamentarian name = f'{parliamentarian.first_name} {parliamentarian.last_name}' if name not in parliamentarian_totals: parliamentarian_totals[name] = { 'Meetings': Decimal('0'), 'Expenses': Decimal('0'), 'Total': Decimal('0'), 'Cost-of-living Allowance': Decimal('0'), 'Final': Decimal('0'), } is_president = any( r.role == 'president' for r in parliamentarian.roles ) base_rate = calculate_rate( rate_set=rate_set, attendence_type=attendence.type, duration_minutes=int(attendence.duration), is_president=is_president, commission_type=( attendence.commission.type if attendence.commission else None ), ) parliamentarian_totals[name]['Meetings'] += Decimal(str(base_rate)) base_total = parliamentarian_totals[name]['Meetings'] expenses = parliamentarian_totals[name]['Expenses'] cola_amount = base_total * (cola_multiplier - 1) parliamentarian_totals[name]['Total'] = base_total + expenses parliamentarian_totals[name]['Cost-of-living Allowance'] = cola_amount parliamentarian_totals[name]['Final'] = ( base_total + expenses + cola_amount ) result = [ ( name, data['Meetings'], data['Expenses'], data['Total'], data['Cost-of-living Allowance'], data['Final'], ) for name, data in sorted(parliamentarian_totals.items()) ] if result: totals = ( sum(r[1] for r in result), # Meetings sum(r[2] for r in result), # Expenses sum(r[3] for r in result), # Total sum(r[4] for r in result), # COLA sum(r[5] for r in result), # Final ) result.append( ( f'Total {party.name}', Decimal(str(totals[0])), Decimal(str(totals[1])), Decimal(str(totals[2])), Decimal(str(totals[3])), Decimal(str(totals[4])), ) ) return result
[docs] def _get_party_settlement_data( settlement_run: SettlementRun, request: TownRequest, party: Party ) -> list[SettlementDataRow]: """Get settlement data for a specific party.""" session = request.session rate_set = get_current_rate_set(session, settlement_run) # Get all attendences in period attendences = ( AttendenceCollection(session) .by_party( party_id=str(party.id), start_date=settlement_run.start, end_date=settlement_run.end ) .query() .all() ) result = [] for attendence in attendences: # Fixme: this check is likely redundant current_party = attendence.parliamentarian.get_party_during_period( settlement_run.start, settlement_run.end, session ) if not current_party or current_party.id != party.id: continue # found an export is_president = is_commission_president( attendence.parliamentarian, attendence, settlement_run ) base_rate = calculate_rate( rate_set=rate_set, attendence_type=attendence.type, duration_minutes=int(attendence.duration), is_president=is_president, commission_type=( attendence.commission.type if attendence.commission else None ) ) cola_multiplier = Decimal( str(1 + (rate_set.cost_of_living_adjustment / 100)) ) rate_with_cola = Decimal(str(base_rate)) * cola_multiplier type_desc = request.translate(TYPES[attendence.type]) if attendence.commission: type_desc = f'{type_desc} - {attendence.commission.name}' result.append(( attendence.date, attendence.parliamentarian, type_desc, attendence.calculate_value(), Decimal(str(base_rate)), rate_with_cola )) return sorted(result, key=itemgetter(0))
@PasApp.view( model=SettlementRunAllExport, permission=Private, name='run-export', request_method='GET' )
[docs] def view_settlement_run_all_export( self: SettlementRunAllExport, request: TownRequest ) -> Response: """Generate export data for a specific entity in a settlement run.""" if self.category == 'all-parties': pdf_bytes = generate_settlement_pdf( settlement_run=self.settlement_run, request=request, entity_type='all', ) filename = normalize_for_filename( request.translate(_('Total all parties'))) return Response( pdf_bytes, content_type='application/pdf', content_disposition=f'attachment; filename={filename}.pdf' ) elif self.category == 'abschlussliste-xlsx-export': filename = 'Abschlussliste.xlsx' output = generate_abschlussliste_xlsx(self.settlement_run, request) return Response( output.read(), content_type=XLSX_MIMETYPE, content_disposition=f'attachment; filename="{filename}"' ) elif self.category == 'buchungen-abrechnungslauf-kontroll-xlsx-export': filename = 'Buchungen_Abrechnungslauf.xlsx' output = generate_buchungen_abrechnungslauf_xlsx( self.settlement_run, request) return Response( output.read(), content_type=XLSX_MIMETYPE, content_disposition=f'attachment; filename="{filename}"' ) elif self.category == 'fibu-csv-export': year = self.settlement_run.end.year filename = f'KR-Entschaedigung - {year}.csv' output = BytesIO() # Use utf-8-sig to ensure proper Excel compatibility with BOM csv_data = list(generate_fibu_export_rows( self.settlement_run, request)) # Create CSV content as string csv_string = '' for row in csv_data: # Convert all values to strings and escape quotes row_strings = [] for value in row: if value is None: row_strings.append('') # type: ignore[unreachable] else: # Convert to string and handle quotes str_value = str(value) if '"' in str_value: str_value = str_value.replace('"', '""') if (',' in str_value or '"' in str_value or '\n' in str_value): str_value = f'"{str_value}"' row_strings.append(str_value) csv_string += ','.join(row_strings) + '\n' # Encode to bytes with BOM for Excel compatibility csv_bytes = '\ufeff'.encode('utf-8') + csv_string.encode('utf-8') return Response( csv_bytes, content_type='text/csv; charset=utf-8', content_disposition=f'attachment; filename="{filename}"' ) else: raise NotImplementedError( f'Export category {self.category} not implemented for all exports' )
@PasApp.view( model=SettlementRunExport, permission=Private, name='run-export', request_method='GET' )
[docs] def view_settlement_run_export( self: SettlementRunExport, request: TownRequest ) -> Response: """Generate export data for a specific entity (commission, party or parliamentarian) in a settlement run.""" if self.category == 'party': assert isinstance(self.entity, Party) pdf_bytes = generate_settlement_pdf( settlement_run=self.settlement_run, request=request, entity_type='party', entity=self.entity, ) filename = normalize_for_filename(f'Partei_{self.entity.name}') elif self.category == 'commission': assert isinstance(self.entity, PASCommission) pdf_bytes = generate_settlement_pdf( settlement_run=self.settlement_run, request=request, entity_type='commission', entity=self.entity, ) filename = normalize_for_filename(f'commission_{self.entity.name}') elif self.category == 'parliamentarian': assert isinstance(self.entity, PASParliamentarian) # PASParliamentarian specific export has it's own rendering function pdf_bytes = generate_parliamentarian_settlement_pdf( self.settlement_run, request, self.entity ) filename = normalize_for_filename( f'Parlamentarier_{self.entity.last_name}_{self.entity.first_name}' ) return Response( pdf_bytes, content_type='application/pdf', content_disposition=f'attachment; filename={filename}.pdf' ) else: raise NotImplementedError( f'Export category {self.category} not implemented' ) return Response( pdf_bytes, content_type='application/pdf', content_disposition=f'attachment; filename={filename}.pdf' )
@PasApp.form( model=SettlementRun, name='edit', template='form.pt', permission=Private, form=SettlementRunForm )
[docs] def edit_settlement_run( self: SettlementRun, request: TownRequest, form: SettlementRunForm ) -> RenderData | Response: if form.submitted(request): form.populate_obj(self) request.success(_('Your changes were saved')) return request.redirect(request.link(self)) form.process(obj=self) layout = SettlementRunLayout(self, request) layout.breadcrumbs.append(Link(_('Edit'), '#')) layout.editbar_links = [] return { 'layout': layout, 'title': layout.title, 'form': form, 'form_width': 'large' }
@PasApp.view( model=SettlementRun, request_method='DELETE', permission=Private )
[docs] def delete_settlement_run( self: SettlementRun, request: TownRequest ) -> None: request.assert_valid_csrf_token() collection = SettlementRunCollection(request.session) collection.delete(self)