from __future__ import annotations import hashlib import json import re import uuid from dataclasses import dataclass from datetime import datetime, timezone from pathlib import Path from sqlite3 import Connection from typing import Any from collections import Counter from jarvis_finance.audit.log import record_audit_event SOURCE_PLATFORMS = {"PostFinance", "True Wealth", "Raiffeisen", "Manuell / Sonstige"} ACCOUNT_TYPES = {"brokerage", "cash", "robo_portfolio", "reserve", "other"} ASSET_CLASSES = {"equity", "etf", "cash", "other"} REVIEW_STATUSES = {"needs_manual_review", "missing_isin", "missing_ticker", "ambiguous"} def utc_now() -> str: return datetime.now(timezone.utc).isoformat() def normalize_name(value: str | None) -> str: text = (value or "").strip().lower() text = re.sub(r"[^a-z0-9äöüéèàç]+", " ", text, flags=re.IGNORECASE) return re.sub(r"\s+", " ", text).strip() def _json(data: dict[str, Any] | list[Any] | None) -> str: return json.dumps(data or {}, ensure_ascii=False, sort_keys=True) def _hash_label(value: str | bytes | Path) -> str: if isinstance(value, Path): return hashlib.sha256(value.read_bytes()).hexdigest() if isinstance(value, bytes): return hashlib.sha256(value).hexdigest() return hashlib.sha256(str(value).encode("utf-8")).hexdigest() def _first_matching_instrument(conn: Connection, *, isin: str | None, ticker: str | None, exchange: str | None, currency: str | None) -> str | None: if isin: row = conn.execute("SELECT instrument_id FROM instruments WHERE upper(isin)=upper(?) AND is_active=1 ORDER BY created_at LIMIT 1", (isin,)).fetchone() if row: return row["instrument_id"] if ticker and exchange: row = conn.execute( """ SELECT instrument_id FROM instruments WHERE upper(ticker)=upper(?) AND upper(COALESCE(exchange,''))=upper(?) AND (? IS NULL OR upper(currency)=upper(?)) AND is_active=1 ORDER BY created_at LIMIT 1 """, (ticker, exchange, currency, currency), ).fetchone() if row: return row["instrument_id"] return None def classify_instrument_mapping( conn: Connection, *, asset_class: str, source_label: str, isin: str | None = None, ticker: str | None = None, exchange: str | None = None, currency: str | None = None, ) -> tuple[str, str, str | None, list[str]]: flags: list[str] = [] asset = asset_class if asset_class in ASSET_CLASSES else "other" instrument_id = _first_matching_instrument(conn, isin=isin, ticker=ticker, exchange=exchange, currency=currency) if asset == "cash": return "mapped", "0.90", instrument_id, flags if asset in {"equity", "etf"}: if isin and instrument_id: return "mapped", "0.98", instrument_id, flags if isin and not instrument_id: flags.append("needs_manual_review") return "needs_manual_review", "0.60", None, flags flags.append("missing_isin") if ticker and exchange and instrument_id: flags.append("ticker_exchange_fallback_requires_review") return "needs_manual_review", "0.75", instrument_id, flags if not ticker: flags.append("missing_ticker") if source_label and not ticker: flags.append("name_only_not_auto_mapped") return "needs_manual_review", "0.20", None, flags if not isin and not ticker: flags.append("name_only_not_auto_mapped") return "needs_manual_review", "0.10", None, flags return "needs_manual_review", "0.40", instrument_id, flags def create_instrument_mapping( conn: Connection, *, source_platform: str, source_label: str, asset_class: str = "other", source_account: str | None = None, isin: str | None = None, ticker: str | None = None, exchange: str | None = None, currency: str | None = None, notes: str | None = None, ) -> dict[str, Any]: normalized = normalize_name(source_label) status, confidence, instrument_id, flags = classify_instrument_mapping( conn, asset_class=asset_class, source_label=source_label, isin=isin, ticker=ticker, exchange=exchange, currency=currency, ) mapping_id = str(uuid.uuid4()) now = utc_now() conn.execute( """ INSERT INTO instrument_mappings( mapping_id, source_name, source_platform, source_account, source_label, normalized_name, isin, ticker, exchange, currency, asset_class, source_instrument_name, source_isin, source_ticker, source_exchange, source_currency, instrument_id, mapping_status, confidence, quality_flags_json, notes, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( mapping_id, source_platform, source_platform, source_account, source_label, normalized, isin, ticker, exchange, currency, asset_class, source_label, isin, ticker, exchange, currency, instrument_id, status, confidence, _json(flags), notes, now, now, ), ) return {"mapping_id": mapping_id, "mapping_status": status, "confidence": confidence, "instrument_id": instrument_id, "quality_flags": flags} def confirm_instrument_mapping( conn: Connection, *, mapping_id: str, instrument_id: str, note: str, created_by: str = "manual_review", ) -> str: if not note or not note.strip(): raise ValueError("Manual mapping confirmation requires a note") row = conn.execute("SELECT * FROM instrument_mappings WHERE mapping_id=?", (mapping_id,)).fetchone() if not row: raise ValueError("instrument mapping not found") old = dict(row) now = utc_now() conn.execute( "UPDATE instrument_mappings SET instrument_id=?, mapping_status='mapped', confidence='1.00', notes=?, updated_at=? WHERE mapping_id=?", (instrument_id, note, now, mapping_id), ) audit_id = record_audit_event( conn, source="manual_review_queue", action="confirm_instrument_mapping", entity_type="instrument_mapping", entity_id=mapping_id, old_values={"mapping_status": old.get("mapping_status"), "instrument_id": old.get("instrument_id")}, new_values={"mapping_status": "mapped", "instrument_id": instrument_id}, user_text_note=note, confirmed=True, created_by=created_by, ) return audit_id def create_platform_account_mapping( conn: Connection, *, source_platform: str, source_account_label: str, account_type: str = "other", currency: str | None = None, internal_platform_id: str | None = None, internal_account_id: str | None = None, notes: str | None = None, ) -> dict[str, Any]: if source_platform not in SOURCE_PLATFORMS: raise ValueError("unsupported source_platform") acct_type = account_type if account_type in ACCOUNT_TYPES else "other" flags: list[str] = [] if source_platform == "Raiffeisen" and acct_type not in {"cash", "reserve", "other"}: flags.append("raiffeisen_instrument_detail_missing") acct_type = "cash" status = "mapped" if internal_platform_id and internal_account_id else "needs_manual_review" mapping_id = str(uuid.uuid4()) now = utc_now() conn.execute( """ INSERT INTO platform_account_mappings( mapping_id, source_name, source_platform, source_platform_label, source_account_label, normalized_platform, normalized_account_name, source_account_type, account_type, source_currency, currency, platform_id, account_id, internal_platform_id, internal_account_id, mapping_status, quality_flags_json, notes, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( mapping_id, source_platform, source_platform, source_platform, source_account_label, normalize_name(source_platform), normalize_name(source_account_label), acct_type, acct_type, currency, currency, internal_platform_id, internal_account_id, internal_platform_id, internal_account_id, status, _json(flags), notes, now, now, ), ) return {"mapping_id": mapping_id, "mapping_status": status, "account_type": acct_type, "quality_flags": flags} @dataclass(frozen=True) class DryRunInput: source_platform: str source_file_type: str source_filename: str detected_snapshot_date: str | None = None snapshot_date_status: str = "missing" rows_total: int = 0 candidate_positions: int = 0 candidate_cash_rows: int = 0 mapped_positions: int = 0 blocked_positions: int = 0 warnings_count: int = 0 errors_count: int = 0 quality_flags: dict[str, int] | None = None summary: dict[str, Any] | None = None detected_sections: list[str] | None = None def create_broker_import_dry_run(conn: Connection, item: DryRunInput) -> str: dry_run_id = str(uuid.uuid4()) now = utc_now() source_hash = _hash_label(item.source_filename) summary = dict(item.summary or {}) summary.pop("rows", None) # explicit guard: no raw rows in persisted summary summary.pop("values", None) conn.execute( """ INSERT INTO broker_import_dry_runs( dry_run_id, source_name, source_platform, source_file_label, source_file_type, source_hash, source_filename_hash, detected_snapshot_date, snapshot_date_status, detected_sections_json, field_coverage_json, rows_total, rows_position_candidates, rows_cash_candidates, candidate_positions, candidate_cash_rows, mapped_positions, blocked_positions, warnings_count, errors_count, quality_flags_json, summary_json, created_at, notes ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( dry_run_id, item.source_platform, item.source_platform, "[runtime-file-redacted]", item.source_file_type, source_hash, source_hash, item.detected_snapshot_date, item.snapshot_date_status, _json(item.detected_sections or []), _json({}), item.rows_total, item.candidate_positions, item.candidate_cash_rows, item.candidate_positions, item.candidate_cash_rows, item.mapped_positions, item.blocked_positions, item.warnings_count, item.errors_count, _json(item.quality_flags or {}), _json(summary), now, "dry-run summary only; no productive ledger writes", ), ) return dry_run_id def _row_hash(item: dict[str, Any]) -> str: return hashlib.sha256(json.dumps(item, ensure_ascii=False, sort_keys=True, default=str).encode("utf-8")).hexdigest() def create_broker_import_review_item( conn: Connection, *, dry_run_id: str, source_platform: str, source_file_type: str, source_row_ref: str, source_label: str, detected_asset_class: str, detected_currency: str | None = None, detected_quantity_present: bool = False, detected_market_value_present: bool = False, isin: str | None = None, ticker: str | None = None, exchange: str | None = None, mapped_instrument_id: str | None = None, mapped_account_id: str | None = None, quality_flags: list[str] | tuple[str, ...] | None = None, review_status: str = "open", ) -> str: review_item_id = str(uuid.uuid4()) now = utc_now() normalized = normalize_name(source_label) row_hash = _row_hash({ "source_platform": source_platform, "source_row_ref": source_row_ref, "source_label": source_label, "asset_class": detected_asset_class, "currency": detected_currency, "isin": isin, "ticker": ticker, "exchange": exchange, }) conn.execute( """ INSERT INTO broker_import_review_items( review_item_id, dry_run_id, source_platform, source_file_type, source_row_ref, row_hash, source_label, normalized_name, detected_asset_class, detected_currency, detected_quantity_present, detected_market_value_present, isin, ticker, exchange, mapped_instrument_id, mapped_account_id, quality_flags_json, review_status, reviewer_note, created_at, updated_at ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, ( review_item_id, dry_run_id, source_platform, source_file_type, source_row_ref, row_hash, source_label, normalized, detected_asset_class, detected_currency, 1 if detected_quantity_present else 0, 1 if detected_market_value_present else 0, isin, ticker, exchange, mapped_instrument_id, mapped_account_id, _json(list(quality_flags or [])), review_status, None, now, now, ), ) return review_item_id def get_broker_review_items(conn: Connection, *, status: str | None = "open", limit: int = 100) -> list[dict[str, str]]: params: list[Any] = [] where = "" if status: where = "WHERE review_status=?" params.append(status) params.append(limit) rows = conn.execute( f""" SELECT review_item_id AS item_id, dry_run_id, source_platform, source_file_type, source_row_ref, source_label, normalized_name, detected_asset_class, detected_currency, detected_quantity_present, detected_market_value_present, isin, ticker, exchange, mapped_instrument_id, mapped_account_id, quality_flags_json, review_status, reviewer_note, created_at FROM broker_import_review_items {where} ORDER BY created_at DESC LIMIT ? """, tuple(params), ).fetchall() return [{k: "" if row[k] is None else str(row[k]) for k in row.keys()} for row in rows] def aggregate_review_quality_flags(conn: Connection) -> dict[str, int]: flags: Counter[str] = Counter() for row in conn.execute("SELECT quality_flags_json FROM broker_import_review_items WHERE review_status IN ('open','blocked')").fetchall(): try: for flag in json.loads(row["quality_flags_json"] or "[]"): flags[str(flag)] += 1 except json.JSONDecodeError: flags["invalid_quality_flags"] += 1 return dict(flags) def run_broker_import_dry_run(conn: Connection, parsed, *, source_filename: str) -> dict[str, Any]: mapped_positions = 0 blocked_positions = 0 review_items = 0 for candidate in parsed.candidates: if candidate.detected_asset_class == "cash": continue status, _confidence, instrument_id, map_flags = classify_instrument_mapping( conn, asset_class=candidate.detected_asset_class, source_label=candidate.source_label, isin=candidate.isin, ticker=candidate.ticker, exchange=candidate.exchange, currency=candidate.detected_currency, ) if status == "mapped" and instrument_id: mapped_positions += 1 else: blocked_positions += 1 quality_summary = parsed.quality_flags_summary dry_run_id = create_broker_import_dry_run(conn, DryRunInput( source_platform=parsed.source_platform, source_file_type=parsed.source_file_type, source_filename=source_filename, detected_snapshot_date=parsed.detected_snapshot_date, snapshot_date_status=parsed.snapshot_date_status, rows_total=parsed.rows_total, candidate_positions=parsed.candidate_positions, candidate_cash_rows=parsed.candidate_cash_rows, mapped_positions=mapped_positions, blocked_positions=blocked_positions, warnings_count=sum(quality_summary.values()), errors_count=len(parsed.errors), quality_flags=quality_summary, summary={ "source_platform": parsed.source_platform, "source_file_type": parsed.source_file_type, "rows_total": parsed.rows_total, "candidate_positions": parsed.candidate_positions, "candidate_cash_rows": parsed.candidate_cash_rows, "mapped_positions": mapped_positions, "blocked_positions": blocked_positions, }, detected_sections=list(parsed.detected_sections), )) for candidate in parsed.candidates: mapped_instrument_id = _first_matching_instrument( conn, isin=candidate.isin, ticker=candidate.ticker, exchange=candidate.exchange, currency=candidate.detected_currency, ) create_broker_import_review_item( conn, dry_run_id=dry_run_id, source_platform=candidate.source_platform, source_file_type=candidate.source_file_type, source_row_ref=candidate.source_row_ref, source_label=candidate.source_label, detected_asset_class=candidate.detected_asset_class, detected_currency=candidate.detected_currency, detected_quantity_present=candidate.detected_quantity_present, detected_market_value_present=candidate.detected_market_value_present, isin=candidate.isin, ticker=candidate.ticker, exchange=candidate.exchange, mapped_instrument_id=mapped_instrument_id, mapped_account_id=None, quality_flags=candidate.quality_flags, review_status="blocked" if "aggregate_only" in candidate.quality_flags else "open", ) review_items += 1 return { "dry_run_id": dry_run_id, "rows_total": parsed.rows_total, "candidate_positions": parsed.candidate_positions, "candidate_cash_rows": parsed.candidate_cash_rows, "mapped_positions": mapped_positions, "blocked_positions": blocked_positions, "review_items_created": review_items, "warnings_count": sum(quality_summary.values()), "errors_count": len(parsed.errors), "quality_flags_summary": quality_summary, } def confirm_review_item_mapping(conn: Connection, *, review_item_id: str, instrument_id: str | None = None, account_id: str | None = None, note: str, created_by: str = "manual_review") -> str: if not note or not note.strip(): raise ValueError("Review item change requires a note") row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") now = utc_now() conn.execute( "UPDATE broker_import_review_items SET mapped_instrument_id=COALESCE(?, mapped_instrument_id), mapped_account_id=COALESCE(?, mapped_account_id), review_status='mapped', reviewer_note=?, updated_at=? WHERE review_item_id=?", (instrument_id, account_id, note, now, review_item_id), ) return record_audit_event( conn, source="manual_review_queue", action="confirm_review_item_mapping", entity_type="broker_import_review_item", entity_id=review_item_id, old_values={"review_status": row["review_status"], "mapped_instrument_id": row["mapped_instrument_id"], "mapped_account_id": row["mapped_account_id"]}, new_values={"review_status": "mapped", "mapped_instrument_id": instrument_id, "mapped_account_id": account_id}, user_text_note=note, confirmed=True, created_by=created_by, ) def get_manual_review_queue(conn: Connection) -> list[dict[str, str]]: rows: list[dict[str, str]] = [] for row in conn.execute( """ SELECT 'broker_import_review_item' AS item_type, review_item_id AS item_id, source_platform, mapped_account_id AS account_label, source_label AS label, review_status AS mapping_status, quality_flags_json, reviewer_note AS notes, created_at FROM broker_import_review_items WHERE review_status IN ('open','blocked') UNION ALL SELECT 'instrument_mapping' AS item_type, mapping_id AS item_id, source_platform, source_account AS account_label, source_label AS label, mapping_status, quality_flags_json, notes, created_at FROM instrument_mappings WHERE mapping_status IN ('needs_manual_review','missing_isin','missing_ticker','ambiguous') UNION ALL SELECT 'platform_account_mapping' AS item_type, mapping_id AS item_id, source_platform, source_account_label AS account_label, source_account_label AS label, mapping_status, quality_flags_json, notes, created_at FROM platform_account_mappings WHERE mapping_status IN ('needs_manual_review','missing_isin','missing_ticker','ambiguous') ORDER BY created_at DESC """ ).fetchall(): rows.append({k: "" if row[k] is None else str(row[k]) for k in row.keys()}) return rows def build_import_wizard_summary(conn: Connection) -> dict[str, str]: last = conn.execute("SELECT * FROM broker_import_dry_runs ORDER BY created_at DESC LIMIT 1").fetchone() pending = len(get_manual_review_queue(conn)) return { "mode": "read-only", "productive_import_enabled": "false", "pending_review_items": str(pending), "open_runtime_review_items": str(conn.execute("SELECT COUNT(*) AS n FROM broker_import_review_items WHERE review_status IN ('open','blocked')").fetchone()["n"]), "last_dry_run_source": last["source_platform"] if last else "", "last_dry_run_status": last["snapshot_date_status"] if last else "none", } def assert_write_enabled(*, write_enabled: bool) -> None: if not write_enabled: raise PermissionError("Import commit is blocked in read-only mode") def _load_flags(row: Any) -> list[str]: try: return [str(x) for x in json.loads(row["quality_flags_json"] or "[]")] except Exception: return [] def _store_flags(flags: list[str]) -> str: return _json(sorted({f for f in flags if f})) def _without_flags(flags: list[str], remove: set[str]) -> list[str]: return [f for f in flags if f not in remove] def _require_note(note: str | None, message: str = "Review metadata change requires a note") -> str: if not note or not note.strip(): raise ValueError(message) return note.strip() def validate_isin(value: str) -> str: isin = (value or "").strip().upper().replace(" ", "") if not re.fullmatch(r"[A-Z]{2}[A-Z0-9]{9}[0-9]", isin): raise ValueError("Invalid ISIN format") return isin def calculate_review_item_readiness(row: Any) -> str: status = str(row["review_status"] or "open") if status == "ignored": return "ignored" flags = set(_load_flags(row)) asset_class = str(row["detected_asset_class"] or "other") if status == "blocked" or "aggregate_only" in flags or "missing_instrument_details" in flags: return "blocked" if asset_class in {"equity", "etf"}: has_identifier = bool(row["isin"]) or (bool(row["ticker"]) and bool(row["exchange"]) and int(row["ticker_exchange_confirmed"] or 0) == 1) required = [ bool(row["source_platform"]), bool(row["mapped_account_id"]), asset_class in {"equity", "etf"}, has_identifier, bool(row["detected_currency"]), (int(row["snapshot_date_confirmed"] or 0) == 1), (int(row["detected_quantity_present"] or 0) == 1), int(row["reviewer_confirmed"] or 0) == 1, ] blocking_flags = {"missing_isin", "missing_ticker", "aggregate_only", "missing_instrument_details", "missing_fx"} if all(required) and not (flags & blocking_flags): return "ready_for_import" return "review_needed" if has_identifier or bool(row["mapped_account_id"]) else "not_ready" if asset_class == "cash": required = [bool(row["mapped_account_id"]), bool(row["detected_currency"]), int(row["snapshot_date_confirmed"] or 0) == 1, int(row["detected_market_value_present"] or 0) == 1, int(row["reviewer_confirmed"] or 0) == 1] return "ready_for_import" if all(required) else "review_needed" return "not_ready" def refresh_review_item_readiness(conn: Connection, review_item_id: str) -> str: row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") readiness = calculate_review_item_readiness(row) conn.execute("UPDATE broker_import_review_items SET import_readiness_status=?, account_mapping_status=CASE WHEN mapped_account_id IS NULL OR mapped_account_id='' THEN 'missing' ELSE 'mapped' END, updated_at=? WHERE review_item_id=?", (readiness, utc_now(), review_item_id)) return readiness def _audit_review_change(conn: Connection, *, review_item_id: str, action: str, old: dict[str, Any], new: dict[str, Any], note: str, created_by: str = "manual_review") -> str: return record_audit_event(conn, source="manual_review_queue", action=action, entity_type="broker_import_review_item", entity_id=review_item_id, old_values=old, new_values=new, user_text_note=note, confirmed=True, created_by=created_by) def update_review_item_isin(conn: Connection, *, review_item_id: str, isin: str, note: str, created_by: str = "manual_review") -> str: note = _require_note(note) isin = validate_isin(isin) row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") flags = _without_flags(_load_flags(row), {"missing_isin", "name_only_not_auto_mapped"}) conn.execute("UPDATE broker_import_review_items SET isin=?, quality_flags_json=?, reviewer_note=?, updated_at=? WHERE review_item_id=?", (isin, _store_flags(flags), note, utc_now(), review_item_id)) readiness = refresh_review_item_readiness(conn, review_item_id) return _audit_review_change(conn, review_item_id=review_item_id, action="update_isin", old={"isin": row["isin"], "quality_flags": _load_flags(row)}, new={"isin": isin, "quality_flags": flags, "readiness": readiness}, note=note, created_by=created_by) def update_review_item_ticker_exchange(conn: Connection, *, review_item_id: str, ticker: str | None = None, exchange: str | None = None, note: str, confirm_ticker_exchange: bool = False, created_by: str = "manual_review") -> str: note = _require_note(note) row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") new_ticker = (ticker or row["ticker"] or "").strip().upper() or None new_exchange = (exchange or row["exchange"] or "").strip().upper() or None flags = _load_flags(row) if new_ticker: flags = _without_flags(flags, {"missing_ticker"}) if new_ticker and not new_exchange and "ticker_without_exchange" not in flags: flags.append("ticker_without_exchange") if new_ticker and new_exchange: flags = _without_flags(flags, {"ticker_without_exchange"}) confirmed = 1 if (confirm_ticker_exchange and new_ticker and new_exchange) else int(row["ticker_exchange_confirmed"] or 0) conn.execute("UPDATE broker_import_review_items SET ticker=?, exchange=?, ticker_exchange_confirmed=?, quality_flags_json=?, reviewer_note=?, updated_at=? WHERE review_item_id=?", (new_ticker, new_exchange, confirmed, _store_flags(flags), note, utc_now(), review_item_id)) readiness = refresh_review_item_readiness(conn, review_item_id) return _audit_review_change(conn, review_item_id=review_item_id, action="update_ticker_exchange", old={"ticker": row["ticker"], "exchange": row["exchange"]}, new={"ticker": new_ticker, "exchange": new_exchange, "ticker_exchange_confirmed": confirmed, "readiness": readiness}, note=note, created_by=created_by) def confirm_review_item_instrument_mapping(conn: Connection, *, review_item_id: str, instrument_id: str | None = None, note: str, confirm_ticker_exchange: bool = False, created_by: str = "manual_review") -> str: note = _require_note(note) row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") if not row["isin"] and not (row["ticker"] and row["exchange"] and confirm_ticker_exchange): raise ValueError("Name-only rows cannot be marked as mapped") if not instrument_id: instrument_id = _first_matching_instrument(conn, isin=row["isin"], ticker=row["ticker"], exchange=row["exchange"], currency=row["detected_currency"]) if not instrument_id: raise ValueError("instrument_id required when no existing instrument matches") flags = _without_flags(_load_flags(row), {"missing_isin", "missing_ticker", "ticker_without_exchange", "name_only_not_auto_mapped"}) conn.execute("UPDATE broker_import_review_items SET mapped_instrument_id=?, review_status='mapped', ticker_exchange_confirmed=?, reviewer_confirmed=1, quality_flags_json=?, reviewer_note=?, updated_at=? WHERE review_item_id=?", (instrument_id, 1 if confirm_ticker_exchange else int(row["ticker_exchange_confirmed"] or 0), _store_flags(flags), note, utc_now(), review_item_id)) readiness = refresh_review_item_readiness(conn, review_item_id) return _audit_review_change(conn, review_item_id=review_item_id, action="confirm_instrument_mapping", old={"review_status": row["review_status"], "mapped_instrument_id": row["mapped_instrument_id"]}, new={"review_status": "mapped", "mapped_instrument_id": instrument_id, "readiness": readiness}, note=note, created_by=created_by) def confirm_review_item_account_mapping(conn: Connection, *, review_item_id: str, account_id: str, note: str, created_by: str = "manual_review") -> str: note = _require_note(note) row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() acct = conn.execute("SELECT account_id, account_type FROM accounts WHERE account_id=? AND is_active=1", (account_id,)).fetchone() if not row: raise ValueError("review item not found") if not acct: raise ValueError("account not found") if row["detected_asset_class"] == "cash" and acct["account_type"] not in {"cash", "reserve", "other"}: raise ValueError("cash review items must map to cash/reserve/other accounts") if row["detected_asset_class"] in {"equity", "etf"} and acct["account_type"] == "cash": raise ValueError("brokerage review items must not map to cash-only accounts") conn.execute("UPDATE broker_import_review_items SET mapped_account_id=?, account_mapping_status='mapped', reviewer_note=?, updated_at=? WHERE review_item_id=?", (account_id, note, utc_now(), review_item_id)) readiness = refresh_review_item_readiness(conn, review_item_id) return _audit_review_change(conn, review_item_id=review_item_id, action="confirm_account_mapping", old={"mapped_account_id": row["mapped_account_id"]}, new={"mapped_account_id": account_id, "readiness": readiness}, note=note, created_by=created_by) def confirm_review_item_snapshot_date(conn: Connection, *, review_item_id: str, note: str, created_by: str = "manual_review") -> str: note = _require_note(note) row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") conn.execute("UPDATE broker_import_review_items SET snapshot_date_confirmed=1, reviewer_note=?, updated_at=? WHERE review_item_id=?", (note, utc_now(), review_item_id)) readiness = refresh_review_item_readiness(conn, review_item_id) return _audit_review_change(conn, review_item_id=review_item_id, action="confirm_snapshot_date", old={"snapshot_date_confirmed": row["snapshot_date_confirmed"]}, new={"snapshot_date_confirmed": 1, "readiness": readiness}, note=note, created_by=created_by) def ignore_review_item(conn: Connection, *, review_item_id: str, note: str, created_by: str = "manual_review") -> str: note = _require_note(note, "Ignore requires a note") row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") conn.execute("UPDATE broker_import_review_items SET review_status='ignored', import_readiness_status='ignored', reviewer_note=?, updated_at=? WHERE review_item_id=?", (note, utc_now(), review_item_id)) return _audit_review_change(conn, review_item_id=review_item_id, action="ignore_review_item", old={"review_status": row["review_status"]}, new={"review_status": "ignored", "readiness": "ignored"}, note=note, created_by=created_by) def mark_review_item_resolved(conn: Connection, *, review_item_id: str, note: str, created_by: str = "manual_review") -> str: note = _require_note(note) row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") readiness = calculate_review_item_readiness(row) if readiness not in {"ready_for_import", "ignored"}: raise ValueError("Review item cannot be resolved until mandatory issues are solved") conn.execute("UPDATE broker_import_review_items SET review_status='resolved', import_readiness_status=?, reviewer_note=?, updated_at=? WHERE review_item_id=?", (readiness, note, utc_now(), review_item_id)) return _audit_review_change(conn, review_item_id=review_item_id, action="mark_resolved", old={"review_status": row["review_status"]}, new={"review_status": "resolved", "readiness": readiness}, note=note, created_by=created_by) def get_broker_review_items(conn: Connection, *, status: str | None = None, source_platform: str | None = None, dry_run_id: str | None = None, quality_flag: str | None = None, asset_class: str | None = None, currency: str | None = None, account_mapping_status: str | None = None, readiness_status: str | None = None, limit: int = 100) -> list[dict[str, str]]: clauses=[]; params=[] for col,val in [("review_status",status),("source_platform",source_platform),("dry_run_id",dry_run_id),("detected_asset_class",asset_class),("detected_currency",currency),("account_mapping_status",account_mapping_status),("import_readiness_status",readiness_status)]: if val: clauses.append(f"{col}=?"); params.append(val) if quality_flag: clauses.append("quality_flags_json LIKE ?"); params.append(f'%"{quality_flag}"%') where = "WHERE " + " AND ".join(clauses) if clauses else "" params.append(limit) rows = conn.execute(f""" SELECT review_item_id AS item_id, dry_run_id, source_platform, source_file_type, source_row_ref, source_label, normalized_name, detected_asset_class, detected_currency, detected_quantity_present, detected_market_value_present, isin, ticker, exchange, mapped_instrument_id, mapped_account_id, account_mapping_status, quality_flags_json, review_status, import_readiness_status, reviewer_confirmed, snapshot_date_confirmed, ticker_exchange_confirmed, reviewer_note, created_at, updated_at FROM broker_import_review_items {where} ORDER BY created_at DESC LIMIT ? """, tuple(params)).fetchall() return [{k: "" if row[k] is None else str(row[k]) for k in row.keys()} for row in rows] def get_review_item_detail(conn: Connection, review_item_id: str) -> dict[str, Any]: row = conn.execute("SELECT * FROM broker_import_review_items WHERE review_item_id=?", (review_item_id,)).fetchone() if not row: raise ValueError("review item not found") audit = conn.execute("SELECT timestamp, action, entity_type, entity_id, user_text_note, created_by FROM audit_log WHERE entity_id=? ORDER BY timestamp DESC", (review_item_id,)).fetchall() return {"item": {k: "" if row[k] is None else str(row[k]) for k in row.keys()}, "audit_history": [{k: "" if a[k] is None else str(a[k]) for k in a.keys()} for a in audit]} def get_dry_run_sessions(conn: Connection, *, limit: int = 50) -> list[dict[str, str]]: rows = conn.execute(""" SELECT d.dry_run_id, d.source_platform, d.source_file_type, d.created_at, d.snapshot_date_status, d.rows_total, d.candidate_positions, d.candidate_cash_rows, d.warnings_count, d.errors_count, COALESCE(d.session_status,'active') AS session_status, COALESCE(d.is_current,0) AS is_current, SUM(CASE WHEN r.review_item_id IS NOT NULL THEN 1 ELSE 0 END) AS review_items, SUM(CASE WHEN r.import_readiness_status='ready_for_import' THEN 1 ELSE 0 END) AS ready_for_import, SUM(CASE WHEN r.review_status='blocked' OR r.import_readiness_status='blocked' THEN 1 ELSE 0 END) AS blocked, SUM(CASE WHEN r.review_status='ignored' THEN 1 ELSE 0 END) AS ignored, SUM(CASE WHEN r.review_status='open' THEN 1 ELSE 0 END) AS open FROM broker_import_dry_runs d LEFT JOIN broker_import_review_items r ON r.dry_run_id=d.dry_run_id GROUP BY d.dry_run_id ORDER BY d.created_at DESC LIMIT ? """, (limit,)).fetchall() return [{k: "" if row[k] is None else str(row[k]) for k in row.keys()} for row in rows] def update_dry_run_session_status(conn: Connection, *, dry_run_id: str, action: str, note: str, created_by: str = "manual_review") -> str: note = _require_note(note) row = conn.execute("SELECT * FROM broker_import_dry_runs WHERE dry_run_id=?", (dry_run_id,)).fetchone() if not row: raise ValueError("dry run not found") now=utc_now() if action == "archive": conn.execute("UPDATE broker_import_dry_runs SET session_status='archived', is_current=0, archived_at=?, notes=? WHERE dry_run_id=?", (now, note, dry_run_id)) new={"session_status":"archived"} elif action == "discard": conn.execute("UPDATE broker_import_dry_runs SET session_status='discarded', is_current=0, discarded_at=?, notes=? WHERE dry_run_id=?", (now, note, dry_run_id)) new={"session_status":"discarded"} elif action == "mark_current": conn.execute("UPDATE broker_import_dry_runs SET is_current=0 WHERE source_platform=?", (row["source_platform"],)) conn.execute("UPDATE broker_import_dry_runs SET session_status='active', is_current=1, notes=? WHERE dry_run_id=?", (note, dry_run_id)) new={"session_status":"active","is_current":1} else: raise ValueError("unsupported dry run session action") return record_audit_event(conn, source="manual_review_queue", action=f"dry_run_{action}", entity_type="broker_import_dry_run", entity_id=dry_run_id, old_values={"session_status": row["session_status"] if "session_status" in row.keys() else "active", "is_current": row["is_current"] if "is_current" in row.keys() else 0}, new_values=new, user_text_note=note, confirmed=True, created_by=created_by) def get_review_readiness_counts(conn: Connection) -> list[dict[str, str]]: rows=conn.execute("SELECT import_readiness_status AS readiness_status, COUNT(*) AS count FROM broker_import_review_items GROUP BY import_readiness_status ORDER BY import_readiness_status").fetchall() return [{"readiness_status": row["readiness_status"] or "not_ready", "count": str(row["count"])} for row in rows] def get_ready_for_import_by_source(conn: Connection) -> list[dict[str, str]]: rows=conn.execute("SELECT source_platform, SUM(CASE WHEN import_readiness_status='ready_for_import' THEN 1 ELSE 0 END) AS ready_for_import, COUNT(*) AS total FROM broker_import_review_items GROUP BY source_platform ORDER BY source_platform").fetchall() return [{"source_platform": row["source_platform"], "ready_for_import": str(row["ready_for_import"] or 0), "total": str(row["total"])} for row in rows] def assert_productive_import_disabled() -> bool: return True