Mateusz Gruszczyński committed
1 parent
02d0b13ec1
commit
db0d230d3b
12 files changed
+1146
-17
No files matched your search
Executable
+942
@@ -0,0 +1,942 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Standalone Roundcube -> RustBox contacts/identities migrator.
|
||||
|
||||
The tool does not import or start RustBox. By default it reads RustBox's database
|
||||
configuration from the same TOML file as the application, detects SQLite/MySQL/
|
||||
PostgreSQL, and writes directly to the already provisioned per-account tables.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import hashlib
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import sys
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Any, Iterable, Mapping, Sequence
|
||||
from urllib.parse import parse_qs, quote, unquote, urlparse
|
||||
|
||||
try:
|
||||
import tomllib
|
||||
except ModuleNotFoundError: # pragma: no cover - Python < 3.11
|
||||
try:
|
||||
import tomli as tomllib # type: ignore[no-redef]
|
||||
except ModuleNotFoundError as exc: # pragma: no cover
|
||||
raise SystemExit(
|
||||
"Python 3.11+ is recommended. On older Python install 'tomli'."
|
||||
) from exc
|
||||
|
||||
|
||||
EMAIL_RE = re.compile(
|
||||
r"[a-z0-9.!#$%&'*+/=?^_`{|}~-]+@"
|
||||
r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?"
|
||||
r"(?:\.[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?)+",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
IDENT_RE = re.compile(r"^[A-Za-z0-9_]+$")
|
||||
BREAK_RE = re.compile(r"<br\s*/?>|</p\s*>|</div\s*>", re.IGNORECASE)
|
||||
TAG_RE = re.compile(r"<[^>]*>", re.DOTALL)
|
||||
|
||||
|
||||
@dataclass
|
||||
class Contact:
|
||||
id: str
|
||||
email: str
|
||||
name: str
|
||||
company: str
|
||||
notes: str
|
||||
phone: str
|
||||
job_title: str
|
||||
address: str
|
||||
website: str
|
||||
birthday: str
|
||||
emails_json: str
|
||||
phones_json: str
|
||||
addresses_json: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class Identity:
|
||||
display_name: str
|
||||
email: str
|
||||
signature_html: str
|
||||
signature_text: str
|
||||
is_default: bool
|
||||
reply_to: str
|
||||
bcc: str
|
||||
organization: str
|
||||
|
||||
|
||||
@dataclass
|
||||
class VCard:
|
||||
name: str = ""
|
||||
emails: list[str] | None = None
|
||||
phones: list[str] | None = None
|
||||
addresses: list[str] | None = None
|
||||
company: str = ""
|
||||
notes: str = ""
|
||||
job_title: str = ""
|
||||
website: str = ""
|
||||
birthday: str = ""
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self.emails = self.emails or []
|
||||
self.phones = self.phones or []
|
||||
self.addresses = self.addresses or []
|
||||
|
||||
|
||||
class Db:
|
||||
"""Small DB-API adapter for SQLite, PyMySQL and psycopg/psycopg2."""
|
||||
|
||||
def __init__(self, backend: str, conn: Any):
|
||||
self.backend = backend
|
||||
self.conn = conn
|
||||
|
||||
@property
|
||||
def placeholder(self) -> str:
|
||||
return "?" if self.backend == "sqlite" else "%s"
|
||||
|
||||
def ident(self, value: str) -> str:
|
||||
validate_identifier(value, "SQL identifier")
|
||||
if self.backend == "mysql":
|
||||
return f"`{value}`"
|
||||
return f'"{value}"'
|
||||
|
||||
def execute(self, sql: str, params: Sequence[Any] = ()) -> Any:
|
||||
cur = self.conn.cursor()
|
||||
cur.execute(sql, tuple(params))
|
||||
return cur
|
||||
|
||||
def fetchone(self, sql: str, params: Sequence[Any] = ()) -> Any | None:
|
||||
cur = self.execute(sql, params)
|
||||
row = cur.fetchone()
|
||||
cur.close()
|
||||
return row
|
||||
|
||||
def fetchall(self, sql: str, params: Sequence[Any] = ()) -> list[Any]:
|
||||
cur = self.execute(sql, params)
|
||||
rows = list(cur.fetchall())
|
||||
cur.close()
|
||||
return rows
|
||||
|
||||
def commit(self) -> None:
|
||||
self.conn.commit()
|
||||
|
||||
def rollback(self) -> None:
|
||||
self.conn.rollback()
|
||||
|
||||
def close(self) -> None:
|
||||
self.conn.close()
|
||||
|
||||
|
||||
def parser() -> argparse.ArgumentParser:
|
||||
p = argparse.ArgumentParser(
|
||||
description="Migrate Roundcube MySQL contacts and identities directly into RustBox.",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Default RustBox config precedence:
|
||||
--config > RUSTBOX_CONFIG > /etc/rustbox/rustbox.toml
|
||||
|
||||
Target DB precedence:
|
||||
--target-backend/--target-url > DATABASE_BACKEND/DATABASE_URL > config.toml
|
||||
|
||||
Examples:
|
||||
# all users, RustBox target read from config.toml
|
||||
scripts/migrate_roundcube.py --roundcube-url mysql://rc:secret@db/roundcube --dry-run
|
||||
scripts/migrate_roundcube.py --roundcube-url mysql://rc:secret@db/roundcube
|
||||
|
||||
# override both sides
|
||||
scripts/migrate_roundcube.py \\
|
||||
--config ./config.toml \\
|
||||
--target-backend mysql \\
|
||||
--target-url mysql://rustbox:secret@db/rustbox \\
|
||||
--roundcube-host rc-db --roundcube-database roundcube \\
|
||||
--roundcube-db-user rc --roundcube-db-password secret
|
||||
|
||||
Dependencies:
|
||||
MySQL source/target: PyMySQL (pip install pymysql)
|
||||
PostgreSQL target: psycopg (pip install 'psycopg[binary]')
|
||||
""",
|
||||
)
|
||||
p.add_argument("--config", help="RustBox TOML config path")
|
||||
p.add_argument(
|
||||
"--target-backend",
|
||||
choices=("sqlite", "mysql", "pgsql", "postgres", "postgresql"),
|
||||
help="override RustBox DB backend",
|
||||
)
|
||||
p.add_argument("--target-url", help="override RustBox DB URL/path")
|
||||
|
||||
rc = p.add_argument_group("Roundcube MySQL source")
|
||||
rc.add_argument("--roundcube-url", help="mysql://user:password@host:3306/database")
|
||||
rc.add_argument("--roundcube-host", help="Roundcube MySQL host")
|
||||
rc.add_argument("--roundcube-port", type=int, default=None, help="Roundcube MySQL port (default 3306)")
|
||||
rc.add_argument("--roundcube-database", help="Roundcube database name")
|
||||
rc.add_argument("--roundcube-db-user", help="Roundcube database user")
|
||||
rc.add_argument("--roundcube-db-password", help="Roundcube database password")
|
||||
rc.add_argument("--roundcube-prefix", default=os.getenv("ROUNDCUBE_TABLE_PREFIX", ""), help="table prefix, e.g. rc_")
|
||||
|
||||
mode = p.add_mutually_exclusive_group()
|
||||
mode.add_argument("--all-users", action="store_true", help="migrate every matching Roundcube user (default)")
|
||||
mode.add_argument("--roundcube-account", help="migrate only this Roundcube users.username")
|
||||
|
||||
p.add_argument("--rustbox-user", help="single-user target login if different from Roundcube username")
|
||||
p.add_argument("--user-hash", help="single-user RustBox user hash override")
|
||||
p.add_argument("--target-prefix", help="single-user RustBox tenant prefix override")
|
||||
p.add_argument("--dry-run", action="store_true", help="read/map only; do not write")
|
||||
p.add_argument("--fail-fast", action="store_true", help="stop all-users migration after first user failure")
|
||||
return p
|
||||
|
||||
|
||||
def validate_identifier(value: str, label: str, allow_empty: bool = False) -> None:
|
||||
if value == "" and allow_empty:
|
||||
return
|
||||
if not value or not IDENT_RE.fullmatch(value):
|
||||
raise ValueError(f"{label} may contain only ASCII letters, digits and underscore")
|
||||
|
||||
|
||||
def normalize_backend(raw: str | None) -> str | None:
|
||||
if raw is None:
|
||||
return None
|
||||
value = raw.strip().lower()
|
||||
aliases = {"postgres": "pgsql", "postgresql": "pgsql"}
|
||||
value = aliases.get(value, value)
|
||||
if value not in {"sqlite", "mysql", "pgsql"}:
|
||||
raise ValueError(f"unsupported database backend: {raw}")
|
||||
return value
|
||||
|
||||
|
||||
def infer_backend_from_url(raw: str | None) -> str | None:
|
||||
if not raw:
|
||||
return None
|
||||
value = raw.strip().lower()
|
||||
if value.startswith("mysql://"):
|
||||
return "mysql"
|
||||
if value.startswith("postgres://") or value.startswith("postgresql://"):
|
||||
return "pgsql"
|
||||
if value.startswith("sqlite:"):
|
||||
return "sqlite"
|
||||
if "://" not in value:
|
||||
return "sqlite"
|
||||
return None
|
||||
|
||||
|
||||
def load_config(path: Path) -> dict[str, Any]:
|
||||
if not path.is_file():
|
||||
return {}
|
||||
with path.open("rb") as fh:
|
||||
data = tomllib.load(fh)
|
||||
if not isinstance(data, dict):
|
||||
raise ValueError(f"invalid TOML root in {path}")
|
||||
return data
|
||||
|
||||
|
||||
def sqlite_path(raw: str) -> Path:
|
||||
value = raw.strip()
|
||||
if value.startswith("sqlite://"):
|
||||
value = value[len("sqlite://") :]
|
||||
elif value.startswith("sqlite:"):
|
||||
value = value[len("sqlite:") :]
|
||||
value = value.split("?", 1)[0].strip()
|
||||
if not value or value == ":memory:":
|
||||
raise ValueError("SQLite target must point to a persistent database file")
|
||||
return Path(value)
|
||||
|
||||
|
||||
def resolve_target(args: argparse.Namespace) -> tuple[Path, dict[str, Any], str, str]:
|
||||
config_path = Path(
|
||||
args.config
|
||||
or os.getenv("RUSTBOX_CONFIG")
|
||||
or "/etc/rustbox/rustbox.toml"
|
||||
)
|
||||
config = load_config(config_path)
|
||||
|
||||
env_backend = os.getenv("DATABASE_BACKEND") or os.getenv("RUSTBOX_DATABASE_BACKEND")
|
||||
env_url = os.getenv("DATABASE_URL") or os.getenv("RUSTBOX_DATABASE_URL")
|
||||
env_data_dir = os.getenv("DATA_DIR")
|
||||
|
||||
explicit_url_backend = infer_backend_from_url(args.target_url) if args.target_url else None
|
||||
backend = normalize_backend(
|
||||
args.target_backend
|
||||
or explicit_url_backend
|
||||
or env_backend
|
||||
or config.get("database_backend")
|
||||
or "sqlite"
|
||||
)
|
||||
assert backend is not None
|
||||
|
||||
target_url = args.target_url or env_url or config.get("database_url")
|
||||
if backend == "sqlite":
|
||||
if target_url:
|
||||
target_url = str(sqlite_path(str(target_url)))
|
||||
else:
|
||||
data_dir = str(env_data_dir or config.get("data_dir") or "/data")
|
||||
target_url = str(Path(data_dir) / "rustbox.sqlite")
|
||||
elif not target_url:
|
||||
raise ValueError(
|
||||
f"database_url is required for {backend}; set it in config.toml or pass --target-url"
|
||||
)
|
||||
|
||||
inferred = infer_backend_from_url(str(target_url))
|
||||
if inferred is not None and inferred != backend:
|
||||
raise ValueError(
|
||||
f"target backend is {backend}, but target URL looks like {inferred}; "
|
||||
"override both --target-backend and --target-url together"
|
||||
)
|
||||
if not config and not (args.target_url or env_url):
|
||||
raise FileNotFoundError(
|
||||
f"RustBox config not found: {config_path}; pass --config or --target-url"
|
||||
)
|
||||
return config_path, config, backend, str(target_url)
|
||||
|
||||
|
||||
def build_roundcube_url(args: argparse.Namespace) -> str:
|
||||
env_url = os.getenv("ROUNDCUBE_DATABASE_URL")
|
||||
if args.roundcube_url:
|
||||
return args.roundcube_url
|
||||
if env_url and not any(
|
||||
[
|
||||
args.roundcube_host,
|
||||
args.roundcube_database,
|
||||
args.roundcube_db_user,
|
||||
args.roundcube_db_password,
|
||||
args.roundcube_port,
|
||||
]
|
||||
):
|
||||
return env_url
|
||||
|
||||
host = args.roundcube_host or os.getenv("ROUNDCUBE_HOST")
|
||||
port = args.roundcube_port or int(os.getenv("ROUNDCUBE_PORT", "3306"))
|
||||
database = args.roundcube_database or os.getenv("ROUNDCUBE_DATABASE")
|
||||
user = args.roundcube_db_user or os.getenv("ROUNDCUBE_DB_USER")
|
||||
password = (
|
||||
args.roundcube_db_password
|
||||
if args.roundcube_db_password is not None
|
||||
else os.getenv("ROUNDCUBE_DB_PASSWORD", "")
|
||||
)
|
||||
if not all([host, database, user]):
|
||||
raise ValueError(
|
||||
"Roundcube source is required: pass --roundcube-url or "
|
||||
"--roundcube-host/--roundcube-database/--roundcube-db-user"
|
||||
)
|
||||
return (
|
||||
f"mysql://{quote(str(user), safe='')}:{quote(str(password), safe='')}@"
|
||||
f"{host}:{port}/{quote(str(database), safe='')}"
|
||||
)
|
||||
|
||||
|
||||
def parse_mysql_url(raw: str) -> dict[str, Any]:
|
||||
parsed = urlparse(raw)
|
||||
if parsed.scheme.lower() not in {"mysql", "mariadb"}:
|
||||
raise ValueError("MySQL URL must start with mysql://")
|
||||
database = unquote(parsed.path.lstrip("/"))
|
||||
if not parsed.hostname or not parsed.username or not database:
|
||||
raise ValueError("MySQL URL must contain user, host and database")
|
||||
query = parse_qs(parsed.query)
|
||||
out: dict[str, Any] = {
|
||||
"host": parsed.hostname,
|
||||
"port": parsed.port or 3306,
|
||||
"user": unquote(parsed.username),
|
||||
"password": unquote(parsed.password or ""),
|
||||
"database": database,
|
||||
"charset": query.get("charset", ["utf8mb4"])[0],
|
||||
"autocommit": False,
|
||||
}
|
||||
if "unix_socket" in query:
|
||||
out["unix_socket"] = query["unix_socket"][0]
|
||||
if "connect_timeout" in query:
|
||||
out["connect_timeout"] = int(query["connect_timeout"][0])
|
||||
return out
|
||||
|
||||
|
||||
def connect_mysql(raw: str, *, dict_rows: bool) -> Db:
|
||||
try:
|
||||
import pymysql # type: ignore
|
||||
except ModuleNotFoundError as exc:
|
||||
raise RuntimeError(
|
||||
"PyMySQL is required for Roundcube/MySQL. Install: python3 -m pip install pymysql"
|
||||
) from exc
|
||||
kwargs = parse_mysql_url(raw)
|
||||
if dict_rows:
|
||||
kwargs["cursorclass"] = pymysql.cursors.DictCursor
|
||||
conn = pymysql.connect(**kwargs)
|
||||
return Db("mysql", conn)
|
||||
|
||||
|
||||
def connect_pgsql(raw: str) -> Db:
|
||||
try:
|
||||
import psycopg # type: ignore
|
||||
|
||||
return Db("pgsql", psycopg.connect(raw, autocommit=False))
|
||||
except ModuleNotFoundError:
|
||||
try:
|
||||
import psycopg2 # type: ignore
|
||||
|
||||
return Db("pgsql", psycopg2.connect(raw))
|
||||
except ModuleNotFoundError as exc:
|
||||
raise RuntimeError(
|
||||
"PostgreSQL target requires psycopg. Install: python3 -m pip install 'psycopg[binary]'"
|
||||
) from exc
|
||||
|
||||
|
||||
def connect_target(backend: str, raw: str) -> Db:
|
||||
if backend == "sqlite":
|
||||
path = sqlite_path(raw)
|
||||
conn = sqlite3.connect(path, timeout=10)
|
||||
conn.execute("PRAGMA foreign_keys=ON")
|
||||
return Db("sqlite", conn)
|
||||
if backend == "mysql":
|
||||
return connect_mysql(raw, dict_rows=False)
|
||||
if backend == "pgsql":
|
||||
return connect_pgsql(raw)
|
||||
raise ValueError(f"unsupported target backend: {backend}")
|
||||
|
||||
|
||||
def row_first(row: Any) -> Any:
|
||||
if row is None:
|
||||
return None
|
||||
if isinstance(row, Mapping):
|
||||
return next(iter(row.values()))
|
||||
return row[0]
|
||||
|
||||
|
||||
def row_tuple(row: Any) -> tuple[Any, ...]:
|
||||
if isinstance(row, Mapping):
|
||||
return tuple(row.values())
|
||||
return tuple(row)
|
||||
|
||||
|
||||
def hash_user(value: str) -> str:
|
||||
return hashlib.sha256(value.encode("utf-8")).hexdigest()
|
||||
|
||||
|
||||
def dedupe(values: Iterable[str]) -> list[str]:
|
||||
result: list[str] = []
|
||||
seen: set[str] = set()
|
||||
for raw in values:
|
||||
value = str(raw).strip()
|
||||
if not value:
|
||||
continue
|
||||
key = value.lower()
|
||||
if key not in seen:
|
||||
seen.add(key)
|
||||
result.append(value)
|
||||
return result
|
||||
|
||||
|
||||
def first_nonempty(*values: str) -> str:
|
||||
for value in values:
|
||||
if value and value.strip():
|
||||
return value.strip()
|
||||
return ""
|
||||
|
||||
|
||||
def row_mapping(row: Any) -> Mapping[str, Any]:
|
||||
if isinstance(row, Mapping):
|
||||
return row
|
||||
if isinstance(row, sqlite3.Row):
|
||||
return {key: row[key] for key in row.keys()}
|
||||
raise TypeError("Roundcube query did not return named columns")
|
||||
|
||||
|
||||
def row_text(row: Mapping[str, Any], name: str) -> str:
|
||||
value = row.get(name)
|
||||
if value is None:
|
||||
return ""
|
||||
if isinstance(value, bytes):
|
||||
return value.decode("utf-8", errors="replace")
|
||||
return str(value)
|
||||
|
||||
|
||||
def row_int(row: Mapping[str, Any], name: str) -> int | None:
|
||||
value = row.get(name)
|
||||
if value is None or value == "":
|
||||
return None
|
||||
try:
|
||||
return int(value)
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def unescape_vcard(value: str) -> str:
|
||||
return (
|
||||
value.replace("\\n", "\n")
|
||||
.replace("\\N", "\n")
|
||||
.replace("\\,", ",")
|
||||
.replace("\\;", ";")
|
||||
.replace("\\\\", "\\")
|
||||
.strip()
|
||||
)
|
||||
|
||||
|
||||
def parse_vcard(raw: str) -> VCard:
|
||||
out = VCard()
|
||||
if not raw.strip():
|
||||
return out
|
||||
unfolded: list[str] = []
|
||||
for line in raw.replace("\r\n", "\n").replace("\r", "\n").split("\n"):
|
||||
if line.startswith((" ", "\t")) and unfolded:
|
||||
unfolded[-1] += line.lstrip()
|
||||
else:
|
||||
unfolded.append(line)
|
||||
for line in unfolded:
|
||||
if ":" not in line:
|
||||
continue
|
||||
left, raw_value = line.split(":", 1)
|
||||
prop = left.split(";", 1)[0].rsplit(".", 1)[-1].upper()
|
||||
value = unescape_vcard(raw_value)
|
||||
if prop == "FN" and not out.name:
|
||||
out.name = value
|
||||
elif prop == "EMAIL":
|
||||
out.emails.append(value)
|
||||
elif prop == "TEL":
|
||||
out.phones.append(value)
|
||||
elif prop == "ADR":
|
||||
address = ", ".join(part.strip() for part in value.split(";") if part.strip())
|
||||
if address:
|
||||
out.addresses.append(address)
|
||||
elif prop == "ORG" and not out.company:
|
||||
out.company = " / ".join(part.strip() for part in value.split(";") if part.strip())
|
||||
elif prop == "NOTE" and not out.notes:
|
||||
out.notes = value
|
||||
elif prop == "TITLE" and not out.job_title:
|
||||
out.job_title = value
|
||||
elif prop == "URL" and not out.website:
|
||||
out.website = value
|
||||
elif prop == "BDAY" and not out.birthday:
|
||||
out.birthday = value
|
||||
out.emails = dedupe(out.emails)
|
||||
out.phones = dedupe(out.phones)
|
||||
out.addresses = dedupe(out.addresses)
|
||||
return out
|
||||
|
||||
|
||||
def email_candidates(raw: str) -> list[str]:
|
||||
return dedupe(match.group(0) for match in EMAIL_RE.finditer(raw or ""))
|
||||
|
||||
|
||||
def stable_contact_id(roundcube_user: str, source_id: str, email_addr: str) -> str:
|
||||
stable_key = source_id.strip() or email_addr
|
||||
digest = hashlib.sha256(f"{roundcube_user}:{stable_key}".encode("utf-8")).hexdigest()
|
||||
return f"rc-{digest[:32]}"
|
||||
|
||||
|
||||
def plain_to_html(value: str) -> str:
|
||||
return html.escape(value, quote=True).replace("\r\n", "<br>").replace("\n", "<br>")
|
||||
|
||||
|
||||
def html_to_plain(value: str) -> str:
|
||||
text = BREAK_RE.sub("\n", value)
|
||||
text = TAG_RE.sub("", text)
|
||||
return html.unescape(text.replace(" ", " ")).strip()
|
||||
|
||||
|
||||
def source_table(prefix: str, name: str) -> str:
|
||||
validate_identifier(prefix, "Roundcube table prefix", allow_empty=True)
|
||||
validate_identifier(name, "Roundcube table name")
|
||||
return f"`{prefix}{name}`"
|
||||
|
||||
|
||||
def load_roundcube_users(source: Db, prefix: str) -> list[str]:
|
||||
table = source_table(prefix, "users")
|
||||
rows = source.fetchall(
|
||||
f"SELECT username FROM {table} WHERE username IS NOT NULL AND username<>'' ORDER BY user_id"
|
||||
)
|
||||
return [str(row_first(row)) for row in rows]
|
||||
|
||||
|
||||
def load_roundcube_data(source: Db, prefix: str, username: str) -> tuple[list[Contact], list[Identity], int]:
|
||||
users = source_table(prefix, "users")
|
||||
contacts_table = source_table(prefix, "contacts")
|
||||
identities_table = source_table(prefix, "identities")
|
||||
ph = source.placeholder
|
||||
|
||||
row = source.fetchone(f"SELECT user_id FROM {users} WHERE username={ph} LIMIT 1", [username])
|
||||
if row is None:
|
||||
raise ValueError(f"Roundcube user not found: {username}")
|
||||
user_id = row_first(row)
|
||||
|
||||
contact_rows = source.fetchall(f"SELECT * FROM {contacts_table} WHERE user_id={ph}", [user_id])
|
||||
contacts: list[Contact] = []
|
||||
skipped_contacts = 0
|
||||
for raw_row in contact_rows:
|
||||
row_map = row_mapping(raw_row)
|
||||
if row_int(row_map, "del") not in (None, 0):
|
||||
continue
|
||||
vcard = parse_vcard(row_text(row_map, "vcard"))
|
||||
emails = list(vcard.emails)
|
||||
emails.extend(email_candidates(row_text(row_map, "email")))
|
||||
emails = dedupe(email_candidates(" ".join(emails)))
|
||||
if not emails:
|
||||
skipped_contacts += 1
|
||||
continue
|
||||
primary_email = emails[0]
|
||||
|
||||
full_name = " ".join(
|
||||
part for part in [row_text(row_map, "firstname"), row_text(row_map, "surname")] if part.strip()
|
||||
)
|
||||
name = first_nonempty(vcard.name, row_text(row_map, "name"), full_name, primary_email)
|
||||
company = first_nonempty(vcard.company, row_text(row_map, "organization"), row_text(row_map, "company"))
|
||||
notes = first_nonempty(vcard.notes, row_text(row_map, "notes"))
|
||||
job_title = first_nonempty(vcard.job_title, row_text(row_map, "job_title"), row_text(row_map, "title"))
|
||||
website = first_nonempty(vcard.website, row_text(row_map, "website"))
|
||||
birthday = first_nonempty(vcard.birthday, row_text(row_map, "birthday"))
|
||||
|
||||
phones = list(vcard.phones)
|
||||
direct_phone = row_text(row_map, "phone")
|
||||
if direct_phone.strip():
|
||||
phones.append(direct_phone)
|
||||
phones = dedupe(phones)
|
||||
|
||||
addresses = list(vcard.addresses)
|
||||
direct_address = row_text(row_map, "address")
|
||||
if direct_address.strip():
|
||||
addresses.append(direct_address)
|
||||
addresses = dedupe(addresses)
|
||||
|
||||
source_id = first_nonempty(row_text(row_map, "contact_id"), primary_email)
|
||||
contacts.append(
|
||||
Contact(
|
||||
id=stable_contact_id(username, source_id, primary_email),
|
||||
email=primary_email,
|
||||
name=name,
|
||||
company=company,
|
||||
notes=notes,
|
||||
phone=phones[0] if phones else "",
|
||||
job_title=job_title,
|
||||
address=addresses[0] if addresses else "",
|
||||
website=website,
|
||||
birthday=birthday,
|
||||
emails_json=json.dumps(emails[1:], ensure_ascii=False, separators=(",", ":")),
|
||||
phones_json=json.dumps(phones[1:], ensure_ascii=False, separators=(",", ":")),
|
||||
addresses_json=json.dumps(addresses[1:], ensure_ascii=False, separators=(",", ":")),
|
||||
)
|
||||
)
|
||||
|
||||
identity_rows = source.fetchall(f"SELECT * FROM {identities_table} WHERE user_id={ph}", [user_id])
|
||||
identities: list[Identity] = []
|
||||
for raw_row in identity_rows:
|
||||
row_map = row_mapping(raw_row)
|
||||
if row_int(row_map, "del") not in (None, 0):
|
||||
continue
|
||||
candidates = email_candidates(row_text(row_map, "email"))
|
||||
if not candidates:
|
||||
continue
|
||||
signature = row_text(row_map, "signature")
|
||||
html_signature = (row_int(row_map, "html_signature") or 0) != 0
|
||||
signature_html, signature_text = (
|
||||
(signature, html_to_plain(signature))
|
||||
if html_signature
|
||||
else (plain_to_html(signature), signature)
|
||||
)
|
||||
identities.append(
|
||||
Identity(
|
||||
display_name=row_text(row_map, "name"),
|
||||
email=candidates[0],
|
||||
signature_html=signature_html,
|
||||
signature_text=signature_text,
|
||||
is_default=(row_int(row_map, "standard") or 0) != 0,
|
||||
reply_to=first_nonempty(row_text(row_map, "reply-to"), row_text(row_map, "reply_to")),
|
||||
bcc=row_text(row_map, "bcc"),
|
||||
organization=row_text(row_map, "organization"),
|
||||
)
|
||||
)
|
||||
return contacts, identities, skipped_contacts
|
||||
|
||||
|
||||
def target_prefix_for_hash(target: Db, user_hash: str) -> str | None:
|
||||
ph = target.placeholder
|
||||
row = target.fetchone(
|
||||
f"SELECT table_prefix FROM rustbox_accounts WHERE user_hash={ph}", [user_hash]
|
||||
)
|
||||
if row is None:
|
||||
return None
|
||||
prefix = str(row_first(row))
|
||||
validate_identifier(prefix, "RustBox table prefix")
|
||||
return prefix
|
||||
|
||||
|
||||
def target_account_for_login(target: Db, login: str) -> tuple[str, str, str] | None:
|
||||
exact_hash = hash_user(login)
|
||||
prefix = target_prefix_for_hash(target, exact_hash)
|
||||
if prefix is not None:
|
||||
return exact_hash, prefix, login
|
||||
|
||||
ph = target.placeholder
|
||||
row = target.fetchone(
|
||||
"SELECT a.user_hash, a.table_prefix, s.email "
|
||||
"FROM sessions s JOIN rustbox_accounts a ON a.user_hash=s.user_hash "
|
||||
f"WHERE lower(s.email)=lower({ph}) ORDER BY s.last_accessed_unix DESC LIMIT 1",
|
||||
[login],
|
||||
)
|
||||
if row is None:
|
||||
return None
|
||||
user_hash, prefix, email_addr = row_tuple(row)[:3]
|
||||
validate_identifier(str(prefix), "RustBox table prefix")
|
||||
return str(user_hash), str(prefix), str(email_addr)
|
||||
|
||||
|
||||
def now_expr(backend: str) -> str:
|
||||
if backend == "sqlite":
|
||||
return "datetime('now')"
|
||||
if backend == "mysql":
|
||||
return "CAST(CURRENT_TIMESTAMP AS CHAR)"
|
||||
return "CURRENT_TIMESTAMP::text"
|
||||
|
||||
|
||||
def upsert_contact(target: Db, prefix: str, contact: Contact) -> None:
|
||||
table = target.ident(f"{prefix}contacts")
|
||||
ph = target.placeholder
|
||||
existing = target.fetchone(
|
||||
f"SELECT id FROM {table} WHERE lower(email)=lower({ph}) ORDER BY created_at LIMIT 1",
|
||||
[contact.email],
|
||||
)
|
||||
values = [
|
||||
contact.email,
|
||||
contact.name,
|
||||
contact.company,
|
||||
contact.notes,
|
||||
contact.phone,
|
||||
contact.job_title,
|
||||
contact.address,
|
||||
contact.website,
|
||||
contact.birthday,
|
||||
contact.emails_json,
|
||||
contact.phones_json,
|
||||
contact.addresses_json,
|
||||
]
|
||||
if existing is not None:
|
||||
contact_id = row_first(existing)
|
||||
assigns = ",".join(
|
||||
f"{column}={ph}"
|
||||
for column in (
|
||||
"email", "name", "company", "notes", "phone", "job_title",
|
||||
"address", "website", "birthday", "emails_json", "phones_json", "addresses_json",
|
||||
)
|
||||
)
|
||||
target.execute(
|
||||
f"UPDATE {table} SET {assigns},source='roundcube',updated_at={now_expr(target.backend)} WHERE id={ph}",
|
||||
[*values, contact_id],
|
||||
).close()
|
||||
return
|
||||
|
||||
columns = (
|
||||
"id,email,name,company,notes,phone,job_title,address,website,birthday,"
|
||||
"emails_json,phones_json,addresses_json,is_favorite,contact_count,source"
|
||||
)
|
||||
placeholders = ",".join([ph] * 13)
|
||||
target.execute(
|
||||
f"INSERT INTO {table} ({columns}) VALUES ({placeholders},0,0,'roundcube')",
|
||||
[contact.id, *values],
|
||||
).close()
|
||||
|
||||
|
||||
def upsert_identity(target: Db, prefix: str, identity: Identity) -> None:
|
||||
table = target.ident(f"{prefix}identities")
|
||||
ph = target.placeholder
|
||||
if identity.is_default:
|
||||
target.execute(f"UPDATE {table} SET is_default=0 WHERE is_default<>0").close()
|
||||
|
||||
existing = target.fetchone(
|
||||
f"SELECT id FROM {table} WHERE email={ph} AND display_name={ph} AND organization={ph} ORDER BY id LIMIT 1",
|
||||
[identity.email, identity.display_name, identity.organization],
|
||||
)
|
||||
if existing is None:
|
||||
existing = target.fetchone(
|
||||
f"SELECT id FROM {table} WHERE email={ph} AND display_name='' AND organization='' "
|
||||
"AND signature_html='' AND signature_text='' ORDER BY id LIMIT 1",
|
||||
[identity.email],
|
||||
)
|
||||
|
||||
values = [
|
||||
identity.display_name,
|
||||
identity.email,
|
||||
identity.signature_html,
|
||||
1 if identity.is_default else 0,
|
||||
identity.reply_to,
|
||||
identity.bcc,
|
||||
identity.organization,
|
||||
identity.signature_text,
|
||||
]
|
||||
if existing is not None:
|
||||
identity_id = row_first(existing)
|
||||
assignments = ",".join(
|
||||
f"{column}={ph}"
|
||||
for column in (
|
||||
"display_name", "email", "signature_html", "is_default", "reply_to", "bcc",
|
||||
"organization", "signature_text",
|
||||
)
|
||||
)
|
||||
target.execute(
|
||||
f"UPDATE {table} SET {assignments},updated_at={now_expr(target.backend)} WHERE id={ph}",
|
||||
[*values, identity_id],
|
||||
).close()
|
||||
return
|
||||
|
||||
placeholders = ",".join([ph] * 8)
|
||||
target.execute(
|
||||
f"INSERT INTO {table} "
|
||||
"(display_name,email,signature_html,is_default,reply_to,bcc,organization,signature_text,"
|
||||
"signature_new,signature_reply,signature_forward,signature_position) "
|
||||
f"VALUES ({placeholders},1,1,1,'above')",
|
||||
values,
|
||||
).close()
|
||||
|
||||
|
||||
def migrate_one(
|
||||
source: Db,
|
||||
target: Db,
|
||||
roundcube_prefix: str,
|
||||
roundcube_user: str,
|
||||
rustbox_user: str,
|
||||
*,
|
||||
dry_run: bool,
|
||||
user_hash_override: str | None = None,
|
||||
target_prefix_override: str | None = None,
|
||||
) -> tuple[int, int, int]:
|
||||
contacts, identities, skipped_contacts = load_roundcube_data(
|
||||
source, roundcube_prefix, roundcube_user
|
||||
)
|
||||
|
||||
if target_prefix_override:
|
||||
validate_identifier(target_prefix_override, "RustBox table prefix")
|
||||
prefix = target_prefix_override
|
||||
else:
|
||||
user_hash = user_hash_override or hash_user(rustbox_user)
|
||||
prefix = target_prefix_for_hash(target, user_hash)
|
||||
if prefix is None:
|
||||
match = target_account_for_login(target, rustbox_user)
|
||||
if match is None:
|
||||
raise ValueError(
|
||||
f"RustBox account not provisioned for {rustbox_user!r}; log in once first"
|
||||
)
|
||||
_, prefix, _ = match
|
||||
|
||||
print(
|
||||
f"{roundcube_user} -> {rustbox_user}: {len(contacts)} contact(s), "
|
||||
f"{len(identities)} identity/identities, {skipped_contacts} contact(s) skipped without email"
|
||||
)
|
||||
if dry_run:
|
||||
return len(contacts), len(identities), skipped_contacts
|
||||
|
||||
try:
|
||||
for contact in contacts:
|
||||
upsert_contact(target, prefix, contact)
|
||||
for identity in identities:
|
||||
upsert_identity(target, prefix, identity)
|
||||
target.commit()
|
||||
except Exception:
|
||||
target.rollback()
|
||||
raise
|
||||
return len(contacts), len(identities), skipped_contacts
|
||||
|
||||
|
||||
def run(args: argparse.Namespace) -> int:
|
||||
validate_identifier(args.roundcube_prefix, "Roundcube table prefix", allow_empty=True)
|
||||
if args.target_prefix:
|
||||
validate_identifier(args.target_prefix, "RustBox table prefix")
|
||||
|
||||
all_users = args.all_users or not args.roundcube_account
|
||||
if all_users and any([args.rustbox_user, args.user_hash, args.target_prefix]):
|
||||
raise ValueError(
|
||||
"--rustbox-user, --user-hash and --target-prefix are only valid with --roundcube-account"
|
||||
)
|
||||
|
||||
config_path, _config, backend, target_url = resolve_target(args)
|
||||
roundcube_url = build_roundcube_url(args)
|
||||
|
||||
print(f"RustBox config: {config_path}")
|
||||
print(f"RustBox target backend: {backend}")
|
||||
if backend == "sqlite":
|
||||
print(f"RustBox target database: {target_url}")
|
||||
print(f"Mode: {'all users' if all_users else 'single user'}{' (dry-run)' if args.dry_run else ''}")
|
||||
|
||||
source = connect_mysql(roundcube_url, dict_rows=True)
|
||||
target = connect_target(backend, target_url)
|
||||
try:
|
||||
# Fail early with a clear error if this is not a RustBox database.
|
||||
target.fetchall("SELECT user_hash, table_prefix FROM rustbox_accounts LIMIT 1")
|
||||
|
||||
if not all_users:
|
||||
roundcube_user = args.roundcube_account
|
||||
assert roundcube_user is not None
|
||||
rustbox_user = args.rustbox_user or roundcube_user
|
||||
migrate_one(
|
||||
source,
|
||||
target,
|
||||
args.roundcube_prefix,
|
||||
roundcube_user,
|
||||
rustbox_user,
|
||||
dry_run=args.dry_run,
|
||||
user_hash_override=args.user_hash,
|
||||
target_prefix_override=args.target_prefix,
|
||||
)
|
||||
print("Dry run complete." if args.dry_run else "Migration complete.")
|
||||
return 0
|
||||
|
||||
users = load_roundcube_users(source, args.roundcube_prefix)
|
||||
print(f"Roundcube: found {len(users)} user(s).")
|
||||
migrated = skipped = failed = 0
|
||||
total_contacts = total_identities = total_skipped_contacts = 0
|
||||
|
||||
for roundcube_user in users:
|
||||
try:
|
||||
match = target_account_for_login(target, roundcube_user)
|
||||
if match is None:
|
||||
skipped += 1
|
||||
print(
|
||||
f"SKIP {roundcube_user}: matching RustBox account is not provisioned",
|
||||
file=sys.stderr,
|
||||
)
|
||||
continue
|
||||
_, prefix, rustbox_user = match
|
||||
counts = migrate_one(
|
||||
source,
|
||||
target,
|
||||
args.roundcube_prefix,
|
||||
roundcube_user,
|
||||
rustbox_user,
|
||||
dry_run=args.dry_run,
|
||||
target_prefix_override=prefix,
|
||||
)
|
||||
migrated += 1
|
||||
total_contacts += counts[0]
|
||||
total_identities += counts[1]
|
||||
total_skipped_contacts += counts[2]
|
||||
except Exception as exc:
|
||||
target.rollback()
|
||||
failed += 1
|
||||
print(f"FAIL {roundcube_user}: {exc}", file=sys.stderr)
|
||||
if args.fail_fast:
|
||||
raise
|
||||
|
||||
label = "Dry run complete" if args.dry_run else "Migration complete"
|
||||
print(
|
||||
f"{label}: {migrated} user(s) processed, {skipped} skipped without RustBox account, "
|
||||
f"{failed} failed; {total_contacts} contact(s), {total_identities} identity/identities, "
|
||||
f"{total_skipped_contacts} contact(s) skipped without email."
|
||||
)
|
||||
return 2 if failed else 0
|
||||
finally:
|
||||
target.close()
|
||||
source.close()
|
||||
|
||||
|
||||
def main() -> int:
|
||||
args = parser().parse_args()
|
||||
try:
|
||||
return run(args)
|
||||
except KeyboardInterrupt:
|
||||
print("Migration interrupted.", file=sys.stderr)
|
||||
return 130
|
||||
except Exception as exc:
|
||||
print(f"Roundcube migration failed: {exc}", file=sys.stderr)
|
||||
return 1
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in new issue
Block a user