221 lines
6.9 KiB
Python
221 lines
6.9 KiB
Python
"""Configuration loading and validation."""
|
|
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from typing import Dict, Optional
|
|
|
|
import yaml
|
|
|
|
|
|
class ConfigError(Exception):
|
|
pass
|
|
|
|
|
|
def _check_type(val, typ, path: str):
|
|
if not isinstance(val, typ):
|
|
raise ConfigError(f"{path}: expected {typ.__name__}, got {type(val).__name__}")
|
|
|
|
|
|
def _check_optional(val, typ, path: str):
|
|
if val is not None and not isinstance(val, typ):
|
|
raise ConfigError(f"{path}: expected {typ.__name__} or null, got {type(val).__name__}")
|
|
|
|
|
|
@dataclass
|
|
class SVNConfig:
|
|
url: str
|
|
layout: str = "std"
|
|
trunk: str = "trunk"
|
|
branches: str = "branches"
|
|
tags: str = "tags"
|
|
username: Optional[str] = None
|
|
password: Optional[str] = None
|
|
|
|
def auth_args(self) -> list:
|
|
"""Return ``['--username', u, '--password', p]`` if configured, else []."""
|
|
if not self.username:
|
|
return []
|
|
args = ["--username", self.username]
|
|
if self.password:
|
|
args += ["--password", self.password]
|
|
return args
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict, prefix: str) -> "SVNConfig":
|
|
url = d.get("url")
|
|
if not url:
|
|
raise ConfigError(f"{prefix}.url: required")
|
|
_check_type(url, str, f"{prefix}.url")
|
|
|
|
layout = d.get("layout", "std")
|
|
_check_type(layout, str, f"{prefix}.layout")
|
|
if layout not in ("std", "custom"):
|
|
raise ConfigError(f"{prefix}.layout: must be 'std' or 'custom'")
|
|
|
|
trunk = d.get("trunk", "trunk")
|
|
branches = d.get("branches", "branches")
|
|
tags = d.get("tags", "tags")
|
|
for name, val in [("trunk", trunk), ("branches", branches), ("tags", tags)]:
|
|
_check_type(val, str, f"{prefix}.{name}")
|
|
|
|
username = d.get("username")
|
|
_check_optional(username, str, f"{prefix}.username")
|
|
password = d.get("password")
|
|
_check_optional(password, str, f"{prefix}.password")
|
|
|
|
return cls(url=url, layout=layout, trunk=trunk, branches=branches,
|
|
tags=tags, username=username, password=password)
|
|
|
|
|
|
@dataclass
|
|
class GiteaConfig:
|
|
owner: str
|
|
repo: str
|
|
repos_path: Optional[str] = None
|
|
webhook_secret: Optional[str] = None
|
|
webhook_host: str = "0.0.0.0"
|
|
webhook_port: int = 8080
|
|
api_url: Optional[str] = None
|
|
api_token: Optional[str] = None
|
|
|
|
@property
|
|
def repo_dir(self) -> str:
|
|
"""Return the expected on-disk path to Gitea's bare repo."""
|
|
if self.repos_path:
|
|
return f"{self.repos_path.rstrip('/')}/{self.owner}/{self.repo}.git"
|
|
# Common defaults if repos_path not configured
|
|
for base in ("/var/lib/gitea/data/repositories",
|
|
"/data/git/repositories",
|
|
"/home/git/repositories"):
|
|
candidate = f"{base}/{self.owner}/{self.repo}.git"
|
|
import os
|
|
if os.path.isdir(candidate):
|
|
return candidate
|
|
raise ConfigError(
|
|
f"gitea.repos_path not set and no default Gitea repo found "
|
|
f"for {self.owner}/{self.repo}"
|
|
)
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict, prefix: str) -> "GiteaConfig":
|
|
owner = d.get("owner")
|
|
if not owner:
|
|
raise ConfigError(f"{prefix}.owner: required")
|
|
_check_type(owner, str, f"{prefix}.owner")
|
|
|
|
repo = d.get("repo")
|
|
if not repo:
|
|
raise ConfigError(f"{prefix}.repo: required")
|
|
_check_type(repo, str, f"{prefix}.repo")
|
|
|
|
repos_path = d.get("repos_path")
|
|
_check_optional(repos_path, str, f"{prefix}.repos_path")
|
|
|
|
webhook_secret = d.get("webhook_secret")
|
|
_check_optional(webhook_secret, str, f"{prefix}.webhook_secret")
|
|
|
|
webhook_host = d.get("webhook_host", "0.0.0.0")
|
|
_check_type(webhook_host, str, f"{prefix}.webhook_host")
|
|
|
|
webhook_port = d.get("webhook_port", 8080)
|
|
_check_type(webhook_port, int, f"{prefix}.webhook_port")
|
|
|
|
api_url = d.get("api_url")
|
|
_check_optional(api_url, str, f"{prefix}.api_url")
|
|
|
|
api_token = d.get("api_token")
|
|
_check_optional(api_token, str, f"{prefix}.api_token")
|
|
|
|
return cls(
|
|
owner=owner, repo=repo,
|
|
repos_path=repos_path,
|
|
webhook_secret=webhook_secret,
|
|
webhook_host=webhook_host,
|
|
webhook_port=webhook_port,
|
|
api_url=api_url, api_token=api_token,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class MirrorConfig:
|
|
id: str
|
|
svn: SVNConfig
|
|
gitea: GiteaConfig
|
|
authors: Dict[str, str] = field(default_factory=dict)
|
|
sync_interval: int = 120
|
|
enabled: bool = True
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> "MirrorConfig":
|
|
mid = d.get("id")
|
|
if not mid:
|
|
raise ConfigError("mirror.id: required")
|
|
_check_type(mid, str, "mirror.id")
|
|
|
|
svn_raw = d.get("svn")
|
|
if not svn_raw:
|
|
raise ConfigError(f"mirror '{mid}'.svn: required")
|
|
_check_type(svn_raw, dict, f"mirror '{mid}'.svn")
|
|
|
|
gitea_raw = d.get("gitea")
|
|
if not gitea_raw:
|
|
raise ConfigError(f"mirror '{mid}'.gitea: required")
|
|
_check_type(gitea_raw, dict, f"mirror '{mid}'.gitea")
|
|
|
|
svn = SVNConfig.from_dict(svn_raw, f"mirror '{mid}'.svn")
|
|
gitea = GiteaConfig.from_dict(gitea_raw, f"mirror '{mid}'.gitea")
|
|
|
|
authors = d.get("authors", {})
|
|
_check_type(authors, dict, f"mirror '{mid}'.authors")
|
|
for k, v in authors.items():
|
|
_check_type(k, str, f"mirror '{mid}'.authors key")
|
|
_check_type(v, str, f"mirror '{mid}'.authors.{k}")
|
|
|
|
sync_interval = d.get("sync_interval", 120)
|
|
_check_type(sync_interval, int, f"mirror '{mid}'.sync_interval")
|
|
|
|
enabled = d.get("enabled", True)
|
|
_check_type(enabled, bool, f"mirror '{mid}'.enabled")
|
|
|
|
return cls(
|
|
id=mid,
|
|
svn=svn,
|
|
gitea=gitea,
|
|
authors=authors,
|
|
sync_interval=sync_interval,
|
|
enabled=enabled,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class AppConfig:
|
|
data_dir: str = "/var/svn-mirror"
|
|
mirrors: Dict[str, MirrorConfig] = field(default_factory=dict)
|
|
|
|
@classmethod
|
|
def load(cls, path: str) -> "AppConfig":
|
|
if not os.path.exists(path):
|
|
raise ConfigError(f"Config file not found: {path}")
|
|
|
|
with open(path) as f:
|
|
raw = yaml.safe_load(f)
|
|
|
|
if not raw:
|
|
raise ConfigError("Empty config file")
|
|
|
|
data_dir = raw.get("data_dir", "/var/svn-mirror")
|
|
_check_type(data_dir, str, "data_dir")
|
|
|
|
mirrors_raw = raw.get("mirrors", [])
|
|
_check_type(mirrors_raw, list, "mirrors")
|
|
|
|
mirrors = {}
|
|
for i, entry in enumerate(mirrors_raw):
|
|
_check_type(entry, dict, f"mirrors[{i}]")
|
|
mc = MirrorConfig.from_dict(entry)
|
|
if mc.id in mirrors:
|
|
raise ConfigError(f"Duplicate mirror id: {mc.id}")
|
|
mirrors[mc.id] = mc
|
|
|
|
return cls(data_dir=data_dir, mirrors=mirrors)
|