k1nq
202 lines
7.4 KiB
Python
202 lines
7.4 KiB
Python
import csv
|
|
import io
|
|
import json
|
|
import uuid
|
|
import zipfile
|
|
from collections.abc import Sequence
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.audit.models import AuditEvent
|
|
from app.budgeting.models import BudgetPeriod, ExpectedIncome, FundingMovement, PlanAssignment
|
|
from app.config import get_settings
|
|
from app.errors import conflict, not_found
|
|
from app.exports.models import DataExport, DeletionLedger
|
|
from app.imports.blob import get_blob_store, new_object_key
|
|
from app.imports.crypto import decrypt_file, encrypt_file
|
|
from app.imports.models import StatementDocument
|
|
from app.ledger.models import Account, Category, Operation
|
|
from app.spaces.models import FinancialSpace
|
|
|
|
|
|
def request_export(db: Session, *, space_id: uuid.UUID, user_id: uuid.UUID) -> DataExport:
|
|
job = DataExport(space_id=space_id, requested_by_user_id=user_id, status="queued")
|
|
db.add(job)
|
|
db.flush()
|
|
return job
|
|
|
|
|
|
def get_export(db: Session, *, space_id: uuid.UUID, export_id: uuid.UUID) -> DataExport:
|
|
job = db.query(DataExport).filter_by(id=export_id, space_id=space_id).one_or_none()
|
|
if job is None:
|
|
raise not_found("EXPORT_NOT_FOUND", "Data export not found")
|
|
return job
|
|
|
|
|
|
def _json_rows(rows: Sequence[object], fields: tuple[str, ...]) -> list[dict]:
|
|
return [{field: getattr(row, field) for field in fields} for row in rows]
|
|
|
|
|
|
def _csv_bytes(rows: list[dict]) -> bytes:
|
|
output = io.StringIO()
|
|
if rows:
|
|
writer = csv.DictWriter(output, fieldnames=list(rows[0]))
|
|
writer.writeheader()
|
|
writer.writerows(rows)
|
|
return output.getvalue().encode()
|
|
|
|
|
|
def build_export(db: Session, *, space_id: uuid.UUID, export_id: uuid.UUID) -> None:
|
|
job = get_export(db, space_id=space_id, export_id=export_id)
|
|
if job.status == "ready":
|
|
return
|
|
job.status = "processing"
|
|
datasets = {
|
|
"accounts": _json_rows(
|
|
db.query(Account).filter_by(space_id=space_id).all(),
|
|
("id", "name", "kind", "currency", "opening_amount_kopecks", "opened_on", "status"),
|
|
),
|
|
"operations": _json_rows(
|
|
db.query(Operation).filter_by(space_id=space_id).all(),
|
|
(
|
|
"id",
|
|
"type",
|
|
"status",
|
|
"amount_kopecks",
|
|
"account_id",
|
|
"counter_account_id",
|
|
"category_id",
|
|
"occurred_on",
|
|
"posted_on",
|
|
"description",
|
|
),
|
|
),
|
|
"categories": _json_rows(
|
|
db.query(Category).filter_by(space_id=space_id).all(),
|
|
("id", "kind", "system_key", "name_override", "custom_name", "archived"),
|
|
),
|
|
"budget_periods": _json_rows(
|
|
db.query(BudgetPeriod).filter_by(space_id=space_id).all(),
|
|
("id", "year", "month", "status"),
|
|
),
|
|
"plan_assignments": _json_rows(
|
|
db.query(PlanAssignment).filter_by(space_id=space_id).all(),
|
|
("id", "period_id", "category_id", "planned_kopecks"),
|
|
),
|
|
"funding": _json_rows(
|
|
db.query(FundingMovement).filter_by(space_id=space_id).all(),
|
|
("id", "period_id", "category_id", "kind", "amount_kopecks"),
|
|
),
|
|
"expected_incomes": _json_rows(
|
|
db.query(ExpectedIncome).filter_by(space_id=space_id).all(),
|
|
("id", "period_id", "expected_on", "amount_kopecks", "source_name"),
|
|
),
|
|
"audit": _json_rows(
|
|
db.query(AuditEvent).filter_by(space_id=space_id).all(),
|
|
("id", "event_type", "created_at", "detail"),
|
|
),
|
|
}
|
|
raw = io.BytesIO()
|
|
with zipfile.ZipFile(raw, "w", compression=zipfile.ZIP_DEFLATED) as archive:
|
|
for name, rows in datasets.items():
|
|
archive.writestr(f"{name}.csv", _csv_bytes(rows))
|
|
archive.writestr(
|
|
"data.json", json.dumps(datasets, default=str, ensure_ascii=False, indent=2)
|
|
)
|
|
encrypted = encrypt_file(raw.getvalue())
|
|
key = new_object_key()
|
|
get_blob_store().put(key, encrypted.ciphertext)
|
|
job.object_key = key
|
|
job.dek_wrapped = encrypted.dek_wrapped
|
|
job.dek_wrap_nonce = encrypted.wrap_nonce
|
|
job.content_nonce = encrypted.content_nonce
|
|
job.expires_at = datetime.now(UTC) + timedelta(hours=1)
|
|
job.status = "ready"
|
|
|
|
|
|
def consume_export(db: Session, *, space_id: uuid.UUID, export_id: uuid.UUID) -> bytes:
|
|
job = get_export(db, space_id=space_id, export_id=export_id)
|
|
now = datetime.now(UTC)
|
|
if (
|
|
job.status != "ready"
|
|
or job.consumed_at is not None
|
|
or not job.expires_at
|
|
or job.expires_at <= now
|
|
):
|
|
raise conflict("EXPORT_UNAVAILABLE", "Export is not ready, expired, or already downloaded")
|
|
if not all((job.object_key, job.dek_wrapped, job.dek_wrap_nonce, job.content_nonce)):
|
|
raise conflict("EXPORT_UNAVAILABLE", "Export key material is unavailable")
|
|
assert job.object_key is not None
|
|
assert job.dek_wrapped is not None
|
|
assert job.dek_wrap_nonce is not None
|
|
assert job.content_nonce is not None
|
|
raw = decrypt_file(
|
|
ciphertext=get_blob_store().get(job.object_key),
|
|
content_nonce=job.content_nonce,
|
|
dek_wrapped=job.dek_wrapped,
|
|
wrap_nonce=job.dek_wrap_nonce,
|
|
)
|
|
job.consumed_at = now
|
|
job.status = "consumed"
|
|
job.dek_wrapped = None
|
|
job.dek_wrap_nonce = None
|
|
job.content_nonce = None
|
|
return raw
|
|
|
|
|
|
def schedule_deletion(db: Session, *, space: FinancialSpace) -> datetime:
|
|
if space.status == "blocked" and space.deletion_scheduled_at:
|
|
return space.deletion_scheduled_at
|
|
when = datetime.now(UTC) + timedelta(days=get_settings().deletion_window_days)
|
|
space.status = "blocked"
|
|
space.deletion_scheduled_at = when
|
|
marker = db.query(DeletionLedger).filter_by(space_id=space.id).one_or_none()
|
|
if marker is None:
|
|
db.add(DeletionLedger(space_id=space.id, scheduled_at=when))
|
|
else:
|
|
marker.scheduled_at = when
|
|
marker.completed_at = None
|
|
return when
|
|
|
|
|
|
def cancel_deletion(db: Session, *, space: FinancialSpace) -> None:
|
|
if space.status != "blocked" or not space.deletion_scheduled_at:
|
|
raise conflict("SPACE_DELETION_NOT_SCHEDULED", "Space deletion is not scheduled")
|
|
space.status = "active"
|
|
space.deletion_scheduled_at = None
|
|
marker = db.query(DeletionLedger).filter_by(space_id=space.id).one_or_none()
|
|
if marker:
|
|
db.delete(marker)
|
|
|
|
|
|
def purge_due_spaces(db: Session) -> int:
|
|
now = datetime.now(UTC)
|
|
spaces = (
|
|
db.query(FinancialSpace)
|
|
.filter(
|
|
FinancialSpace.status == "blocked",
|
|
FinancialSpace.deletion_scheduled_at.is_not(None),
|
|
FinancialSpace.deletion_scheduled_at <= now,
|
|
)
|
|
.all()
|
|
)
|
|
store = get_blob_store()
|
|
purged = 0
|
|
for space in spaces:
|
|
from app.db.session import set_space_context
|
|
|
|
set_space_context(db, space.id)
|
|
for document in db.query(StatementDocument).filter_by(space_id=space.id).all():
|
|
store.delete(document.object_key)
|
|
document.dek_wrapped = None
|
|
for export in db.query(DataExport).filter_by(space_id=space.id).all():
|
|
if export.object_key:
|
|
store.delete(export.object_key)
|
|
export.dek_wrapped = None
|
|
marker = db.query(DeletionLedger).filter_by(space_id=space.id).one()
|
|
marker.completed_at = now
|
|
db.delete(space)
|
|
purged += 1
|
|
return purged
|