mirror of
https://github.com/lrnselfreliance/wrolpi.git
synced 2026-10-26 19:01:39 -04:00
f49288b8ec
Branch names are passed to git as arguments and come from the API, the config, a `/tree/<branch>` URL, or the remote itself (its default branch). A name starting with `-` would be read as an option, so every branch must pass git's ref-name rules, including in the downloader just before use. A repo URL with a `.` or `..` path segment (or a `..` name) could format a directory outside its template (e.g. `repos/<tag>/..` is the whole repos directory). Such URLs and names are rejected, and every repo directory, including one from repos.yaml, must be strictly inside the media directory with no `..`. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
575 lines
23 KiB
Python
575 lines
23 KiB
Python
import asyncio
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import shutil
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Optional
|
|
from urllib.parse import urlparse
|
|
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
|
|
from wrolpi.cmd import GIT_BIN
|
|
from wrolpi.collections.models import Collection
|
|
from wrolpi.common import ConfigFile, get_media_directory, get_relative_to_media_directory, logger, \
|
|
escape_file_name, background_task
|
|
from wrolpi.db import get_db_session
|
|
from wrolpi.downloader import DownloadFrequency, save_downloads_config
|
|
from wrolpi.errors import UnknownDirectory, ValidationError
|
|
from wrolpi.events import Events
|
|
from wrolpi.switches import register_switch_handler, ActivateSwitchMethod
|
|
from wrolpi.tags import Tag
|
|
from .errors import InvalidRepo, RepoConflict
|
|
from .models import Repository, REPO_MODES
|
|
|
|
logger = logger.getChild(__name__)
|
|
|
|
__all__ = ['ParsedRepoURL', 'parse_repo_url', 'repo_url_key', 'git_command', 'git_env', 'create_repository',
|
|
'update_repository', 'delete_repository', 'get_repositories', 'list_repo_tree', 'find_readme',
|
|
'repo_destination_variables', 'is_valid_branch', 'ReposConfig', 'repos_config', 'get_repos_config', 'save_repos_config',
|
|
'import_repos_config', 'DEFAULT_REPO_FREQUENCY']
|
|
|
|
DEFAULT_REPO_FREQUENCY = DownloadFrequency.weekly.value
|
|
|
|
# The only transports git may use. Submodules and redirects cannot reach `file://`, `ext::`, ssh, etc.
|
|
ALLOWED_PROTOCOLS = ('https',)
|
|
|
|
# README files in the order they are preferred; matched case-insensitively.
|
|
README_NAMES = ('readme.md', 'readme.markdown', 'readme.rst', 'readme.txt', 'readme')
|
|
|
|
# A forge's web URL of a branch: GitHub `/owner/repo/tree/<branch>`, GitLab `/group/repo/-/tree/<branch>`.
|
|
TREE_URL_RE = re.compile(r'^(?P<repo>/.+?)(?:/-)?/tree/(?P<branch>[^?#]+)$')
|
|
|
|
|
|
@dataclass
|
|
class ParsedRepoURL:
|
|
url: str
|
|
host: str
|
|
owner: str
|
|
name: str
|
|
branch: Optional[str] = None
|
|
|
|
|
|
def parse_repo_url(url: str) -> ParsedRepoURL:
|
|
"""Validate an https git URL and split it into its parts.
|
|
|
|
A forge's branch URL (e.g. `https://github.com/owner/repo/tree/dev`) becomes the repo URL and its branch."""
|
|
url = (url or '').strip()
|
|
parsed = urlparse(url)
|
|
if parsed.scheme not in ALLOWED_PROTOCOLS:
|
|
raise InvalidRepo('Only https:// repo URLs are supported')
|
|
if parsed.username or parsed.password:
|
|
# Credentials would be saved in the configs.
|
|
raise InvalidRepo('Repo URL must not contain a username or password')
|
|
if not parsed.hostname:
|
|
raise InvalidRepo('Repo URL must have a host')
|
|
if parsed.query or parsed.fragment:
|
|
raise InvalidRepo('Repo URL must not contain a query or fragment')
|
|
|
|
path = parsed.path.rstrip('/')
|
|
branch = None
|
|
if match := TREE_URL_RE.match(path):
|
|
path, branch = match.group('repo'), match.group('branch')
|
|
|
|
parts = [i for i in path.split('/') if i]
|
|
if not parts:
|
|
raise InvalidRepo('Repo URL must have a path')
|
|
if any(i in ('.', '..') for i in parts):
|
|
# The name and owner are used in the Repo's directory.
|
|
raise InvalidRepo('Repo URL must not contain "." or ".." in its path')
|
|
name = parts[-1].removesuffix('.git')
|
|
if not name:
|
|
raise InvalidRepo('Repo URL must have a name')
|
|
owner = '/'.join(parts[:-1])
|
|
|
|
host = parsed.hostname.lower()
|
|
netloc = f'{host}:{parsed.port}' if parsed.port else host
|
|
return ParsedRepoURL(url=f'{parsed.scheme}://{netloc}{path}', host=host, owner=owner, name=name, branch=branch)
|
|
|
|
|
|
# Characters git forbids in a ref name (`git check-ref-format`), plus whitespace.
|
|
INVALID_BRANCH_CHARACTERS_RE = re.compile(r'[\x00-\x20\x7f~^:?*\[\\]')
|
|
|
|
|
|
def is_valid_branch(branch: str) -> bool:
|
|
"""Is `branch` a valid git branch name (the rules of `git check-ref-format --branch`)?
|
|
|
|
Branch names are passed to git as arguments, so a name starting with `-` (an option) is never valid."""
|
|
if not branch or not isinstance(branch, str):
|
|
return False
|
|
if branch.startswith('-') or branch == '@' or '@{' in branch or '..' in branch:
|
|
return False
|
|
if INVALID_BRANCH_CHARACTERS_RE.search(branch):
|
|
return False
|
|
if branch.startswith('/') or branch.endswith('/') or '//' in branch or branch.endswith('.'):
|
|
return False
|
|
for component in branch.split('/'):
|
|
if component.startswith('.') or component.endswith('.lock'):
|
|
return False
|
|
return True
|
|
|
|
|
|
def _validate_branch(branch: Optional[str]):
|
|
if branch is not None and not is_valid_branch(branch):
|
|
raise ValidationError(f'Invalid branch name: {branch!r}')
|
|
|
|
|
|
def repo_url_key(url: str) -> str:
|
|
"""Two URLs of the same repo have the same key (`.git` suffix and case of the host are ignored)."""
|
|
parsed = urlparse(url.strip())
|
|
return f'{(parsed.hostname or "").lower()}{parsed.path.rstrip("/").removesuffix(".git")}'
|
|
|
|
|
|
def git_command(*args, config: tuple = ()) -> tuple:
|
|
"""Build a git command that is safe to run against an untrusted remote repo."""
|
|
if not GIT_BIN:
|
|
raise InvalidRepo('git is not installed')
|
|
safety = [
|
|
'-c', 'protocol.allow=never',
|
|
*(i for p in ALLOWED_PROTOCOLS for i in ('-c', f'protocol.{p}.allow=always')),
|
|
# Never run the repo's hooks.
|
|
'-c', 'core.hooksPath=/dev/null',
|
|
# A symlink in a repo could point outside the repo (e.g. /etc/shadow), and media files are served.
|
|
# Symlinks are checked out as plain files containing the link's target.
|
|
'-c', 'core.symlinks=false',
|
|
'-c', 'core.fsmonitor=false',
|
|
# The media directory may be owned by another user (e.g. a Docker volume).
|
|
'-c', 'safe.directory=*',
|
|
'-c', 'credential.helper=',
|
|
'-c', 'submodule.recurse=false',
|
|
]
|
|
for i in config:
|
|
safety.extend(['-c', i])
|
|
return GIT_BIN, *safety, *args
|
|
|
|
|
|
def git_env() -> dict:
|
|
"""Environment for git: never prompt for credentials, ignore the user's and system's git configs."""
|
|
env = dict(os.environ)
|
|
env.update(
|
|
GIT_TERMINAL_PROMPT='0',
|
|
GIT_ASKPASS='true',
|
|
SSH_ASKPASS='true',
|
|
GIT_CONFIG_NOSYSTEM='1',
|
|
GIT_CONFIG_GLOBAL=os.devnull,
|
|
GIT_LFS_SKIP_SMUDGE='1',
|
|
LC_ALL='C',
|
|
)
|
|
return env
|
|
|
|
|
|
def find_readme(directory: pathlib.Path) -> Optional[str]:
|
|
"""Return the name of the preferred README file in `directory`, if any."""
|
|
try:
|
|
files = {i.name.lower(): i.name for i in directory.iterdir() if i.is_file() and not i.is_symlink()}
|
|
except FileNotFoundError:
|
|
return None
|
|
for name in README_NAMES:
|
|
if name in files:
|
|
return files[name]
|
|
return None
|
|
|
|
|
|
def repo_destination_variables(collection: Collection) -> dict:
|
|
"""Variables for `repos_destination` which are stored on the Repository (not the Collection)."""
|
|
session = Session.object_session(collection)
|
|
repo = Repository.get_by_collection_id(session, collection.id) if session and collection.id else None
|
|
return dict(
|
|
repo_owner=escape_file_name(repo.owner) if repo and repo.owner else '',
|
|
repo_host=escape_file_name(repo.host) if repo and repo.host else '',
|
|
)
|
|
|
|
|
|
def get_repository_by_url(session: Session, url: str) -> Optional[Repository]:
|
|
key = repo_url_key(url)
|
|
for id_, repo_url in session.query(Repository.id, Repository.url):
|
|
if repo_url_key(repo_url) == key:
|
|
return session.query(Repository).filter_by(id=id_).one()
|
|
return None
|
|
|
|
|
|
def _validate_mode(mode: str):
|
|
if mode not in REPO_MODES:
|
|
raise ValidationError(f'Repo mode must be one of {", ".join(REPO_MODES)}')
|
|
|
|
|
|
def _validate_directory(directory: pathlib.Path):
|
|
"""A Repo's directory is strictly inside the media directory; it is cloned into, and may be deleted."""
|
|
media_directory = get_media_directory()
|
|
directory = pathlib.Path(directory)
|
|
if '..' in directory.parts or not directory.is_absolute():
|
|
raise InvalidRepo(f'Invalid repo directory: {directory}')
|
|
try:
|
|
relative = directory.relative_to(media_directory)
|
|
except ValueError:
|
|
raise InvalidRepo('Repo directory must be in the media directory')
|
|
if not relative.parts:
|
|
raise InvalidRepo('Repo directory can not be the media directory')
|
|
|
|
|
|
def _validate_frequency(frequency: int):
|
|
if not isinstance(frequency, int) or frequency <= 0:
|
|
raise ValidationError('Repo frequency must be a positive number of seconds')
|
|
|
|
|
|
def create_repository(session: Session, url: str, tag_name: str = None, frequency: int = DEFAULT_REPO_FREQUENCY,
|
|
mode: str = 'full', branch: str = None, name: str = None, description: str = None,
|
|
directory: pathlib.Path = None) -> Repository:
|
|
"""Create a Repo (Collection + Repository + recurring Download). The download clones it."""
|
|
parsed = parse_repo_url(url)
|
|
_validate_mode(mode)
|
|
_validate_frequency(frequency)
|
|
branch = branch or parsed.branch or None
|
|
_validate_branch(branch)
|
|
name = (name or parsed.name).strip()
|
|
if name in ('', '.', '..'):
|
|
raise InvalidRepo(f'Invalid repo name: {name!r}')
|
|
|
|
if get_repository_by_url(session, parsed.url):
|
|
raise RepoConflict(f'A repo with this URL already exists: {parsed.url}')
|
|
if session.query(Collection).filter_by(kind='repo', name=name).one_or_none():
|
|
suggestion = f'{parsed.owner.replace("/", "-")}-{name}' if parsed.owner else f'{parsed.host}-{name}'
|
|
raise RepoConflict(f'A repo named {name!r} already exists. Try the name {suggestion!r}')
|
|
|
|
tag = Tag.get_or_create_tag(session, tag_name) if tag_name else None
|
|
collection = Collection(kind='repo', name=name, description=description, tag=tag)
|
|
session.add(collection)
|
|
session.flush([collection])
|
|
repo = Repository(url=parsed.url, host=parsed.host, owner=parsed.owner, mode=mode, branch=branch,
|
|
collection_id=collection.id)
|
|
session.add(repo)
|
|
session.flush([repo])
|
|
|
|
directory = directory or collection.format_destination(tag_name)
|
|
_validate_directory(directory)
|
|
if session.query(Collection).filter(Collection.directory == directory, Collection.id != collection.id).first():
|
|
raise RepoConflict(f'Another collection already uses the directory {get_relative_to_media_directory(directory)}')
|
|
collection.directory = directory
|
|
|
|
download = collection.get_or_create_download(session, parsed.url, frequency, downloader_name='git')
|
|
# The Download may already exist (e.g. imported from download_manager.yaml).
|
|
download.frequency = frequency
|
|
session.commit()
|
|
|
|
save_repos_config.activate_switch()
|
|
save_downloads_config.activate_switch()
|
|
return repo
|
|
|
|
|
|
def update_repository(session: Session, repo_id: int, description: str = None, frequency: int = None,
|
|
mode: str = None, branch: str = None) -> Repository:
|
|
"""Change a Repo's settings. An empty `branch` follows the remote's default branch."""
|
|
repo = Repository.find_by_id(session, repo_id)
|
|
if description is not None:
|
|
repo.collection.description = description or None
|
|
if mode is not None:
|
|
_validate_mode(mode)
|
|
repo.mode = mode
|
|
if branch is not None:
|
|
branch = branch.strip() or None
|
|
_validate_branch(branch)
|
|
repo.branch = branch
|
|
if frequency is not None:
|
|
_validate_frequency(frequency)
|
|
if download := repo.download:
|
|
download.frequency = frequency
|
|
else:
|
|
repo.collection.get_or_create_download(session, repo.url, frequency, downloader_name='git')
|
|
session.commit()
|
|
|
|
save_repos_config.activate_switch()
|
|
save_downloads_config.activate_switch()
|
|
return repo
|
|
|
|
|
|
def _delete_repository(session: Session, repo: Repository):
|
|
collection = repo.collection
|
|
for download in list(collection.downloads):
|
|
session.delete(download)
|
|
session.delete(collection)
|
|
session.flush()
|
|
|
|
|
|
async def delete_repository(session: Session, repo_id: int, delete_files: bool = False) -> dict:
|
|
"""Delete a Repo and its Download. The repo's files are kept unless `delete_files`."""
|
|
repo = Repository.find_by_id(session, repo_id)
|
|
directory = repo.directory
|
|
result = dict(id=repo.id, name=repo.name, directory=str(get_relative_to_media_directory(directory)))
|
|
_delete_repository(session, repo)
|
|
session.commit()
|
|
|
|
if delete_files and directory and directory.is_dir():
|
|
media_directory = get_media_directory().resolve()
|
|
if directory.resolve() == media_directory or media_directory not in directory.resolve().parents:
|
|
raise InvalidRepo('Refusing to delete a directory outside the media directory')
|
|
background_task(_delete_directory(directory))
|
|
|
|
save_repos_config.activate_switch()
|
|
save_downloads_config.activate_switch()
|
|
return result
|
|
|
|
|
|
async def _delete_directory(directory: pathlib.Path):
|
|
await asyncio.to_thread(shutil.rmtree, directory)
|
|
logger.info(f'Deleted repo directory {directory}')
|
|
|
|
|
|
def get_repositories(session: Session) -> List[dict]:
|
|
repos = session.query(Repository).options(*Repository.json_options()) \
|
|
.join(Collection, Collection.id == Repository.collection_id) \
|
|
.order_by(func.lower(Collection.name)).all()
|
|
return [i.__json__() for i in repos]
|
|
|
|
|
|
@dataclass
|
|
class RepoTreeEntry:
|
|
name: str
|
|
path: str
|
|
is_dir: bool
|
|
size: Optional[int] = None
|
|
|
|
def __json__(self) -> dict:
|
|
return dict(name=self.name, path=self.path, is_dir=self.is_dir, size=self.size)
|
|
|
|
|
|
def list_repo_tree(repo: Repository, path: str = '') -> dict:
|
|
"""List one directory of a Repo's checked out files, directories first. `.git` is hidden."""
|
|
root = repo.directory
|
|
if not root or not root.is_dir():
|
|
raise UnknownDirectory('Repo has not been downloaded')
|
|
root = root.resolve()
|
|
|
|
path = (path or '').strip('/')
|
|
directory = (root / path).resolve()
|
|
if directory != root and root not in directory.parents:
|
|
raise InvalidRepo('Path is not in the repo')
|
|
if not directory.is_dir():
|
|
raise UnknownDirectory(f'No directory {path!r} in repo')
|
|
|
|
entries = []
|
|
for entry in directory.iterdir():
|
|
if entry.name == '.git' or entry.is_symlink():
|
|
continue
|
|
relative = str(entry.relative_to(root))
|
|
if entry.is_dir():
|
|
entries.append(RepoTreeEntry(name=entry.name, path=relative, is_dir=True))
|
|
else:
|
|
entries.append(RepoTreeEntry(name=entry.name, path=relative, is_dir=False, size=entry.stat().st_size))
|
|
entries.sort(key=lambda i: (not i.is_dir, i.name.lower()))
|
|
|
|
readme = find_readme(directory)
|
|
return dict(
|
|
path=path,
|
|
entries=[i.__json__() for i in entries],
|
|
readme_path=str((directory / readme).relative_to(root)) if readme else None,
|
|
)
|
|
|
|
|
|
def get_repo_size(directory: pathlib.Path) -> int:
|
|
"""Total size of every file in the repo, including `.git`."""
|
|
total = 0
|
|
for dir_path, _, file_names in os.walk(directory):
|
|
for name in file_names:
|
|
try:
|
|
total += os.lstat(os.path.join(dir_path, name)).st_size
|
|
except FileNotFoundError:
|
|
pass
|
|
return total
|
|
|
|
|
|
@dataclass
|
|
class ReposConfigValidator:
|
|
version: int = 0
|
|
repos: List[dict] = field(default_factory=list)
|
|
|
|
|
|
class ReposConfig(ConfigFile):
|
|
"""Config file for git Repos (Collections with kind='repo').
|
|
|
|
Format:
|
|
repos:
|
|
- name: kiwix-tools
|
|
url: https://github.com/kiwix/kiwix-tools
|
|
directory: repos/software/kiwix-tools
|
|
tag_name: software
|
|
mode: full
|
|
branch: null
|
|
frequency: 604800
|
|
"""
|
|
file_name = 'repos.yaml'
|
|
validator = ReposConfigValidator
|
|
default_config = dict(version=0, repos=[])
|
|
width = 120
|
|
|
|
@property
|
|
def repos(self) -> List[dict]:
|
|
return self._config.get('repos', [])
|
|
|
|
def import_config(self, file: pathlib.Path = None, send_events=False):
|
|
"""Import Repos from the config into the database."""
|
|
file = file or self.get_file()
|
|
file_str = str(self.get_relative_file())
|
|
|
|
# A missing config is not an empty config; never delete Repos because a file is lost.
|
|
if not file.is_file():
|
|
logger.info(f'No repos config file, skipping import: {file_str}')
|
|
self.successful_import = True
|
|
return
|
|
|
|
super().import_config(file, send_events)
|
|
repos_data = self._config.get('repos', [])
|
|
|
|
# An empty repos list never deletes DB records. Deleting all repos is done through the API.
|
|
if not repos_data:
|
|
logger.info(f'No repos in config, preserving existing DB repos: {file_str}')
|
|
self.successful_import = True
|
|
return
|
|
|
|
try:
|
|
with get_db_session(commit=True) as session:
|
|
imported_ids = set()
|
|
for data in repos_data:
|
|
try:
|
|
repo = self._import_repo(session, data)
|
|
except Exception as e:
|
|
logger.error(f'Failed to import repo from config: {data}', exc_info=e)
|
|
self.import_skipped += 1
|
|
continue
|
|
imported_ids.add(repo.id)
|
|
|
|
for repo in session.query(Repository).all():
|
|
if repo.id not in imported_ids:
|
|
logger.info(f'Deleting repo {repo.name!r} (no longer in config)')
|
|
_delete_repository(session, repo)
|
|
|
|
logger.info(f'Imported {len(imported_ids)} repos from {file_str}')
|
|
self.successful_import = True
|
|
except Exception as e:
|
|
self.successful_import = False
|
|
message = f'Failed to import {file_str} config!'
|
|
logger.error(message, exc_info=e)
|
|
if send_events:
|
|
Events.send_config_import_failed(message)
|
|
raise
|
|
|
|
@staticmethod
|
|
def _import_repo(session: Session, data: dict) -> Repository:
|
|
parsed = parse_repo_url(data['url'])
|
|
name = data.get('name') or parsed.name
|
|
mode = data.get('mode') or 'full'
|
|
_validate_mode(mode)
|
|
frequency = data.get('frequency') or DEFAULT_REPO_FREQUENCY
|
|
_validate_frequency(frequency)
|
|
directory = data.get('directory')
|
|
directory = get_media_directory() / directory if directory else None
|
|
if directory:
|
|
_validate_directory(directory)
|
|
tag_name = data.get('tag_name')
|
|
|
|
repo = get_repository_by_url(session, parsed.url)
|
|
if not repo:
|
|
return create_repository(session, parsed.url, tag_name=tag_name, frequency=frequency, mode=mode,
|
|
branch=data.get('branch'), name=name, description=data.get('description'),
|
|
directory=directory)
|
|
|
|
collection = repo.collection
|
|
collection.name = name
|
|
collection.description = data.get('description')
|
|
collection.tag = Tag.get_or_create_tag(session, tag_name) if tag_name else None
|
|
if directory:
|
|
collection.directory = directory
|
|
repo.mode = mode
|
|
branch = data.get('branch') or None
|
|
_validate_branch(branch)
|
|
repo.branch = branch
|
|
if download := repo.download:
|
|
download.frequency = frequency
|
|
else:
|
|
collection.get_or_create_download(session, repo.url, frequency, downloader_name='git')
|
|
session.flush()
|
|
return repo
|
|
|
|
def dump_config(self, file: pathlib.Path = None, send_events=False, overwrite=False):
|
|
"""Dump all Repos to the config file."""
|
|
with get_db_session() as session:
|
|
repos = session.query(Repository).options(*Repository.json_options()) \
|
|
.join(Collection, Collection.id == Repository.collection_id) \
|
|
.order_by(func.lower(Collection.name)).all()
|
|
data = []
|
|
for repo in repos:
|
|
download = repo.download
|
|
entry = dict(
|
|
name=repo.name,
|
|
url=repo.url,
|
|
directory=str(get_relative_to_media_directory(repo.directory)) if repo.directory else None,
|
|
tag_name=repo.tag_name,
|
|
description=repo.collection.description,
|
|
mode=repo.mode,
|
|
branch=repo.branch,
|
|
frequency=download.frequency if download else None,
|
|
)
|
|
data.append(entry)
|
|
|
|
self._config['repos'] = data
|
|
logger.info(f'Dumping {len(data)} repos to config')
|
|
self.save(file, send_events, overwrite)
|
|
|
|
def preview_backup_import(self, backup_date: str, mode: str) -> dict:
|
|
"""Preview which Repos a backup import would add/remove (keyed by URL)."""
|
|
backup_data = self.read_config_file(self._get_backup_file(backup_date))
|
|
current_data = self.read_config_file() if self.get_file().is_file() else dict(repos=[], version=0)
|
|
|
|
backup_repos = {repo_url_key(i['url']): i for i in backup_data.get('repos', []) if i.get('url')}
|
|
current_repos = {repo_url_key(i['url']): i for i in current_data.get('repos', []) if i.get('url')}
|
|
|
|
add = [dict(type='repo', name=i.get('name', ''), url=i['url'])
|
|
for key, i in backup_repos.items() if key not in current_repos]
|
|
unchanged = len([i for i in backup_repos if i in current_repos])
|
|
remove = []
|
|
if mode == 'overwrite':
|
|
remove = [dict(type='repo', name=i.get('name', ''), url=i['url'])
|
|
for key, i in sorted(current_repos.items()) if key not in backup_repos]
|
|
return dict(mode=mode, add=add, remove=remove, unchanged=unchanged)
|
|
|
|
def import_backup(self, backup_date: str, mode: str, send_events: bool = False):
|
|
"""Restore Repos from a dated backup of repos.yaml ('overwrite' or 'merge' by URL)."""
|
|
self._preserve_current_config()
|
|
backup_file = self._get_backup_file(backup_date)
|
|
config_file = self.get_file()
|
|
config_file.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if mode == 'overwrite':
|
|
shutil.copy2(backup_file, config_file)
|
|
elif mode == 'merge':
|
|
backup_data = self.read_config_file(backup_file)
|
|
current_data = self.read_config_file() if config_file.is_file() else dict(repos=[], version=0)
|
|
current_keys = {repo_url_key(i['url']) for i in current_data.get('repos', []) if i.get('url')}
|
|
merged = list(current_data.get('repos', []))
|
|
merged.extend(i for i in backup_data.get('repos', [])
|
|
if i.get('url') and repo_url_key(i['url']) not in current_keys)
|
|
self.write_config_data(dict(repos=merged, version=current_data.get('version', 0) + 1), config_file)
|
|
|
|
self.import_config(send_events=send_events)
|
|
save_downloads_config.activate_switch()
|
|
|
|
|
|
repos_config = ReposConfig()
|
|
|
|
|
|
def get_repos_config() -> ReposConfig:
|
|
return repos_config
|
|
|
|
|
|
@register_switch_handler('save_repos_config')
|
|
def save_repos_config():
|
|
"""Save the repos config when the switch is activated."""
|
|
repos_config.background_dump.activate_switch()
|
|
|
|
|
|
save_repos_config: ActivateSwitchMethod
|
|
|
|
|
|
def import_repos_config():
|
|
logger.info('Importing repos config')
|
|
repos_config.import_config()
|