Utility Functions

from app.utils.time import get_current_financial_year

# If current date is June 2024, returns 2024
# If current date is February 2024, returns 2023
financial_year = get_current_financial_year()

from app.utils.time import percentage_through_current_financial_year

# Returns value between 0.001 and 100
percentage = percentage_through_current_financial_year()
print(f"We are {percentage}% through the financial year")

from app.utils.time import is_less_than_days_ago
from datetime import datetime, UTC

processing_started = datetime.now(UTC)
if is_less_than_days_ago(processing_started, 7):
    print("Within data retention period")

from app.utils.time import to_utc_string
from datetime import datetime, UTC

aware_dt = datetime.now(UTC)
utc_string = to_utc_string(aware_dt)
# Returns: "2024-12-15T14:30:45.123456Z"

from app.utils.time import str_no_tz

aware_dt = datetime.now(UTC)
string_dt = str_no_tz(aware_dt)
# Returns: "2024-12-15 14:30:45.123456"

from app.utils.time import isoformat_no_tz

aware_dt = datetime.now(UTC)
iso_string = isoformat_no_tz(aware_dt)
# Returns: "2024-12-15T14:30:45.123456"

from app.utils.user import user_is_logged_in

@app.route('/dashboard')
@user_is_logged_in
def dashboard():
    return render_template('dashboard.html')

from app.utils.user import user_has_permissions

@app.route('/send')
@user_has_permissions('send_messages', 'manage_templates')
def send_messages():
    # User must have both permissions
    return render_template('send.html')

from app.utils.user import user_is_gov_user

@app.route('/admin')
@user_is_gov_user
def admin_panel():
    # Only users with .gov.uk or approved org emails
    return render_template('admin.html')

from app.utils.user import user_is_platform_admin

@app.route('/platform-admin')
@user_is_platform_admin
def platform_admin():
    # Only platform admins can access
    return render_template('platform_admin.html')

from app.utils.user import is_gov_user

if is_gov_user("[email protected]"):
    print("Government user")
else:
    print("Non-government user")

from app.utils.csv import get_errors_for_csv
from notifications_utils.recipients import RecipientCSV

recipients = RecipientCSV(file_data, template=template)
errors = get_errors_for_csv(recipients, "email")

if errors:
    print(f"CSV has errors: {', '.join(errors)}")
    # Output: "fix 3 email addresses, enter missing data in 2 rows"

from app.utils.csv import generate_notifications_csv

# Generate CSV for job
for chunk in generate_notifications_csv(
    service_id=service_id,
    job_id=job_id,
    template_type="email",
    page_size=50
):
    yield chunk

from app.utils.branding import get_email_choices

choices = list(get_email_choices(service))
# Returns: [("govuk", "GOV.UK"), ("nhs", "NHS"), ...]

from app.utils.branding import get_letter_choices

choices = list(get_letter_choices(service))
# Returns: [("nhs", "NHS"), ("org123", "Department Name"), ...]

from app.utils.branding import letter_filename_for_db_from_logo_key

logo_key = "letters/logos/department-logo.svg"
filename = letter_filename_for_db_from_logo_key(logo_key)
# Returns: "department-logo"

from app.utils.pagination import get_page_from_request

page = get_page_from_request()
if page is None:
    # Invalid page parameter
    abort(400)

from app.utils.pagination import generate_previous_dict

previous = generate_previous_dict(
    'main.view_jobs',
    service_id,
    page=2,
    url_args={'status': 'completed'}
)
# Returns: {
#   "url": "/services/123/jobs?page=1&status=completed",
#   "title": "Previous page",
#   "label": "page 1"
# }

from app.utils.pagination import generate_next_dict

next_page = generate_next_dict(
    'main.view_jobs',
    service_id,
    page=2,
    url_args={'status': 'completed'}
)
# Returns: {
#   "url": "/services/123/jobs?page=3&status=completed",
#   "title": "Next page",
#   "label": "page 3"
# }

from app.utils.pagination import generate_optional_previous_and_next_dicts

previous, next_page = generate_optional_previous_and_next_dicts(
    view='main.view_jobs',
    service_id=service_id,
    page=2,
    num_pages=5,
    url_args={'status': 'completed'}
)

if previous:
    # Render previous link
if next_page:
    # Render next link

from app.utils.templates import get_sample_template

email_template = get_sample_template("email")
sms_template = get_sample_template("sms")
letter_template = get_sample_template("letter")

from app.utils.templates import get_template

template_obj = get_template(
    template=template_data,
    service=service,
    show_recipient=True,
    redact_missing_personalisation=False,
    email_reply_to="[email protected]",
    sms_sender="Department"
)

rendered = str(template_obj)

from app.utils.templates import TemplateChange

change = TemplateChange(
    old_template,
    new_template,
    service_has_api_keys=True
)

if change.is_breaking_change:
    # Warn user about breaking changes
    print(f"Added placeholders: {change.placeholders_added}")
    print(f"Removed placeholders: {change.placeholders_removed}")

from app.utils.user import is_gov_user
from app.utils.time import to_utc_string
from datetime import datetime, UTC

def register_new_user(email, name, mobile):
    # Validate government email
    if not is_gov_user(email):
        raise ValueError("Must use government email")
    
    # Register with timestamp
    user = user_api_client.register_user(
        name=name,
        email_address=email,
        mobile_number=mobile,
        password=generate_password(),
        auth_type="sms_auth"
    )
    
    # Log registration time
    registration_time = to_utc_string(datetime.now(UTC))
    logger.info(f"User {user['id']} registered at {registration_time}")
    
    return user

from app.utils.csv import get_errors_for_csv
from app.utils.pagination import generate_optional_previous_and_next_dicts
from notifications_utils.recipients import RecipientCSV

def process_csv_upload(file_data, template, service_id):
    # Parse and validate CSV
    recipients = RecipientCSV(file_data, template=template)
    errors = get_errors_for_csv(recipients, template.template_type)
    
    if errors:
        return {"errors": errors}
    
    # Create job
    job = job_api_client.create_job(
        job_id=generate_job_id(),
        service_id=service_id
    )
    
    return {"job_id": job['id']}

def view_job_notifications(service_id, job_id, page):
    # Get paginated notifications
    notifications = notification_api_client.get_notifications_for_service(
        service_id,
        job_id=job_id,
        page=page,
        page_size=50
    )
    
    # Generate pagination links
    previous, next_page = generate_optional_previous_and_next_dicts(
        view='main.view_job',
        service_id=service_id,
        page=page,
        num_pages=notifications['total_pages']
    )
    
    return render_template(
        'notifications.html',
        notifications=notifications['data'],
        previous=previous,
        next=next_page
    )