Harden duplicate review and CSV handling

This commit is contained in:
2026-09-03 12:19:01 -04:00
parent 2cebb0db2e
commit b5317e297a
7 changed files with 52 additions and 4 deletions
+3
View File
@@ -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"
+12
View File
@@ -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")
+2 -2
View File
@@ -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)
+1
View File
@@ -35,3 +35,4 @@ class MergeResult:
included: list[Record]
removed_exact: list[Record]
review_groups: list[DuplicateGroup]
warnings: list[str] = field(default_factory=list)
+10 -2
View File
@@ -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", ""),
+10
View File
@@ -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()
+14
View File
@@ -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])