nordabiz/blueprints/board/routes.py
Maciej Pienczyn 86d10c8d70
Some checks are pending
NordaBiz Tests / Unit & Integration Tests (push) Waiting to run
NordaBiz Tests / E2E Tests (Playwright) (push) Blocked by required conditions
NordaBiz Tests / Smoke Tests (Production) (push) Blocked by required conditions
NordaBiz Tests / Send Failure Notification (push) Blocked by required conditions
fix: Use SessionLocal instead of db_session in board routes
Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
2026-02-03 18:43:20 +01:00

266 lines
8.9 KiB
Python

"""
Board Routes (Rada Izby)
========================
Routes for board document management.
Endpoints:
- GET /rada/ - Document list
- GET/POST /rada/upload - Upload new document (office_manager+)
- GET /rada/<id>/download - Download document
- POST /rada/<id>/delete - Soft delete document (office_manager+)
"""
import os
from datetime import datetime
from flask import (
render_template, request, redirect, url_for, flash,
send_file, current_app
)
from flask_login import login_required, current_user
from sqlalchemy import desc, extract
from . import bp
from database import SessionLocal, BoardDocument, SystemRole
from utils.decorators import rada_member_required, office_manager_required
from services.document_upload_service import DocumentUploadService
@bp.route('/')
@login_required
@rada_member_required
def index():
"""Display list of board documents with filtering"""
# Get filter parameters
year = request.args.get('year', type=int)
month = request.args.get('month', type=int)
doc_type = request.args.get('type', '')
db = SessionLocal()
try:
# Build query
query = db.query(BoardDocument).filter(BoardDocument.is_active == True)
if year:
query = query.filter(extract('year', BoardDocument.meeting_date) == year)
if month:
query = query.filter(extract('month', BoardDocument.meeting_date) == month)
if doc_type and doc_type in BoardDocument.DOCUMENT_TYPES:
query = query.filter(BoardDocument.document_type == doc_type)
# Order by meeting date descending
documents = query.order_by(desc(BoardDocument.meeting_date)).all()
# Get available years for filter
all_docs = db.query(BoardDocument).filter(BoardDocument.is_active == True).all()
available_years = sorted(set(
doc.meeting_date.year for doc in all_docs if doc.meeting_date
), reverse=True)
# Check if user can upload (office_manager or admin)
can_upload = current_user.has_role(SystemRole.OFFICE_MANAGER)
return render_template(
'board/index.html',
documents=documents,
available_years=available_years,
document_types=BoardDocument.DOCUMENT_TYPES,
type_labels=BoardDocument.DOCUMENT_TYPE_LABELS,
current_year=year,
current_month=month,
current_type=doc_type,
can_upload=can_upload
)
finally:
db.close()
@bp.route('/upload', methods=['GET', 'POST'])
@login_required
@office_manager_required
def upload():
"""Upload new board document"""
if request.method == 'POST':
# Get form data
title = request.form.get('title', '').strip()
description = request.form.get('description', '').strip()
document_type = request.form.get('document_type', 'protocol')
meeting_date_str = request.form.get('meeting_date', '')
meeting_number = request.form.get('meeting_number', type=int)
# Validate required fields
errors = []
if not title:
errors.append('Tytuł dokumentu jest wymagany.')
if not meeting_date_str:
errors.append('Data posiedzenia jest wymagana.')
# Parse meeting date
meeting_date = None
if meeting_date_str:
try:
meeting_date = datetime.strptime(meeting_date_str, '%Y-%m-%d').date()
except ValueError:
errors.append('Nieprawidłowy format daty.')
# Validate document type
if document_type not in BoardDocument.DOCUMENT_TYPES:
document_type = 'other'
# Get uploaded file
file = request.files.get('document')
if not file or file.filename == '':
errors.append('Plik dokumentu jest wymagany.')
# Validate file
if file and file.filename:
is_valid, error_msg = DocumentUploadService.validate_file(file)
if not is_valid:
errors.append(error_msg)
if errors:
for error in errors:
flash(error, 'error')
return render_template(
'board/upload.html',
document_types=BoardDocument.DOCUMENT_TYPES,
type_labels=BoardDocument.DOCUMENT_TYPE_LABELS,
form_data={
'title': title,
'description': description,
'document_type': document_type,
'meeting_date': meeting_date_str,
'meeting_number': meeting_number
}
)
# Save file
try:
stored_filename, file_path, file_size, mime_type = DocumentUploadService.save_file(file)
except Exception as e:
current_app.logger.error(f"Failed to save board document: {e}")
flash('Błąd podczas zapisywania pliku. Spróbuj ponownie.', 'error')
return redirect(url_for('board.upload'))
# Get file extension
file_extension = file.filename.rsplit('.', 1)[-1].lower() if '.' in file.filename else ''
# Create database record
document = BoardDocument(
title=title,
description=description if description else None,
document_type=document_type,
meeting_date=meeting_date,
meeting_number=meeting_number if meeting_number else None,
original_filename=file.filename,
stored_filename=stored_filename,
file_extension=file_extension,
file_size=file_size,
mime_type=mime_type,
uploaded_by=current_user.id
)
db = SessionLocal()
try:
db.add(document)
db.commit()
flash(f'Dokument "{title}" został dodany.', 'success')
current_app.logger.info(f"Board document uploaded: {title} by user {current_user.id}")
return redirect(url_for('board.index'))
except Exception as e:
db.rollback()
# Clean up uploaded file
DocumentUploadService.delete_file(stored_filename)
current_app.logger.error(f"Failed to create board document record: {e}")
flash('Błąd podczas tworzenia rekordu. Spróbuj ponownie.', 'error')
return redirect(url_for('board.upload'))
finally:
db.close()
# GET request - show upload form
return render_template(
'board/upload.html',
document_types=BoardDocument.DOCUMENT_TYPES,
type_labels=BoardDocument.DOCUMENT_TYPE_LABELS,
form_data={}
)
@bp.route('/<int:doc_id>/download')
@login_required
@rada_member_required
def download(doc_id):
"""Download board document (secure, authenticated)"""
db = SessionLocal()
try:
document = db.query(BoardDocument).filter(
BoardDocument.id == doc_id,
BoardDocument.is_active == True
).first()
if not document:
flash('Dokument nie został znaleziony.', 'error')
return redirect(url_for('board.index'))
# Get file path
file_path = DocumentUploadService.get_file_path(
document.stored_filename,
document.uploaded_at
)
if not os.path.exists(file_path):
current_app.logger.error(f"Board document file not found: {file_path}")
flash('Plik dokumentu nie został znaleziony.', 'error')
return redirect(url_for('board.index'))
# Log download
current_app.logger.info(
f"Board document downloaded: {document.title} (ID: {doc_id}) by user {current_user.id}"
)
# Send file with original filename
return send_file(
file_path,
as_attachment=True,
download_name=document.original_filename,
mimetype=document.mime_type
)
finally:
db.close()
@bp.route('/<int:doc_id>/delete', methods=['POST'])
@login_required
@office_manager_required
def delete(doc_id):
"""Soft delete board document"""
db = SessionLocal()
try:
document = db.query(BoardDocument).filter(
BoardDocument.id == doc_id,
BoardDocument.is_active == True
).first()
if not document:
flash('Dokument nie został znaleziony.', 'error')
return redirect(url_for('board.index'))
# Soft delete
document.is_active = False
document.updated_at = datetime.now()
document.updated_by = current_user.id
db.commit()
flash(f'Dokument "{document.title}" został usunięty.', 'success')
current_app.logger.info(
f"Board document deleted: {document.title} (ID: {doc_id}) by user {current_user.id}"
)
except Exception as e:
db.rollback()
current_app.logger.error(f"Failed to delete board document: {e}")
flash('Błąd podczas usuwania dokumentu.', 'error')
finally:
db.close()
return redirect(url_for('board.index'))