645 lines
24 KiB
Python
645 lines
24 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import UTC, date, datetime
|
|
from decimal import Decimal
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
from mousehold_domain import (
|
|
EventObservationRelation,
|
|
FinancialEventStatus,
|
|
FinancialEventType,
|
|
ImportConnectionStatus,
|
|
ObservationStatus,
|
|
)
|
|
from mousehold_matching import compute_normalized_transaction_fingerprint
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from .. import models
|
|
from ..auth import get_optional_local_session, require_household_access
|
|
from ..db import get_session
|
|
from ..schemas import (
|
|
EventRead,
|
|
ImportRunDetailRead,
|
|
ImportRunRead,
|
|
ManualObservationCreate,
|
|
ObservationAcceptRequest,
|
|
ObservationBatchRequest,
|
|
ObservationBatchResponse,
|
|
ObservationIgnoreRequest,
|
|
ObservationImportCreate,
|
|
ObservationLinkRequest,
|
|
ObservationReviewItem,
|
|
ObservationUpdateRequest,
|
|
RawObservationRead,
|
|
SyncObservationsRequest,
|
|
SyncObservationsResponse,
|
|
)
|
|
from ..services.ingestion import create_observation_with_candidate
|
|
from .events import _attach_event_tags
|
|
|
|
router = APIRouter(tags=["observations"])
|
|
|
|
|
|
def _get_observation(session: Session, observation_id: str) -> models.RawObservation:
|
|
observation = session.get(models.RawObservation, observation_id)
|
|
if not observation:
|
|
raise HTTPException(status_code=404, detail="Observation not found")
|
|
return observation
|
|
|
|
|
|
def _events_for_observation(
|
|
session: Session, observation_id: str
|
|
) -> list[models.FinancialEvent]:
|
|
events = list(
|
|
session.scalars(
|
|
select(models.FinancialEvent)
|
|
.join(
|
|
models.EventObservationLink,
|
|
models.EventObservationLink.event_id == models.FinancialEvent.id,
|
|
)
|
|
.where(models.EventObservationLink.observation_id == observation_id)
|
|
.options(selectinload(models.FinancialEvent.split_lines))
|
|
.order_by(models.FinancialEvent.created_at.desc())
|
|
).all()
|
|
)
|
|
_attach_event_tags(session, events)
|
|
return events
|
|
|
|
|
|
def _review_item(session: Session, observation: models.RawObservation) -> ObservationReviewItem:
|
|
events = _events_for_observation(session, observation.id)
|
|
if observation.status == ObservationStatus.DUPLICATE_SOURCE:
|
|
suggested_action = "duplicate"
|
|
elif observation.status == ObservationStatus.IGNORED:
|
|
suggested_action = "ignored"
|
|
elif events:
|
|
suggested_action = "review_event"
|
|
else:
|
|
suggested_action = "create_event"
|
|
action_required = observation.status not in {
|
|
ObservationStatus.IGNORED,
|
|
ObservationStatus.DUPLICATE_SOURCE,
|
|
}
|
|
return ObservationReviewItem(
|
|
observation=RawObservationRead.model_validate(observation),
|
|
events=[EventRead.model_validate(event) for event in events],
|
|
action_required=action_required,
|
|
suggested_action=suggested_action,
|
|
)
|
|
|
|
|
|
def _date_from_parsed(value: object, field: str) -> date | None:
|
|
if not value:
|
|
return None
|
|
try:
|
|
return date.fromisoformat(str(value))
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=400, detail=f"Observation has invalid {field}") from exc
|
|
|
|
|
|
def _create_event_from_observation(
|
|
session: Session,
|
|
observation: models.RawObservation,
|
|
event_status: FinancialEventStatus | None,
|
|
) -> models.FinancialEvent:
|
|
parsed = observation.parsed_json or {}
|
|
event_date = _date_from_parsed(parsed.get("event_date"), "event_date")
|
|
if not event_date:
|
|
raise HTTPException(status_code=400, detail="Observation does not contain an event date")
|
|
amount = parsed.get("amount")
|
|
if amount is None:
|
|
raise HTTPException(status_code=400, detail="Observation does not contain an amount")
|
|
event_type = FinancialEventType(str(parsed.get("event_type") or FinancialEventType.EXPENSE))
|
|
status = event_status or FinancialEventStatus(str(parsed.get("status") or FinancialEventStatus.CANDIDATE))
|
|
fingerprint = compute_normalized_transaction_fingerprint(
|
|
amount=Decimal(str(amount)),
|
|
currency=str(parsed.get("currency") or "EUR"),
|
|
event_date=event_date,
|
|
merchant=parsed.get("merchant"),
|
|
counterparty=parsed.get("counterparty"),
|
|
external_id=observation.provider_transaction_id,
|
|
)
|
|
event = models.FinancialEvent(
|
|
household_id=observation.household_id,
|
|
source_account_id=observation.source_account_id,
|
|
payment_account_id=observation.source_account_id,
|
|
payer_user_id=observation.owner_user_id,
|
|
event_type=event_type,
|
|
status=status,
|
|
event_date=event_date,
|
|
booking_date=_date_from_parsed(parsed.get("booking_date"), "booking_date"),
|
|
due_date=_date_from_parsed(parsed.get("due_date"), "due_date"),
|
|
amount=Decimal(str(amount)),
|
|
currency=str(parsed.get("currency") or "EUR"),
|
|
merchant=parsed.get("merchant"),
|
|
counterparty=parsed.get("counterparty"),
|
|
description=parsed.get("description"),
|
|
created_by_user_id=observation.owner_user_id,
|
|
fingerprint=fingerprint,
|
|
)
|
|
session.add(event)
|
|
session.flush()
|
|
session.add(
|
|
models.EventObservationLink(
|
|
event_id=event.id,
|
|
observation_id=observation.id,
|
|
relation=EventObservationRelation.IMPORTED_FROM,
|
|
confidence=Decimal("1.00"),
|
|
)
|
|
)
|
|
session.flush()
|
|
return event
|
|
|
|
|
|
def _require_account(session: Session, household_id: str, account_id: str | None) -> None:
|
|
if not account_id:
|
|
return
|
|
account = session.get(models.Account, account_id)
|
|
if not account or account.household_id != household_id:
|
|
raise HTTPException(status_code=404, detail="Account not found")
|
|
|
|
|
|
def _require_user(session: Session, household_id: str, user_id: str | None) -> None:
|
|
if not user_id:
|
|
return
|
|
user = session.get(models.User, user_id)
|
|
if not user or user.household_id != household_id:
|
|
raise HTTPException(status_code=404, detail="User not found")
|
|
|
|
|
|
def _parsed_update_payload(payload: ObservationUpdateRequest) -> dict[str, object]:
|
|
parsed_fields = {
|
|
"amount",
|
|
"currency",
|
|
"event_date",
|
|
"booking_date",
|
|
"due_date",
|
|
"merchant",
|
|
"counterparty",
|
|
"description",
|
|
"event_type",
|
|
"status",
|
|
}
|
|
dumped = payload.model_dump(exclude_unset=True, mode="json")
|
|
return {key: value for key, value in dumped.items() if key in parsed_fields}
|
|
|
|
|
|
def _sync_event_from_observation(
|
|
event: models.FinancialEvent,
|
|
observation: models.RawObservation,
|
|
parsed_update: dict[str, object],
|
|
) -> None:
|
|
parsed = observation.parsed_json or {}
|
|
if "amount" in parsed_update:
|
|
event.amount = Decimal(str(parsed["amount"]))
|
|
if "currency" in parsed_update:
|
|
event.currency = str(parsed["currency"] or "EUR").upper()
|
|
if "event_date" in parsed_update:
|
|
parsed_date = _date_from_parsed(parsed.get("event_date"), "event_date")
|
|
if parsed_date:
|
|
event.event_date = parsed_date
|
|
if "booking_date" in parsed_update:
|
|
event.booking_date = _date_from_parsed(parsed.get("booking_date"), "booking_date")
|
|
if "due_date" in parsed_update:
|
|
event.due_date = _date_from_parsed(parsed.get("due_date"), "due_date")
|
|
if "merchant" in parsed_update:
|
|
event.merchant = parsed.get("merchant")
|
|
if "counterparty" in parsed_update:
|
|
event.counterparty = parsed.get("counterparty")
|
|
if "description" in parsed_update:
|
|
event.description = parsed.get("description")
|
|
if "event_type" in parsed_update and parsed.get("event_type"):
|
|
event.event_type = FinancialEventType(str(parsed["event_type"]))
|
|
if "status" in parsed_update and parsed.get("status"):
|
|
event.status = FinancialEventStatus(str(parsed["status"]))
|
|
event.fingerprint = compute_normalized_transaction_fingerprint(
|
|
amount=event.amount,
|
|
currency=event.currency,
|
|
event_date=event.event_date,
|
|
merchant=event.merchant,
|
|
counterparty=event.counterparty,
|
|
external_id=observation.provider_transaction_id,
|
|
)
|
|
|
|
|
|
def _apply_observation_update(
|
|
session: Session,
|
|
observation: models.RawObservation,
|
|
payload: ObservationUpdateRequest,
|
|
) -> None:
|
|
if "source_account_id" in payload.model_fields_set:
|
|
_require_account(session, observation.household_id, payload.source_account_id)
|
|
observation.source_account_id = payload.source_account_id
|
|
if "owner_user_id" in payload.model_fields_set:
|
|
_require_user(session, observation.household_id, payload.owner_user_id)
|
|
observation.owner_user_id = payload.owner_user_id
|
|
if "raw_text" in payload.model_fields_set:
|
|
observation.raw_text = payload.raw_text
|
|
parsed_update = _parsed_update_payload(payload)
|
|
if parsed_update:
|
|
parsed = dict(observation.parsed_json or {})
|
|
parsed.update(parsed_update)
|
|
if "currency" in parsed and parsed["currency"]:
|
|
parsed["currency"] = str(parsed["currency"]).upper()
|
|
observation.parsed_json = parsed
|
|
session.add(observation)
|
|
session.flush()
|
|
if payload.sync_linked_events:
|
|
for event in _events_for_observation(session, observation.id):
|
|
_sync_event_from_observation(event, observation, parsed_update)
|
|
if "source_account_id" in payload.model_fields_set:
|
|
event.source_account_id = observation.source_account_id
|
|
event.payment_account_id = observation.source_account_id
|
|
if "owner_user_id" in payload.model_fields_set:
|
|
event.payer_user_id = observation.owner_user_id
|
|
session.add(event)
|
|
session.flush()
|
|
|
|
|
|
def _accept_observation(
|
|
session: Session,
|
|
observation: models.RawObservation,
|
|
event_status: FinancialEventStatus | None,
|
|
create_event_if_missing: bool = True,
|
|
) -> None:
|
|
events = _events_for_observation(session, observation.id)
|
|
if not events and create_event_if_missing:
|
|
events = [_create_event_from_observation(session, observation, event_status)]
|
|
for event in events:
|
|
if event_status:
|
|
event.status = event_status
|
|
session.add(event)
|
|
if events and observation.status not in {
|
|
ObservationStatus.DUPLICATE_SOURCE,
|
|
ObservationStatus.IGNORED,
|
|
}:
|
|
observation.status = ObservationStatus.CANDIDATE_CREATED
|
|
session.add(observation)
|
|
session.flush()
|
|
|
|
|
|
def _ignore_observation(
|
|
session: Session,
|
|
observation: models.RawObservation,
|
|
ignore_linked_events: bool,
|
|
actor_user_id: str | None,
|
|
reason: str | None,
|
|
) -> None:
|
|
observation.status = ObservationStatus.IGNORED
|
|
session.add(observation)
|
|
if ignore_linked_events:
|
|
for event in _events_for_observation(session, observation.id):
|
|
event.status = FinancialEventStatus.IGNORED
|
|
session.add(event)
|
|
session.add(
|
|
models.AuditLog(
|
|
household_id=observation.household_id,
|
|
actor_user_id=actor_user_id,
|
|
action="observation.ignore",
|
|
entity_type="raw_observation",
|
|
entity_id=observation.id,
|
|
details_json={"reason": reason},
|
|
)
|
|
)
|
|
session.flush()
|
|
|
|
|
|
@router.post("/observations/manual", response_model=RawObservationRead)
|
|
def create_manual_observation(
|
|
payload: ManualObservationCreate,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> models.RawObservation:
|
|
require_household_access(payload.household_id, local_session)
|
|
observation, _event = create_observation_with_candidate(session, payload)
|
|
session.commit()
|
|
session.refresh(observation)
|
|
return observation
|
|
|
|
|
|
@router.post("/observations/import", response_model=RawObservationRead)
|
|
def import_observation(
|
|
payload: ObservationImportCreate,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> models.RawObservation:
|
|
require_household_access(payload.household_id, local_session)
|
|
observation, _event = create_observation_with_candidate(session, payload)
|
|
session.commit()
|
|
session.refresh(observation)
|
|
return observation
|
|
|
|
|
|
@router.post("/observations/sync", response_model=SyncObservationsResponse)
|
|
def sync_observations(
|
|
payload: SyncObservationsRequest,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> SyncObservationsResponse:
|
|
imported_ids: list[str] = []
|
|
event_ids: list[str] = []
|
|
duplicate_ids: list[str] = []
|
|
connection = session.get(models.ImportConnection, payload.connection_id)
|
|
household_id = (
|
|
payload.observations[0].household_id
|
|
if payload.observations
|
|
else connection.household_id if connection else None
|
|
)
|
|
if not household_id:
|
|
raise HTTPException(status_code=400, detail="No household available for import")
|
|
require_household_access(household_id, local_session)
|
|
run = models.ImportRun(
|
|
household_id=household_id,
|
|
connection_id=connection.id if connection else None,
|
|
actor_user_id=local_session.user_id if local_session else None,
|
|
source_type=str(payload.observations[0].source_type) if payload.observations else None,
|
|
status="running",
|
|
summary_json={"requested_count": len(payload.observations)},
|
|
)
|
|
session.add(run)
|
|
session.flush()
|
|
for observation_payload in payload.observations:
|
|
if observation_payload.household_id != household_id:
|
|
raise HTTPException(status_code=400, detail="Import payload mixes households")
|
|
observation, event = create_observation_with_candidate(session, observation_payload)
|
|
imported_ids.append(observation.id)
|
|
if observation.status == ObservationStatus.DUPLICATE_SOURCE:
|
|
duplicate_ids.append(observation.id)
|
|
if event:
|
|
event_ids.append(event.id)
|
|
if connection:
|
|
connection.status = ImportConnectionStatus.ACTIVE
|
|
connection.last_sync_at = datetime.now(UTC)
|
|
connection.last_error = None
|
|
config = dict(connection.connector_config_json or {})
|
|
config["last_imported_count"] = len(imported_ids) - len(duplicate_ids)
|
|
config["last_duplicate_count"] = len(duplicate_ids)
|
|
config["last_created_event_count"] = len(event_ids)
|
|
config["last_import_run_id"] = run.id
|
|
connection.connector_config_json = config
|
|
session.add(connection)
|
|
run.status = "done"
|
|
run.imported_count = len(imported_ids) - len(duplicate_ids)
|
|
run.duplicate_count = len(duplicate_ids)
|
|
run.created_event_count = len(event_ids)
|
|
run.finished_at = datetime.now(UTC)
|
|
run.summary_json = {
|
|
"requested_count": len(payload.observations),
|
|
"imported_observation_ids": imported_ids,
|
|
"created_event_ids": event_ids,
|
|
"duplicate_observation_ids": duplicate_ids,
|
|
}
|
|
session.add(run)
|
|
session.commit()
|
|
session.refresh(run)
|
|
return SyncObservationsResponse(
|
|
imported_observation_ids=imported_ids,
|
|
created_event_ids=event_ids,
|
|
duplicate_observation_ids=duplicate_ids,
|
|
connection_id=connection.id if connection else None,
|
|
imported_count=len(imported_ids) - len(duplicate_ids),
|
|
duplicate_count=len(duplicate_ids),
|
|
created_event_count=len(event_ids),
|
|
run=ImportRunRead.model_validate(run),
|
|
)
|
|
|
|
|
|
@router.get("/observations/inbox", response_model=list[RawObservationRead])
|
|
def get_observation_inbox(
|
|
household_id: str,
|
|
status: ObservationStatus | None = None,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> list[models.RawObservation]:
|
|
require_household_access(household_id, local_session)
|
|
query = (
|
|
select(models.RawObservation)
|
|
.where(models.RawObservation.household_id == household_id)
|
|
.order_by(models.RawObservation.received_at.desc())
|
|
)
|
|
if status:
|
|
query = query.where(models.RawObservation.status == status)
|
|
return list(session.scalars(query).all())
|
|
|
|
|
|
@router.get("/observations/review", response_model=list[ObservationReviewItem])
|
|
def get_observation_review_items(
|
|
household_id: str,
|
|
status: ObservationStatus | None = None,
|
|
limit: int = 100,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> list[ObservationReviewItem]:
|
|
require_household_access(household_id, local_session)
|
|
query = (
|
|
select(models.RawObservation)
|
|
.where(models.RawObservation.household_id == household_id)
|
|
.order_by(models.RawObservation.received_at.desc())
|
|
.limit(limit)
|
|
)
|
|
if status:
|
|
query = query.where(models.RawObservation.status == status)
|
|
return [_review_item(session, observation) for observation in session.scalars(query).all()]
|
|
|
|
|
|
@router.patch("/observations/{observation_id}", response_model=ObservationReviewItem)
|
|
def update_observation(
|
|
observation_id: str,
|
|
payload: ObservationUpdateRequest,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> ObservationReviewItem:
|
|
observation = _get_observation(session, observation_id)
|
|
require_household_access(observation.household_id, local_session)
|
|
_apply_observation_update(session, observation, payload)
|
|
session.commit()
|
|
session.refresh(observation)
|
|
return _review_item(session, observation)
|
|
|
|
|
|
@router.post("/observations/{observation_id}/accept", response_model=ObservationReviewItem)
|
|
def accept_observation(
|
|
observation_id: str,
|
|
payload: ObservationAcceptRequest,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> ObservationReviewItem:
|
|
observation = _get_observation(session, observation_id)
|
|
require_household_access(observation.household_id, local_session)
|
|
_accept_observation(
|
|
session,
|
|
observation,
|
|
event_status=payload.event_status,
|
|
create_event_if_missing=payload.create_event_if_missing,
|
|
)
|
|
session.commit()
|
|
session.refresh(observation)
|
|
return _review_item(session, observation)
|
|
|
|
|
|
@router.post("/observations/{observation_id}/ignore", response_model=ObservationReviewItem)
|
|
def ignore_observation(
|
|
observation_id: str,
|
|
payload: ObservationIgnoreRequest,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> ObservationReviewItem:
|
|
observation = _get_observation(session, observation_id)
|
|
require_household_access(observation.household_id, local_session)
|
|
_ignore_observation(
|
|
session,
|
|
observation,
|
|
ignore_linked_events=payload.ignore_linked_events,
|
|
actor_user_id=local_session.user_id if local_session else None,
|
|
reason=payload.reason,
|
|
)
|
|
session.commit()
|
|
session.refresh(observation)
|
|
return _review_item(session, observation)
|
|
|
|
|
|
@router.post("/observations/{observation_id}/link", response_model=ObservationReviewItem)
|
|
def link_observation(
|
|
observation_id: str,
|
|
payload: ObservationLinkRequest,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> ObservationReviewItem:
|
|
observation = _get_observation(session, observation_id)
|
|
require_household_access(observation.household_id, local_session)
|
|
event = session.get(models.FinancialEvent, payload.event_id)
|
|
if not event or event.household_id != observation.household_id:
|
|
raise HTTPException(status_code=404, detail="Event not found")
|
|
exists = session.scalars(
|
|
select(models.EventObservationLink)
|
|
.where(models.EventObservationLink.event_id == event.id)
|
|
.where(models.EventObservationLink.observation_id == observation.id)
|
|
.where(models.EventObservationLink.relation == payload.relation)
|
|
.limit(1)
|
|
).first()
|
|
if not exists:
|
|
session.add(
|
|
models.EventObservationLink(
|
|
event_id=event.id,
|
|
observation_id=observation.id,
|
|
relation=payload.relation,
|
|
confidence=Decimal("1.00"),
|
|
)
|
|
)
|
|
if observation.status == ObservationStatus.NEW:
|
|
observation.status = ObservationStatus.PARSED
|
|
session.add(observation)
|
|
session.commit()
|
|
session.refresh(observation)
|
|
return _review_item(session, observation)
|
|
|
|
|
|
@router.post("/observations/batch", response_model=ObservationBatchResponse)
|
|
def batch_review_observations(
|
|
payload: ObservationBatchRequest,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> ObservationBatchResponse:
|
|
if payload.action not in {"accept", "confirm", "ignore"}:
|
|
raise HTTPException(status_code=400, detail="Unsupported batch action")
|
|
if not payload.observation_ids:
|
|
return ObservationBatchResponse(updated_count=0, items=[])
|
|
items: list[ObservationReviewItem] = []
|
|
household_id: str | None = None
|
|
for observation_id in payload.observation_ids:
|
|
observation = _get_observation(session, observation_id)
|
|
if household_id is None:
|
|
household_id = observation.household_id
|
|
require_household_access(household_id, local_session)
|
|
elif observation.household_id != household_id:
|
|
raise HTTPException(status_code=400, detail="Batch payload mixes households")
|
|
if payload.action == "ignore":
|
|
_ignore_observation(
|
|
session,
|
|
observation,
|
|
ignore_linked_events=payload.ignore_linked_events,
|
|
actor_user_id=local_session.user_id if local_session else None,
|
|
reason="Batch review",
|
|
)
|
|
else:
|
|
_accept_observation(
|
|
session,
|
|
observation,
|
|
event_status=FinancialEventStatus.CONFIRMED
|
|
if payload.action == "confirm"
|
|
else payload.event_status,
|
|
create_event_if_missing=True,
|
|
)
|
|
items.append(_review_item(session, observation))
|
|
session.commit()
|
|
return ObservationBatchResponse(updated_count=len(items), items=items)
|
|
|
|
|
|
@router.get("/import-runs", response_model=list[ImportRunRead])
|
|
def list_import_runs(
|
|
household_id: str,
|
|
connection_id: str | None = None,
|
|
limit: int = 25,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> list[models.ImportRun]:
|
|
require_household_access(household_id, local_session)
|
|
query = (
|
|
select(models.ImportRun)
|
|
.where(models.ImportRun.household_id == household_id)
|
|
.order_by(models.ImportRun.started_at.desc())
|
|
.limit(limit)
|
|
)
|
|
if connection_id:
|
|
query = query.where(models.ImportRun.connection_id == connection_id)
|
|
return list(session.scalars(query).all())
|
|
|
|
|
|
@router.get("/import-runs/{run_id}", response_model=ImportRunDetailRead)
|
|
def get_import_run_detail(
|
|
run_id: str,
|
|
session: Session = Depends(get_session),
|
|
local_session: models.LocalSession | None = Depends(get_optional_local_session),
|
|
) -> ImportRunDetailRead:
|
|
run = session.get(models.ImportRun, run_id)
|
|
if not run:
|
|
raise HTTPException(status_code=404, detail="Import run not found")
|
|
require_household_access(run.household_id, local_session)
|
|
summary = run.summary_json or {}
|
|
observation_ids = list(
|
|
dict.fromkeys(
|
|
[
|
|
*summary.get("imported_observation_ids", []),
|
|
*summary.get("duplicate_observation_ids", []),
|
|
]
|
|
)
|
|
)
|
|
event_ids = summary.get("created_event_ids", [])
|
|
observations = (
|
|
list(
|
|
session.scalars(
|
|
select(models.RawObservation).where(models.RawObservation.id.in_(observation_ids))
|
|
).all()
|
|
)
|
|
if observation_ids
|
|
else []
|
|
)
|
|
events = (
|
|
list(
|
|
session.scalars(
|
|
select(models.FinancialEvent)
|
|
.where(models.FinancialEvent.id.in_(event_ids))
|
|
.options(selectinload(models.FinancialEvent.split_lines))
|
|
).all()
|
|
)
|
|
if event_ids
|
|
else []
|
|
)
|
|
_attach_event_tags(session, events)
|
|
return ImportRunDetailRead(
|
|
run=ImportRunRead.model_validate(run),
|
|
observations=[RawObservationRead.model_validate(observation) for observation in observations],
|
|
events=[EventRead.model_validate(event) for event in events],
|
|
)
|