"""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