ilia
934b674995
Record the completed migration (channel map, mmetl/mmctl steps, caveats) and teach prepare-export to apply mattermost_channel renames plus user_emails for matching existing MM accounts.
254 lines
8.1 KiB
Python
254 lines
8.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Build a filtered Slack-shaped export from manifest.json for mmetl/mmctl."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
MANIFEST_PATH = REPO_ROOT / "manifest.json"
|
|
DEFAULT_EXPORT = Path.home() / "Downloads/Levkin Slack export enriched"
|
|
DEFAULT_OUT = REPO_ROOT / "prepared-export"
|
|
|
|
|
|
def message_key(channel: str, ts: str | float) -> str:
|
|
return f"{channel}:{ts}"
|
|
|
|
|
|
def channel_included(entry: dict) -> bool:
|
|
status = entry.get("status")
|
|
if status == "skip":
|
|
return False
|
|
if status == "import" and entry.get("import_mode", "all") != "pick":
|
|
return True
|
|
if status in ("import", "partial"):
|
|
picks = entry.get("messages") or {}
|
|
return any(
|
|
(sel.get("action") in ("import", "merge") or sel.get("checked"))
|
|
for sel in picks.values()
|
|
if isinstance(sel, dict)
|
|
)
|
|
return False
|
|
|
|
|
|
def message_included(entry: dict, key: str) -> bool:
|
|
status = entry.get("status")
|
|
if status == "skip":
|
|
return False
|
|
if status == "import" and entry.get("import_mode", "all") != "pick":
|
|
return True
|
|
picks = entry.get("messages") or {}
|
|
sel = picks.get(key)
|
|
if not sel:
|
|
return False
|
|
if sel.get("action") == "skip":
|
|
return False
|
|
return sel.get("action") in ("import", "merge") or bool(sel.get("checked"))
|
|
|
|
|
|
def filter_files_on_message(entry: dict, key: str, raw: dict) -> dict:
|
|
"""Return a copy of the Slack message with only manifest-selected files."""
|
|
picks = entry.get("messages") or {}
|
|
sel = picks.get(key) or {}
|
|
file_flags = sel.get("files") if isinstance(sel.get("files"), dict) else {}
|
|
files = raw.get("files")
|
|
if not files:
|
|
return raw
|
|
out = dict(raw)
|
|
kept = []
|
|
for f in files:
|
|
if not isinstance(f, dict):
|
|
continue
|
|
fid = f.get("id") or f.get("name")
|
|
if not fid:
|
|
continue
|
|
if file_flags and file_flags.get(fid) is False:
|
|
continue
|
|
kept.append(f)
|
|
if not kept:
|
|
out.pop("files", None)
|
|
else:
|
|
out["files"] = kept
|
|
return out
|
|
|
|
|
|
def prepare_export(
|
|
export_root: Path,
|
|
manifest: dict,
|
|
out_root: Path,
|
|
*,
|
|
export_mod=None,
|
|
) -> dict[str, int]:
|
|
"""
|
|
Write filtered export to out_root. Returns per-channel message counts.
|
|
"""
|
|
sys.path.insert(0, str(REPO_ROOT))
|
|
from slack_export import SlackExport, load_json
|
|
|
|
export = export_mod or SlackExport(export_root)
|
|
out_root = out_root.resolve()
|
|
if out_root.exists():
|
|
shutil.rmtree(out_root)
|
|
out_root.mkdir(parents=True)
|
|
|
|
channels_meta = load_json(export.paths.channels_json)
|
|
if not isinstance(channels_meta, list):
|
|
channels_meta = []
|
|
|
|
manifest_channels = manifest.get("channels") or {}
|
|
email_map = manifest.get("user_emails") or {}
|
|
included_names: set[str] = set()
|
|
stats: dict[str, int] = {}
|
|
rename_map: dict[str, str] = {}
|
|
|
|
for ch in channels_meta:
|
|
name = ch.get("name")
|
|
if not name:
|
|
continue
|
|
entry = manifest_channels.get(name) or {}
|
|
if not channel_included(entry):
|
|
continue
|
|
included_names.add(name)
|
|
mm_name = (entry.get("mattermost_channel") or "").strip()
|
|
if mm_name and mm_name != name:
|
|
rename_map[name] = mm_name
|
|
|
|
# channels.json subset (optionally renamed for Mattermost)
|
|
filtered_meta = []
|
|
for c in channels_meta:
|
|
name = c.get("name")
|
|
if name not in included_names:
|
|
continue
|
|
out_ch = dict(c)
|
|
if name in rename_map:
|
|
out_ch["name"] = rename_map[name]
|
|
filtered_meta.append(out_ch)
|
|
(out_root / "channels.json").write_text(
|
|
json.dumps(filtered_meta, indent=2, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
# users.json — copy + inject emails so mmetl can match existing MM users
|
|
users_src = export.paths.users_json
|
|
if users_src.is_file():
|
|
users = load_json(users_src)
|
|
if isinstance(users, list) and email_map:
|
|
for u in users:
|
|
if not isinstance(u, dict):
|
|
continue
|
|
uid = u.get("id")
|
|
uname = u.get("name")
|
|
email = email_map.get(uid) or email_map.get(uname)
|
|
if email:
|
|
profile = dict(u.get("profile") or {})
|
|
profile["email"] = email
|
|
u["profile"] = profile
|
|
(out_root / "users.json").write_text(
|
|
json.dumps(users, indent=2, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
uploads_out = out_root / "__uploads"
|
|
uploads_src = export.paths.root / "__uploads"
|
|
|
|
for name in sorted(included_names):
|
|
entry = manifest_channels[name]
|
|
out_name = rename_map.get(name, name)
|
|
ch_out = out_root / out_name
|
|
ch_out.mkdir(parents=True)
|
|
count = 0
|
|
for date in export.list_channel_date_files(name):
|
|
raw_list = export.read_messages_file(name, date)
|
|
if not isinstance(raw_list, list):
|
|
continue
|
|
kept: list[dict] = []
|
|
for raw in raw_list:
|
|
if not isinstance(raw, dict):
|
|
continue
|
|
ts = raw.get("ts")
|
|
if ts is None:
|
|
continue
|
|
key = message_key(name, ts)
|
|
if not message_included(entry, key):
|
|
continue
|
|
msg = filter_files_on_message(entry, key, raw)
|
|
kept.append(msg)
|
|
count += 1
|
|
for f in msg.get("files") or []:
|
|
if not isinstance(f, dict):
|
|
continue
|
|
fid = f.get("id")
|
|
fname = f.get("name")
|
|
if fid and fname and uploads_src.is_dir():
|
|
src = uploads_src / fid / fname
|
|
if src.is_file():
|
|
dest = uploads_out / fid / fname
|
|
dest.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.copy2(src, dest)
|
|
if kept:
|
|
(ch_out / f"{date}.json").write_text(
|
|
json.dumps(kept, indent=2, ensure_ascii=False) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
stats[out_name] = count
|
|
|
|
summary = {
|
|
"channels": len(stats),
|
|
"messages": sum(stats.values()),
|
|
"per_channel": stats,
|
|
"renames": rename_map,
|
|
}
|
|
(out_root / "prepare-export-summary.json").write_text(
|
|
json.dumps(summary, indent=2) + "\n",
|
|
encoding="utf-8",
|
|
)
|
|
return stats
|
|
|
|
|
|
def main() -> int:
|
|
export_path = DEFAULT_EXPORT
|
|
out_path = DEFAULT_OUT
|
|
manifest_path = MANIFEST_PATH
|
|
|
|
args = sys.argv[1:]
|
|
i = 0
|
|
while i < len(args):
|
|
if args[i] == "--export" and i + 1 < len(args):
|
|
export_path = Path(args[i + 1]).expanduser()
|
|
i += 2
|
|
elif args[i] == "--out" and i + 1 < len(args):
|
|
out_path = Path(args[i + 1]).expanduser()
|
|
i += 2
|
|
elif args[i] == "--manifest" and i + 1 < len(args):
|
|
manifest_path = Path(args[i + 1]).expanduser()
|
|
i += 2
|
|
elif not args[i].startswith("-"):
|
|
export_path = Path(args[i]).expanduser()
|
|
i += 1
|
|
else:
|
|
i += 1
|
|
|
|
export_path = export_path.resolve()
|
|
if not export_path.is_dir():
|
|
print(f"Export not found: {export_path}", file=sys.stderr)
|
|
return 1
|
|
if not manifest_path.is_file():
|
|
print(f"Manifest not found: {manifest_path}", file=sys.stderr)
|
|
return 1
|
|
|
|
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
stats = prepare_export(export_path, manifest, out_path.resolve())
|
|
total = sum(stats.values())
|
|
print(f"Wrote {out_path}")
|
|
print(f" {len(stats)} channels, {total} messages")
|
|
for name, n in sorted(stats.items()):
|
|
print(f" #{name}: {n}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|