From b5317e297a9eeb833a91ab90658b6a08c2aee25e Mon Sep 17 00:00:00 2001 From: ledadmin Date: Thu, 3 Sep 2026 12:19:01 -0400 Subject: [PATCH] Harden duplicate review and CSV handling --- pyproject.toml | 3 +++ src/dataloom/core/exporters.py | 12 ++++++++++++ src/dataloom/core/merge.py | 4 ++-- src/dataloom/core/models.py | 1 + src/dataloom/core/parsers.py | 12 ++++++++++-- src/dataloom/ui/main_window.py | 10 ++++++++++ tests/test_core.py | 14 ++++++++++++++ 7 files changed, 52 insertions(+), 4 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 733f40a..09dffb5 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,6 +10,9 @@ readme = "README.md" requires-python = ">=3.11" dependencies = ["PySide6>=6.6"] +[project.optional-dependencies] +test = ["pytest>=8.0"] + [project.scripts] dataloom = "dataloom.app:main" diff --git a/src/dataloom/core/exporters.py b/src/dataloom/core/exporters.py index 199fde7..9eb4d27 100644 --- a/src/dataloom/core/exporters.py +++ b/src/dataloom/core/exporters.py @@ -8,6 +8,18 @@ from typing import Iterable from .models import Record +def csv_loss_warnings(records: Iterable[Record]) -> list[str]: + warnings = [] + for record in records: + unsupported = [key for key in record.data if key not in {"type", "name", "notes", "favorite", "login"}] + login = record.data.get("login") or {} + unsupported += [f"login.{key}" for key in login if key not in {"username", "password", "totp", "uris"}] + if len(login.get("uris") or []) > 1: unsupported.append("multiple login URIs") + if unsupported: + warnings.append(f"{record.name or '(unnamed)'}: {', '.join(unsupported)} cannot be represented losslessly in CSV") + return warnings + + def export_json(records: Iterable[Record], destination: str | Path) -> Path: path = Path(destination) path.write_text(json.dumps({"encrypted": False, "folders": [], "items": [r.data for r in records]}, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") diff --git a/src/dataloom/core/merge.py b/src/dataloom/core/merge.py index 4bbff36..dc647f5 100644 --- a/src/dataloom/core/merge.py +++ b/src/dataloom/core/merge.py @@ -1,7 +1,7 @@ """Duplicate classification and conservative merging.""" from collections import defaultdict -from datetime import datetime +from datetime import datetime, timezone from typing import Iterable from .models import DuplicateGroup, MergeResult, Record @@ -55,6 +55,6 @@ def merge_records(records: Iterable[Record]) -> MergeResult: if len(matches) < 2 or not matches[0].data.get("login"): continue revisions = [_revision(item) for item in matches] - selected = max(matches, key=lambda item: _revision(item) or datetime.min) if all(revisions) else None + selected = max(matches, key=lambda item: _revision(item) or datetime.min.replace(tzinfo=timezone.utc)) if all(revisions) else None review.append(DuplicateGroup("likely", matches, "Same type, normalized URL, and username; values differ", selected)) return MergeResult(included, removed, review) diff --git a/src/dataloom/core/models.py b/src/dataloom/core/models.py index 45508b5..417a60c 100644 --- a/src/dataloom/core/models.py +++ b/src/dataloom/core/models.py @@ -35,3 +35,4 @@ class MergeResult: included: list[Record] removed_exact: list[Record] review_groups: list[DuplicateGroup] + warnings: list[str] = field(default_factory=list) diff --git a/src/dataloom/core/parsers.py b/src/dataloom/core/parsers.py index 185c900..90df8f0 100644 --- a/src/dataloom/core/parsers.py +++ b/src/dataloom/core/parsers.py @@ -14,6 +14,10 @@ class ImportError(ValueError): pass +CSV_FIELDS = {"type", "name", "notes", "favorite", "login_username", "login_password", "login_totp", "login_uri", + "folder", "collection", "reprompt"} + + def detect_format(path: str | Path) -> str: suffix = Path(path).suffix.casefold() if suffix == ".json": return "json" @@ -32,8 +36,12 @@ def load_file(path: str | Path) -> tuple[str, list[Record]]: if not isinstance(items, list): raise ImportError(f"{path.name} has no Bitwarden items list") return fmt, [Record(item, path.name, i, item.get("id"), fmt) for i, item in enumerate(items, 1)] with path.open(newline="", encoding="utf-8-sig") as stream: - rows = list(csv.DictReader(stream)) - if not rows and not stream.tell(): raise ImportError(f"{path.name} is empty") + reader = csv.DictReader(stream) + headers = set(reader.fieldnames or []) + if not headers: raise ImportError(f"{path.name} is empty or has no header") + rows = list(reader) + if not (headers & {"name", "login_username", "login_uri"}): + raise ImportError(f"{path.name} does not look like a Bitwarden CSV export") records = [] for i, row in enumerate(rows, 1): item = {"type": row.get("type", "login"), "name": row.get("name", ""), "notes": row.get("notes", ""), diff --git a/src/dataloom/ui/main_window.py b/src/dataloom/ui/main_window.py index a93777b..6702fbf 100644 --- a/src/dataloom/ui/main_window.py +++ b/src/dataloom/ui/main_window.py @@ -24,8 +24,10 @@ class MainWindow(QMainWindow): self.summary = QTextEdit(); self.summary.setReadOnly(True) self.duplicates = QListWidget(); self.duplicates.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection) self.log_view = QTextEdit(); self.log_view.setReadOnly(True) + self.entries_view = QListWidget() tabs = QTabWidget() tabs.addTab(self._page("Import Summary", self.summary), "Import Summary") + tabs.addTab(self._page("Merged Entries", self.entries_view), "Merged Entries") tabs.addTab(self._page("Duplicate Review", self.duplicates), "Duplicate Review") tabs.addTab(self._page("Audit Log", self.log_view), "Audit Log") import_button = QPushButton("Import Bitwarden exports…") @@ -61,6 +63,9 @@ class MainWindow(QMainWindow): self.audit.save() self.export_button.setEnabled(True) self.summary.setPlainText(f"Format: {formats[0].upper()}\nFiles: {len(paths)}\nRecords loaded: {len(records)}\nMerged records: {len(self.result.included)}\nExact duplicates removed: {len(self.result.removed_exact)}\nGroups requiring review: {len(self.result.review_groups)}\n\nAudit log: {self.audit.path}") + self.entries_view.clear() + for record in self.result.included: + self.entries_view.addItem(f"{record.name or '(unnamed)'} — type {record.item_type} — {record.source_file}:{record.source_index}") self.duplicates.clear() for group in self.result.review_groups: selected = f"; selected: {group.selected.name}" if group.selected else "; no automatic selection" @@ -76,6 +81,11 @@ class MainWindow(QMainWindow): if not path: return try: from dataloom.core.exporters import export_csv, export_json + if fmt == "csv": + from dataloom.core.exporters import csv_loss_warnings + warnings = csv_loss_warnings(self.result.included) + if warnings: + self.summary.append("\nCSV loss warnings:\n" + "\n".join(f"- {warning}" for warning in warnings)) output = export_json(self.result.included, path) if fmt == "json" else export_csv(self.result.included, path) self.audit.record("EXPORT", f"Wrote merged {fmt.upper()} export to {output}") self.audit.save() diff --git a/tests/test_core.py b/tests/test_core.py index 594db39..db3e2de 100644 --- a/tests/test_core.py +++ b/tests/test_core.py @@ -6,6 +6,7 @@ from pathlib import Path from dataloom.core.merge import merge_records from dataloom.core.models import Record from dataloom.core.normalize import normalize_url +from dataloom.core.exporters import csv_loss_warnings from dataloom.core.parsers import load_file @@ -40,3 +41,16 @@ def test_conflicting_passwords_are_retained_for_review(): result = merge_records([Record(base, "a.json", 1), Record(other, "b.json", 1)]) assert len(result.included) == 2 assert any(group.classification == "likely" for group in result.review_groups) + + +def test_newest_revision_is_selected_without_timezone_error(): + left = {"type": 1, "name": "X", "revisionDate": "2026-01-01T00:00:00Z", "login": {"username": "u", "password": "a", "uris": [{"uri": "https://x.test"}]}} + right = {"type": 1, "name": "X", "revisionDate": "2026-02-01T00:00:00Z", "login": {"username": "u", "password": "b", "uris": [{"uri": "https://x.test"}]}} + result = merge_records([Record(left, "a.json", 1), Record(right, "b.json", 1)]) + assert result.review_groups[-1].selected is not None + assert result.review_groups[-1].selected.source_file == "b.json" + + +def test_csv_loss_warning_for_extra_uri(): + record = Record({"type": 1, "name": "X", "login": {"uris": [{"uri": "https://a"}, {"uri": "https://b"}]}}, "a.json", 1) + assert csv_loss_warnings([record])