582 lines
20 KiB
Python
582 lines
20 KiB
Python
from __future__ import annotations
|
|
|
|
import imaplib
|
|
import os
|
|
import threading
|
|
import uuid
|
|
from datetime import date, timedelta
|
|
from pathlib import Path
|
|
from typing import Any, Literal
|
|
|
|
import uvicorn
|
|
from fastapi import APIRouter, FastAPI, HTTPException
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fints.client import FinTS3PinTanClient, NeedTANResponse
|
|
from pydantic import BaseModel, Field
|
|
|
|
from .fints_connector import (
|
|
PRODUCT_ID_HELP,
|
|
FinTSConfig,
|
|
FinTSConfigurationError,
|
|
account_to_dict,
|
|
balance_to_dict,
|
|
create_client,
|
|
persist_client,
|
|
select_account,
|
|
tan_challenge_to_dict,
|
|
transaction_to_observation,
|
|
)
|
|
from .imap_importer import DEFAULT_SENDER_FILTERS, IMAPConfig, list_imap_folders, scan_imap_mailbox
|
|
from .paypal_connector import PayPalAPIError, PayPalConfig, fetch_paypal_transactions, inspect_paypal_credentials
|
|
from .statement_importer import parse_klarna_statement, parse_paypal_statement
|
|
|
|
LOCAL_AGENT_ORIGINS = [
|
|
"http://127.0.0.1:5173",
|
|
"http://localhost:5173",
|
|
]
|
|
|
|
|
|
class FinTSRequest(BaseModel):
|
|
blz: str
|
|
server: str
|
|
bank_user_id: str
|
|
pin: str
|
|
customer_id: str | None = None
|
|
product_id: str | None = None
|
|
product_version: str | None = None
|
|
tan_mechanism: str | None = None
|
|
tan_medium: str | None = None
|
|
profile: str = "browser"
|
|
state_dir: str = "~/.config/mousehold/fints"
|
|
store_state: bool = True
|
|
|
|
|
|
class FinTSAccountsRequest(BaseModel):
|
|
credentials: FinTSRequest
|
|
with_balances: bool = False
|
|
|
|
|
|
class FinTSTransactionsRequest(BaseModel):
|
|
credentials: FinTSRequest
|
|
iban: str | None = None
|
|
days: int = Field(default=30, ge=1, le=730)
|
|
start_date: date | None = None
|
|
end_date: date | None = None
|
|
include_pending: bool = False
|
|
household_id: str | None = None
|
|
source_account_id: str | None = None
|
|
owner_user_id: str | None = None
|
|
|
|
|
|
class FinTSChoiceRequest(BaseModel):
|
|
value: str
|
|
|
|
|
|
class FinTSTanRequest(BaseModel):
|
|
tan: str = ""
|
|
|
|
|
|
class IMAPCredentials(BaseModel):
|
|
host: str
|
|
port: int = Field(default=993, ge=1, le=65535)
|
|
username: str
|
|
password: str
|
|
use_ssl: bool = True
|
|
|
|
|
|
class IMAPFoldersRequest(BaseModel):
|
|
credentials: IMAPCredentials
|
|
|
|
|
|
class IMAPScanRequest(BaseModel):
|
|
credentials: IMAPCredentials
|
|
folder: str = "INBOX"
|
|
search_query: str = "ALL"
|
|
sender_filters: list[str] = Field(default_factory=lambda: list(DEFAULT_SENDER_FILTERS))
|
|
limit: int = Field(default=25, ge=1, le=500)
|
|
household_id: str
|
|
source_account_id: str | None = None
|
|
owner_user_id: str | None = None
|
|
|
|
|
|
class PayPalCredentials(BaseModel):
|
|
client_id: str
|
|
client_secret: str
|
|
environment: Literal["live", "sandbox"] = "live"
|
|
base_url: str | None = None
|
|
|
|
|
|
class PayPalTransactionsRequest(BaseModel):
|
|
credentials: PayPalCredentials
|
|
days: int = Field(default=30, ge=1, le=31)
|
|
start_date: date | None = None
|
|
end_date: date | None = None
|
|
transaction_currency: str | None = None
|
|
page_size: int = Field(default=100, ge=1, le=500)
|
|
max_pages: int = Field(default=10, ge=1, le=100)
|
|
household_id: str
|
|
source_account_id: str | None = None
|
|
owner_user_id: str | None = None
|
|
|
|
|
|
class PayPalAuthCheckRequest(BaseModel):
|
|
credentials: PayPalCredentials
|
|
|
|
|
|
class StatementParseRequest(BaseModel):
|
|
csv_text: str
|
|
filename: str | None = None
|
|
household_id: str
|
|
source_account_id: str | None = None
|
|
owner_user_id: str | None = None
|
|
|
|
|
|
JobStatus = Literal["running", "needs_tan", "needs_choice", "done", "error"]
|
|
|
|
|
|
class FinTSBrowserJob:
|
|
def __init__(self, kind: str) -> None:
|
|
self.id = str(uuid.uuid4())
|
|
self.kind = kind
|
|
self.status: JobStatus = "running"
|
|
self.error: str | None = None
|
|
self.result: dict[str, Any] | None = None
|
|
self.challenge: dict[str, Any] | None = None
|
|
self.choice_kind: str | None = None
|
|
self.choice_options: list[dict[str, Any]] = []
|
|
self._lock = threading.Lock()
|
|
self._event = threading.Event()
|
|
self._input_value: str | None = None
|
|
|
|
def snapshot(self) -> dict[str, Any]:
|
|
with self._lock:
|
|
return {
|
|
"id": self.id,
|
|
"kind": self.kind,
|
|
"status": self.status,
|
|
"error": self.error,
|
|
"result": self.result,
|
|
"challenge": self.challenge,
|
|
"choice_kind": self.choice_kind,
|
|
"choice_options": self.choice_options,
|
|
}
|
|
|
|
def set_done(self, result: dict[str, Any]) -> None:
|
|
with self._lock:
|
|
self.status = "done"
|
|
self.result = result
|
|
self.challenge = None
|
|
self.choice_kind = None
|
|
self.choice_options = []
|
|
self._input_value = None
|
|
self._event.clear()
|
|
|
|
def set_error(self, error: BaseException | str) -> None:
|
|
with self._lock:
|
|
self.status = "error"
|
|
self.error = str(error)
|
|
self.challenge = None
|
|
self.choice_kind = None
|
|
self.choice_options = []
|
|
self._input_value = None
|
|
self._event.clear()
|
|
|
|
def wait_for_tan(self, response: NeedTANResponse, timeout_seconds: int = 600) -> str:
|
|
with self._lock:
|
|
self.status = "needs_tan"
|
|
self.challenge = tan_challenge_to_dict(response)
|
|
self.choice_kind = None
|
|
self.choice_options = []
|
|
self._input_value = None
|
|
self._event.clear()
|
|
if not self._event.wait(timeout_seconds):
|
|
raise TimeoutError("Timed out waiting for TAN input.")
|
|
with self._lock:
|
|
value = self._input_value or ""
|
|
self.status = "running"
|
|
self.challenge = None
|
|
self._input_value = None
|
|
self._event.clear()
|
|
return value
|
|
|
|
def wait_for_choice(
|
|
self,
|
|
kind: str,
|
|
options: list[dict[str, Any]],
|
|
timeout_seconds: int = 600,
|
|
) -> str:
|
|
with self._lock:
|
|
self.status = "needs_choice"
|
|
self.choice_kind = kind
|
|
self.choice_options = options
|
|
self.challenge = None
|
|
self._input_value = None
|
|
self._event.clear()
|
|
if not self._event.wait(timeout_seconds):
|
|
raise TimeoutError(f"Timed out waiting for {kind} selection.")
|
|
with self._lock:
|
|
value = self._input_value
|
|
self.status = "running"
|
|
self.choice_kind = None
|
|
self.choice_options = []
|
|
self._input_value = None
|
|
self._event.clear()
|
|
if value is None:
|
|
raise RuntimeError(f"No {kind} selection received.")
|
|
return value
|
|
|
|
def provide_value(self, value: str) -> None:
|
|
with self._lock:
|
|
if self.status not in {"needs_tan", "needs_choice"}:
|
|
raise RuntimeError("Job is not waiting for input.")
|
|
self._input_value = value
|
|
self._event.set()
|
|
|
|
|
|
jobs: dict[str, FinTSBrowserJob] = {}
|
|
|
|
|
|
def create_router() -> APIRouter:
|
|
router = APIRouter()
|
|
|
|
@router.get("/health")
|
|
def health() -> dict[str, str]:
|
|
return {"status": "ok"}
|
|
|
|
@router.post("/fints/accounts/start")
|
|
def start_accounts(payload: FinTSAccountsRequest) -> dict[str, Any]:
|
|
job = _start_job("accounts", lambda current_job: _run_accounts(current_job, payload))
|
|
return job.snapshot()
|
|
|
|
@router.post("/fints/transactions/start")
|
|
def start_transactions(payload: FinTSTransactionsRequest) -> dict[str, Any]:
|
|
job = _start_job("transactions", lambda current_job: _run_transactions(current_job, payload))
|
|
return job.snapshot()
|
|
|
|
@router.get("/fints/jobs/{job_id}")
|
|
def get_job(job_id: str) -> dict[str, Any]:
|
|
return _get_job(job_id).snapshot()
|
|
|
|
@router.post("/fints/jobs/{job_id}/tan")
|
|
def provide_tan(job_id: str, payload: FinTSTanRequest) -> dict[str, Any]:
|
|
job = _get_job(job_id)
|
|
if job.snapshot()["status"] != "needs_tan":
|
|
raise HTTPException(status_code=409, detail="Job is not waiting for a TAN.")
|
|
try:
|
|
job.provide_value(payload.tan)
|
|
except RuntimeError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
return job.snapshot()
|
|
|
|
@router.post("/fints/jobs/{job_id}/choice")
|
|
def provide_choice(job_id: str, payload: FinTSChoiceRequest) -> dict[str, Any]:
|
|
job = _get_job(job_id)
|
|
if job.snapshot()["status"] != "needs_choice":
|
|
raise HTTPException(status_code=409, detail="Job is not waiting for a choice.")
|
|
try:
|
|
job.provide_value(payload.value)
|
|
except RuntimeError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
|
return job.snapshot()
|
|
|
|
@router.post("/imap/folders")
|
|
def imap_folders(payload: IMAPFoldersRequest) -> dict[str, Any]:
|
|
try:
|
|
return {"folders": list_imap_folders(_imap_config_from_payload(payload.credentials))}
|
|
except (OSError, RuntimeError, imaplib.IMAP4.error) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
@router.post("/imap/scan")
|
|
def imap_scan(payload: IMAPScanRequest) -> dict[str, Any]:
|
|
try:
|
|
return scan_imap_mailbox(
|
|
_imap_config_from_payload(payload.credentials),
|
|
household_id=payload.household_id,
|
|
owner_user_id=payload.owner_user_id,
|
|
source_account_id=payload.source_account_id,
|
|
limit=payload.limit,
|
|
folder=payload.folder,
|
|
search_query=payload.search_query,
|
|
sender_filters=payload.sender_filters,
|
|
)
|
|
except (OSError, RuntimeError, imaplib.IMAP4.error) as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
@router.post("/paypal/transactions")
|
|
def paypal_transactions(payload: PayPalTransactionsRequest) -> dict[str, Any]:
|
|
end_date = payload.end_date or date.today()
|
|
start_date = payload.start_date or (end_date - timedelta(days=payload.days - 1))
|
|
try:
|
|
result = fetch_paypal_transactions(
|
|
_paypal_config_from_payload(payload.credentials),
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
household_id=payload.household_id,
|
|
owner_user_id=payload.owner_user_id,
|
|
source_account_id=payload.source_account_id,
|
|
transaction_currency=payload.transaction_currency,
|
|
page_size=payload.page_size,
|
|
max_pages=payload.max_pages,
|
|
)
|
|
except PayPalAPIError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
return {
|
|
"start_date": result.start_date.isoformat(),
|
|
"end_date": result.end_date.isoformat(),
|
|
"transactions": result.transactions,
|
|
"observations": result.observations,
|
|
"count": len(result.observations),
|
|
"raw_response_count": result.raw_response_count,
|
|
}
|
|
|
|
@router.post("/paypal/auth-check")
|
|
def paypal_auth_check(payload: PayPalAuthCheckRequest) -> dict[str, Any]:
|
|
try:
|
|
return inspect_paypal_credentials(_paypal_config_from_payload(payload.credentials))
|
|
except PayPalAPIError as exc:
|
|
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
|
|
|
@router.post("/paypal/statement/parse")
|
|
def paypal_statement_parse(payload: StatementParseRequest) -> dict[str, Any]:
|
|
return parse_paypal_statement(
|
|
payload.csv_text,
|
|
household_id=payload.household_id,
|
|
owner_user_id=payload.owner_user_id,
|
|
source_account_id=payload.source_account_id,
|
|
filename=payload.filename,
|
|
)
|
|
|
|
@router.post("/klarna/statement/parse")
|
|
def klarna_statement_parse(payload: StatementParseRequest) -> dict[str, Any]:
|
|
return parse_klarna_statement(
|
|
payload.csv_text,
|
|
household_id=payload.household_id,
|
|
owner_user_id=payload.owner_user_id,
|
|
source_account_id=payload.source_account_id,
|
|
filename=payload.filename,
|
|
)
|
|
|
|
return router
|
|
|
|
|
|
def create_app() -> FastAPI:
|
|
app = FastAPI(title="Mousehold Local Agent", version="0.1.0")
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=_cors_origins(),
|
|
allow_credentials=False,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
app.include_router(create_router())
|
|
return app
|
|
|
|
|
|
def run(host: str = "127.0.0.1", port: int = 8765) -> None:
|
|
uvicorn.run("mousehold_agent.server:app", host=host, port=port, reload=False)
|
|
|
|
|
|
def _start_job(kind: str, target: Any) -> FinTSBrowserJob:
|
|
job = FinTSBrowserJob(kind)
|
|
jobs[job.id] = job
|
|
|
|
def wrapped() -> None:
|
|
try:
|
|
target(job)
|
|
except BaseException as exc:
|
|
job.set_error(exc)
|
|
|
|
thread = threading.Thread(target=wrapped, name=f"fints-{kind}-{job.id}", daemon=True)
|
|
thread.start()
|
|
return job
|
|
|
|
|
|
def _get_job(job_id: str) -> FinTSBrowserJob:
|
|
job = jobs.get(job_id)
|
|
if not job:
|
|
raise HTTPException(status_code=404, detail="FinTS job not found")
|
|
return job
|
|
|
|
|
|
def _run_accounts(job: FinTSBrowserJob, payload: FinTSAccountsRequest) -> None:
|
|
config = _config_from_payload(payload.credentials)
|
|
|
|
def operation(client: FinTS3PinTanClient) -> dict[str, Any]:
|
|
accounts: list[dict[str, Any]] = []
|
|
for index, account in enumerate(client.get_sepa_accounts()):
|
|
row = account_to_dict(account)
|
|
row["index"] = index
|
|
if payload.with_balances:
|
|
balance = _resolve_tan_response(client, client.get_balance(account), job)
|
|
row["balance"] = balance_to_dict(balance)
|
|
accounts.append(row)
|
|
return {"accounts": accounts}
|
|
|
|
job.set_done(_with_browser_dialog(config, job, operation))
|
|
|
|
|
|
def _run_transactions(job: FinTSBrowserJob, payload: FinTSTransactionsRequest) -> None:
|
|
config = _config_from_payload(payload.credentials)
|
|
end_date = payload.end_date or date.today()
|
|
start_date = payload.start_date or (end_date - timedelta(days=payload.days))
|
|
|
|
def operation(client: FinTS3PinTanClient) -> dict[str, Any]:
|
|
account = select_account(client.get_sepa_accounts(), payload.iban)
|
|
transactions = _resolve_tan_response(
|
|
client,
|
|
client.get_transactions(
|
|
account,
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
include_pending=payload.include_pending,
|
|
),
|
|
job,
|
|
)
|
|
observations = [
|
|
transaction_to_observation(
|
|
transaction,
|
|
account=account,
|
|
household_id=payload.household_id,
|
|
source_account_id=payload.source_account_id,
|
|
owner_user_id=payload.owner_user_id,
|
|
)
|
|
for transaction in transactions
|
|
]
|
|
return {
|
|
"account": account_to_dict(account),
|
|
"observations": observations,
|
|
"count": len(observations),
|
|
}
|
|
|
|
job.set_done(_with_browser_dialog(config, job, operation))
|
|
|
|
|
|
def _with_browser_dialog(config: FinTSConfig, job: FinTSBrowserJob, operation: Any) -> dict[str, Any]:
|
|
client = create_client(config)
|
|
_browser_bootstrap(client, config, job)
|
|
try:
|
|
with client:
|
|
while isinstance(client.init_tan_response, NeedTANResponse):
|
|
client.init_tan_response = _answer_tan(client, client.init_tan_response, job)
|
|
return operation(client)
|
|
finally:
|
|
persist_client(client, config)
|
|
|
|
|
|
def _browser_bootstrap(client: FinTS3PinTanClient, config: FinTSConfig, job: FinTSBrowserJob) -> None:
|
|
if not client.get_current_tan_mechanism():
|
|
client.fetch_tan_mechanisms()
|
|
mechanisms = client.get_tan_mechanisms()
|
|
if config.tan_mechanism:
|
|
client.set_tan_mechanism(config.tan_mechanism)
|
|
elif len(mechanisms) == 1:
|
|
client.set_tan_mechanism(next(iter(mechanisms)))
|
|
elif len(mechanisms) > 1:
|
|
options = [
|
|
{
|
|
"value": security_function,
|
|
"label": getattr(parameters, "name", security_function) or security_function,
|
|
"description": f"Function {security_function}",
|
|
}
|
|
for security_function, parameters in mechanisms.items()
|
|
]
|
|
client.set_tan_mechanism(job.wait_for_choice("tan_mechanism", options))
|
|
|
|
if client.selected_tan_medium is None and client.is_tan_media_required():
|
|
usage, media = client.get_tan_media()
|
|
del usage
|
|
if config.tan_medium:
|
|
client.selected_tan_medium = config.tan_medium
|
|
elif len(media) == 1:
|
|
client.set_tan_medium(media[0])
|
|
elif len(media) == 0:
|
|
client.selected_tan_medium = ""
|
|
else:
|
|
options = [
|
|
{
|
|
"value": str(index),
|
|
"label": getattr(item, "tan_medium_name", None) or f"Medium {index + 1}",
|
|
"description": _tan_medium_description(item),
|
|
}
|
|
for index, item in enumerate(media)
|
|
]
|
|
client.set_tan_medium(media[int(job.wait_for_choice("tan_medium", options))])
|
|
|
|
|
|
def _resolve_tan_response(client: FinTS3PinTanClient, response: Any, job: FinTSBrowserJob) -> Any:
|
|
while isinstance(response, NeedTANResponse):
|
|
response = _answer_tan(client, response, job)
|
|
return response
|
|
|
|
|
|
def _answer_tan(
|
|
client: FinTS3PinTanClient,
|
|
response: NeedTANResponse,
|
|
job: FinTSBrowserJob,
|
|
) -> Any:
|
|
if response.decoupled:
|
|
tan = job.wait_for_tan(response)
|
|
del tan
|
|
return client.send_tan(response, "")
|
|
return client.send_tan(response, job.wait_for_tan(response))
|
|
|
|
|
|
def _tan_medium_description(item: Any) -> str:
|
|
parts = []
|
|
mobile = getattr(item, "mobile_number_masked", None)
|
|
if mobile:
|
|
parts.append(str(mobile))
|
|
last_use = getattr(item, "last_use", None)
|
|
if last_use:
|
|
parts.append(f"Last used {last_use}")
|
|
return " · ".join(parts)
|
|
|
|
|
|
def _config_from_payload(payload: FinTSRequest) -> FinTSConfig:
|
|
if not payload.product_id:
|
|
raise FinTSConfigurationError(PRODUCT_ID_HELP)
|
|
return FinTSConfig(
|
|
bank_identifier=payload.blz,
|
|
user_id=payload.bank_user_id,
|
|
pin=payload.pin,
|
|
server=payload.server,
|
|
customer_id=payload.customer_id or None,
|
|
product_id=payload.product_id,
|
|
product_version=payload.product_version or None,
|
|
tan_mechanism=payload.tan_mechanism or None,
|
|
tan_medium=payload.tan_medium or None,
|
|
profile=payload.profile,
|
|
state_dir=Path(payload.state_dir).expanduser(),
|
|
store_state=payload.store_state,
|
|
interactive_bootstrap=False,
|
|
)
|
|
|
|
|
|
def _imap_config_from_payload(payload: IMAPCredentials) -> IMAPConfig:
|
|
return IMAPConfig(
|
|
host=payload.host,
|
|
port=payload.port,
|
|
username=payload.username,
|
|
password=payload.password,
|
|
use_ssl=payload.use_ssl,
|
|
)
|
|
|
|
|
|
def _paypal_config_from_payload(payload: PayPalCredentials) -> PayPalConfig:
|
|
return PayPalConfig(
|
|
client_id=payload.client_id,
|
|
client_secret=payload.client_secret,
|
|
environment=payload.environment,
|
|
base_url=payload.base_url or None,
|
|
)
|
|
|
|
|
|
def _cors_origins() -> list[str]:
|
|
configured = os.environ.get("MOUSEHOLD_LOCAL_AGENT_CORS_ORIGINS")
|
|
if not configured:
|
|
return LOCAL_AGENT_ORIGINS
|
|
return [origin.strip() for origin in configured.split(",") if origin.strip()]
|
|
|
|
|
|
app = create_app()
|