busya/ss-tools
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

semantics

Browse Source
This commit is contained in:
busya
2026-03-27 21:27:31 +03:00
parent 202823aba5
commit 7901ce0f39
182 changed files with 21186 additions and 10254 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
backend/src/scripts/clean_release_tui.py
View File

@@ -1,10 +1,10 @@
# [DEF:backend.src.scripts.clean_release_tui:Module]
# [DEF:CleanReleaseTuiScript:Module]
# @COMPLEXITY: 3
# @SEMANTICS: clean-release, tui, ncurses, interactive-validator
# @PURPOSE: Interactive terminal interface for Enterprise Clean Release compliance validation.
# @LAYER: UI
# @RELATION: DEPENDS_ON -> backend.src.services.clean_release.compliance_orchestrator
# @RELATION: DEPENDS_ON -> backend.src.services.clean_release.repository
# @RELATION: DEPENDS_ON -> [compliance_orchestrator]
# @RELATION: DEPENDS_ON -> [repository]
# @INVARIANT: TUI refuses startup in non-TTY environments; headless flow is CLI/API only.
import curses
@@ -37,12 +37,15 @@ from src.models.clean_release import (
)
from src.services.clean_release.approval_service import approve_candidate
from src.services.clean_release.artifact_catalog_loader import load_bootstrap_artifacts
from src.services.clean_release.compliance_execution_service import ComplianceExecutionService
from src.services.clean_release.compliance_execution_service import (
ComplianceExecutionService,
)
from src.services.clean_release.enums import CandidateStatus
from src.services.clean_release.manifest_service import build_manifest_snapshot
from src.services.clean_release.publication_service import publish_candidate
from src.services.clean_release.repository import CleanReleaseRepository
# [DEF:TuiFacadeAdapter:Class]
# @PURPOSE: Thin TUI adapter that routes business mutations through application services.
# @PRE: repository contains candidate and trusted policy/registry snapshots for execution.
@@ -67,15 +70,25 @@ class TuiFacadeAdapter:
manifests = self.repository.get_manifests_by_candidate(candidate_id)
if not manifests:
raise ValueError("Manifest required before compliance run")
latest_manifest = sorted(manifests, key=lambda item: item.manifest_version, reverse=True)[0]
latest_manifest = sorted(
manifests, key=lambda item: item.manifest_version, reverse=True
)[0]
service = ComplianceExecutionService(
repository=self.repository,
config_manager=self._build_config_manager(),
)
return service.execute_run(candidate_id=candidate_id, requested_by=actor, manifest_id=latest_manifest.id)
return service.execute_run(
candidate_id=candidate_id,
requested_by=actor,
manifest_id=latest_manifest.id,
)
def approve_latest(self, *, candidate_id: str, actor: str):
reports = [item for item in self.repository.reports.values() if item.candidate_id == candidate_id]
reports = [
item
for item in self.repository.reports.values()
if item.candidate_id == candidate_id
]
if not reports:
raise ValueError("No compliance report available for approval")
report = sorted(reports, key=lambda item: item.generated_at, reverse=True)[0]
@@ -88,7 +101,11 @@ class TuiFacadeAdapter:
)
def publish_latest(self, *, candidate_id: str, actor: str):
reports = [item for item in self.repository.reports.values() if item.candidate_id == candidate_id]
reports = [
item
for item in self.repository.reports.values()
if item.candidate_id == candidate_id
]
if not reports:
raise ValueError("No compliance report available for publication")
report = sorted(reports, key=lambda item: item.generated_at, reverse=True)[0]
@@ -111,24 +128,55 @@ class TuiFacadeAdapter:
def get_overview(self, *, candidate_id: str) -> Dict[str, Any]:
candidate = self.repository.get_candidate(candidate_id)
manifests = self.repository.get_manifests_by_candidate(candidate_id)
latest_manifest = sorted(manifests, key=lambda item: item.manifest_version, reverse=True)[0] if manifests else None
runs = [item for item in self.repository.check_runs.values() if item.candidate_id == candidate_id]
latest_run = sorted(runs, key=lambda item: item.requested_at, reverse=True)[0] if runs else None
latest_report = next((item for item in self.repository.reports.values() if latest_run and item.run_id == latest_run.id), None)
latest_manifest = (
sorted(manifests, key=lambda item: item.manifest_version, reverse=True)[0]
if manifests
else None
)
runs = [
item
for item in self.repository.check_runs.values()
if item.candidate_id == candidate_id
]
latest_run = (
sorted(runs, key=lambda item: item.requested_at, reverse=True)[0]
if runs
else None
)
latest_report = next(
(
item
for item in self.repository.reports.values()
if latest_run and item.run_id == latest_run.id
),
None,
)
approvals = getattr(self.repository, "approval_decisions", [])
latest_approval = sorted(
[item for item in approvals if item.candidate_id == candidate_id],
key=lambda item: item.decided_at,
reverse=True,
)[0] if any(item.candidate_id == candidate_id for item in approvals) else None
latest_approval = (
sorted(
[item for item in approvals if item.candidate_id == candidate_id],
key=lambda item: item.decided_at,
reverse=True,
)[0]
if any(item.candidate_id == candidate_id for item in approvals)
else None
)
publications = getattr(self.repository, "publication_records", [])
latest_publication = sorted(
[item for item in publications if item.candidate_id == candidate_id],
key=lambda item: item.published_at,
reverse=True,
)[0] if any(item.candidate_id == candidate_id for item in publications) else None
latest_publication = (
sorted(
[item for item in publications if item.candidate_id == candidate_id],
key=lambda item: item.published_at,
reverse=True,
)[0]
if any(item.candidate_id == candidate_id for item in publications)
else None
)
policy = self.repository.get_active_policy()
registry = self.repository.get_registry(policy.internal_source_registry_ref) if policy else None
registry = (
self.repository.get_registry(policy.internal_source_registry_ref)
if policy
else None
)
return {
"candidate": candidate,
"manifest": latest_manifest,
@@ -139,6 +187,8 @@ class TuiFacadeAdapter:
"policy": policy,
"registry": registry,
}
# [/DEF:TuiFacadeAdapter:Class]
@@ -166,11 +216,11 @@ class CleanReleaseTUI:
curses.start_color()
curses.use_default_colors()
curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLUE) # Header/Footer
curses.init_pair(2, curses.COLOR_GREEN, -1) # PASS
curses.init_pair(3, curses.COLOR_RED, -1) # FAIL/BLOCKED
curses.init_pair(4, curses.COLOR_YELLOW, -1) # RUNNING
curses.init_pair(5, curses.COLOR_CYAN, -1) # Text
curses.init_pair(1, curses.COLOR_WHITE, curses.COLOR_BLUE) # Header/Footer
curses.init_pair(2, curses.COLOR_GREEN, -1) # PASS
curses.init_pair(3, curses.COLOR_RED, -1) # FAIL/BLOCKED
curses.init_pair(4, curses.COLOR_YELLOW, -1) # RUNNING
curses.init_pair(5, curses.COLOR_CYAN, -1) # Text
def _build_repository(self, mode: str) -> CleanReleaseRepository:
repo = CleanReleaseRepository()
@@ -317,7 +367,9 @@ class CleanReleaseTUI:
"prohibited_artifact_categories",
["test-data", "demo", "load-test"],
),
required_system_categories=payload.get("required_system_categories", ["core"]),
required_system_categories=payload.get(
"required_system_categories", ["core"]
),
effective_from=now,
)
)
@@ -354,18 +406,21 @@ class CleanReleaseTUI:
self.stdscr.addstr(4, 3, "Checks:")
check_defs = [
(CheckStageName.DATA_PURITY, "Data Purity (no test/demo payloads)"),
(CheckStageName.INTERNAL_SOURCES_ONLY, "Internal Sources Only (company servers)"),
(
CheckStageName.INTERNAL_SOURCES_ONLY,
"Internal Sources Only (company servers)",
),
(CheckStageName.NO_EXTERNAL_ENDPOINTS, "No External Internet Endpoints"),
(CheckStageName.MANIFEST_CONSISTENCY, "Release Manifest Consistency"),
]
row = 5
drawn_checks = {c["stage"]: c for c in self.checks_progress}
for stage, desc in check_defs:
status_text = " "
color = curses.color_pair(5)
if stage in drawn_checks:
c = drawn_checks[stage]
if c["status"] == "RUNNING":
@@ -377,7 +432,7 @@ class CleanReleaseTUI:
elif c["status"] == CheckStageStatus.FAIL:
status_text = "FAIL"
color = curses.color_pair(3)
self.stdscr.addstr(row, 4, f"[{status_text:^4}] {desc}")
if status_text != " ":
self.stdscr.addstr(row, 50, f"{status_text:>10}", color | curses.A_BOLD)
@@ -396,12 +451,18 @@ class CleanReleaseTUI:
def draw_status(self):
color = curses.color_pair(5)
if self.status == CheckFinalStatus.COMPLIANT: color = curses.color_pair(2)
elif self.status == CheckFinalStatus.BLOCKED: color = curses.color_pair(3)
stat_str = str(self.status.value if hasattr(self.status, "value") else self.status)
self.stdscr.addstr(18, 3, f"FINAL STATUS: {stat_str.upper()}", color | curses.A_BOLD)
if self.status == CheckFinalStatus.COMPLIANT:
color = curses.color_pair(2)
elif self.status == CheckFinalStatus.BLOCKED:
color = curses.color_pair(3)
stat_str = str(
self.status.value if hasattr(self.status, "value") else self.status
)
self.stdscr.addstr(
18, 3, f"FINAL STATUS: {stat_str.upper()}", color | curses.A_BOLD
)
if self.report_id:
self.stdscr.addstr(19, 3, f"Report ID: {self.report_id}")
@@ -413,22 +474,36 @@ class CleanReleaseTUI:
self.stdscr.addstr(20, 32, f"Publication: {publication.status}")
if self.violations_list:
self.stdscr.addstr(21, 3, f"Violations Details ({len(self.violations_list)} total):", curses.color_pair(3) | curses.A_BOLD)
self.stdscr.addstr(
21,
3,
f"Violations Details ({len(self.violations_list)} total):",
curses.color_pair(3) | curses.A_BOLD,
)
row = 22
for i, v in enumerate(self.violations_list[:5]):
v_cat = str(getattr(v, "code", "VIOLATION"))
msg = str(getattr(v, "message", "Violation detected"))
location = str(
getattr(v, "artifact_path", "")
or getattr(getattr(v, "evidence_json", {}), "get", lambda *_: "")("location", "")
or getattr(getattr(v, "evidence_json", {}), "get", lambda *_: "")(
"location", ""
)
)
msg_text = f"[{v_cat}] {msg} (Loc: {location})"
self.stdscr.addstr(row + i, 5, msg_text[:70], curses.color_pair(3))
if self.last_error:
self.stdscr.addstr(27, 3, f"Error: {self.last_error}"[:100], curses.color_pair(3) | curses.A_BOLD)
self.stdscr.addstr(
27,
3,
f"Error: {self.last_error}"[:100],
curses.color_pair(3) | curses.A_BOLD,
)
def draw_footer(self, max_y: int, max_x: int):
footer_text = " F5 Run F6 Manifest F7 Refresh F8 Approve F9 Publish F10 Exit ".center(max_x)
footer_text = " F5 Run F6 Manifest F7 Refresh F8 Approve F9 Publish F10 Exit ".center(
max_x
)
self.stdscr.attron(curses.color_pair(1))
self.stdscr.addstr(max_y - 1, 0, footer_text[:max_x])
self.stdscr.attroff(curses.color_pair(1))
@@ -446,7 +521,9 @@ class CleanReleaseTUI:
self.refresh_screen()
try:
result = self.facade.run_compliance(candidate_id=self.candidate_id, actor="operator")
result = self.facade.run_compliance(
candidate_id=self.candidate_id, actor="operator"
)
except Exception as exc: # noqa: BLE001
self.status = CheckFinalStatus.FAILED
self.last_error = str(exc)
@@ -456,7 +533,9 @@ class CleanReleaseTUI:
self.checks_progress = [
{
"stage": stage.stage_name,
"status": CheckStageStatus.PASS if str(stage.decision).upper() == "PASSED" else CheckStageStatus.FAIL,
"status": CheckStageStatus.PASS
if str(stage.decision).upper() == "PASSED"
else CheckStageStatus.FAIL,
}
for stage in result.stage_runs
]
@@ -472,11 +551,14 @@ class CleanReleaseTUI:
self.status = CheckFinalStatus.FAILED
self.refresh_overview()
self.refresh_screen()
# [/DEF:run_checks:Function]
def build_manifest(self):
try:
manifest = self.facade.build_manifest(candidate_id=self.candidate_id, actor="operator")
manifest = self.facade.build_manifest(
candidate_id=self.candidate_id, actor="operator"
)
self.status = "READY"
self.report_id = None
self.violations_list = []
@@ -570,11 +652,13 @@ class CleanReleaseTUI:
self.approve_latest()
elif char == curses.KEY_F9:
self.publish_latest()
# [/DEF:CleanReleaseTUI:Class]
def tui_main(stdscr: curses.window):
curses.curs_set(0) # Hide cursor
curses.curs_set(0) # Hide cursor
app = CleanReleaseTUI(stdscr)
app.loop()
@@ -597,4 +681,4 @@ def main() -> int:
if __name__ == "__main__":
sys.exit(main())
# [/DEF:backend.src.scripts.clean_release_tui:Module]
# [/DEF:CleanReleaseTuiScript:Module]