ilia
4b034d19f6
Cherry-pick heuristics, manifest merge, global file search, scrollbar styling, and pytest/Playwright coverage unblock review and filtered export for mmetl.
610 lines
22 KiB
Python
610 lines
22 KiB
Python
"""Read a standard Slack workspace export directory."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
from dataclasses import dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
DEFAULT_EXPORT = Path.home() / "Downloads" / "slack-export"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExportPaths:
|
|
root: Path
|
|
channels_json: Path
|
|
users_json: Path
|
|
manifest_json: Path
|
|
|
|
|
|
def resolve_paths(export_root: str | Path | None = None) -> ExportPaths:
|
|
root = Path(export_root or DEFAULT_EXPORT).expanduser().resolve()
|
|
tool_dir = Path(__file__).resolve().parent
|
|
return ExportPaths(
|
|
root=root,
|
|
channels_json=root / "channels.json",
|
|
users_json=root / "users.json",
|
|
manifest_json=tool_dir / "manifest.json",
|
|
)
|
|
|
|
|
|
def load_json(path: Path) -> Any:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
|
|
def slack_ts_to_iso(ts: str | float | None) -> str | None:
|
|
if ts is None:
|
|
return None
|
|
try:
|
|
seconds = float(str(ts).split(".")[0])
|
|
return datetime.fromtimestamp(seconds, tz=timezone.utc).isoformat()
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def user_display(user: dict[str, Any] | None) -> str:
|
|
if not user:
|
|
return "unknown"
|
|
profile = user.get("profile") or {}
|
|
return (
|
|
profile.get("display_name")
|
|
or profile.get("real_name")
|
|
or user.get("real_name")
|
|
or user.get("name")
|
|
or user.get("id")
|
|
or "unknown"
|
|
)
|
|
|
|
|
|
class SlackExport:
|
|
def __init__(self, export_root: str | Path | None = None) -> None:
|
|
self.paths = resolve_paths(export_root)
|
|
if not self.paths.root.is_dir():
|
|
raise FileNotFoundError(f"Slack export not found: {self.paths.root}")
|
|
|
|
self.channels_meta: list[dict[str, Any]] = []
|
|
if self.paths.channels_json.is_file():
|
|
self.channels_meta = load_json(self.paths.channels_json)
|
|
|
|
self.users_raw: list[dict[str, Any]] = []
|
|
if self.paths.users_json.is_file():
|
|
self.users_raw = load_json(self.paths.users_json)
|
|
|
|
self.users_by_id = {u["id"]: u for u in self.users_raw if u.get("id")}
|
|
self.channel_meta_by_name = {
|
|
c["name"]: c for c in self.channels_meta if c.get("name")
|
|
}
|
|
self._file_index: dict[str, dict[str, Any]] | None = None
|
|
|
|
@staticmethod
|
|
def upload_relpath(file_id: str, file_name: str) -> str:
|
|
return f"__uploads/{file_id}/{file_name}"
|
|
|
|
def local_upload_path(self, file_id: str, file_name: str) -> Path:
|
|
return self.paths.root / self.upload_relpath(file_id, file_name)
|
|
|
|
def _register_file_raw(self, f: dict[str, Any]) -> None:
|
|
file_id = f.get("id")
|
|
file_name = f.get("name")
|
|
if not file_id or not file_name:
|
|
return
|
|
rel = self.upload_relpath(file_id, file_name)
|
|
local = self.paths.root / rel
|
|
self._file_index[file_id] = {
|
|
"id": file_id,
|
|
"name": file_name,
|
|
"mimetype": f.get("mimetype"),
|
|
"filetype": f.get("filetype"),
|
|
"url_private_download": f.get("url_private_download") or f.get("url_private"),
|
|
"local_relpath": rel,
|
|
"local_exists": local.is_file(),
|
|
}
|
|
|
|
def ensure_file_index(self) -> dict[str, dict[str, Any]]:
|
|
if self._file_index is not None:
|
|
return self._file_index
|
|
self._file_index = {}
|
|
canvases_path = self.paths.root / "canvases.json"
|
|
if canvases_path.is_file():
|
|
data = load_json(canvases_path)
|
|
if isinstance(data, list):
|
|
for entry in data:
|
|
if isinstance(entry, dict):
|
|
self._register_file_raw(entry)
|
|
for ch_dir in self.channel_dirs():
|
|
for json_path in ch_dir.glob("*.json"):
|
|
data = load_json(json_path)
|
|
if not isinstance(data, list):
|
|
continue
|
|
for msg in data:
|
|
if not isinstance(msg, dict):
|
|
continue
|
|
for raw_file in msg.get("files") or []:
|
|
if isinstance(raw_file, dict):
|
|
self._register_file_raw(raw_file)
|
|
return self._file_index
|
|
|
|
def file_record(self, file_id: str) -> dict[str, Any] | None:
|
|
return self.ensure_file_index().get(file_id)
|
|
|
|
def fetch_file_bytes(self, file_id: str) -> tuple[bytes, dict[str, Any]]:
|
|
"""Read cached upload or download from Slack using export token in URL."""
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
rec = self.file_record(file_id)
|
|
if not rec:
|
|
raise FileNotFoundError(f"Unknown file id: {file_id}")
|
|
|
|
local = self.paths.root / rec["local_relpath"]
|
|
if local.is_file():
|
|
return local.read_bytes(), rec
|
|
|
|
url = rec.get("url_private_download")
|
|
if not url:
|
|
raise FileNotFoundError(f"No download URL for {file_id}")
|
|
|
|
headers: dict[str, str] = {}
|
|
if "token=" in url:
|
|
token = url.split("token=", 1)[1].split("&")[0]
|
|
headers["Authorization"] = f"Bearer {token}"
|
|
|
|
req = urllib.request.Request(url, headers=headers)
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=90) as resp:
|
|
data = resp.read()
|
|
except urllib.error.HTTPError as exc:
|
|
raise FileNotFoundError(
|
|
f"Slack download failed ({exc.code}). Export token may be expired."
|
|
) from exc
|
|
|
|
local.parent.mkdir(parents=True, exist_ok=True)
|
|
local.write_bytes(data)
|
|
rec["local_exists"] = True
|
|
return data, rec
|
|
|
|
def channel_dirs(self) -> list[Path]:
|
|
skip = {
|
|
"__uploads",
|
|
"channels.json",
|
|
"users.json",
|
|
"canvases.json",
|
|
"file_conversations.json",
|
|
"huddle_transcripts.json",
|
|
"integration_logs.json",
|
|
"lists.json",
|
|
}
|
|
dirs = []
|
|
for entry in sorted(self.paths.root.iterdir(), key=lambda p: p.name.lower()):
|
|
if entry.is_dir() and entry.name not in skip:
|
|
dirs.append(entry)
|
|
return dirs
|
|
|
|
def channel_folder(self, name: str) -> Path | None:
|
|
folder = self.paths.root / name
|
|
return folder if folder.is_dir() else None
|
|
|
|
def list_channel_date_files(self, channel_name: str) -> list[str]:
|
|
folder = self.channel_folder(channel_name)
|
|
if not folder:
|
|
return []
|
|
return sorted(f.stem for f in folder.glob("*.json"))
|
|
|
|
def read_messages_file(self, channel_name: str, date: str) -> list[dict[str, Any]]:
|
|
folder = self.channel_folder(channel_name)
|
|
if not folder:
|
|
return []
|
|
path = folder / f"{date}.json"
|
|
if not path.is_file():
|
|
return []
|
|
data = load_json(path)
|
|
return data if isinstance(data, list) else []
|
|
|
|
def normalize_message(
|
|
self, msg: dict[str, Any], channel_name: str, date_file: str
|
|
) -> dict[str, Any]:
|
|
user_id = msg.get("user") or msg.get("bot_id")
|
|
user = self.users_by_id.get(user_id) if user_id else None
|
|
subtype = msg.get("subtype")
|
|
text = (msg.get("text") or "").strip()
|
|
|
|
if subtype == "channel_join" and not text:
|
|
who = user_display(user)
|
|
text = f"{who} joined the channel"
|
|
elif subtype == "channel_leave" and not text:
|
|
who = user_display(user)
|
|
text = f"{who} left the channel"
|
|
elif subtype == "channel_topic" and msg.get("topic"):
|
|
text = f"set topic: {msg['topic']}"
|
|
elif subtype == "channel_purpose" and msg.get("purpose"):
|
|
text = f"set purpose: {msg['purpose']}"
|
|
elif subtype == "file_share" and msg.get("files"):
|
|
names = ", ".join(f.get("name", "file") for f in msg["files"])
|
|
text = text or f"shared file(s): {names}"
|
|
elif not text and msg.get("attachments"):
|
|
text = "[attachment]"
|
|
elif not text and msg.get("blocks"):
|
|
text = "[rich message]"
|
|
|
|
files = []
|
|
for f in msg.get("files") or []:
|
|
file_id = f.get("id")
|
|
file_name = f.get("name")
|
|
rel = (
|
|
self.upload_relpath(file_id, file_name)
|
|
if file_id and file_name
|
|
else None
|
|
)
|
|
local_exists = False
|
|
if rel:
|
|
local_exists = (self.paths.root / rel).is_file()
|
|
files.append(
|
|
{
|
|
"id": file_id,
|
|
"name": file_name,
|
|
"mimetype": f.get("mimetype"),
|
|
"filetype": f.get("filetype"),
|
|
"url_private": f.get("url_private"),
|
|
"url_private_download": f.get("url_private_download"),
|
|
"local_relpath": rel,
|
|
"local_exists": local_exists,
|
|
}
|
|
)
|
|
|
|
ts = msg.get("ts")
|
|
thread_ts = msg.get("thread_ts")
|
|
is_thread_reply = bool(
|
|
thread_ts and ts and str(thread_ts) != str(ts)
|
|
)
|
|
thread_root_ts = str(thread_ts) if thread_ts else str(ts or "")
|
|
|
|
return {
|
|
"id": f"{channel_name}:{ts}",
|
|
"channel": channel_name,
|
|
"date_file": date_file,
|
|
"ts": ts,
|
|
"time": slack_ts_to_iso(ts),
|
|
"user_id": user_id,
|
|
"user_name": user_display(user),
|
|
"username": (user or {}).get("name"),
|
|
"is_bot": bool((user or {}).get("is_bot")) or bool(msg.get("bot_id")),
|
|
"subtype": subtype,
|
|
"type": msg.get("type"),
|
|
"text": text,
|
|
"thread_ts": thread_ts,
|
|
"thread_root_ts": thread_root_ts,
|
|
"is_thread_reply": is_thread_reply,
|
|
"is_thread_parent": bool(msg.get("reply_count")),
|
|
"reply_count": msg.get("reply_count") or 0,
|
|
"files": files,
|
|
"reactions": msg.get("reactions") or [],
|
|
"raw_has_blocks": bool(msg.get("blocks")),
|
|
}
|
|
|
|
def attach_thread_reply_counts(
|
|
self, messages: list[dict[str, Any]]
|
|
) -> list[dict[str, Any]]:
|
|
"""Add thread_reply_count on parent messages from export data."""
|
|
counts: dict[str, int] = {}
|
|
for msg in messages:
|
|
if msg.get("is_thread_reply"):
|
|
root = str(msg.get("thread_root_ts", ""))
|
|
counts[root] = counts.get(root, 0) + 1
|
|
for msg in messages:
|
|
ts = str(msg.get("ts", ""))
|
|
if msg.get("is_thread_parent") or counts.get(ts, 0) > 0:
|
|
msg["thread_reply_count"] = max(
|
|
int(msg.get("reply_count") or 0),
|
|
counts.get(ts, 0),
|
|
)
|
|
else:
|
|
msg["thread_reply_count"] = 0
|
|
return messages
|
|
|
|
def channel_stats(self, channel_name: str) -> dict[str, Any]:
|
|
folder = self.channel_folder(channel_name)
|
|
meta = self.channel_meta_by_name.get(channel_name, {})
|
|
if not folder:
|
|
return {
|
|
"name": channel_name,
|
|
"exists_in_export": False,
|
|
"message_count": 0,
|
|
"file_count": 0,
|
|
"first_date": None,
|
|
"last_date": None,
|
|
"is_archived": meta.get("is_archived", False),
|
|
"is_general": meta.get("is_general", False),
|
|
"purpose": (meta.get("purpose") or {}).get("value", ""),
|
|
"topic": (meta.get("topic") or {}).get("value", ""),
|
|
"member_count": len(meta.get("members") or []),
|
|
"slack_id": meta.get("id"),
|
|
}
|
|
|
|
dates = self.list_channel_date_files(channel_name)
|
|
message_count = 0
|
|
for date in dates:
|
|
message_count += len(self.read_messages_file(channel_name, date))
|
|
|
|
return {
|
|
"name": channel_name,
|
|
"exists_in_export": True,
|
|
"message_count": message_count,
|
|
"file_count": len(dates),
|
|
"first_date": dates[0] if dates else None,
|
|
"last_date": dates[-1] if dates else None,
|
|
"is_archived": meta.get("is_archived", False),
|
|
"is_general": meta.get("is_general", False),
|
|
"purpose": (meta.get("purpose") or {}).get("value", ""),
|
|
"topic": (meta.get("topic") or {}).get("value", ""),
|
|
"member_count": len(meta.get("members") or []),
|
|
"slack_id": meta.get("id"),
|
|
}
|
|
|
|
def build_summary(self) -> dict[str, Any]:
|
|
channels = []
|
|
total_messages = 0
|
|
for folder in self.channel_dirs():
|
|
stats = self.channel_stats(folder.name)
|
|
channels.append(stats)
|
|
total_messages += stats["message_count"]
|
|
|
|
users = []
|
|
for user in self.users_raw:
|
|
users.append(
|
|
{
|
|
"id": user.get("id"),
|
|
"name": user.get("name"),
|
|
"real_name": user_display(user),
|
|
"is_bot": bool(user.get("is_bot")),
|
|
"deleted": bool(user.get("deleted")),
|
|
"is_admin": bool(user.get("is_admin")),
|
|
}
|
|
)
|
|
|
|
return {
|
|
"export_path": str(self.paths.root),
|
|
"channel_count": len(channels),
|
|
"user_count": len(users),
|
|
"total_messages": total_messages,
|
|
"channels": channels,
|
|
"users": users,
|
|
}
|
|
|
|
@staticmethod
|
|
def classify_file(file: dict[str, Any]) -> str:
|
|
"""Rough file bucket for search/filter UI."""
|
|
mt = (file.get("mimetype") or "").lower()
|
|
name = (file.get("name") or "").lower()
|
|
ext = name.rsplit(".", 1)[-1] if "." in name else ""
|
|
|
|
if mt.startswith("image/") or ext in {
|
|
"png",
|
|
"jpg",
|
|
"jpeg",
|
|
"gif",
|
|
"webp",
|
|
"bmp",
|
|
"svg",
|
|
"heic",
|
|
}:
|
|
return "image"
|
|
if ext == "pdf" or mt == "application/pdf":
|
|
return "pdf"
|
|
if ext in {"doc", "docx"} or "word" in mt:
|
|
return "doc"
|
|
if ext in {"xls", "xlsx", "csv"} or "spreadsheet" in mt or "excel" in mt:
|
|
return "spreadsheet"
|
|
if ext in {"ppt", "pptx"}:
|
|
return "slides"
|
|
if ext in {"zip", "rar", "7z", "gz"}:
|
|
return "archive"
|
|
if ext in {"mp4", "mov", "webm", "avi"} or mt.startswith("video/"):
|
|
return "video"
|
|
if ext in {"mp3", "m4a", "wav"} or mt.startswith("audio/"):
|
|
return "audio"
|
|
if ext in {"txt", "md", "log", "json"} or mt.startswith("text/"):
|
|
return "text"
|
|
return "other"
|
|
|
|
def message_file_kinds(self, msg: dict[str, Any]) -> set[str]:
|
|
return {self.classify_file(f) for f in msg.get("files") or []}
|
|
|
|
def search_files_global(
|
|
self,
|
|
query: str = "",
|
|
*,
|
|
file_type: str = "all",
|
|
limit: int = 80,
|
|
) -> dict[str, Any]:
|
|
"""Find messages with attachments across all channels."""
|
|
q = query.strip()
|
|
pattern = re.compile(re.escape(q), re.IGNORECASE) if q else None
|
|
type_filter = (file_type or "all").strip().lower()
|
|
hits: list[dict[str, Any]] = []
|
|
|
|
for folder in self.channel_dirs():
|
|
if len(hits) >= limit:
|
|
break
|
|
channel_name = folder.name
|
|
for date in self.list_channel_date_files(channel_name):
|
|
if len(hits) >= limit:
|
|
break
|
|
for raw in self.read_messages_file(channel_name, date):
|
|
normalized = self.normalize_message(raw, channel_name, date)
|
|
files = normalized.get("files") or []
|
|
if not files:
|
|
continue
|
|
kinds = self.message_file_kinds(normalized)
|
|
if type_filter != "all" and type_filter not in kinds:
|
|
continue
|
|
if pattern:
|
|
hay = " ".join(
|
|
[
|
|
normalized.get("text") or "",
|
|
normalized.get("user_name") or "",
|
|
channel_name,
|
|
" ".join(f.get("name") or "" for f in files),
|
|
]
|
|
)
|
|
if not pattern.search(hay):
|
|
continue
|
|
matching = files
|
|
if type_filter != "all":
|
|
matching = [
|
|
f
|
|
for f in files
|
|
if self.classify_file(f) == type_filter
|
|
]
|
|
hits.append(
|
|
{
|
|
**normalized,
|
|
"matching_files": matching,
|
|
"file_kinds": sorted(kinds),
|
|
}
|
|
)
|
|
if len(hits) >= limit:
|
|
break
|
|
|
|
return {
|
|
"query": q,
|
|
"file_type": type_filter,
|
|
"hits": hits,
|
|
"total": len(hits),
|
|
}
|
|
|
|
def search_channels(self, query: str) -> list[dict[str, Any]]:
|
|
q = query.strip().lower()
|
|
summary = self.build_summary()["channels"]
|
|
if not q:
|
|
return summary
|
|
return [
|
|
c
|
|
for c in summary
|
|
if q in c["name"].lower()
|
|
or q in (c.get("purpose") or "").lower()
|
|
or q in (c.get("topic") or "").lower()
|
|
]
|
|
|
|
def iter_channel_messages(self, channel_name: str) -> list[dict[str, Any]]:
|
|
"""All messages in a channel, oldest first."""
|
|
messages: list[dict[str, Any]] = []
|
|
for date in self.list_channel_date_files(channel_name):
|
|
for raw in self.read_messages_file(channel_name, date):
|
|
messages.append(self.normalize_message(raw, channel_name, date))
|
|
|
|
def sort_key(msg: dict[str, Any]) -> float:
|
|
try:
|
|
return float(str(msg.get("ts", "0")).split(".")[0])
|
|
except (TypeError, ValueError):
|
|
return 0.0
|
|
|
|
messages.sort(key=sort_key)
|
|
return self.attach_thread_reply_counts(messages)
|
|
|
|
def channel_messages_all(self, channel_name: str) -> dict[str, Any]:
|
|
messages = self.iter_channel_messages(channel_name)
|
|
thread_parents = sum(
|
|
1 for m in messages if m.get("is_thread_parent") or m.get("thread_reply_count")
|
|
)
|
|
with_files = sum(1 for m in messages if m.get("files"))
|
|
return {
|
|
"channel": channel_name,
|
|
"messages": messages,
|
|
"total": len(messages),
|
|
"thread_parents": thread_parents,
|
|
"messages_with_files": with_files,
|
|
}
|
|
|
|
def search_global(
|
|
self, query: str, *, limit: int = 80
|
|
) -> dict[str, Any]:
|
|
pattern = re.compile(re.escape(query.strip()), re.IGNORECASE)
|
|
if not query.strip():
|
|
return {"query": query, "channels": [], "messages": []}
|
|
|
|
channel_hits: list[dict[str, Any]] = []
|
|
message_hits: list[dict[str, Any]] = []
|
|
|
|
for ch in self.build_summary()["channels"]:
|
|
name = ch["name"]
|
|
if pattern.search(name) or pattern.search(ch.get("purpose") or ""):
|
|
channel_hits.append(ch)
|
|
|
|
for folder in self.channel_dirs():
|
|
if len(message_hits) >= limit:
|
|
break
|
|
name = folder.name
|
|
for date in self.list_channel_date_files(name):
|
|
if len(message_hits) >= limit:
|
|
break
|
|
for raw in self.read_messages_file(name, date):
|
|
normalized = self.normalize_message(raw, name, date)
|
|
if pattern.search(normalized.get("text") or ""):
|
|
message_hits.append(normalized)
|
|
if len(message_hits) >= limit:
|
|
break
|
|
|
|
return {
|
|
"query": query,
|
|
"channels": channel_hits[:30],
|
|
"messages": message_hits,
|
|
}
|
|
|
|
def channel_messages_page(
|
|
self,
|
|
channel_name: str,
|
|
*,
|
|
date: str | None = None,
|
|
offset: int = 0,
|
|
limit: int = 100,
|
|
) -> dict[str, Any]:
|
|
dates = self.list_channel_date_files(channel_name)
|
|
if not dates:
|
|
return {
|
|
"channel": channel_name,
|
|
"dates": [],
|
|
"date": None,
|
|
"messages": [],
|
|
"offset": 0,
|
|
"limit": limit,
|
|
"total_in_date": 0,
|
|
"has_more": False,
|
|
}
|
|
|
|
active_date = date if date in dates else dates[-1]
|
|
raw = self.read_messages_file(channel_name, active_date)
|
|
normalized = [
|
|
self.normalize_message(m, channel_name, active_date) for m in raw
|
|
]
|
|
page = normalized[offset : offset + limit]
|
|
return {
|
|
"channel": channel_name,
|
|
"dates": dates,
|
|
"date": active_date,
|
|
"messages": page,
|
|
"offset": offset,
|
|
"limit": limit,
|
|
"total_in_date": len(normalized),
|
|
"has_more": offset + limit < len(normalized),
|
|
}
|
|
|
|
def search_messages(
|
|
self, channel_name: str, query: str, *, limit: int = 50
|
|
) -> list[dict[str, Any]]:
|
|
pattern = re.compile(re.escape(query.strip()), re.IGNORECASE)
|
|
if not query.strip():
|
|
return []
|
|
hits: list[dict[str, Any]] = []
|
|
for date in reversed(self.list_channel_date_files(channel_name)):
|
|
for msg in self.read_messages_file(channel_name, date):
|
|
normalized = self.normalize_message(msg, channel_name, date)
|
|
if pattern.search(normalized.get("text") or ""):
|
|
hits.append(normalized)
|
|
if len(hits) >= limit:
|
|
return hits
|
|
return hits
|