from __future__ import annotations
from onegov.pas.models.attendence import Attendence
from onegov.pas.models.commission import PASCommission
from onegov.pas.models.commission_membership import PASCommissionMembership
from onegov.pas.models.party import Party
from onegov.pas.models.parliamentarian import PASParliamentarian
from onegov.pas.models.parliamentarian_role import PASParliamentarianRole
from onegov.pas.models.presidential_allowance import (
PresidentialAllowance,
)
from onegov.pas.collections import PASParliamentarianCollection
from decimal import Decimal, ROUND_HALF_UP
from babel.numbers import format_decimal
from datetime import date
from lxml import html
from uuid import UUID
import requests
from sqlalchemy.orm import selectinload
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from onegov.core import Framework
from onegov.parliament.models.parliamentarian import Parliamentarian
from onegov.parliament.models.parliamentarian_role import (
ParliamentarianRole,
)
from onegov.pas.models import SettlementRun
from onegov.pas.models.attendence import Attendence
from onegov.user import User
from sqlalchemy.orm import Session
[docs]
def _is_kantonsrat_role(
role: PASParliamentarianRole | ParliamentarianRole,
) -> bool:
"""A Kantonsrat role has meta.org_type == 'Kantonsrat' (set
by the KUB importer).
"""
org_type = role.meta.get('org_type')
if org_type is not None:
return org_type == 'Kantonsrat'
return False
[docs]
def is_active_kantonsrat_member(
parliamentarian: PASParliamentarian | Parliamentarian,
reference_date: date | None = None,
) -> bool:
if reference_date is None:
reference_date = date.today()
return any(
_is_kantonsrat_role(r) and (r.end is None or r.end >= reference_date)
for r in parliamentarian.roles
)
[docs]
def get_active_kantonsrat_parliamentarians(
app: Framework,
reference_date: date | None = None,
) -> list[PASParliamentarian]:
parliamentarians = (
PASParliamentarianCollection(app)
.query()
.options(selectinload(PASParliamentarian.roles))
.all()
)
return [
p
for p in parliamentarians
if is_active_kantonsrat_member(p, reference_date)
]
[docs]
CLEX_URL = 'https://zg-compwork.clex.ch/frontend/people'
[docs]
def fetch_clex_kantonsrat_members(
reference_date: date | None = None,
) -> set[tuple[str, str]]:
"""Fetch active Kantonsrat members from the CLEX frontend.
Returns set of (last_name, first_name) tuples for members
without an exit date or with an exit date >= reference_date.
"""
if reference_date is None:
reference_date = date.today()
resp = requests.get(CLEX_URL, timeout=30)
resp.raise_for_status()
tree = html.fromstring(resp.content)
rows = tree.xpath('//table//tr')
members: set[tuple[str, str]] = set()
for row in rows:
cells = row.xpath('td')
if len(cells) < 6:
continue
last_name = (cells[1].text_content() or '').strip()
first_name = (cells[2].text_content() or '').strip()
exit_text = (cells[6].text_content() or '').strip()
if not last_name:
continue
if exit_text and exit_text != '—':
try:
parts = exit_text.split('.')
exit_date = date(int(parts[2]), int(parts[1]), int(parts[0]))
if exit_date < reference_date:
continue
except (ValueError, IndexError):
pass
members.add((last_name, first_name))
return members
[docs]
def round_to_five_rappen(value: Decimal | int) -> Decimal:
"""Round a decimal value to the nearest 5 Rappen (0.05 CHF)."""
if isinstance(value, int):
value = Decimal(value)
return (value / Decimal('0.05')).quantize(
Decimal('1'), rounding=ROUND_HALF_UP
) * Decimal('0.05')
[docs]
def is_commission_president(
parliamentarian: PASParliamentarian,
attendance_or_commission_id: Attendence | UUID,
settlement_run: SettlementRun
) -> bool:
"""
Check if a parliamentarian is president of the commission for the given
attendance or commission_id during the settlement run period.
"""
if isinstance(attendance_or_commission_id, UUID):
commission_id = attendance_or_commission_id
return any(
cm.role == 'president'
for cm in parliamentarian.commission_memberships
if (
cm.commission_id == commission_id and (
cm.end is None or cm.end >= settlement_run.start
) and (
cm.start is None or cm.start <= settlement_run.end
)
)
)
else:
attendance = attendance_or_commission_id
return any(
cm.role == 'president'
for cm in parliamentarian.commission_memberships
if (
attendance.commission and
cm.commission_id == attendance.commission.id and (
cm.end is None or cm.end >= settlement_run.start
) and (
cm.start is None or cm.start <= settlement_run.end
)
)
)
[docs]
def get_parliamentarians_with_settlements(
session: Session,
start_date: date,
end_date: date,
settlement_run_id: UUID | None = None,
) -> list[PASParliamentarian]:
"""
Get all parliamentarians who were active and had settlements
(attendances or presidential allowances) during the specified
period.
"""
active_parliamentarians = session.query(PASParliamentarian).filter(
PASParliamentarian.id.in_(
session.query(PASParliamentarianRole.parliamentarian_id).filter(
(PASParliamentarianRole.start.is_(None) | (
PASParliamentarianRole.start <= end_date)),
(PASParliamentarianRole.end.is_(None) | (
PASParliamentarianRole.end >= start_date))
)
)
).order_by(
PASParliamentarian.last_name,
PASParliamentarian.first_name
).all()
parliamentarians_with_attendances = {
pid[0] for pid in
session.query(Attendence.parliamentarian_id).filter(
Attendence.date >= start_date,
Attendence.date <= end_date
).distinct()
}
parliamentarians_with_allowances: set[UUID] = set()
if settlement_run_id:
parliamentarians_with_allowances = {
pid[0]
for pid in session.query(PresidentialAllowance.parliamentarian_id)
.filter(
PresidentialAllowance.settlement_run_id == settlement_run_id
)
.distinct()
}
eligible = (
parliamentarians_with_attendances | parliamentarians_with_allowances
)
return [p for p in active_parliamentarians if p.id in eligible]
[docs]
def get_parties_with_settlements(
session: Session,
start_date: date,
end_date: date
) -> list[Party]:
"""
Get all parties that had active members with attendances during the
specified period.
This function ensures accurate party filtering by checking that
parliamentarians were active members of their party at the time of each
attendance, properly handling cases where parliamentarians switch parties
or have changing membership dates.
"""
return (
session.query(Party)
.filter(
Party.id.in_(
session.query(PASParliamentarianRole.party_id)
.join(PASParliamentarian)
.join(
Attendence,
PASParliamentarian.id == Attendence.parliamentarian_id,
)
.filter(
Attendence.date >= start_date,
Attendence.date <= end_date,
PASParliamentarianRole.party_id.isnot(None),
(
PASParliamentarianRole.start.is_(None)
| (PASParliamentarianRole.start <= Attendence.date)
),
(
PASParliamentarianRole.end.is_(None)
| (PASParliamentarianRole.end >= Attendence.date)
),
)
.distinct()
)
)
.order_by(Party.name)
.all()
)
[docs]
def is_parliamentarian(user: User | None) -> bool:
"""Check if a user has parliamentarian role."""
parls = {'parliamentarian', 'commission_president'}
if not user:
return False
if user.role in parls:
return True
return False
[docs]
def is_parliamentarian_role(role: str | None) -> bool:
"""Check if a role is a parliamentarian role."""
parls = {'parliamentarian', 'commission_president'}
return role in parls if role else False
[docs]
def get_active_commission_memberships(
user: User | None
) -> list[PASCommissionMembership]:
"""Get active commission memberships for a parliamentarian user."""
if (not user or not hasattr(user, 'parliamentarian')
or not user.parliamentarian):
return []
parliamentarian = user.parliamentarian
return [
m for m in parliamentarian.commission_memberships
if not m.end or m.end >= date.today()
]
[docs]
def get_commissions_with_memberships(
session: Session,
start_date: date,
end_date: date
) -> list[PASCommission]:
"""
Get all commissions that had active memberships during the
specified period.
This function ensures accurate commission filtering by checking that
commissions had active members during the period, properly handling
cases where memberships have changing dates.
"""
return (
session.query(PASCommission)
.filter(
PASCommission.id.in_(
session.query(PASCommissionMembership.commission_id)
.filter(
(
PASCommissionMembership.start.is_(None)
| (PASCommissionMembership.start <= end_date)
),
(
PASCommissionMembership.end.is_(None)
| (PASCommissionMembership.end >= start_date)
),
)
.distinct()
)
)
.order_by(PASCommission.name)
.all()
)