diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..282fcf1 --- /dev/null +++ b/.gitignore @@ -0,0 +1,16 @@ +# Local credentials and session state +config.json +cookies.pkl + +# Export output and search index +results/ +mattermost_index.sqlite + +# Python +.venv/ +__pycache__/ +*.py[cod] +*.pyo + +# IDE +.cursor/ diff --git a/mattermost-dl.py b/mattermost-dl.py index 1de62b4..083ea9b 100644 --- a/mattermost-dl.py +++ b/mattermost-dl.py @@ -1,7 +1,9 @@ +import argparse import os import pickle import random import sqlite3 +import sys import time from datetime import datetime, date, timezone from typing import Tuple, Dict, List, Optional @@ -246,6 +248,15 @@ def _login_to_mattermost_via_gitlab(session: requests.Session, host: str, gitlab return "MMAUTHTOKEN" in session.cookies +def try_get_token_from_cookies(host: str, cookie_file: str = COOKIE_FILE_DEFAULT) -> Optional[str]: + session = requests.Session() + if not _load_cookies(session, cookie_file): + return None + if not _get_current_user(session, host): + return None + return session.cookies.get("MMAUTHTOKEN") + + def get_token_via_gitlab(host: str, gitlab_url: str, username: str, password: str, cookie_file: str = COOKIE_FILE_DEFAULT) -> Optional[str]: session = requests.Session() @@ -310,6 +321,71 @@ def save_users_json(output_base: str, user_profiles: List[dict]) -> str: return str(output_path) +def _enrich_direct_message_channels(d: Driver, channels: List[dict], my_user_id: str, + user_id_to_name: Dict[str, str]) -> None: + for channel in channels: + if channel["type"] != "D": + continue + + user_ids = channel["name"].split("__") + other_user_id = user_ids[1] if user_ids[0] == my_user_id else user_ids[0] + if other_user_id not in user_id_to_name: + try: + user_id_to_name[other_user_id] = d.users.get_user(other_user_id)["username"] + except Exception: + user_id_to_name[other_user_id] = channel.get("display_name") or other_user_id + channel["display_name"] = user_id_to_name[other_user_id] + + +def get_channels_for_team(d: Driver, team: dict, my_user_id: str, + user_id_to_name: Dict[str, str]) -> List[dict]: + channels = d.channels.get_channels_for_user(my_user_id, team["id"]) + _enrich_direct_message_channels(d, channels, my_user_id, user_id_to_name) + return sorted(channels, key=lambda x: x["display_name"].lower()) + + +def find_team(d: Driver, my_user_id: str, team_name: str) -> dict: + teams = d.teams.get_user_teams(my_user_id) + for team in teams: + if team["name"] == team_name: + return team + available = ", ".join(team["name"] for team in teams) + raise SystemExit(f"Team {team_name!r} not found. Available teams: {available}") + + +def find_channels(channels: List[dict], channel_spec: str) -> List[dict]: + if channel_spec.lower() == "all": + return channels + + requested = [part.strip() for part in channel_spec.split(",") if part.strip()] + if not requested: + raise SystemExit("No channels specified (use channel names or 'all').") + + by_name = {} + by_display = {} + for channel in channels: + by_name[channel["name"].lower()] = channel + by_display[channel["display_name"].lower()] = channel + + selected = [] + missing = [] + for name in requested: + key = name.lower() + channel = by_name.get(key) or by_display.get(key) + if channel is None: + missing.append(name) + else: + selected.append(channel) + + if missing: + available = ", ".join(channel["display_name"] for channel in channels) + raise SystemExit( + f"Channel(s) not found: {', '.join(missing)}. " + f"Available channels: {available}" + ) + return selected + + def select_team(d: Driver, my_user_id: str) -> str: print("Downloading all team information... ", end="") teams = d.teams.get_user_teams(my_user_id) @@ -325,23 +401,7 @@ def select_team(d: Driver, my_user_id: str) -> str: def select_channel(d: Driver, team: str, my_user_id: str, user_id_to_name: Dict[str, str], verbose: bool = False) -> List[str]: print("Downloading all channel information... ", end="") - channels = d.channels.get_channels_for_user(my_user_id, team["id"]) - # Add display name to direct messages - for channel in channels: - if channel["type"] != "D": - continue - - # The channel name consists of two user ids connected by a double underscore - user_ids = channel["name"].split("__") - other_user_id = user_ids[1] if user_ids[0] == my_user_id else user_ids[0] - if other_user_id not in user_id_to_name: - try: - user_id_to_name[other_user_id] = d.users.get_user(other_user_id)["username"] - except Exception: - user_id_to_name[other_user_id] = channel.get("display_name") or other_user_id - channel["display_name"] = user_id_to_name[other_user_id] - # Sort channels by name for easier search - channels = sorted(channels, key=lambda x: x["display_name"].lower()) + channels = get_channels_for_team(d, team, my_user_id, user_id_to_name) print(f"Found {len(channels)} channels!") for i_channel, channel in enumerate(channels): @@ -561,7 +621,47 @@ def get_config_from_json(config_filename: str = "config.json") -> dict: return config -def complete_config(config: dict, config_filename: str = "config.json") -> dict: +def complete_config_noninteractive(config: dict) -> dict: + required = ("host", "login_mode") + missing = [key for key in required if not config.get(key)] + if missing: + raise SystemExit( + f"Missing required config field(s) for non-interactive mode: {', '.join(missing)}" + ) + + login_mode = config["login_mode"] + if login_mode not in ("password", "token", "gitlab"): + raise SystemExit(f"Unsupported login_mode in config: {login_mode!r}") + + token = config.get("token") + if login_mode == "gitlab": + token = token or try_get_token_from_cookies( + config["host"], config.get("cookie_file", COOKIE_FILE_DEFAULT) + ) + if token: + config["token"] = token + elif login_mode == "password": + if not config.get("username") or not config.get("password"): + raise SystemExit( + "Non-interactive password login requires 'username' and 'password' in config." + ) + else: + raise SystemExit( + "No valid token in config or cookies. " + "Run interactively once to authenticate." + ) + + if "download_files" not in config: + config["download_files"] = True + + config["password"] = None + return config + + +def complete_config(config: dict, config_filename: str = "config.json", *, interactive: bool = True) -> dict: + if not interactive: + return complete_config_noninteractive(config) + config_changed = False if config.get("host", False): print(f"Using host '{config['host']}' from config") @@ -709,13 +809,48 @@ def find_mmauthtoken_firefox(host): return None +def parse_args(argv: Optional[List[str]] = None) -> argparse.Namespace: + parser = argparse.ArgumentParser( + description="Download Mattermost channel history to local JSON files.", + ) + parser.add_argument( + "--config", + default="config.json", + help="Path to config file (default: config.json).", + ) + parser.add_argument( + "--team", + help="Team name for non-interactive export (e.g. esr-daily).", + ) + parser.add_argument( + "--channel", + help="Channel name(s), comma-separated, or 'all' for every channel in the team.", + ) + parser.add_argument( + "--output", + help="Output directory (overrides config output_base and default date-based path).", + ) + return parser.parse_args(argv) + + if __name__ == '__main__': - config_filename = "config.json" + args = parse_args() + non_interactive = bool(args.team or args.channel) + if non_interactive and not (args.team and args.channel): + print("Non-interactive mode requires both --team and --channel.", file=sys.stderr) + raise SystemExit(2) + + config_filename = args.config config = get_config_from_json(config_filename) - config = complete_config(config, config_filename) + config = complete_config(config, config_filename, interactive=not non_interactive) incremental = config.get("incremental", bool(config.get("last_export_at"))) - if incremental and config.get("last_export_at"): + if args.output: + output_base = args.output + after = config.get("after") + before = config.get("before") + merge = False + elif incremental and config.get("last_export_at"): print(f"Incremental update from {config['last_export_at']}") output_base = config.get("output_base") or ("results/" + date.today().strftime("%Y%m%d")) after = config["last_export_at"] @@ -739,6 +874,8 @@ if __name__ == '__main__': ) apply_throttle(d, config) except Exception as e: + if non_interactive: + raise SystemExit(f"Authentication failed: {e}") from e # If token-based authentication failed, drop the token and ask # the user to rerun the script so that fresh credentials can be # obtained or entered. @@ -751,8 +888,16 @@ if __name__ == '__main__': raise user_id_to_name, my_user_id, user_profiles = get_users(d) save_users_json(output_base, user_profiles) - team = select_team(d, my_user_id) - channels = select_channel(d, team, my_user_id, user_id_to_name) + if non_interactive: + team = find_team(d, my_user_id, args.team) + print(f"Selected team {team['name']}") + channels = get_channels_for_team(d, team, my_user_id, user_id_to_name) + print(f"Found {len(channels)} channels") + channels = find_channels(channels, args.channel) + print("Selected channel(s):", ", ".join(channel["display_name"] for channel in channels)) + else: + team = select_team(d, my_user_id) + channels = select_channel(d, team, my_user_id, user_id_to_name) for i_channel, channel in enumerate(channels): print(f"Start export of channel {i_channel + 1}/{len(channels)}") export_channel(d, channel, user_id_to_name, output_base, config["download_files"], diff --git a/renew.sh b/renew.sh new file mode 100755 index 0000000..7e75059 --- /dev/null +++ b/renew.sh @@ -0,0 +1,2 @@ +cd /home/grayhook/Documents/mattermost-dl && .venv/bin/python mattermost-dl.py --team esr-daily --channel all +cd /home/grayhook/Documents/mattermost-dl && .venv/bin/python mm_index.py --reindex-all