from __future__ import annotations import hashlib import json import os import re from pathlib import Path from sqlite3 import Connection from typing import Any from urllib.parse import urlencode from jarvis_finance.services.budget_common import new_id, now, row_to_dict from jarvis_finance.services.budget_csv_imports import IMPORT_PROFILES, import_budget_csv_text, parse_budget_csv_text from jarvis_finance.services.budget_monthly_import import _source_from_profile MAX_UPLOAD_BYTES = 5 * 1024 * 1024 PROFILE_ALIASES = { "auto": None, "": None, "visa": "visa_credit_card", "visa_credit_card": "visa_credit_card", "migros": "migros_receipts", "migros_receipts": "migros_receipts", "raiffeisen": "raiffeisen_bank", "raiffeisen_bank": "raiffeisen_bank", "akb": "akb_bank", "akb_bank": "akb_bank", } def _runtime_dir() -> Path: root = Path(os.environ.get("JARVIS_FINANCE_RUNTIME_DIR") or Path.home() / "jarvis_runtime" / "finance-system") path = root / "import_uploads" path.mkdir(parents=True, exist_ok=True) return path def _profile_name(profile: str | None) -> str | None: key = str(profile or "auto").strip().lower() if key not in PROFILE_ALIASES: return key return PROFILE_ALIASES[key] def _safe_public_file_name(filename: str, profile: str | None, digest: str) -> str: suffix = ".csv" raw_suffix = Path(str(filename or "")).suffix.lower() if raw_suffix == ".csv": suffix = ".csv" source = _source_from_profile(profile, filename) return re.sub(r"[^A-Za-z0-9_.-]+", "_", f"{source.lower()}_upload_{digest[:12]}{suffix}")[:120] def _decode_csv_text(csv_text: str | bytes) -> str: if isinstance(csv_text, str): data = csv_text.encode("utf-8") if len(data) > MAX_UPLOAD_BYTES: raise ValueError("CSV-Datei ist zu gross; Maximum ist 5 MB.") return csv_text if len(csv_text) > MAX_UPLOAD_BYTES: raise ValueError("CSV-Datei ist zu gross; Maximum ist 5 MB.") for enc in ("utf-8-sig", "cp1252", "latin-1"): try: return csv_text.decode(enc) except UnicodeDecodeError: continue return csv_text.decode("utf-8", errors="replace") def _period_from_rows(rows: list[dict[str, Any]]) -> tuple[str | None, str | None]: dates: list[str] = [] for row in rows: for key, value in row.items(): text = str(value or "").strip() if re.match(r"^\d{4}-\d{2}-\d{2}", text): dates.append(text[:10]) else: m = re.match(r"^(\d{1,2})\.(\d{1,2})\.(\d{4})", text) if m: d, mo, y = m.groups() dates.append(f"{y}-{int(mo):02d}-{int(d):02d}") return (min(dates), max(dates)) if dates else (None, None) def _review_url(*, import_session_id: str, source_type: str, year: str) -> str: return "/planning/budget/expenses/review?" + urlencode({ "import_session_id": import_session_id, "source_type": source_type, "year": year, "budget_year": year, "status": "open", }) def _insert_or_update_session(conn: Connection, *, sid: str, source_hash: str, file_name: str, profile: str | None, rows_total: int, totals: dict[str, int], status: str, review_url: str, period: tuple[str | None, str | None]) -> None: ts = now() source = _source_from_profile(profile, file_name) existing = conn.execute("SELECT 1 FROM budget_import_sessions WHERE import_session_id=?", (sid,)).fetchone() values = ( source, source_hash, file_name, None, period[0], period[1], profile, rows_total, int(totals.get("new_candidates") or totals.get("candidate_count") or totals.get("would_create_candidate_count") or 0), int(totals.get("already_processed") or totals.get("already_processed_count") or 0), int(totals.get("possible_duplicates") or totals.get("possible_duplicate_count") or totals.get("duplicate_count") or 0), int(totals.get("ignored_count") or 0), int(totals.get("covered_by_source") or totals.get("covered_by_source_count") or totals.get("covered_by_migros_count") or 0), int(totals.get("error_rows") or totals.get("error_count") or 0), status, review_url, ts, ) if existing: conn.execute( """ UPDATE budget_import_sessions SET source=?, file_id=?, file_name=?, file_modified_time=?, file_period_start=?, file_period_end=?, profile=?, rows_total=?, new_candidates=?, already_known=?, duplicate_count=?, ignored_count=?, covered_by_source_count=?, error_count=?, status=?, review_url=?, updated_at=? WHERE import_session_id=? """, values + (sid,), ) else: conn.execute( """ INSERT INTO budget_import_sessions(import_session_id, source, file_id, file_name, file_modified_time, file_period_start, file_period_end, profile, rows_total, new_candidates, already_known, duplicate_count, ignored_count, covered_by_source_count, error_count, status, review_url, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, (sid,) + values[:-1] + (ts, ts), ) def preview_uploaded_budget_import(conn: Connection, *, filename: str, csv_text: str | bytes, profile: str | None = "auto", default_category_id: str | None = None, income_category_id: str | None = None) -> dict[str, Any]: try: if not str(filename or "").lower().endswith(".csv"): raise ValueError("Nur CSV-Dateien werden unterstützt.") text = _decode_csv_text(csv_text) digest = hashlib.sha256(text.encode("utf-8", errors="replace")).hexdigest() requested_profile = _profile_name(profile) parsed = parse_budget_csv_text(text, requested_profile) detected_profile = str(parsed["profile"]) if detected_profile not in IMPORT_PROFILES: raise ValueError("Unbekanntes Importprofil.") rows = parsed["rows"] public_name = _safe_public_file_name(filename, detected_profile, digest) archive = _runtime_dir() / f"{digest}.csv" if not archive.exists(): archive.write_text(text, encoding="utf-8") preview_id = "bimport_" + digest[:24] source_label = f"upload_{digest[:16]}.csv" dry = import_budget_csv_text(conn, detected_profile, text, source_file_label=source_label, dry_run=True, default_category_id=default_category_id, income_category_id=income_category_id) year = str(((_period_from_rows(rows)[0] or "2026")[:4]) or "2026") review = _review_url(import_session_id=preview_id, source_type=IMPORT_PROFILES[detected_profile].source_type, year=year) totals = { "rows_total": len(rows), "new_candidates": int(dry.get("would_create_candidate_count") or 0), "already_processed": int(dry.get("already_processed_count") or 0), "possible_duplicates": int(dry.get("possible_duplicate_count") or 0), "safe_duplicates": int(dry.get("duplicate_count") or 0), "error_rows": int(dry.get("error_count") or 0), "transfer_candidates": int(dry.get("transfer_candidate_count") or 0), "income_candidates": int(dry.get("income_candidate_count") or 0), "expense_candidates": max(0, int(dry.get("would_create_candidate_count") or 0) - int(dry.get("income_candidate_count") or 0) - int(dry.get("transfer_candidate_count") or 0)), "covered_by_source": int(dry.get("covered_by_source_count") or dry.get("covered_by_migros_count") or 0), } _insert_or_update_session(conn, sid=preview_id, source_hash=digest, file_name=public_name, profile=detected_profile, rows_total=len(rows), totals=totals, status="dry_run", review_url=review, period=_period_from_rows(rows)) conn.commit() return { "purpose": "budget_import_upload_preview_v1", "preview_id": preview_id, "status": "dry_run", "file": {"detected": True, "name": public_name, "source_hash": digest[:16]}, "profile": detected_profile, "source_type": IMPORT_PROFILES[detected_profile].source_type, "period": {"start": _period_from_rows(rows)[0], "end": _period_from_rows(rows)[1]}, "totals": totals, "review_url": review, "warnings": ["Dry Run: Es wurden keine produktiven Buchungen geschrieben."], "errors": [], } except Exception as exc: return { "purpose": "budget_import_upload_preview_v1", "preview_id": "", "status": "failed", "file": {"detected": False, "name": _safe_public_file_name(filename, None, "unknown000000")}, "profile": None, "source_type": None, "period": {"start": None, "end": None}, "totals": {"rows_total": 0, "new_candidates": 0, "already_processed": 0, "possible_duplicates": 0, "safe_duplicates": 0, "error_rows": 1, "transfer_candidates": 0, "income_candidates": 0, "expense_candidates": 0, "covered_by_source": 0}, "review_url": "", "warnings": [], "errors": [f"Import-Profil konnte nicht erkannt werden: {type(exc).__name__}"], } def confirm_uploaded_budget_import_candidates(conn: Connection, preview_id: str, *, default_category_id: str | None = None, income_category_id: str | None = None) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_import_sessions WHERE import_session_id=?", (preview_id,)).fetchone() if not row: raise ValueError("Import-Preview nicht gefunden; bitte Dry Run erneut starten.") digest = str(row["file_id"] or "") archive = _runtime_dir() / f"{digest}.csv" if not archive.exists(): raise ValueError("Temporäre Upload-Datei nicht gefunden; bitte Datei erneut hochladen.") text = archive.read_text(encoding="utf-8", errors="replace") profile = str(row["profile"] or "") source_label = f"upload_{digest[:16]}.csv" before = int(conn.execute("SELECT COUNT(*) FROM budget_transaction_candidates WHERE import_session_id=?", (preview_id,)).fetchone()[0] or 0) result = import_budget_csv_text(conn, profile, text, source_file_label=source_label, dry_run=False, default_category_id=default_category_id, income_category_id=income_category_id) conn.execute("UPDATE budget_transaction_candidates SET import_session_id=? WHERE source_file_label=? AND import_session_id IS NULL", (preview_id, source_label)) after = int(conn.execute("SELECT COUNT(*) FROM budget_transaction_candidates WHERE import_session_id=?", (preview_id,)).fetchone()[0] or 0) new_count = max(0, after - before) already = int(result.get("already_processed_count") or 0) if before and new_count == 0: already = max(already, before) year = str((row["file_period_start"] or "2026")[:4] or "2026") review = _review_url(import_session_id=preview_id, source_type=IMPORT_PROFILES[profile].source_type, year=year) totals = { "rows_total": int(row["rows_total"] or result.get("parsed_rows") or 0), "new_candidates": new_count, "already_processed": already, "possible_duplicates": int(result.get("possible_duplicate_count") or 0), "safe_duplicates": int(result.get("duplicate_count") or 0), "error_rows": int(result.get("error_count") or 0), "transfer_candidates": int(result.get("transfer_candidate_count") or 0), "income_candidates": int(result.get("income_candidate_count") or 0), "expense_candidates": max(0, new_count - int(result.get("income_candidate_count") or 0) - int(result.get("transfer_candidate_count") or 0)), "covered_by_source": int(result.get("covered_by_source_count") or result.get("covered_by_migros_count") or 0), } _insert_or_update_session(conn, sid=preview_id, source_hash=digest, file_name=str(row["file_name"]), profile=profile, rows_total=int(row["rows_total"] or 0), totals=totals, status="candidates_created", review_url=review, period=(row["file_period_start"], row["file_period_end"])) conn.commit() return {"purpose": "budget_import_upload_confirm_candidates_v1", "status": "candidates_created", "import_session_id": preview_id, "profile": profile, "source_type": IMPORT_PROFILES[profile].source_type, "totals": totals, "review_url": review, "warnings": ["Kandidaten erzeugt; Confirm bleibt eine separate User-Aktion."]} def list_budget_import_sessions(conn: Connection, *, limit: int = 30) -> list[dict[str, Any]]: rows = conn.execute("SELECT * FROM budget_import_sessions ORDER BY created_at DESC, import_session_id DESC LIMIT ?", (limit,)).fetchall() return [row_to_dict(r) for r in rows] def get_budget_import_session(conn: Connection, import_session_id: str) -> dict[str, Any]: row = conn.execute("SELECT * FROM budget_import_sessions WHERE import_session_id=?", (import_session_id,)).fetchone() if not row: raise ValueError("Import-Session nicht gefunden") data = row_to_dict(row) candidates = conn.execute("SELECT status, classification, COUNT(*) AS count FROM budget_transaction_candidates WHERE import_session_id=? GROUP BY status, classification", (import_session_id,)).fetchall() data["candidate_groups"] = [row_to_dict(r) for r in candidates] return data