ilia
1ce803fc9c
Web UI for channel/message cherry-pick, attachment preview, hydrate script, and workspace skip presets.
459 lines
17 KiB
Python
Executable File
459 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Local UI server for reviewing Slack export before Mattermost import."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import html
|
|
import json
|
|
import mimetypes
|
|
import os
|
|
import sys
|
|
import zipfile
|
|
import xml.etree.ElementTree as ET
|
|
from datetime import datetime, timezone
|
|
from http import HTTPStatus
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs, unquote, urlparse
|
|
|
|
from slack_export import SlackExport, resolve_paths
|
|
|
|
TOOL_DIR = Path(__file__).resolve().parent
|
|
STATIC_DIR = TOOL_DIR / "static"
|
|
|
|
|
|
def default_manifest(export_path: str) -> dict:
|
|
return {
|
|
"version": 1,
|
|
"export_path": export_path,
|
|
"updated_at": None,
|
|
"global_notes": "",
|
|
"channels": {},
|
|
"users": {},
|
|
}
|
|
|
|
|
|
def load_manifest(path: Path, export_path: str) -> dict:
|
|
if path.is_file():
|
|
data = json.loads(path.read_text(encoding="utf-8"))
|
|
if isinstance(data, dict):
|
|
return data
|
|
return default_manifest(export_path)
|
|
|
|
|
|
def save_manifest(path: Path, data: dict) -> None:
|
|
data["updated_at"] = datetime.now(tz=timezone.utc).isoformat()
|
|
path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
|
|
|
|
_W_NS = "{http://schemas.openxmlformats.org/wordprocessingml/2006/main}"
|
|
|
|
|
|
def docx_to_html(path: Path) -> str:
|
|
"""Minimal DOCX → HTML using stdlib only (fallback when browser mammoth fails)."""
|
|
with zipfile.ZipFile(path) as zf:
|
|
xml_bytes = zf.read("word/document.xml")
|
|
root = ET.fromstring(xml_bytes)
|
|
blocks: list[str] = []
|
|
for para in root.iter(f"{_W_NS}p"):
|
|
runs = [node.text for node in para.iter(f"{_W_NS}t") if node.text]
|
|
line = "".join(runs).strip()
|
|
if line:
|
|
blocks.append(f"<p>{html.escape(line)}</p>")
|
|
return "".join(blocks) or "<p><em>Empty document</em></p>"
|
|
|
|
|
|
def file_preview_payload_bytes(raw: bytes, name: str) -> dict:
|
|
lower_name = name.lower()
|
|
|
|
if raw[:2] == b"PK":
|
|
try:
|
|
import io
|
|
|
|
with zipfile.ZipFile(io.BytesIO(raw)) as zf:
|
|
xml_bytes = zf.read("word/document.xml")
|
|
root = ET.fromstring(xml_bytes)
|
|
blocks: list[str] = []
|
|
for para in root.iter(f"{_W_NS}p"):
|
|
runs = [node.text for node in para.iter(f"{_W_NS}t") if node.text]
|
|
line = "".join(runs).strip()
|
|
if line:
|
|
blocks.append(f"<p>{html.escape(line)}</p>")
|
|
body = "".join(blocks) or "<p><em>Empty document</em></p>"
|
|
return {"ok": True, "kind": "html", "html": body}
|
|
except (zipfile.BadZipFile, KeyError, ET.ParseError) as exc:
|
|
return {"ok": False, "error": f"Could not read DOCX: {exc}"}
|
|
|
|
try:
|
|
text = raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return {"ok": False, "error": "Binary file — use download."}
|
|
|
|
lower = text.lower()
|
|
if (
|
|
"quip-canvas-content" in lower
|
|
or lower.lstrip().startswith(("<!doctype", "<html", "<div", "<h1"))
|
|
):
|
|
return {"ok": True, "kind": "html", "html": text}
|
|
|
|
if lower_name.endswith((".txt", ".md", ".csv", ".log", ".json")):
|
|
return {
|
|
"ok": True,
|
|
"kind": "text",
|
|
"html": f"<pre class='text-preview'>{html.escape(text)}</pre>",
|
|
}
|
|
|
|
return {"ok": False, "error": "Unsupported format for preview."}
|
|
|
|
|
|
def file_preview_payload(path: Path) -> dict:
|
|
return file_preview_payload_bytes(path.read_bytes(), path.name)
|
|
|
|
|
|
class ReviewHandler(BaseHTTPRequestHandler):
|
|
export: SlackExport
|
|
manifest_path: Path
|
|
|
|
def log_message(self, format: str, *args) -> None: # noqa: A003
|
|
if os.environ.get("SLACK_IMPORT_QUIET") != "1":
|
|
super().log_message(format, *args)
|
|
|
|
def _send_json(self, payload: object, status: int = HTTPStatus.OK) -> None:
|
|
body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", "application/json; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def _send_file(self, path: Path) -> None:
|
|
if not path.is_file():
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
return
|
|
content = path.read_bytes()
|
|
mime, _ = mimetypes.guess_type(str(path))
|
|
self.send_response(HTTPStatus.OK)
|
|
self.send_header("Content-Type", mime or "application/octet-stream")
|
|
self.send_header("Content-Length", str(len(content)))
|
|
self.end_headers()
|
|
self.wfile.write(content)
|
|
|
|
def _safe_export_path(self, relpath: str) -> Path | None:
|
|
"""
|
|
Resolve a relative path within the Slack export root.
|
|
Prevent path traversal and absolute paths.
|
|
"""
|
|
if not relpath or relpath.startswith(("/", "\\")) or ".." in relpath:
|
|
return None
|
|
root = self.export.paths.root.resolve()
|
|
candidate = (root / relpath).resolve()
|
|
try:
|
|
candidate.relative_to(root)
|
|
except ValueError:
|
|
return None
|
|
return candidate
|
|
|
|
def _read_json_body(self) -> dict | None:
|
|
length = int(self.headers.get("Content-Length", 0))
|
|
if length <= 0:
|
|
return None
|
|
raw = self.rfile.read(length)
|
|
try:
|
|
data = json.loads(raw.decode("utf-8"))
|
|
return data if isinstance(data, dict) else None
|
|
except json.JSONDecodeError:
|
|
return None
|
|
|
|
def do_OPTIONS(self) -> None: # noqa: N802
|
|
self.send_response(HTTPStatus.NO_CONTENT)
|
|
self.send_header("Access-Control-Allow-Origin", "*")
|
|
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
|
|
self.send_header("Access-Control-Allow-Headers", "Content-Type")
|
|
self.end_headers()
|
|
|
|
def do_GET(self) -> None: # noqa: N802
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path
|
|
qs = parse_qs(parsed.query)
|
|
|
|
if path in ("/", "/index.html"):
|
|
return self._send_file(STATIC_DIR / "index.html")
|
|
if path.startswith("/static/"):
|
|
rel = path.removeprefix("/static/")
|
|
return self._send_file(STATIC_DIR / rel)
|
|
|
|
if path.startswith("/uploads/"):
|
|
rel = unquote(path.removeprefix("/uploads/"))
|
|
export_file = self._safe_export_path(rel)
|
|
if not export_file:
|
|
self.send_error(HTTPStatus.BAD_REQUEST)
|
|
return
|
|
return self._send_file(export_file)
|
|
|
|
if path == "/api/file-preview":
|
|
file_id = (qs.get("file_id") or [""])[0].strip()
|
|
if file_id:
|
|
try:
|
|
raw, rec = self.export.fetch_file_bytes(file_id)
|
|
return self._send_json(
|
|
file_preview_payload_bytes(raw, rec.get("name") or "file")
|
|
)
|
|
except FileNotFoundError as exc:
|
|
return self._send_json(
|
|
{"ok": False, "error": str(exc)},
|
|
status=HTTPStatus.NOT_FOUND,
|
|
)
|
|
|
|
rel = unquote((qs.get("rel") or [""])[0])
|
|
export_file = self._safe_export_path(rel)
|
|
if not export_file or not export_file.is_file():
|
|
return self._send_json(
|
|
{
|
|
"ok": False,
|
|
"error": "File not on disk — standard Slack exports omit attachments; use file preview (fetches from Slack once).",
|
|
},
|
|
status=HTTPStatus.NOT_FOUND,
|
|
)
|
|
return self._send_json(file_preview_payload(export_file))
|
|
|
|
if path == "/api/file-download":
|
|
file_id = (qs.get("file_id") or [""])[0].strip()
|
|
if not file_id:
|
|
self.send_error(HTTPStatus.BAD_REQUEST)
|
|
return
|
|
try:
|
|
raw, rec = self.export.fetch_file_bytes(file_id)
|
|
except FileNotFoundError:
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
return
|
|
mime, _ = mimetypes.guess_type(rec.get("name") or "")
|
|
self.send_response(HTTPStatus.OK)
|
|
self.send_header("Content-Type", mime or "application/octet-stream")
|
|
self.send_header(
|
|
"Content-Disposition",
|
|
f'inline; filename="{rec.get("name", "file")}"',
|
|
)
|
|
self.send_header("Content-Length", str(len(raw)))
|
|
self.end_headers()
|
|
self.wfile.write(raw)
|
|
return
|
|
|
|
if path == "/api/health":
|
|
return self._send_json(
|
|
{
|
|
"ok": True,
|
|
"export_path": str(self.export.paths.root),
|
|
"manifest_path": str(self.manifest_path),
|
|
}
|
|
)
|
|
|
|
if path == "/api/summary":
|
|
summary = self.export.build_summary()
|
|
manifest = load_manifest(
|
|
self.manifest_path, str(self.export.paths.root)
|
|
)
|
|
selected_channels = sum(
|
|
1
|
|
for c in manifest.get("channels", {}).values()
|
|
if c.get("import")
|
|
)
|
|
return self._send_json(
|
|
{
|
|
**summary,
|
|
"manifest": {
|
|
"updated_at": manifest.get("updated_at"),
|
|
"selected_channels": selected_channels,
|
|
"global_notes": manifest.get("global_notes", ""),
|
|
},
|
|
}
|
|
)
|
|
|
|
if path == "/api/channels":
|
|
query = (qs.get("q") or [""])[0]
|
|
channels = self.export.search_channels(query)
|
|
manifest = load_manifest(
|
|
self.manifest_path, str(self.export.paths.root)
|
|
)
|
|
channel_manifest = manifest.get("channels") or {}
|
|
for ch in channels:
|
|
saved = channel_manifest.get(ch["name"], {})
|
|
ch["import"] = bool(saved.get("import"))
|
|
ch["status"] = saved.get("status") or (
|
|
"import" if saved.get("import") else "undecided"
|
|
)
|
|
ch["reviewed"] = bool(saved.get("reviewed")) or ch["status"] in (
|
|
"import",
|
|
"skip",
|
|
)
|
|
ch["notes"] = saved.get("notes", "")
|
|
ch["mattermost_channel"] = saved.get("mattermost_channel", "")
|
|
ch["mattermost_team"] = saved.get("mattermost_team", "")
|
|
ch["import_mode"] = saved.get("import_mode") or "all"
|
|
ch["messages"] = saved.get("messages") or {}
|
|
ch["message_selections"] = ch["messages"]
|
|
return self._send_json({"channels": channels})
|
|
|
|
if path.startswith("/api/channels/") and path.endswith("/messages/all"):
|
|
channel_name = path.split("/")[3]
|
|
return self._send_json(self.export.channel_messages_all(channel_name))
|
|
|
|
if path.startswith("/api/channels/") and path.endswith("/messages"):
|
|
channel_name = path.split("/")[3]
|
|
date = (qs.get("date") or [None])[0]
|
|
offset = int((qs.get("offset") or ["0"])[0])
|
|
limit = min(int((qs.get("limit") or ["100"])[0]), 500)
|
|
return self._send_json(
|
|
self.export.channel_messages_page(
|
|
channel_name, date=date, offset=offset, limit=limit
|
|
)
|
|
)
|
|
|
|
if path == "/api/search":
|
|
query = (qs.get("q") or [""])[0]
|
|
limit = min(int((qs.get("limit") or ["80"])[0]), 200)
|
|
return self._send_json(self.export.search_global(query, limit=limit))
|
|
|
|
if path.startswith("/api/channels/") and path.endswith("/search"):
|
|
channel_name = path.split("/")[3]
|
|
query = (qs.get("q") or [""])[0]
|
|
limit = min(int((qs.get("limit") or ["50"])[0]), 200)
|
|
hits = self.export.search_messages(channel_name, query, limit=limit)
|
|
return self._send_json({"channel": channel_name, "query": query, "hits": hits})
|
|
|
|
if path == "/api/users":
|
|
manifest = load_manifest(
|
|
self.manifest_path, str(self.export.paths.root)
|
|
)
|
|
user_manifest = manifest.get("users") or {}
|
|
users = []
|
|
for user in self.export.users_raw:
|
|
uid = user.get("id")
|
|
saved = user_manifest.get(uid, {})
|
|
users.append(
|
|
{
|
|
"id": uid,
|
|
"name": user.get("name"),
|
|
"real_name": user.get("profile", {}).get("real_name")
|
|
or user.get("real_name"),
|
|
"is_bot": bool(user.get("is_bot")),
|
|
"deleted": bool(user.get("deleted")),
|
|
"import": bool(saved.get("import")),
|
|
"notes": saved.get("notes", ""),
|
|
"mattermost_username": saved.get("mattermost_username", ""),
|
|
}
|
|
)
|
|
return self._send_json({"users": users})
|
|
|
|
if path == "/api/manifest":
|
|
manifest = load_manifest(
|
|
self.manifest_path, str(self.export.paths.root)
|
|
)
|
|
return self._send_json(manifest)
|
|
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
|
|
def do_POST(self) -> None: # noqa: N802
|
|
parsed = urlparse(self.path)
|
|
if parsed.path != "/api/manifest":
|
|
self.send_error(HTTPStatus.NOT_FOUND)
|
|
return
|
|
|
|
body = self._read_json_body()
|
|
if body is None:
|
|
return self._send_json(
|
|
{"error": "invalid JSON body"}, status=HTTPStatus.BAD_REQUEST
|
|
)
|
|
|
|
export_path = str(self.export.paths.root)
|
|
current = load_manifest(self.manifest_path, export_path)
|
|
|
|
if "global_notes" in body:
|
|
current["global_notes"] = body.get("global_notes") or ""
|
|
|
|
if isinstance(body.get("channels"), dict):
|
|
current.setdefault("channels", {})
|
|
for name, entry in body["channels"].items():
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
status = entry.get("status") or (
|
|
"import" if entry.get("import") else "undecided"
|
|
)
|
|
import_mode = entry.get("import_mode") or "all"
|
|
if import_mode not in ("all", "pick"):
|
|
import_mode = "all"
|
|
saved_messages = entry.get("messages") or entry.get("message_selections")
|
|
if not isinstance(saved_messages, dict):
|
|
saved_messages = {}
|
|
current["channels"][name] = {
|
|
"import": status in ("import", "partial"),
|
|
"status": status,
|
|
"reviewed": bool(entry.get("reviewed"))
|
|
or status in ("import", "skip", "partial"),
|
|
"import_mode": import_mode,
|
|
"notes": entry.get("notes") or "",
|
|
"mattermost_channel": entry.get("mattermost_channel") or "",
|
|
"mattermost_team": entry.get("mattermost_team") or "",
|
|
"messages": saved_messages,
|
|
}
|
|
|
|
if isinstance(body.get("users"), dict):
|
|
current.setdefault("users", {})
|
|
for uid, entry in body["users"].items():
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
current["users"][uid] = {
|
|
"import": bool(entry.get("import")),
|
|
"notes": entry.get("notes") or "",
|
|
"mattermost_username": entry.get("mattermost_username") or "",
|
|
}
|
|
|
|
current["export_path"] = export_path
|
|
current["version"] = 1
|
|
save_manifest(self.manifest_path, current)
|
|
return self._send_json({"ok": True, "manifest": current})
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument(
|
|
"--export",
|
|
default=os.environ.get("SLACK_EXPORT_PATH"),
|
|
help="Path to Slack export directory",
|
|
)
|
|
parser.add_argument("--host", default="127.0.0.1")
|
|
parser.add_argument("--port", type=int, default=8765)
|
|
args = parser.parse_args()
|
|
|
|
paths = resolve_paths(args.export)
|
|
try:
|
|
export = SlackExport(paths.root)
|
|
except FileNotFoundError as exc:
|
|
print(exc, file=sys.stderr)
|
|
print(
|
|
"\nSet SLACK_EXPORT_PATH or pass --export to your Slack export folder.",
|
|
file=sys.stderr,
|
|
)
|
|
return 1
|
|
|
|
handler_cls = type(
|
|
"BoundReviewHandler",
|
|
(ReviewHandler,),
|
|
{"export": export, "manifest_path": paths.manifest_json},
|
|
)
|
|
|
|
server = ThreadingHTTPServer((args.host, args.port), handler_cls)
|
|
print(f"Slack import review UI: http://{args.host}:{args.port}")
|
|
print(f"Export: {paths.root}")
|
|
print(f"Manifest: {paths.manifest_json}")
|
|
try:
|
|
server.serve_forever()
|
|
except KeyboardInterrupt:
|
|
print("\nStopped.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|