""" Provides commands used to initialize election day websites. """
from __future__ import annotations
import click
import os
from onegov.core.cli import abort
from onegov.core.cli import command_group
from onegov.core.cli import pass_group_context
from onegov.core.sms_processor import SmsQueueProcessor
from onegov.election_day.collections import ArchivedResultCollection
from onegov.election_day.models import ArchivedResult
from onegov.election_day.models import Subscriber
from onegov.election_day.utils import add_local_results
from onegov.election_day.utils.archive_generator import ArchiveGenerator
from onegov.election_day.utils.d3_renderer import D3Renderer
from onegov.election_day.utils.pdf_generator import PdfGenerator
from onegov.election_day.utils.svg_generator import SvgGenerator
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from collections.abc import Callable
from onegov.core.cli.core import GroupContext
from onegov.election_day.app import ElectionDayApp
from onegov.election_day.request import ElectionDayRequest
from typing import TypeAlias
[docs]
Processor: TypeAlias = Callable[[ElectionDayRequest, ElectionDayApp], None]
@cli.command(context_settings={'creates_path': True})
@pass_group_context
[docs]
def add(group_context: GroupContext) -> Processor:
""" Adds an election day instance with to the database. For example:
.. code-block:: bash
onegov-election-day --select '/onegov_election_day/zg' add
"""
def add_instance(
request: ElectionDayRequest,
app: ElectionDayApp
) -> None:
app.cache.flush()
if not app.principal:
click.secho('principal.yml not found', fg='yellow')
click.echo('Instance was created successfully')
return add_instance
@cli.command()
@pass_group_context
[docs]
def fetch(group_context: GroupContext) -> Processor:
""" Fetches the results from other instances as defined in the
principal.yml. Only fetches results from the same namespace.
.. code-block:: bash
onegov-election-day --select '/onegov_election_day/zg' fetch
"""
def fetch_results(
request: ElectionDayRequest,
app: ElectionDayApp
) -> None:
if not app.principal:
return
local_session = app.session()
assert local_session.info['schema'] == app.schema
for key in app.principal.fetch:
schema = '{}-{}'.format(app.namespace, key)
assert schema in app.session_manager.list_schemas()
app.session_manager.set_current_schema(schema)
remote_session = app.session_manager.session()
assert remote_session.info['schema'] == schema
items = local_session.query(ArchivedResult)
items = items.filter_by(schema=schema)
for item in items:
local_session.delete(item)
for domain in app.principal.fetch[key]:
items = remote_session.query(ArchivedResult)
items = items.filter_by(schema=schema, domain=domain)
for item in items:
new_item = ArchivedResult()
new_item.copy_from(item)
add_local_results(
item, new_item, app.principal, remote_session
)
local_session.add(new_item)
return fetch_results
# TODO: Get rid of this command, there is now an equivalent command in
# core as well as an option to run a daemon
@cli.command('send-sms')
@click.argument('username')
@click.argument('password')
@click.option('--originator')
@pass_group_context
[docs]
def send_sms(
group_context: GroupContext,
username: str,
password: str,
originator: str | None
) -> Processor:
r""" Sends the SMS in the smsdir for a given instance. For example:
.. code-block:: bash
onegov-election-day --select '/onegov_election_day/zg' \
send_sms 'info@seantis.ch' 'top-secret'
"""
def send(
request: ElectionDayRequest,
app: ElectionDayApp
) -> None:
if 'sms_directory' in app.configuration:
path = os.path.join(app.configuration['sms_directory'], app.schema)
if os.path.exists(path):
qp = SmsQueueProcessor(
path,
username,
password,
originator
)
qp.send_messages()
return send
@cli.command('generate-media')
@cli.command('generate-archive')
[docs]
def generate_archive() -> Processor:
""" Generates a zipped file of the entire archive.
.. code-block:: bash
onegov-election-day --select '/onegov_election_day/zg' generate-archive
"""
def generate(request: ElectionDayRequest, app: ElectionDayApp) -> None:
click.secho('Starting archive.zip generation.')
archive_generator = ArchiveGenerator(app)
archive_zip = archive_generator.generate_archive()
if not archive_zip:
abort('generate_archive returned None.')
archive_filesize = archive_generator.archive_dir.getinfo(
archive_zip, namespaces=['details']).size
if archive_filesize == 0:
click.secho('Generated archive is empty', fg='red')
else:
click.secho('Archive generated successfully:', fg='green')
absolute_path = archive_generator.archive_system_path
if absolute_path:
click.secho(f'file://{absolute_path}')
return generate
@cli.command('update-archived-results')
@click.option('--host', default='localhost:8080')
@click.option('--scheme', default='http')
[docs]
def update_archived_results(host: str, scheme: str) -> Processor:
""" Update the archive results, e.g. after a database transfer. """
def generate(request: ElectionDayRequest, app: ElectionDayApp) -> None:
click.secho(f'Updating {app.schema}', fg='yellow')
request.host = host
request.environ['wsgi.url_scheme'] = scheme
archive = ArchivedResultCollection(request.session)
archive.update_all(request)
return generate
@cli.command('migrate-subscribers')
[docs]
def migrate_subscribers() -> Processor:
def migrate(request: ElectionDayRequest, app: ElectionDayApp) -> None:
if not app.principal or not app.principal.segmented_notifications:
return
click.secho(f'Migrating {app.schema}', fg='yellow')
session = request.app.session()
subscribers = session.query(Subscriber).filter_by(domain=None)
count = 0
for subscriber in subscribers:
subscriber.domain = 'canton'
count += 1
click.echo(f'Migrated {count} subscribers')
return migrate