Сергей Маринкевич
767 lines
29 KiB
Python
767 lines
29 KiB
Python
import os
|
|
import pickle
|
|
import random
|
|
import sqlite3
|
|
import time
|
|
from datetime import datetime, date, timezone
|
|
from typing import Tuple, Dict, List, Optional
|
|
import getpass
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
from mattermostdriver import Driver
|
|
from mattermostdriver.exceptions import ResourceNotFound
|
|
import pathlib
|
|
import json
|
|
|
|
GITLAB_URL_DEFAULT = "https://mmgit.eltex-co.ru"
|
|
COOKIE_FILE_DEFAULT = "cookies.pkl"
|
|
|
|
THROTTLE_DEFAULTS = {
|
|
"min_delay": 0.4,
|
|
"max_delay": 1.8,
|
|
"burst_pause_every": 12,
|
|
"burst_pause_min": 3.0,
|
|
"burst_pause_max": 8.0,
|
|
"file_delay_multiplier": 1.5,
|
|
}
|
|
|
|
RETRY_DEFAULTS = {
|
|
"max_attempts": 5,
|
|
"base_delay": 2.0,
|
|
"max_delay": 60.0,
|
|
}
|
|
|
|
RETRYABLE_REQUEST_ERRORS = (
|
|
requests.exceptions.ConnectionError,
|
|
requests.exceptions.Timeout,
|
|
requests.exceptions.ChunkedEncodingError,
|
|
)
|
|
|
|
|
|
class RequestThrottler:
|
|
"""Paces API calls with jitter and occasional longer pauses."""
|
|
|
|
def __init__(self, **opts):
|
|
self.min_delay = opts["min_delay"]
|
|
self.max_delay = opts["max_delay"]
|
|
self.burst_pause_every = opts["burst_pause_every"]
|
|
self.burst_pause_min = opts["burst_pause_min"]
|
|
self.burst_pause_max = opts["burst_pause_max"]
|
|
self.file_delay_multiplier = opts["file_delay_multiplier"]
|
|
self._request_count = 0
|
|
self._last_request_at = 0.0
|
|
|
|
def wait(self, endpoint: str = "") -> None:
|
|
self._request_count += 1
|
|
if self.burst_pause_every and self._request_count % self.burst_pause_every == 0:
|
|
delay = random.uniform(self.burst_pause_min, self.burst_pause_max)
|
|
else:
|
|
delay = random.uniform(self.min_delay, self.max_delay)
|
|
|
|
if endpoint.startswith("/files/"):
|
|
delay *= self.file_delay_multiplier
|
|
|
|
elapsed = time.monotonic() - self._last_request_at
|
|
if elapsed < delay:
|
|
time.sleep(delay - elapsed)
|
|
self._last_request_at = time.monotonic()
|
|
|
|
|
|
def _parse_retry_config(config: dict) -> Optional[dict]:
|
|
retry = config.get("retry")
|
|
if retry is False:
|
|
return None
|
|
if retry is None or retry is True:
|
|
return dict(RETRY_DEFAULTS)
|
|
if isinstance(retry, dict):
|
|
if not retry.get("enabled", True):
|
|
return None
|
|
opts = dict(RETRY_DEFAULTS)
|
|
opts.update({k: v for k, v in retry.items() if k != "enabled"})
|
|
return opts
|
|
return None
|
|
|
|
|
|
def _parse_throttle_config(config: dict) -> Optional[dict]:
|
|
throttle = config.get("throttle")
|
|
if not throttle:
|
|
return None
|
|
if throttle is True:
|
|
return dict(THROTTLE_DEFAULTS)
|
|
if isinstance(throttle, dict):
|
|
if not throttle.get("enabled", True):
|
|
return None
|
|
opts = dict(THROTTLE_DEFAULTS)
|
|
opts.update({k: v for k, v in throttle.items() if k != "enabled"})
|
|
return opts
|
|
return None
|
|
|
|
|
|
def apply_throttle(d: Driver, config: dict) -> None:
|
|
throttle_opts = _parse_throttle_config(config)
|
|
retry_opts = _parse_retry_config(config)
|
|
if not throttle_opts and not retry_opts:
|
|
return
|
|
|
|
throttler = RequestThrottler(**(throttle_opts or THROTTLE_DEFAULTS))
|
|
client = d.client
|
|
original_make_request = client.make_request
|
|
|
|
def throttled_make_request(method, endpoint, options=None, params=None, data=None, files=None, basepath=None):
|
|
if throttle_opts:
|
|
throttler.wait(endpoint)
|
|
|
|
attempts = retry_opts["max_attempts"] if retry_opts else 1
|
|
last_error = None
|
|
for attempt in range(attempts):
|
|
try:
|
|
return original_make_request(method, endpoint, options, params, data, files, basepath)
|
|
except RETRYABLE_REQUEST_ERRORS as error:
|
|
last_error = error
|
|
if attempt + 1 >= attempts:
|
|
break
|
|
delay = min(
|
|
retry_opts["base_delay"] * (2 ** attempt),
|
|
retry_opts["max_delay"],
|
|
) + random.uniform(0, 1)
|
|
print(
|
|
f"Request failed ({error}), retrying in {delay:.1f}s "
|
|
f"({attempt + 2}/{attempts})..."
|
|
)
|
|
time.sleep(delay)
|
|
raise last_error
|
|
|
|
client.make_request = throttled_make_request
|
|
if throttle_opts:
|
|
print(
|
|
"Request throttling enabled "
|
|
f"(delay {throttle_opts['min_delay']}-{throttle_opts['max_delay']}s, "
|
|
f"pause every {throttle_opts['burst_pause_every']} requests)"
|
|
)
|
|
if retry_opts:
|
|
print(
|
|
"Request retries enabled "
|
|
f"(up to {retry_opts['max_attempts']} attempts, "
|
|
f"backoff from {retry_opts['base_delay']}s)"
|
|
)
|
|
|
|
|
|
def connect(host: str, login_token: str = None, username: str = None, password: str = None) -> Driver:
|
|
d = Driver({
|
|
"url": host,
|
|
"port": 443,
|
|
"token": login_token,
|
|
"username": username,
|
|
"password": password
|
|
})
|
|
d.login()
|
|
return d
|
|
|
|
|
|
def _mattermost_url(host: str) -> str:
|
|
return f"https://{host}"
|
|
|
|
|
|
def _save_cookies(session: requests.Session, filename: str) -> None:
|
|
try:
|
|
with open(filename, "wb") as f:
|
|
pickle.dump(session.cookies, f)
|
|
except OSError as e:
|
|
print(f"Warning: failed to save cookies to {filename}: {e}")
|
|
|
|
|
|
def _load_cookies(session: requests.Session, filename: str) -> bool:
|
|
if not os.path.exists(filename):
|
|
return False
|
|
try:
|
|
with open(filename, "rb") as f:
|
|
session.cookies.update(pickle.load(f))
|
|
return True
|
|
except (OSError, pickle.PickleError) as e:
|
|
print(f"Warning: failed to load cookies from {filename}: {e}")
|
|
return False
|
|
|
|
|
|
def _remove_cookie_file(filename: str) -> None:
|
|
try:
|
|
if os.path.exists(filename):
|
|
os.remove(filename)
|
|
print(f"Removed invalid cookie file: {filename}")
|
|
except OSError as e:
|
|
print(f"Warning: failed to remove cookie file {filename}: {e}")
|
|
|
|
|
|
def _get_current_user(session: requests.Session, host: str) -> Optional[dict]:
|
|
try:
|
|
response = session.get(f"{_mattermost_url(host)}/api/v4/users/me", timeout=10)
|
|
return response.json() if response.status_code == 200 else None
|
|
except requests.exceptions.RequestException:
|
|
return None
|
|
|
|
|
|
def _login_to_mattermost_via_gitlab(session: requests.Session, host: str, gitlab_url: str,
|
|
username: str, password: str) -> bool:
|
|
login_url = f"{_mattermost_url(host)}/oauth/gitlab/login"
|
|
headers = {"User-Agent": "Mozilla/5.0"}
|
|
|
|
try:
|
|
response = session.get(login_url, headers=headers, allow_redirects=True, timeout=15)
|
|
response.raise_for_status()
|
|
|
|
if "/users/sign_in" in response.url or "/users/auth/ldapmain" in response.url:
|
|
soup = BeautifulSoup(response.text, "html.parser")
|
|
auth_form = soup.find("form", {"id": "new_ldap_user"})
|
|
if not auth_form:
|
|
return False
|
|
auth_token_input = auth_form.find("input", {"name": "authenticity_token"})
|
|
if not auth_token_input:
|
|
return False
|
|
|
|
auth_token = auth_token_input["value"]
|
|
payload = {"authenticity_token": auth_token, "username": username, "password": password}
|
|
callback_url = f"{gitlab_url}/users/auth/ldapmain/callback"
|
|
response = session.post(
|
|
callback_url, data=payload, headers={"Referer": response.url},
|
|
allow_redirects=True, timeout=15
|
|
)
|
|
response.raise_for_status()
|
|
|
|
if response.status_code == 200 and "/oauth/authorize" in response.url:
|
|
soup = BeautifulSoup(response.text, "html.parser")
|
|
redirect_link = soup.find("a", href=True)
|
|
if not redirect_link:
|
|
return False
|
|
|
|
response = session.get(
|
|
redirect_link["href"], headers={"Referer": response.url},
|
|
allow_redirects=True, timeout=15
|
|
)
|
|
response.raise_for_status()
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
print(f"GitLab login failed: {e}")
|
|
return False
|
|
|
|
return "MMAUTHTOKEN" in session.cookies
|
|
|
|
|
|
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()
|
|
|
|
if _load_cookies(session, cookie_file):
|
|
print("Checking saved GitLab/Mattermost session...")
|
|
if _get_current_user(session, host):
|
|
token = session.cookies.get("MMAUTHTOKEN")
|
|
if token:
|
|
print("Authenticated using saved cookies.")
|
|
return token
|
|
print("Saved cookies are invalid or expired.")
|
|
_remove_cookie_file(cookie_file)
|
|
|
|
print("Logging in to Mattermost via GitLab...")
|
|
if not _login_to_mattermost_via_gitlab(session, host, gitlab_url, username, password):
|
|
print("Failed to authenticate via GitLab.")
|
|
return None
|
|
|
|
_save_cookies(session, cookie_file)
|
|
print(f"Session saved to {cookie_file}")
|
|
return session.cookies.get("MMAUTHTOKEN")
|
|
|
|
|
|
def get_users(d: Driver) -> Tuple[Dict[str, str], str, List[dict]]:
|
|
my_user = d.users.get_user("me")
|
|
my_username = my_user["username"]
|
|
my_user_id = my_user["id"]
|
|
print(f"Successfully logged in as {my_username} ({my_user_id})")
|
|
|
|
# Get all usernames as we want to use those instead of the user ids
|
|
user_id_to_name = {}
|
|
user_profiles = []
|
|
page = 0
|
|
print("Downloading all user information... ", end="")
|
|
while True:
|
|
users_resp = d.users.get_users(params={"per_page": 200, "page": page})
|
|
if len(users_resp) == 0:
|
|
break
|
|
for user in users_resp:
|
|
user_id_to_name[user["id"]] = user["username"]
|
|
user_profiles.append(
|
|
{
|
|
"username": user.get("username") or "",
|
|
"first_name": user.get("first_name") or "",
|
|
"last_name": user.get("last_name") or "",
|
|
"nickname": user.get("nickname") or "",
|
|
}
|
|
)
|
|
page += 1
|
|
print(f"Found {len(user_id_to_name)} users!")
|
|
|
|
return user_id_to_name, my_user_id, user_profiles
|
|
|
|
|
|
def save_users_json(output_base: str, user_profiles: List[dict]) -> str:
|
|
output_path = pathlib.Path(output_base) / "users.json"
|
|
output_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(output_path, "w", encoding="utf-8") as f:
|
|
json.dump(user_profiles, f, indent=2, ensure_ascii=False)
|
|
print(f"Saved user directory to '{output_path}'")
|
|
return str(output_path)
|
|
|
|
|
|
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)
|
|
print(f"Found {len(teams)} teams!")
|
|
for i_team, team in enumerate(teams):
|
|
print(f"{i_team}\t{team['name']}\t({team['id']})")
|
|
team_idx = int(input("Select team by idx: "))
|
|
team = teams[team_idx]
|
|
print(f"Selected team {team['name']}")
|
|
return team
|
|
|
|
|
|
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())
|
|
print(f"Found {len(channels)} channels!")
|
|
|
|
for i_channel, channel in enumerate(channels):
|
|
if verbose:
|
|
channel_id = f"\t({channel['id']})"
|
|
else:
|
|
channel_id = ""
|
|
print(f"{i_channel}\t{channel['display_name']}{channel_id}")
|
|
channel_input = input("Select channels by idx separated by comma or type 'all' for downloading all channels: ")
|
|
if channel_input == "all":
|
|
channel_idxs = list(range(len(channels)))
|
|
else:
|
|
channel_idxs = channel_input.replace(" ", "").split(",")
|
|
selected_channels = [channels[int(idx)] for idx in channel_idxs]
|
|
print("Selected channel(s):", ", ".join([channel["display_name"] for channel in selected_channels]))
|
|
return selected_channels
|
|
|
|
|
|
def _resolve_team_name(d: Driver, channel: dict, fallback_team: Optional[dict] = None) -> str:
|
|
team_id = channel.get("team_id") or ""
|
|
if team_id:
|
|
try:
|
|
return d.teams.get_team(team_id)["name"]
|
|
except ResourceNotFound:
|
|
pass
|
|
if fallback_team:
|
|
return fallback_team["name"]
|
|
return ""
|
|
|
|
|
|
def _parse_filter_timestamp(s: str) -> float:
|
|
for fmt in ('%Y-%m-%dT%H:%M:%SZ', '%Y-%m-%d'):
|
|
try:
|
|
dt = datetime.strptime(s, fmt)
|
|
if dt.tzinfo is None:
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
|
return dt.timestamp()
|
|
except ValueError:
|
|
continue
|
|
raise ValueError(f"Invalid date format: {s!r} (expected YYYY-MM-DD or YYYY-MM-DDTHH:MM:SSZ)")
|
|
|
|
|
|
def save_config(config: dict, config_filename: str = "config.json") -> None:
|
|
to_save = {k: v for k, v in config.items() if k != "password"}
|
|
with open(config_filename, "w") as f:
|
|
json.dump(to_save, f, indent=2)
|
|
|
|
|
|
def export_channel(d: Driver, channel: str, user_id_to_name: Dict[str, str], output_base: str,
|
|
download_files: bool = True, before: str = None, after: str = None,
|
|
team: Optional[dict] = None, merge: bool = False):
|
|
# Sanitize channel name
|
|
channel_name = channel["display_name"].replace("\\", "").replace("/", "")
|
|
|
|
print("Exporting channel", channel_name)
|
|
after_ts = _parse_filter_timestamp(after) if after else None
|
|
before_ts = _parse_filter_timestamp(before) if before else None
|
|
filtered_channel_name = ''.join(filter(lambda ch: ch not in "?!/\\.;:*\"<>|", channel_name))
|
|
output_filepath = pathlib.Path(output_base) / channel_name / (filtered_channel_name + ".json")
|
|
if merge and after_ts is not None and not output_filepath.exists():
|
|
print(" no existing export for this channel, fetching full history")
|
|
after_ts = None
|
|
elif after:
|
|
print(f" fetching posts after {after}")
|
|
|
|
# Get all posts for selected channel
|
|
page = 0
|
|
all_posts = []
|
|
while True:
|
|
print(f"Requesting channel page {page}")
|
|
params = {"per_page": 200, "page": page}
|
|
if after_ts is not None:
|
|
# Ask server only for posts changed since last export.
|
|
# For incremental updates we typically expect few new posts,
|
|
# so a single page is enough and avoids walking the whole history.
|
|
params["since"] = int(after_ts * 1000)
|
|
posts = d.posts.get_posts_for_channel(channel["id"], params=params)
|
|
|
|
if len(posts["posts"]) == 0:
|
|
# If no posts are returned, we have reached the end
|
|
break
|
|
|
|
all_posts.extend([posts["posts"][post] for post in posts["order"]])
|
|
|
|
# In incremental mode (after_ts is set) we intentionally stop
|
|
# after the first non-empty page to avoid downloading the
|
|
# entire channel history again. For full dumps, keep paging.
|
|
if after_ts is not None:
|
|
break
|
|
|
|
page += 1
|
|
print(f"Found {len(all_posts)} posts")
|
|
# Create output directory
|
|
output_base = pathlib.Path(output_base) / channel_name
|
|
output_base.mkdir(parents=True, exist_ok=True)
|
|
|
|
existing_post_ids = set()
|
|
if merge and output_filepath.exists():
|
|
with open(output_filepath, encoding='utf8') as f:
|
|
existing_export = json.load(f)
|
|
existing_post_ids = {post["id"] for post in existing_export.get("posts", [])}
|
|
if existing_post_ids:
|
|
print(f" skipping {len(existing_post_ids)} posts already exported")
|
|
|
|
# Simplify all posts to contain only username, date, message and files in chronological order
|
|
simple_posts = []
|
|
for i_post, post in enumerate(reversed(all_posts)):
|
|
|
|
# Filter posts by date range
|
|
created = post["create_at"] / 1000
|
|
if (before_ts and created > before_ts) or (after_ts and created <= after_ts):
|
|
continue
|
|
|
|
if post["id"] in existing_post_ids:
|
|
continue
|
|
|
|
user_id = post["user_id"]
|
|
if user_id not in user_id_to_name:
|
|
user_id_to_name[user_id] = d.users.get_user(user_id)["username"]
|
|
username = user_id_to_name[user_id]
|
|
created = datetime.fromtimestamp(
|
|
post["create_at"] / 1000, timezone.utc
|
|
).strftime('%Y-%m-%dT%H:%M:%SZ')
|
|
message = post["message"]
|
|
simple_post = dict(idx=i_post, id=post["id"], created=created, username=username, message=message)
|
|
|
|
# If a code block is given in the message, dump it to file
|
|
if message.count("```") > 1:
|
|
start_pos = message.find("```") + 3
|
|
end_pos = message.rfind("```")
|
|
|
|
cut = message[start_pos:end_pos]
|
|
if not len(cut):
|
|
print("Code has no length")
|
|
else:
|
|
filename = "%03d" % i_post + "_code.txt"
|
|
with open(output_base / filename, "wb") as f:
|
|
f.write(cut.encode())
|
|
|
|
# If any files are attached to the message, download each
|
|
if "files" in post["metadata"]:
|
|
filenames = []
|
|
for file in post["metadata"]["files"]:
|
|
if download_files:
|
|
filename = "%03d" % i_post + "_" + file["name"]
|
|
dest = output_base / filename
|
|
if merge and dest.exists():
|
|
print(f"Skipping already downloaded {file['name']}")
|
|
else:
|
|
print("Downloading", file["name"])
|
|
try:
|
|
resp = d.files.get_file(file["id"])
|
|
except ResourceNotFound:
|
|
print(f"File not found, skipping: {file['name']}")
|
|
filenames.append(file["name"])
|
|
continue
|
|
# Mattermost Driver unfortunately parses json files to dicts
|
|
if isinstance(resp, dict):
|
|
with open(dest, "w") as f:
|
|
json.dump(resp, f)
|
|
else:
|
|
with open(dest, "wb") as f:
|
|
f.write(resp.content)
|
|
|
|
filenames.append(file["name"])
|
|
simple_post["files"] = filenames
|
|
simple_posts.append(simple_post)
|
|
|
|
output_filename = filtered_channel_name + ".json"
|
|
output_filepath = output_base / output_filename
|
|
|
|
merged_posts = simple_posts
|
|
if merge and output_filepath.exists():
|
|
with open(output_filepath, encoding='utf8') as f:
|
|
existing = json.load(f)
|
|
seen_ids = {post["id"] for post in existing.get("posts", [])}
|
|
merged_posts = list(existing.get("posts", []))
|
|
new_count = 0
|
|
for post in simple_posts:
|
|
if post["id"] not in seen_ids:
|
|
merged_posts.append(post)
|
|
seen_ids.add(post["id"])
|
|
new_count += 1
|
|
merged_posts.sort(key=lambda p: p["created"])
|
|
for i, post in enumerate(merged_posts):
|
|
post["idx"] = i
|
|
print(f" merged {new_count} new posts ({len(merged_posts)} total)")
|
|
elif merge:
|
|
print(f" no existing export at '{output_filepath}', writing fresh")
|
|
|
|
output = {
|
|
"channel": {
|
|
"name": channel["name"],
|
|
"display_name": channel["display_name"],
|
|
"header": channel["header"],
|
|
"id": channel["id"],
|
|
"team": _resolve_team_name(d, channel, team),
|
|
"team_id": channel.get("team_id") or "",
|
|
"exported_at": datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
|
|
},
|
|
"posts": merged_posts
|
|
}
|
|
|
|
with open(output_filepath, "w", encoding='utf8') as f:
|
|
json.dump(output, f, indent=2, ensure_ascii=False)
|
|
print(f"Exported channel data to '{output_filepath}'")
|
|
|
|
|
|
def get_config_from_json(config_filename: str = "config.json") -> dict:
|
|
config_path = pathlib.Path(config_filename)
|
|
if not config_path.exists():
|
|
return {}
|
|
|
|
with config_path.open() as f:
|
|
config = json.load(f)
|
|
|
|
return config
|
|
|
|
|
|
def complete_config(config: dict, config_filename: str = "config.json") -> dict:
|
|
config_changed = False
|
|
if config.get("host", False):
|
|
print(f"Using host '{config['host']}' from config")
|
|
else:
|
|
config["host"] = input("Please input host/server address (without https://): ")
|
|
config_changed = True
|
|
|
|
if config.get("login_mode", False):
|
|
print(f"Using login mode '{config['login_mode']}' from config")
|
|
else:
|
|
login_mode = ""
|
|
while login_mode not in ["password", "token", "gitlab"]:
|
|
login_mode = input(
|
|
"Please input login_mode 'password', 'token', or 'gitlab' (GitLab OAuth): "
|
|
)
|
|
config["login_mode"] = login_mode
|
|
config_changed = True
|
|
|
|
password = None
|
|
if config["login_mode"] == "password":
|
|
if config.get("username", False):
|
|
print(f"Using username '{config['username']}' from config")
|
|
else:
|
|
config["username"] = input("Please input your username: ")
|
|
config_changed = True
|
|
|
|
password = getpass.getpass("Enter your password (hidden): ")
|
|
elif config["login_mode"] == "gitlab":
|
|
# If we already have a token in config, reuse it and do not ask
|
|
# for GitLab credentials again.
|
|
if config.get("token"):
|
|
print("Using existing token from config for GitLab login_mode.")
|
|
else:
|
|
if config.get("gitlab_url", False):
|
|
print(f"Using GitLab URL '{config['gitlab_url']}' from config")
|
|
else:
|
|
config["gitlab_url"] = input(
|
|
f"Please input GitLab URL [{GITLAB_URL_DEFAULT}]: "
|
|
) or GITLAB_URL_DEFAULT
|
|
config_changed = True
|
|
|
|
if config.get("username", False):
|
|
print(f"Using username '{config['username']}' from config")
|
|
else:
|
|
config["username"] = input("Please input your GitLab username: ")
|
|
config_changed = True
|
|
|
|
password = getpass.getpass("Enter your GitLab password (hidden): ")
|
|
token = get_token_via_gitlab(
|
|
config["host"], config["gitlab_url"], config["username"], password,
|
|
config.get("cookie_file", COOKIE_FILE_DEFAULT)
|
|
)
|
|
if not token:
|
|
raise SystemExit("GitLab authentication failed.")
|
|
config["token"] = token
|
|
config_changed = True
|
|
else:
|
|
if config.get("token", False):
|
|
print(f"Using token '{config['token']}' from config")
|
|
else:
|
|
print("Token not found in config. Choose how to obtain MMAUTHTOKEN:")
|
|
print(" g - login via GitLab (username/password)")
|
|
print(" f - extract from Firefox cookies")
|
|
print(" m - enter token manually")
|
|
dec = ""
|
|
while dec not in ["g", "f", "m"]:
|
|
dec = input("Choice [g/f/m]: ")
|
|
|
|
token = None
|
|
if dec == "g":
|
|
if config.get("gitlab_url", False):
|
|
print(f"Using GitLab URL '{config['gitlab_url']}' from config")
|
|
else:
|
|
config["gitlab_url"] = input(
|
|
f"Please input GitLab URL [{GITLAB_URL_DEFAULT}]: "
|
|
) or GITLAB_URL_DEFAULT
|
|
config_changed = True
|
|
if config.get("username", False):
|
|
print(f"Using username '{config['username']}' from config")
|
|
else:
|
|
config["username"] = input("Please input your GitLab username: ")
|
|
config_changed = True
|
|
password = getpass.getpass("Enter your GitLab password (hidden): ")
|
|
token = get_token_via_gitlab(
|
|
config["host"], config["gitlab_url"], config["username"], password,
|
|
config.get("cookie_file", COOKIE_FILE_DEFAULT)
|
|
)
|
|
if not token:
|
|
raise SystemExit("GitLab authentication failed.")
|
|
elif dec == "f":
|
|
token = find_mmauthtoken_firefox(config["host"])
|
|
if not token:
|
|
token = input("Please input your login token (MMAUTHTOKEN): ")
|
|
config["token"] = token
|
|
config_changed = True
|
|
|
|
if "download_files" in config:
|
|
print(f"Download files set to '{config['download_files']}' from config")
|
|
else:
|
|
dec = ""
|
|
while not (dec == "y" or dec == "n"):
|
|
dec = input("Should files be downloaded? y/n: ")
|
|
config["download_files"] = dec == "y"
|
|
config_changed = True
|
|
|
|
if config_changed:
|
|
dec = ""
|
|
while not (dec == "y" or dec == "n"):
|
|
dec = input("Config changed! Would you like to store your config (without password) to file? y/n: ")
|
|
if dec == "y":
|
|
with open(config_filename, "w") as f:
|
|
json.dump(config, f, indent=2)
|
|
|
|
print(f"Stored new config to {config_filename}")
|
|
|
|
config["password"] = password
|
|
return config
|
|
|
|
|
|
def find_mmauthtoken_firefox(host):
|
|
appdata_dir = pathlib.Path(os.environ["APPDATA"])
|
|
profiles_dir = appdata_dir / "Mozilla/Firefox/Profiles"
|
|
cookie_files = profiles_dir.rglob("cookies.sqlite")
|
|
|
|
all_tokens = []
|
|
for cookie_file in cookie_files:
|
|
print(f"Opening {cookie_file}")
|
|
connection = sqlite3.connect(str(cookie_file))
|
|
cursor = connection.cursor()
|
|
rows = cursor.execute("SELECT host, value FROM moz_cookies WHERE name = 'MMAUTHTOKEN'").fetchall()
|
|
all_tokens.extend(rows)
|
|
|
|
all_tokens = [token for token in all_tokens if host in token[0]]
|
|
|
|
print(f"Found {len(all_tokens)} token for {host}")
|
|
for token in all_tokens:
|
|
print(f"{token[0]}: {token[1]}")
|
|
|
|
if len(all_tokens) > 1:
|
|
print("Using first token!")
|
|
|
|
if len(all_tokens):
|
|
return all_tokens[0][1]
|
|
else:
|
|
return None
|
|
|
|
|
|
if __name__ == '__main__':
|
|
config_filename = "config.json"
|
|
config = get_config_from_json(config_filename)
|
|
config = complete_config(config, config_filename)
|
|
|
|
incremental = config.get("incremental", bool(config.get("last_export_at")))
|
|
if 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"]
|
|
before = config.get("before")
|
|
merge = True
|
|
else:
|
|
output_base = "results/" + date.today().strftime("%Y%m%d")
|
|
after = config.get("after")
|
|
before = config.get("before")
|
|
merge = False
|
|
|
|
print(f"Storing downloaded data in {output_base}")
|
|
export_started_at = datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ')
|
|
|
|
try:
|
|
d = connect(
|
|
config["host"],
|
|
config.get("token", None),
|
|
config.get("username", None),
|
|
config.get("password", None),
|
|
)
|
|
apply_throttle(d, config)
|
|
except Exception as 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.
|
|
if config.get("token"):
|
|
print("Token-based authentication failed, removing stored token.")
|
|
config.pop("token", None)
|
|
save_config(config, config_filename)
|
|
print("Please restart the script to reconfigure authentication.")
|
|
raise SystemExit(1) from e
|
|
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)
|
|
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"],
|
|
before, after, team=team, merge=merge)
|
|
|
|
config["last_export_at"] = export_started_at
|
|
config["output_base"] = output_base
|
|
config["incremental"] = True
|
|
save_config(config, config_filename)
|
|
print(f"Updated last_export_at to {export_started_at}")
|
|
print("Finished export")
|