ledadmin/DataLoom
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

Add structured duplicate selection review

Browse Source
This commit is contained in:
ledadmin
2026-09-03 13:56:28 -04:00
parent cb82594bd0
commit 5adbe61ebf
1 changed files with 96 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/dataloom/ui/main_window.py
+96 -12
View File
@@ -4,8 +4,9 @@ from pathlib import Path
from PySide6.QtWidgets import (
QAbstractItemView, QApplication, QFileDialog, QLabel, QListWidget,
QMainWindow, QMessageBox, QPushButton, QTabWidget, QTextEdit,
QCheckBox, QVBoxLayout, QWidget,
QComboBox, QMainWindow, QMessageBox, QPushButton, QTabWidget,
QTableWidget, QTableWidgetItem, QTextEdit, QCheckBox, QVBoxLayout,
QHBoxLayout, QWidget,
)
from dataloom.core.audit import AuditLog
@@ -21,9 +22,19 @@ class MainWindow(QMainWindow):
self.records = []
self.result = None
self.audit = None
self.review_decisions = {}
self.summary = QTextEdit(); self.summary.setReadOnly(True)
self.duplicates = QListWidget(); self.duplicates.setSelectionMode(QAbstractItemView.SelectionMode.SingleSelection)
self.duplicate_detail = QTextEdit(); self.duplicate_detail.setReadOnly(True)
self.duplicate_table = QTableWidget()
self.duplicate_table.setEditTriggers(QAbstractItemView.EditTrigger.NoEditTriggers)
self.duplicate_table.setAlternatingRowColors(True)
self.keep_selected = QPushButton("Keep selected record")
self.keep_both = QPushButton("Keep both records")
self.decision_choice = QComboBox()
self.decision_choice.addItem("Choose a record…", None)
self.keep_selected.clicked.connect(self.choose_selected_record)
self.keep_both.clicked.connect(self.choose_both_records)
self.duplicate_status = QLabel("Select a duplicate group to compare it.")
self.duplicates.currentRowChanged.connect(self.show_duplicate_detail)
self.log_view = QTextEdit(); self.log_view.setReadOnly(True)
self.entries_view = QListWidget()
@@ -31,7 +42,8 @@ class MainWindow(QMainWindow):
tabs.addTab(self._page("Import Summary", self.summary), "Import Summary")
tabs.addTab(self._page("Merged Entries", self.entries_view), "Merged Entries")
duplicate_page = QWidget(); duplicate_layout = QVBoxLayout(duplicate_page)
duplicate_layout.addWidget(QLabel("<h2>Duplicate Review</h2>")); duplicate_layout.addWidget(self.duplicates); duplicate_layout.addWidget(self.duplicate_detail)
duplicate_layout.addWidget(QLabel("<h2>Duplicate Review</h2>")); duplicate_layout.addWidget(self.duplicates); duplicate_layout.addWidget(self.duplicate_table)
controls = QHBoxLayout(); controls.addWidget(self.decision_choice); controls.addWidget(self.keep_selected); controls.addWidget(self.keep_both); duplicate_layout.addLayout(controls); duplicate_layout.addWidget(self.duplicate_status)
tabs.addTab(duplicate_page, "Duplicate Review")
tabs.addTab(self._page("Audit Log", self.log_view), "Audit Log")
import_button = QPushButton("Import Bitwarden exports…")
@@ -61,6 +73,7 @@ class MainWindow(QMainWindow):
if len(set(formats)) != 1: raise ImportError("All selected files must use the same format")
self.records = records
self.result = merge_records(records)
self.review_decisions = {}
self.audit = AuditLog(Path(paths[0]).parent / "dataloom-audit.log", include_secrets=self.secret_logging.isChecked())
self.audit.record("IMPORT", f"Loaded {len(paths)} {formats[0]} file(s) and {len(records)} record(s)")
for record in records:
@@ -76,6 +89,7 @@ class MainWindow(QMainWindow):
for group in self.result.review_groups:
selected = f"; selected: {group.selected.name}" if group.selected else "; no automatic selection"
self.duplicates.addItem(f"[{group.classification.upper()}] {len(group.records)} record(s) — {group.reason}{selected}")
self.duplicates.setCurrentRow(0 if self.result.review_groups else -1)
self.log_view.setPlainText("\n".join(self.audit.entries))
except (OSError, ValueError, ImportError) as exc:
QMessageBox.critical(self, "Import failed", str(exc))
@@ -92,7 +106,9 @@ class MainWindow(QMainWindow):
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)
records = self.resolved_records()
output = export_json(records, path) if fmt == "json" else export_csv(records, path)
self.audit.record("DECISIONS", f"Export used {len(records)} record(s) after duplicate review decisions: {self.review_decisions}")
self.audit.record("EXPORT", f"Wrote merged {fmt.upper()} export to {output}")
self.audit.save()
self.log_view.setPlainText("\n".join(self.audit.entries))
@@ -101,16 +117,84 @@ class MainWindow(QMainWindow):
except OSError as exc:
QMessageBox.critical(self, "Export failed", str(exc))
def resolved_records(self):
"""Apply review choices to the records that will be exported."""
if not self.result:
return []
excluded = set()
for index, group in enumerate(self.result.review_groups):
if group.classification == "exact":
continue
decision = self.review_decisions.get(index)
if decision is None and group.selected is not None:
decision = group.records.index(group.selected)
if isinstance(decision, int):
excluded.update(id(record) for record in group.records if record is not group.records[decision])
return [record for record in self.result.included if id(record) not in excluded]
def choose_selected_record(self) -> None:
row = self.duplicates.currentRow()
if not self.result or row < 0: return
group = self.result.review_groups[row]
if group.classification == "exact":
self.duplicate_status.setText("Exact duplicates are removed automatically.")
return
selected_index = self.decision_choice.currentData()
selected = group.records[selected_index] if isinstance(selected_index, int) else (group.selected or group.records[0])
self.review_decisions[row] = group.records.index(selected)
self.audit.record("DECISION", f"Group {row}: keep {selected.source_file}:{selected.source_index}")
self.refresh_duplicate_status(row)
def choose_both_records(self) -> None:
row = self.duplicates.currentRow()
if not self.result or row < 0: return
group = self.result.review_groups[row]
if group.classification == "exact":
self.duplicate_status.setText("Exact duplicates are removed automatically.")
return
self.review_decisions[row] = "both"
self.audit.record("DECISION", f"Group {row}: keep all {len(group.records)} records")
self.refresh_duplicate_status(row)
def refresh_duplicate_status(self, row: int) -> None:
decision = self.review_decisions.get(row, "automatic/newest" if self.result.review_groups[row].selected else "both")
self.duplicate_status.setText(f"Export decision: {decision}")
def show_duplicate_detail(self, row: int) -> None:
if not self.result or row < 0 or row >= len(self.result.review_groups):
self.duplicate_detail.clear(); return
self.duplicate_table.clear(); self.duplicate_table.setRowCount(0); return
group = self.result.review_groups[row]
lines = [f"Classification: {group.classification}", f"Reason: {group.reason}"]
if group.selected:
lines.append(f"Automatic preference: {group.selected.source_file}:{group.selected.source_index} ({group.selected.name})")
for index, record in enumerate(group.records, 1):
lines.extend(["", f"Record {index}: {record.source_file}:{record.source_index} — {record.name}", repr(record.data)])
self.duplicate_detail.setPlainText("\n".join(lines))
self.decision_choice.blockSignals(True)
self.decision_choice.clear()
self.decision_choice.addItem("Choose a record…", None)
for index, record in enumerate(group.records):
label = f"Keep record {index + 1}: {record.source_file}:{record.source_index}"
if group.selected is record:
label += " (newest/recommended)"
self.decision_choice.addItem(label, index)
decision = self.review_decisions.get(row)
if isinstance(decision, int):
self.decision_choice.setCurrentIndex(decision + 1)
elif group.selected is not None:
self.decision_choice.setCurrentIndex(group.records.index(group.selected) + 1)
self.decision_choice.blockSignals(False)
field_rows = ["Name", "Folder", "Username", "Password", "URLs", "TOTP", "Notes", "Custom fields", "Revision date", "Source"]
self.duplicate_table.clear(); self.duplicate_table.setRowCount(len(field_rows)); self.duplicate_table.setColumnCount(len(group.records));
self.duplicate_table.setVerticalHeaderLabels(field_rows)
self.duplicate_table.setHorizontalHeaderLabels([f"Record {i + 1}: {r.source_file}" for i, r in enumerate(group.records)])
for column, record in enumerate(group.records):
login = record.data.get("login") or {}
uris = login.get("uris") or []
values = [record.name, record.data.get("folderId") or "(none)", login.get("username", ""), login.get("password", ""),
"\n".join(str(uri.get("uri", "")) for uri in uris), login.get("totp", ""), record.data.get("notes", ""),
repr(record.data.get("fields", [])), record.data.get("revisionDate", ""), f"{record.source_file}:{record.source_index}"]
for row_index, value in enumerate(values):
item = QTableWidgetItem(str(value))
if any(str(value) != str(other.data.get("name", "")) for other in group.records if other is not record) and row_index == 0:
item.setBackground("#fff2cc")
self.duplicate_table.setItem(row_index, column, item)
self.duplicate_table.resizeColumnsToContents(); self.duplicate_table.resizeRowsToContents()
self.refresh_duplicate_status(row)
def run() -> int: