Implement offline merge and review core
This commit is contained in:
+4
-3
@@ -1,6 +1,7 @@
|
||||
"""Application entry point placeholder for the DataLoom desktop UI."""
|
||||
"""Application entry point for the DataLoom desktop UI."""
|
||||
|
||||
|
||||
def main() -> int:
|
||||
"""Start the desktop application once the UI is implemented."""
|
||||
raise NotImplementedError("DataLoom UI is not implemented yet")
|
||||
"""Start the desktop application."""
|
||||
from dataloom.ui.main_window import run
|
||||
return run()
|
||||
|
||||
@@ -0,0 +1 @@
|
||||
"""Core import, merge, review, and export services."""
|
||||
@@ -0,0 +1,22 @@
|
||||
"""Plain-text audit trail. Secret-bearing logging is opt-in at construction."""
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
class AuditLog:
|
||||
def __init__(self, path: str | Path, include_secrets: bool = True) -> None:
|
||||
self.path = Path(path)
|
||||
self.include_secrets = include_secrets
|
||||
self.entries: list[str] = []
|
||||
|
||||
def record(self, action: str, detail: str, sensitive_detail: str = "") -> None:
|
||||
timestamp = datetime.now(timezone.utc).isoformat()
|
||||
suffix = f" | {sensitive_detail}" if self.include_secrets and sensitive_detail else ""
|
||||
self.entries.append(f"{timestamp} | {action} | {detail}{suffix}")
|
||||
|
||||
def save(self) -> Path:
|
||||
self.path.parent.mkdir(parents=True, exist_ok=True)
|
||||
self.path.write_text("\n".join(self.entries) + "\n", encoding="utf-8")
|
||||
return self.path
|
||||
@@ -0,0 +1,30 @@
|
||||
"""Same-format merged output exporters."""
|
||||
|
||||
import csv
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Iterable
|
||||
|
||||
from .models import Record
|
||||
|
||||
|
||||
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")
|
||||
return path
|
||||
|
||||
|
||||
def export_csv(records: Iterable[Record], destination: str | Path) -> Path:
|
||||
path = Path(destination)
|
||||
fields = ["type", "name", "notes", "favorite", "login_username", "login_password", "login_totp", "login_uri"]
|
||||
with path.open("w", newline="", encoding="utf-8") as stream:
|
||||
writer = csv.DictWriter(stream, fieldnames=fields)
|
||||
writer.writeheader()
|
||||
for record in records:
|
||||
login = record.data.get("login") or {}
|
||||
uris = login.get("uris") or []
|
||||
writer.writerow({"type": record.data.get("type", ""), "name": record.data.get("name", ""), "notes": record.data.get("notes", ""),
|
||||
"favorite": record.data.get("favorite", False), "login_username": login.get("username", ""),
|
||||
"login_password": login.get("password", ""), "login_totp": login.get("totp", ""),
|
||||
"login_uri": (uris[0].get("uri", "") if uris else "")})
|
||||
return path
|
||||
@@ -0,0 +1,60 @@
|
||||
"""Duplicate classification and conservative merging."""
|
||||
|
||||
from collections import defaultdict
|
||||
from datetime import datetime
|
||||
from typing import Iterable
|
||||
|
||||
from .models import DuplicateGroup, MergeResult, Record
|
||||
from .normalize import identity_key, normalize_url, record_fingerprint
|
||||
|
||||
|
||||
def _revision(record: Record) -> datetime | None:
|
||||
value = record.data.get("revisionDate")
|
||||
if not isinstance(value, str):
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(value.replace("Z", "+00:00"))
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def merge_records(records: Iterable[Record]) -> MergeResult:
|
||||
all_records = list(records)
|
||||
by_fingerprint: dict[str, list[Record]] = defaultdict(list)
|
||||
for record in all_records:
|
||||
# For logins, the user's agreed exact-match rule is type + username +
|
||||
# password + normalized URI(s). Do not let Bitwarden item IDs, names,
|
||||
# folders, notes, or revision dates prevent an exact match.
|
||||
login = record.data.get("login")
|
||||
if isinstance(login, dict):
|
||||
urls = tuple(sorted(normalize_url(str(uri.get("uri") or ""))
|
||||
for uri in login.get("uris", [])
|
||||
if isinstance(uri, dict)))
|
||||
fingerprint = repr((record.item_type,
|
||||
str(login.get("username") or "").casefold().strip(),
|
||||
str(login.get("password") or ""), urls))
|
||||
else:
|
||||
fingerprint = record_fingerprint(record.data)
|
||||
by_fingerprint[fingerprint].append(record)
|
||||
|
||||
included: list[Record] = []
|
||||
removed: list[Record] = []
|
||||
review: list[DuplicateGroup] = []
|
||||
survivors: list[Record] = []
|
||||
for matches in by_fingerprint.values():
|
||||
survivors.append(matches[0])
|
||||
removed.extend(matches[1:])
|
||||
if len(matches) > 1:
|
||||
review.append(DuplicateGroup("exact", matches, "Identical normalized record content", matches[0]))
|
||||
|
||||
by_identity: dict[tuple[str, str, str], list[Record]] = defaultdict(list)
|
||||
for record in survivors:
|
||||
by_identity[identity_key(record.data)].append(record)
|
||||
included.append(record)
|
||||
for matches in by_identity.values():
|
||||
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
|
||||
review.append(DuplicateGroup("likely", matches, "Same type, normalized URL, and username; values differ", selected))
|
||||
return MergeResult(included, removed, review)
|
||||
@@ -0,0 +1,37 @@
|
||||
"""Internal models used by DataLoom."""
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
|
||||
|
||||
@dataclass
|
||||
class Record:
|
||||
data: dict[str, Any]
|
||||
source_file: str
|
||||
source_index: int
|
||||
source_id: str | None = None
|
||||
source_format: str = "json"
|
||||
|
||||
@property
|
||||
def item_type(self) -> str:
|
||||
value = self.data.get("type", "unknown")
|
||||
return str(value)
|
||||
|
||||
@property
|
||||
def name(self) -> str:
|
||||
return str(self.data.get("name", ""))
|
||||
|
||||
|
||||
@dataclass
|
||||
class DuplicateGroup:
|
||||
classification: str
|
||||
records: list[Record] = field(default_factory=list)
|
||||
reason: str = ""
|
||||
selected: Record | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
class MergeResult:
|
||||
included: list[Record]
|
||||
removed_exact: list[Record]
|
||||
review_groups: list[DuplicateGroup]
|
||||
@@ -0,0 +1,62 @@
|
||||
"""Conservative, deterministic comparison normalization."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from urllib.parse import urlsplit, urlunsplit
|
||||
import json
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
|
||||
def normalize_text(value: Any) -> Any:
|
||||
if isinstance(value, str):
|
||||
return value.casefold().strip()
|
||||
if isinstance(value, list):
|
||||
return [normalize_text(item) for item in value]
|
||||
if isinstance(value, Mapping):
|
||||
return {str(k): normalize_text(v) for k, v in sorted(value.items())}
|
||||
return value
|
||||
|
||||
|
||||
def normalize_url(value: str) -> str:
|
||||
value = value.strip().casefold()
|
||||
if not value:
|
||||
return ""
|
||||
parsed = urlsplit(value if "://" in value else f"https://{value}")
|
||||
host = parsed.hostname or ""
|
||||
if host.startswith("www."):
|
||||
host = host[4:]
|
||||
port = parsed.port
|
||||
netloc = host
|
||||
if port and not ((parsed.scheme == "http" and port == 80) or (parsed.scheme == "https" and port == 443)):
|
||||
netloc = f"{host}:{port}"
|
||||
path = re.sub(r"/{2,}", "/", parsed.path).rstrip("/") or "/"
|
||||
return urlunsplit((parsed.scheme, netloc, path, parsed.query, parsed.fragment))
|
||||
|
||||
|
||||
def normalized_record(record: Mapping[str, Any]) -> dict[str, Any]:
|
||||
"""Return comparison data while retaining the original record elsewhere."""
|
||||
result = normalize_text(record)
|
||||
login = result.get("login") if isinstance(result, dict) else None
|
||||
if isinstance(login, dict):
|
||||
uris = login.get("uris")
|
||||
if isinstance(uris, list):
|
||||
for uri in uris:
|
||||
if isinstance(uri, dict) and isinstance(uri.get("uri"), str):
|
||||
uri["uri"] = normalize_url(uri["uri"])
|
||||
uris.sort(key=lambda item: json.dumps(item, sort_keys=True))
|
||||
return result
|
||||
|
||||
|
||||
def record_fingerprint(record: Mapping[str, Any]) -> str:
|
||||
payload = json.dumps(normalized_record(record), sort_keys=True, separators=(",", ":"), ensure_ascii=False)
|
||||
return payload
|
||||
|
||||
|
||||
def identity_key(record: Mapping[str, Any]) -> tuple[str, str, str]:
|
||||
login = record.get("login") or {}
|
||||
username = str(login.get("username") or "").casefold().strip()
|
||||
uris = login.get("uris") or []
|
||||
urls = tuple(sorted(normalize_url(str(uri.get("uri") or "")) for uri in uris if isinstance(uri, Mapping)))
|
||||
return str(record.get("type", "unknown")), username, "|".join(urls)
|
||||
@@ -0,0 +1,44 @@
|
||||
"""Strict same-format Bitwarden JSON and CSV import."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import csv
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .models import Record
|
||||
|
||||
|
||||
class ImportError(ValueError):
|
||||
pass
|
||||
|
||||
|
||||
def detect_format(path: str | Path) -> str:
|
||||
suffix = Path(path).suffix.casefold()
|
||||
if suffix == ".json": return "json"
|
||||
if suffix == ".csv": return "csv"
|
||||
raise ImportError(f"Unsupported file extension: {path}")
|
||||
|
||||
|
||||
def load_file(path: str | Path) -> tuple[str, list[Record]]:
|
||||
path = Path(path)
|
||||
fmt = detect_format(path)
|
||||
if fmt == "json":
|
||||
payload: Any = json.loads(path.read_text(encoding="utf-8"))
|
||||
if payload.get("encrypted") is not False:
|
||||
raise ImportError(f"{path.name} is encrypted or is not a standard unencrypted export")
|
||||
items = payload.get("items")
|
||||
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")
|
||||
records = []
|
||||
for i, row in enumerate(rows, 1):
|
||||
item = {"type": row.get("type", "login"), "name": row.get("name", ""), "notes": row.get("notes", ""),
|
||||
"favorite": row.get("favorite", "false"), "login": {"username": row.get("login_username", ""),
|
||||
"password": row.get("login_password", ""), "totp": row.get("login_totp", ""),
|
||||
"uris": [{"uri": row.get("login_uri", "")} ]}}
|
||||
records.append(Record(item, path.name, i, None, fmt))
|
||||
return fmt, records
|
||||
@@ -0,0 +1 @@
|
||||
"""PySide6 desktop user interface."""
|
||||
@@ -0,0 +1,92 @@
|
||||
"""Initial automated review window for DataLoom."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from PySide6.QtWidgets import (
|
||||
QAbstractItemView, QApplication, QFileDialog, QLabel, QListWidget,
|
||||
QMainWindow, QMessageBox, QPushButton, QTabWidget, QTextEdit,
|
||||
QVBoxLayout, QWidget,
|
||||
)
|
||||
|
||||
from dataloom.core.audit import AuditLog
|
||||
from dataloom.core.merge import merge_records
|
||||
from dataloom.core.parsers import ImportError, load_file
|
||||
|
||||
|
||||
class MainWindow(QMainWindow):
|
||||
def __init__(self) -> None:
|
||||
super().__init__()
|
||||
self.setWindowTitle("DataLoom — Bitwarden Review and Merge")
|
||||
self.resize(1000, 700)
|
||||
self.records = []
|
||||
self.result = None
|
||||
self.audit = None
|
||||
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)
|
||||
tabs = QTabWidget()
|
||||
tabs.addTab(self._page("Import Summary", self.summary), "Import Summary")
|
||||
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…")
|
||||
import_button.clicked.connect(self.import_files)
|
||||
export_button = QPushButton("Export merged vault…")
|
||||
export_button.setEnabled(False)
|
||||
export_button.clicked.connect(self.export_merged)
|
||||
self.export_button = export_button
|
||||
layout = QVBoxLayout(); layout.addWidget(QLabel("DataLoom processes exports locally. Do not import encrypted files.")); layout.addWidget(import_button); layout.addWidget(export_button); layout.addWidget(tabs)
|
||||
root = QWidget(); root.setLayout(layout); self.setCentralWidget(root)
|
||||
|
||||
@staticmethod
|
||||
def _page(title: str, widget: QWidget) -> QWidget:
|
||||
page = QWidget(); layout = QVBoxLayout(page); layout.addWidget(QLabel(f"<h2>{title}</h2>")); layout.addWidget(widget); return page
|
||||
|
||||
def import_files(self) -> None:
|
||||
paths, _ = QFileDialog.getOpenFileNames(self, "Select Bitwarden exports", "", "Bitwarden exports (*.json *.csv)")
|
||||
if not paths: return
|
||||
if QMessageBox.warning(self, "Sensitive unencrypted data", "These files and the requested audit log contain passwords and other secrets. Continue?", QMessageBox.StandardButton.Yes | QMessageBox.StandardButton.No) != QMessageBox.StandardButton.Yes:
|
||||
return
|
||||
try:
|
||||
formats = []; records = []
|
||||
for path in paths:
|
||||
fmt, loaded = load_file(path); formats.append(fmt); records.extend(loaded)
|
||||
if len(set(formats)) != 1: raise ImportError("All selected files must use the same format")
|
||||
self.records = records
|
||||
self.result = merge_records(records)
|
||||
self.audit = AuditLog(Path(paths[0]).parent / "dataloom-audit.log", include_secrets=True)
|
||||
self.audit.record("IMPORT", f"Loaded {len(paths)} {formats[0]} file(s) and {len(records)} record(s)")
|
||||
for record in records:
|
||||
self.audit.record("VIEW", f"{record.source_file} record {record.source_index} ({record.item_type}) {record.name}", repr(record.data))
|
||||
self.audit.record("MERGE", f"Included {len(self.result.included)}; removed {len(self.result.removed_exact)} exact duplicate(s); review groups {len(self.result.review_groups)}")
|
||||
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.duplicates.clear()
|
||||
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.log_view.setPlainText("\n".join(self.audit.entries))
|
||||
except (OSError, ValueError, ImportError) as exc:
|
||||
QMessageBox.critical(self, "Import failed", str(exc))
|
||||
|
||||
def export_merged(self) -> None:
|
||||
if not self.result or not self.records: return
|
||||
fmt = self.records[0].source_format
|
||||
path, _ = QFileDialog.getSaveFileName(self, "Save merged Bitwarden export", "dataloom-merged." + fmt, f"Bitwarden {fmt.upper()} (*.{fmt})")
|
||||
if not path: return
|
||||
try:
|
||||
from dataloom.core.exporters import export_csv, export_json
|
||||
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()
|
||||
self.log_view.setPlainText("\n".join(self.audit.entries))
|
||||
self.summary.append(f"\nMerged output: {output}")
|
||||
QMessageBox.information(self, "Export complete", f"Merged export written to:\n{output}\n\nAudit log:\n{self.audit.path}")
|
||||
except OSError as exc:
|
||||
QMessageBox.critical(self, "Export failed", str(exc))
|
||||
|
||||
|
||||
def run() -> int:
|
||||
app = QApplication([])
|
||||
window = MainWindow(); window.show()
|
||||
return app.exec()
|
||||
@@ -0,0 +1,42 @@
|
||||
"""Core tests; run only in an approved disposable development environment."""
|
||||
|
||||
import json
|
||||
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.parsers import load_file
|
||||
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def test_url_normalization_equates_www_and_trailing_slash():
|
||||
assert normalize_url("https://example.com") == normalize_url("https://www.example.com/")
|
||||
|
||||
|
||||
def test_fixture_import_and_exact_merge():
|
||||
_, left = load_file(FIXTURES / "bitwarden-a.json")
|
||||
_, right = load_file(FIXTURES / "bitwarden-b.json")
|
||||
result = merge_records(left + right)
|
||||
assert len(result.included) == 1
|
||||
assert len(result.removed_exact) == 1
|
||||
|
||||
|
||||
def test_exact_login_match_ignores_name_and_revision():
|
||||
left = {"id": "a", "type": 1, "name": "Old name", "revisionDate": "2026-01-01T00:00:00Z",
|
||||
"login": {"username": "U", "password": "P", "uris": [{"uri": "https://example.com"}]}}
|
||||
right = {"id": "b", "type": 1, "name": "New name", "revisionDate": "2026-02-01T00:00:00Z",
|
||||
"login": {"username": "u", "password": "P", "uris": [{"uri": "https://www.example.com/"}]}}
|
||||
result = merge_records([Record(left, "a.json", 1), Record(right, "b.json", 1)])
|
||||
assert len(result.included) == 1
|
||||
assert len(result.removed_exact) == 1
|
||||
|
||||
|
||||
def test_conflicting_passwords_are_retained_for_review():
|
||||
base = {"type": 1, "name": "X", "login": {"username": "u", "password": "a", "uris": [{"uri": "https://x.test"}]}}
|
||||
other = json.loads(json.dumps(base)); other["login"]["password"] = "b"
|
||||
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)
|
||||
Reference in New Issue
Block a user