475 lines
18 KiB
Python
475 lines
18 KiB
Python
from dataclasses import dataclass
|
||
from typing import Any
|
||
|
||
from cli.tui.adapters.base import BaseDetailAdapter
|
||
from cli.tui.adapters.models import (
|
||
TUI_ADAPTER_MODEL_RENDERER,
|
||
DetailPageSnapshot,
|
||
DetailSectionSnapshot,
|
||
)
|
||
from cli.tui.copy import TUI_COPY
|
||
from cli.tui.diagnostics import TUI_DIAGNOSTIC_SERVICE
|
||
from cli.utils import NESTED_CLI_SUPPORT, SHELL_TEXT_FORMATTER
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class DatabaseDetailSourcePayloads:
|
||
"""
|
||
数据库详情页原始数据源快照。
|
||
|
||
:param revision_payload: `db current` 结果
|
||
:param check_payload: `db check` 结果
|
||
:param heads_payload: `db heads` 结果
|
||
:param history_payload: `db history` 结果
|
||
:param config_payload: `app config` 结果
|
||
"""
|
||
|
||
revision_payload: dict[str, Any] | None
|
||
check_payload: dict[str, Any] | None
|
||
heads_payload: dict[str, Any] | None
|
||
history_payload: dict[str, Any] | None
|
||
config_payload: dict[str, Any] | None
|
||
|
||
|
||
class DatabaseDetailSnapshotCollector:
|
||
"""
|
||
数据库详情页数据采集器。
|
||
|
||
该对象负责拉取数据库详情页所需的多路 CLI 原始结果，
|
||
让 `DatabaseDetailAdapter` 保持详情页编排职责。
|
||
"""
|
||
|
||
def collect(self, env: str) -> DatabaseDetailSourcePayloads:
|
||
"""
|
||
采集数据库详情页所需原始结果。
|
||
|
||
:param env: 当前运行环境
|
||
:return: 数据库详情页原始数据源快照
|
||
"""
|
||
return DatabaseDetailSourcePayloads(
|
||
revision_payload=NESTED_CLI_SUPPORT.run(
|
||
'db',
|
||
'current',
|
||
f'--env={env}',
|
||
'--output=json',
|
||
parse_json=True,
|
||
).payload,
|
||
check_payload=NESTED_CLI_SUPPORT.run(
|
||
'db',
|
||
'check',
|
||
f'--env={env}',
|
||
'--output=json',
|
||
parse_json=True,
|
||
).payload,
|
||
heads_payload=NESTED_CLI_SUPPORT.run(
|
||
'db',
|
||
'heads',
|
||
f'--env={env}',
|
||
'--output=json',
|
||
parse_json=True,
|
||
).payload,
|
||
history_payload=NESTED_CLI_SUPPORT.run(
|
||
'db',
|
||
'history',
|
||
f'--env={env}',
|
||
'--limit=8',
|
||
'--output=json',
|
||
parse_json=True,
|
||
).payload,
|
||
config_payload=NESTED_CLI_SUPPORT.run(
|
||
'app',
|
||
'config',
|
||
f'--env={env}',
|
||
'--output=json',
|
||
parse_json=True,
|
||
).payload,
|
||
)
|
||
|
||
|
||
class DatabaseSectionBuilder:
|
||
"""
|
||
数据库详情分区构建器。
|
||
|
||
该构建器负责将数据库相关 CLI 结果负载转换为 TUI 详情页分区，
|
||
使详情页适配器本体只保留采集与编排职责。
|
||
"""
|
||
|
||
@staticmethod
|
||
def extract_revision_items(payload: dict[str, Any] | None) -> list[dict[str, Any]]:
|
||
"""
|
||
从 Alembic 结果中提取修订版本列表。
|
||
|
||
:param payload: `db heads` 或 `db history` JSON 负载
|
||
:return: 修订版本列表
|
||
"""
|
||
items = payload.get('items') if isinstance(payload, dict) else None
|
||
if not isinstance(items, list):
|
||
return []
|
||
return [item for item in items if isinstance(item, dict)]
|
||
|
||
@staticmethod
|
||
def build_revision_item_lines(item: dict[str, Any], *, index: int, item_label: str) -> list[str]:
|
||
"""
|
||
构建单条 Alembic 修订版本详情文本。
|
||
|
||
:param item: 修订版本字典
|
||
:param index: 当前索引
|
||
:param item_label: 条目标识名称
|
||
:return: 文本行列表
|
||
"""
|
||
down_revisions = item.get('downRevisions')
|
||
branch_labels = item.get('branchLabels')
|
||
depends_on = item.get('dependsOn')
|
||
return [
|
||
f'## {item_label} {index:02d} · {item.get("revision", "-")}',
|
||
f'> 下游版本: {",".join(down_revisions) if isinstance(down_revisions, list) and down_revisions else "-"}',
|
||
f'> 分支标签: {",".join(branch_labels) if isinstance(branch_labels, list) and branch_labels else "-"}',
|
||
f'> 依赖版本: {",".join(depends_on) if isinstance(depends_on, list) and depends_on else "-"}',
|
||
f'> 说明: {SHELL_TEXT_FORMATTER.truncate_text(item.get("doc", "-"), 72)}',
|
||
f'> 文件: {SHELL_TEXT_FORMATTER.truncate_text(item.get("path", "-"), 72)}',
|
||
]
|
||
|
||
@staticmethod
|
||
def build_revision_section(payload: dict[str, Any] | None) -> DetailSectionSnapshot:
|
||
"""
|
||
构建数据库迁移状态分区。
|
||
|
||
:param payload: `db current` JSON 负载
|
||
:return: 分区快照
|
||
"""
|
||
current_revision = payload.get('currentRevision', '-') if isinstance(payload, dict) else '-'
|
||
if not isinstance(payload, dict) or not payload.get('ok', False):
|
||
return DetailSectionSnapshot(
|
||
title='迁移版本',
|
||
status='fail',
|
||
lines=TUI_ADAPTER_MODEL_RENDERER.build_failure_lines(
|
||
payload, empty_label='迁移版本', empty_value=str(current_revision or '-')
|
||
),
|
||
)
|
||
return DetailSectionSnapshot(
|
||
title='迁移版本',
|
||
status='ok',
|
||
lines=[
|
||
'## 当前版本',
|
||
f'迁移版本: {current_revision}',
|
||
'',
|
||
'## 状态说明',
|
||
f'结果消息: {TUI_ADAPTER_MODEL_RENDERER.extract_payload_message(payload)}',
|
||
],
|
||
)
|
||
|
||
def build_overview_section(
|
||
self,
|
||
revision_payload: dict[str, Any] | None,
|
||
check_payload: dict[str, Any] | None,
|
||
heads_payload: dict[str, Any] | None,
|
||
history_payload: dict[str, Any] | None,
|
||
) -> DetailSectionSnapshot:
|
||
"""
|
||
构建数据库页总览判断分区。
|
||
|
||
:param revision_payload: `db current` JSON 负载
|
||
:param check_payload: `db check` JSON 负载
|
||
:param heads_payload: `db heads` JSON 负载
|
||
:param history_payload: `db history` JSON 负载
|
||
:return: 分区快照
|
||
"""
|
||
revision_ok = bool(isinstance(revision_payload, dict) and revision_payload.get('ok', False))
|
||
check_ok = bool(isinstance(check_payload, dict) and check_payload.get('ok', False))
|
||
heads_ok = bool(isinstance(heads_payload, dict) and heads_payload.get('ok', False))
|
||
current_revision = revision_payload.get('currentRevision', '-') if isinstance(revision_payload, dict) else '-'
|
||
head_items = self.extract_revision_items(heads_payload)
|
||
history_items = self.extract_revision_items(history_payload)
|
||
|
||
status = 'ok'
|
||
conclusion = '数据库基线正常，可继续查看连接状态、Heads 与版本链路'
|
||
if not revision_ok or not check_ok:
|
||
status = 'fail'
|
||
conclusion = '数据库存在基础异常，优先确认迁移版本与连通性检查结果'
|
||
elif not heads_ok or len(head_items) != 1:
|
||
status = 'warn'
|
||
conclusion = '数据库存在迁移分叉风险，优先确认 Heads 与版本链路是否一致'
|
||
|
||
return DetailSectionSnapshot(
|
||
title='总览判断',
|
||
status=status,
|
||
lines=[
|
||
'## 当前结论',
|
||
conclusion,
|
||
'',
|
||
'## 核心指标',
|
||
f'当前 revision: {current_revision}',
|
||
f'连通性检查: {"正常" if check_ok else "异常"}',
|
||
f'Heads 数量: {len(head_items)}',
|
||
f'版本数量: {history_payload.get("totalCount", len(history_items)) if isinstance(history_payload, dict) else len(history_items)}',
|
||
'',
|
||
'## 建议入口',
|
||
'优先关注：迁移版本 / 连通性检查 / Heads 状态 / 升级入口',
|
||
],
|
||
)
|
||
|
||
@staticmethod
|
||
def build_profile_section(payload: dict[str, Any] | None) -> DetailSectionSnapshot:
|
||
"""
|
||
构建数据库连接配置分区。
|
||
|
||
:param payload: `app config` JSON 负载
|
||
:return: 分区快照
|
||
"""
|
||
config = payload.get('config') if isinstance(payload, dict) else None
|
||
if not isinstance(config, dict):
|
||
return DetailSectionSnapshot(
|
||
title='连接信息',
|
||
status='fail',
|
||
lines=TUI_ADAPTER_MODEL_RENDERER.build_failure_lines(
|
||
payload, empty_label='连接信息', empty_value='不可用'
|
||
),
|
||
)
|
||
return DetailSectionSnapshot(
|
||
title='连接信息',
|
||
status='ok',
|
||
lines=[
|
||
'## 数据库连接',
|
||
f'数据库类型: {config.get("dbType", "-")}',
|
||
f'连接地址: {config.get("dbHost", "-")}:{config.get("dbPort", "-")}',
|
||
f'数据库名: {config.get("dbDatabase", "-")}',
|
||
],
|
||
)
|
||
|
||
@staticmethod
|
||
def build_check_section(payload: dict[str, Any] | None) -> DetailSectionSnapshot:
|
||
"""
|
||
构建数据库连通性检查分区。
|
||
|
||
:param payload: `db check` JSON 负载
|
||
:return: 分区快照
|
||
"""
|
||
if not isinstance(payload, dict) or not payload.get('ok', False):
|
||
return DetailSectionSnapshot(
|
||
title='连通性检查',
|
||
status='fail',
|
||
lines=TUI_ADAPTER_MODEL_RENDERER.build_failure_lines(
|
||
payload, empty_label='数据库连接', empty_value='异常'
|
||
),
|
||
)
|
||
return DetailSectionSnapshot(
|
||
title='连通性检查',
|
||
status='ok',
|
||
lines=[
|
||
'## 当前状态',
|
||
'数据库连接: 正常',
|
||
'',
|
||
'## 结果摘要',
|
||
f'结果消息: {TUI_ADAPTER_MODEL_RENDERER.extract_payload_message(payload)}',
|
||
],
|
||
)
|
||
|
||
def build_heads_section(self, payload: dict[str, Any] | None) -> DetailSectionSnapshot:
|
||
"""
|
||
构建 Alembic heads 分区。
|
||
|
||
:param payload: `db heads` JSON 负载
|
||
:return: 分区快照
|
||
"""
|
||
if not isinstance(payload, dict) or not payload.get('ok', False):
|
||
return DetailSectionSnapshot(
|
||
title='Heads 状态',
|
||
status='fail',
|
||
lines=TUI_ADAPTER_MODEL_RENDERER.build_failure_lines(
|
||
payload, empty_label='Heads', empty_value='不可用'
|
||
),
|
||
)
|
||
items = self.extract_revision_items(payload)
|
||
status = 'ok' if len(items) == 1 else 'warn'
|
||
lines = [
|
||
'## Heads 概况',
|
||
f'Heads 数量: {len(items)}',
|
||
f'结果消息: {TUI_ADAPTER_MODEL_RENDERER.extract_payload_message(payload)}',
|
||
]
|
||
if not items:
|
||
lines.extend(['', '## Heads 列表', '> 当前仓库未返回可用 heads'])
|
||
else:
|
||
lines.extend(['', '## Heads 列表'])
|
||
for index, item in enumerate(items[:4], start=1):
|
||
lines.extend(self.build_revision_item_lines(item, index=index, item_label='Head'))
|
||
lines.append('')
|
||
if len(lines) > 1 and lines[-1] == '':
|
||
lines.pop()
|
||
return DetailSectionSnapshot(
|
||
title='Heads 状态',
|
||
status=status,
|
||
lines=lines,
|
||
)
|
||
|
||
def build_history_section(self, payload: dict[str, Any] | None) -> DetailSectionSnapshot:
|
||
"""
|
||
构建 Alembic 历史版本分区。
|
||
|
||
:param payload: `db history` JSON 负载
|
||
:return: 分区快照
|
||
"""
|
||
if not isinstance(payload, dict) or not payload.get('ok', False):
|
||
return DetailSectionSnapshot(
|
||
title='历史版本',
|
||
status='fail',
|
||
lines=TUI_ADAPTER_MODEL_RENDERER.build_failure_lines(
|
||
payload, empty_label='历史版本', empty_value='不可用'
|
||
),
|
||
)
|
||
items = self.extract_revision_items(payload)
|
||
lines = [
|
||
'## 历史概况',
|
||
f'返回数量: {payload.get("count", len(items))}',
|
||
f'总版本数: {payload.get("totalCount", len(items))}',
|
||
f'查询上限: {payload.get("limit", len(items))}',
|
||
'',
|
||
'## 历史列表',
|
||
]
|
||
if not items:
|
||
lines.append('> 当前仓库未返回可展示的历史版本')
|
||
else:
|
||
for index, item in enumerate(items[:6], start=1):
|
||
lines.extend(self.build_revision_item_lines(item, index=index, item_label='版本'))
|
||
lines.append('')
|
||
if lines[-1] == '':
|
||
lines.pop()
|
||
return DetailSectionSnapshot(
|
||
title='历史版本',
|
||
status='ok',
|
||
lines=lines,
|
||
)
|
||
|
||
@staticmethod
|
||
def build_upgrade_entry_section() -> DetailSectionSnapshot:
|
||
"""
|
||
构建数据库升级向导入口分区。
|
||
|
||
:return: 分区快照
|
||
"""
|
||
return DetailSectionSnapshot(
|
||
title='升级入口',
|
||
status='info',
|
||
lines=TUI_COPY.build_command_hint_lines(
|
||
scenario='准备执行数据库升级、校验迁移路径或在生产前做升级预演时，应先通过向导确认目标版本与 dry-run 结果。',
|
||
command=TUI_COPY.build_cli_command_hint('wizard', 'db-upgrade', '--output=text'),
|
||
guide='向导会继续询问目标环境和目标 revision，并在真正升级前先展示预览信息。',
|
||
),
|
||
)
|
||
|
||
@staticmethod
|
||
def build_init_entry_section() -> DetailSectionSnapshot:
|
||
"""
|
||
构建数据库初始化预演入口分区。
|
||
|
||
:return: 分区快照
|
||
"""
|
||
return DetailSectionSnapshot(
|
||
title='初始化预演入口',
|
||
status='info',
|
||
lines=TUI_COPY.build_command_hint_lines(
|
||
scenario='准备初始化新库、重建本地环境，或先确认初始化命令会执行到哪些迁移版本时，应先执行 dry-run 预演。',
|
||
command=TUI_COPY.build_cli_command_hint('db', 'init', '--dry-run', '--output=text'),
|
||
guide='预演只返回命令摘要与执行目录；确认无误后，再通过外部终端或向导执行真实初始化流程。',
|
||
),
|
||
)
|
||
|
||
def build_sections(
|
||
self,
|
||
*,
|
||
revision_payload: dict[str, Any] | None,
|
||
check_payload: dict[str, Any] | None,
|
||
heads_payload: dict[str, Any] | None,
|
||
history_payload: dict[str, Any] | None,
|
||
config_payload: dict[str, Any] | None,
|
||
) -> list[DetailSectionSnapshot]:
|
||
"""
|
||
构建数据库页全部详情分区。
|
||
|
||
:param revision_payload: `db current` 结果负载
|
||
:param check_payload: `db check` 结果负载
|
||
:param heads_payload: `db heads` 结果负载
|
||
:param history_payload: `db history` 结果负载
|
||
:param config_payload: `app config` 结果负载
|
||
:return: 详情分区列表
|
||
"""
|
||
return [
|
||
self.build_overview_section(revision_payload, check_payload, heads_payload, history_payload),
|
||
self.build_revision_section(revision_payload),
|
||
self.build_profile_section(config_payload),
|
||
self.build_check_section(check_payload),
|
||
self.build_heads_section(heads_payload),
|
||
self.build_history_section(history_payload),
|
||
self.build_init_entry_section(),
|
||
self.build_upgrade_entry_section(),
|
||
]
|
||
|
||
|
||
class DatabaseDetailAdapter(BaseDetailAdapter):
|
||
"""
|
||
数据库详情页适配器。
|
||
|
||
该适配器负责采集数据库相关 CLI 结果，并组装为 TUI 详情页快照。
|
||
页面私有的解析逻辑、分区构建逻辑和快照采集流程统一收口在该类中。
|
||
|
||
:param section_builder: 数据库详情分区构建器
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
section_builder: DatabaseSectionBuilder | None = None,
|
||
snapshot_collector: DatabaseDetailSnapshotCollector | None = None,
|
||
) -> None:
|
||
"""
|
||
初始化数据库详情页适配器。
|
||
|
||
:param section_builder: 数据库详情分区构建器
|
||
:param snapshot_collector: 数据库详情页数据采集器
|
||
:return: None
|
||
"""
|
||
super().__init__(
|
||
page_title='数据库',
|
||
search_view_key='database',
|
||
default_suggestions=[
|
||
'总览判断',
|
||
'迁移版本',
|
||
'连接信息',
|
||
'连通性检查',
|
||
'Heads 状态',
|
||
'历史版本',
|
||
'初始化预演入口',
|
||
'升级入口',
|
||
],
|
||
)
|
||
self.section_builder = section_builder or DatabaseSectionBuilder()
|
||
self.snapshot_collector = snapshot_collector or DatabaseDetailSnapshotCollector()
|
||
|
||
def collect_snapshot(self, env: str, query: str = '') -> DetailPageSnapshot:
|
||
"""
|
||
采集数据库状态页只读快照。
|
||
|
||
:param env: 当前运行环境
|
||
:param query: 当前搜索词
|
||
:return: 页面快照
|
||
"""
|
||
source_payloads = self.snapshot_collector.collect(env)
|
||
sections = self.section_builder.build_sections(
|
||
revision_payload=source_payloads.revision_payload,
|
||
check_payload=source_payloads.check_payload,
|
||
heads_payload=source_payloads.heads_payload,
|
||
history_payload=source_payloads.history_payload,
|
||
config_payload=source_payloads.config_payload,
|
||
)
|
||
return DetailPageSnapshot(
|
||
title='数据库',
|
||
subtitle=TUI_DIAGNOSTIC_SERVICE.build_database_diagnostic_subtitle(
|
||
source_payloads.revision_payload,
|
||
source_payloads.check_payload,
|
||
source_payloads.heads_payload,
|
||
),
|
||
sections=self.filter_sections(sections, query),
|
||
search=self.resolve_search_context(query),
|
||
)
|
||
|
||
|
||
DATABASE_DETAIL_ADAPTER = DatabaseDetailAdapter()
|