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], )