Hyemin Lee
abf8654b52
- list_repos: Pulp next 페이지 끝까지 순회 (첫-페이지-only 누락 수정) - BFF 메모리 검색(이름)·정렬(이름/검증/버전/패키지/동기화)·페이지네이션(20/페이지) - GET /repos/table 조각: 검색창은 표 밖(포커스 유지), 정렬헤더/페이저는 표 안, hx-include 로 검색어·정렬상태 상호 전송 - 폰트 재조정: 데이터 셀 14px, 헤더 13px, 배지/액션버튼 14px (동기화 버튼 대비 균형) - 버전 페이지 제목↔검증상세 마진, 데모 저장소 26개로 확장 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
205 lines
7.5 KiB
Python
205 lines
7.5 KiB
Python
"""Pulp JSON → 화면용 표시 모델 변환 (순수 함수, httpx 의존 없음).
|
||
|
||
GPG 검증 배지 로직(✅/⚠️/⚪)은 화면 4 에서도 재사용하므로 여기 함수로 둔다(스펙 §6 화면4).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
|
||
def gpg_status(repo: dict[str, Any]) -> dict[str, str]:
|
||
"""repo_config 의 gpgcheck / repo-gpgcheck 로 3단계 검증 배지 산출.
|
||
|
||
- pass(✅): 패키지 서명 + repo 메타데이터 서명 모두 검증
|
||
- warn(⚠️): 둘 중 하나만 검증
|
||
- unset(⚪): 검증 미설정
|
||
"""
|
||
cfg = repo.get("repo_config") or {}
|
||
pkg = bool(cfg.get("gpgcheck", 0))
|
||
meta = bool(cfg.get("repo-gpgcheck", 0))
|
||
# level 은 색상 칩(badge-{level})으로 렌더 — 장식 이모지는 쓰지 않는다(design-ref §7).
|
||
if pkg and meta:
|
||
return {"level": "pass", "label": "검증됨"}
|
||
if pkg or meta:
|
||
return {"level": "warn", "label": "부분 검증"}
|
||
return {"level": "unset", "label": "미설정"}
|
||
|
||
|
||
def version_number(version_href: str | None) -> int | None:
|
||
"""latest_version_href 끝의 정수(vN) 추출. 예: .../versions/3/ → 3"""
|
||
if not version_href:
|
||
return None
|
||
parts = [p for p in version_href.split("/") if p]
|
||
try:
|
||
return int(parts[-1])
|
||
except (ValueError, IndexError):
|
||
return None
|
||
|
||
|
||
def _content_count(version: dict[str, Any], section: str) -> int:
|
||
"""content_summary[section] (present/added/removed) 의 카운트 합계."""
|
||
data = (version.get("content_summary") or {}).get(section) or {}
|
||
return sum((info or {}).get("count", 0) for info in data.values())
|
||
|
||
|
||
def package_count(version: dict[str, Any] | None) -> int | None:
|
||
"""RepositoryVersion 의 content_summary.present 카운트 합계."""
|
||
if version is None:
|
||
return None
|
||
return _content_count(version, "present")
|
||
|
||
|
||
def _uuid_from_href(href: str) -> str:
|
||
parts = [p for p in href.split("/") if p]
|
||
return parts[-1] if parts else ""
|
||
|
||
|
||
def build_repo_view(
|
||
repo: dict[str, Any], version: dict[str, Any] | None = None
|
||
) -> dict[str, Any]:
|
||
"""저장소 1건 + (선택) 최신 버전 상세 → 대시보드 행 모델."""
|
||
return {
|
||
"name": repo.get("name", ""),
|
||
"uuid": _uuid_from_href(repo.get("pulp_href", "")),
|
||
"href": repo.get("pulp_href", ""),
|
||
"latest_version": version_number(repo.get("latest_version_href")),
|
||
"package_count": package_count(version),
|
||
# 최신 스냅샷 생성 시각 ≈ 마지막 동기화 시각
|
||
"last_sync": (version or {}).get("pulp_created"),
|
||
"gpg": gpg_status(repo),
|
||
# TODO(단계3/4): 현재 운영 배포 버전은 Distribution→publication→version 역추적 필요.
|
||
# 단계1에선 최신 버전만 표시한다.
|
||
"deployed_version": None,
|
||
}
|
||
|
||
|
||
def build_version_view(
|
||
version: dict[str, Any],
|
||
deployed_number: int | None = None,
|
||
gpg: dict[str, str] | None = None,
|
||
) -> dict[str, Any]:
|
||
"""RepositoryVersion → 버전 목록 행 모델.
|
||
|
||
gpg 검증은 저장소 단위 설정(repo_config)이라 모든 버전이 같은 값을 공유한다 —
|
||
호출부에서 repo 의 gpg_status() 를 주입한다.
|
||
"""
|
||
number = version_number(version.get("pulp_href"))
|
||
is_deployed = number is not None and number == deployed_number
|
||
# 배포 가능: 검증 통과(pass) + 현재 배포 중이 아님 (스펙 §5-2 게이트)
|
||
deployable = (gpg or {}).get("level") == "pass" and not is_deployed
|
||
return {
|
||
"number": number,
|
||
"href": version.get("pulp_href"),
|
||
"created": version.get("pulp_created"),
|
||
"package_count": _content_count(version, "present"),
|
||
"added": _content_count(version, "added"),
|
||
"removed": _content_count(version, "removed"),
|
||
"gpg": gpg,
|
||
"is_deployed": is_deployed,
|
||
"deployable": deployable,
|
||
}
|
||
|
||
|
||
def deploy_preview(
|
||
version_views: list[dict[str, Any]],
|
||
target_number: int | None,
|
||
deployed_number: int | None,
|
||
) -> dict[str, Any]:
|
||
"""배포 확인 모달용 미리보기: 현재 운영 버전 → 대상 버전, 순 패키지 변화."""
|
||
|
||
def find(n: int | None) -> dict[str, Any] | None:
|
||
return next((v for v in version_views if v["number"] == n), None)
|
||
|
||
current = find(deployed_number)
|
||
target = find(target_number)
|
||
current_count = current["package_count"] if current else 0
|
||
target_count = target["package_count"] if target else 0
|
||
net = target_count - current_count
|
||
return {
|
||
"current_version": deployed_number,
|
||
"current_count": current_count,
|
||
"target_version": target_number,
|
||
"target_count": target_count,
|
||
"net": abs(net),
|
||
"net_sign": "+" if net >= 0 else "−",
|
||
}
|
||
|
||
|
||
def verification_detail(repo: dict[str, Any]) -> dict[str, Any]:
|
||
"""검증 상태 상세 (화면 4): GPG/checksum 설정값 + 종합 배지."""
|
||
cfg = repo.get("repo_config") or {}
|
||
return {
|
||
"status": gpg_status(repo),
|
||
"gpgcheck": int(bool(cfg.get("gpgcheck", 0))),
|
||
"repo_gpgcheck": int(bool(cfg.get("repo-gpgcheck", 0))),
|
||
"checksum_type": repo.get("metadata_checksum_type")
|
||
or cfg.get("checksum_type")
|
||
or "—",
|
||
}
|
||
|
||
|
||
_REPO_SORT_KEYS = {
|
||
"name": lambda r: (r["name"] or "").lower(),
|
||
"verified": lambda r: {"pass": 0, "warn": 1, "unset": 2}.get(
|
||
(r["gpg"] or {}).get("level"), 3
|
||
),
|
||
"version": lambda r: r["latest_version"] if r["latest_version"] is not None else -1,
|
||
"packages": lambda r: r["package_count"] if r["package_count"] is not None else -1,
|
||
"last_sync": lambda r: r["last_sync"] or "",
|
||
}
|
||
|
||
|
||
def sort_repos(
|
||
repo_views: list[dict[str, Any]], key: str, direction: str = "asc"
|
||
) -> list[dict[str, Any]]:
|
||
"""대시보드 헤더 정렬. 알 수 없는 key 는 이름순으로 폴백."""
|
||
fn = _REPO_SORT_KEYS.get(key) or _REPO_SORT_KEYS["name"]
|
||
return sorted(repo_views, key=fn, reverse=(direction == "desc"))
|
||
|
||
|
||
def summarize(repo_views: list[dict[str, Any]], syncing: int = 0) -> dict[str, int]:
|
||
"""상단 요약 카드 수치. syncing 은 task_store.active_count() 에서 주입."""
|
||
return {
|
||
"total": len(repo_views),
|
||
"syncing": syncing,
|
||
"gpg_pass": sum(1 for r in repo_views if r["gpg"]["level"] == "pass"),
|
||
"total_packages": sum(r["package_count"] or 0 for r in repo_views),
|
||
}
|
||
|
||
|
||
def task_progress(task: dict[str, Any]) -> dict[str, Any]:
|
||
"""Pulp task → 진행률 표시 모델.
|
||
|
||
progress_reports[] 의 done/total 을 합산해 퍼센트를 낸다. 완료 시
|
||
created_resources 에서 새 RepositoryVersion 번호(vN)를 추출한다.
|
||
"""
|
||
state = task.get("state", "waiting")
|
||
reports = task.get("progress_reports") or []
|
||
done = sum((r or {}).get("done") or 0 for r in reports)
|
||
total = sum((r or {}).get("total") or 0 for r in reports)
|
||
percent = int(done / total * 100) if total else 0
|
||
|
||
new_version = None
|
||
for href in task.get("created_resources") or []:
|
||
if "/versions/" in href:
|
||
new_version = version_number(href)
|
||
break
|
||
|
||
error = None
|
||
if state == "failed":
|
||
err = task.get("error") or {}
|
||
error = err.get("description") or "동기화 작업이 실패했습니다."
|
||
|
||
return {
|
||
"state": state,
|
||
"done": done,
|
||
"total": total,
|
||
"percent": percent,
|
||
"new_version": new_version,
|
||
"error": error,
|
||
"completed": state == "completed",
|
||
"failed": state == "failed",
|
||
"running": state in ("waiting", "running"),
|
||
}
|