44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
|
"""Diff пакетов между серверами."""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import Dict, List
|
||
|
|
|
||
|
|
from app.core.packages import Package, list_packages
|
||
|
|
from app.core.ssh import SSHClient
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class VersionDiff:
|
||
|
|
name: str
|
||
|
|
version_a: str
|
||
|
|
version_b: str
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass
|
||
|
|
class PackageDiff:
|
||
|
|
only_a: List[Package]
|
||
|
|
only_b: List[Package]
|
||
|
|
version_mismatch: List[VersionDiff]
|
||
|
|
|
||
|
|
|
||
|
|
async def diff_servers(ca: SSHClient, cb: SSHClient,
|
||
|
|
user_only: bool = True) -> PackageDiff:
|
||
|
|
pa = await list_packages(ca)
|
||
|
|
pb = await list_packages(cb)
|
||
|
|
if user_only:
|
||
|
|
pa = [p for p in pa if not p.is_system and p.manual]
|
||
|
|
pb = [p for p in pb if not p.is_system and p.manual]
|
||
|
|
|
||
|
|
map_a: Dict[str, Package] = {p.name: p for p in pa}
|
||
|
|
map_b: Dict[str, Package] = {p.name: p for p in pb}
|
||
|
|
|
||
|
|
only_a = [map_a[n] for n in sorted(map_a.keys() - map_b.keys())]
|
||
|
|
only_b = [map_b[n] for n in sorted(map_b.keys() - map_a.keys())]
|
||
|
|
mismatch = [
|
||
|
|
VersionDiff(n, map_a[n].version, map_b[n].version)
|
||
|
|
for n in sorted(map_a.keys() & map_b.keys())
|
||
|
|
if map_a[n].version != map_b[n].version
|
||
|
|
]
|
||
|
|
return PackageDiff(only_a, only_b, mismatch)
|