import asyncio from collections.abc import Awaitable, Callable from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from threading import BoundedSemaphore from typing import Any, TypeVar from cli.completion.doctor import COMPLETION_DOCTOR, CompletionDoctorService from cli.completion.installers import COMPLETION_INSTALLER, CompletionInstallerService from cli.core.app_builder import CliApplicationBuilder from cli.exit_codes import DEPENDENCY_ERROR, RUNTIME_ERROR from cli.main import CLI_APPLICATION_BUILDER from cli.runtime.app import APP_RUNTIME, AppRuntimeService from cli.runtime.cache import CACHE_RUNTIME, CacheRuntimeService from cli.runtime.config import CONFIG_RUNTIME, ConfigRuntimeService from cli.runtime.crypto import CRYPTO_RUNTIME, CryptoRuntimeService from cli.runtime.db import DATABASE_RUNTIME, DatabaseRuntimeService from cli.runtime.gen import GEN_RUNTIME, GenRuntimeService from cli.runtime.job import JOB_RUNTIME, JobRuntimeService from cli.runtime.ops import OPERATIONS_RUNTIME, OperationsRuntimeService TUI_REMOTE_QUERY_TIMEOUT_SECONDS = 3.0 TUI_LOCAL_QUERY_TIMEOUT_SECONDS = 2.0 TUI_PAGE_TIMEOUT_SECONDS = 5.0 _ResultT = TypeVar('_ResultT') @dataclass(frozen=True) class TuiGeneratedOutput: """ 进程内生成文本结果。 :param stdout: 生成的标准文本 :param stderr: 生成失败信息 :param returncode: 生成结果码 """ stdout: str stderr: str = '' returncode: int = 0 @dataclass(frozen=True) class TuiQueryExecutor: """ TUI 查询执行器。 该对象统一施加页面查询超时,并将短时间本地同步检查移出 Textual 事件循环。网络查询必须通过 ``run_async`` 执行。 """ remote_timeout_seconds: float = TUI_REMOTE_QUERY_TIMEOUT_SECONDS local_timeout_seconds: float = TUI_LOCAL_QUERY_TIMEOUT_SECONDS local_executor: ThreadPoolExecutor = field( default_factory=lambda: ThreadPoolExecutor(max_workers=2, thread_name_prefix='ruoyi-tui-query'), compare=False, repr=False, ) local_slots: BoundedSemaphore = field( default_factory=lambda: BoundedSemaphore(2), compare=False, repr=False, ) async def run_async( self, operation: Callable[[], Awaitable[_ResultT]], *, label: str, ) -> _ResultT | dict[str, Any]: """ 执行带超时的异步查询。 :param operation: 异步查询工厂 :param label: 查询名称 :return: 查询结果或超时负载 """ try: return await asyncio.wait_for(operation(), timeout=self.remote_timeout_seconds) except asyncio.TimeoutError: return self.build_timeout_payload(label) except asyncio.CancelledError: raise except Exception as exc: return self.build_failure_payload(label, exc) async def run_local( self, operation: Callable[[], _ResultT], *, label: str, ) -> _ResultT | dict[str, Any]: """ 执行带超时的短时间本地同步查询。 :param operation: 同步查询函数 :param label: 查询名称 :return: 查询结果或失败负载 """ loop = asyncio.get_running_loop() deadline = loop.time() + self.local_timeout_seconds while not self.local_slots.acquire(blocking=False): remaining = deadline - loop.time() if remaining <= 0: return self.build_timeout_payload(label) await asyncio.sleep(min(0.01, remaining)) try: future = loop.run_in_executor(self.local_executor, operation) except Exception: self.local_slots.release() raise future.add_done_callback(lambda completed: self.local_slots.release()) try: remaining = max(deadline - loop.time(), 0) return await asyncio.wait_for( asyncio.shield(future), timeout=remaining, ) except asyncio.TimeoutError: return self.build_timeout_payload(label) except asyncio.CancelledError: raise except Exception as exc: return self.build_failure_payload(label, exc) @staticmethod def build_timeout_payload(label: str) -> dict[str, Any]: """ 构建超时结果。 :param label: 查询名称 :return: 标准失败负载 """ return { 'ok': False, 'message': f'{label}超时', 'error': f'查询未在规定时间内完成:{label}', 'timeout': True, 'exit_code': RUNTIME_ERROR, } @staticmethod def build_failure_payload(label: str, exc: Exception) -> dict[str, Any]: """ 构建异常结果。 :param label: 查询名称 :param exc: 查询异常 :return: 标准失败负载 """ return { 'ok': False, 'message': f'{label}失败', 'error': str(exc) or exc.__class__.__name__, 'exit_code': RUNTIME_ERROR, } @dataclass class TuiRuntimeQueryService: """ TUI 进程内查询服务。 所有 TUI 页面通过该服务直接调用 runtime,不再启动嵌套 CLI 进程。 """ app_runtime: AppRuntimeService = field(default_factory=lambda: APP_RUNTIME) database_runtime: DatabaseRuntimeService = field(default_factory=lambda: DATABASE_RUNTIME) operations_runtime: OperationsRuntimeService = field(default_factory=lambda: OPERATIONS_RUNTIME) cache_runtime: CacheRuntimeService = field(default_factory=lambda: CACHE_RUNTIME) job_runtime: JobRuntimeService = field(default_factory=lambda: JOB_RUNTIME) gen_runtime: GenRuntimeService = field(default_factory=lambda: GEN_RUNTIME) config_runtime: ConfigRuntimeService = field(default_factory=lambda: CONFIG_RUNTIME) crypto_runtime: CryptoRuntimeService = field(default_factory=lambda: CRYPTO_RUNTIME) completion_doctor: CompletionDoctorService = field(default_factory=lambda: COMPLETION_DOCTOR) completion_installer: CompletionInstallerService = field(default_factory=lambda: COMPLETION_INSTALLER) application_builder: CliApplicationBuilder = field(default_factory=lambda: CLI_APPLICATION_BUILDER) executor: TuiQueryExecutor = field(default_factory=TuiQueryExecutor) async def get_app_env(self, env: str) -> dict[str, Any]: result = await self.executor.run_local( self.app_runtime.get_app_env_snapshot, label='读取应用环境', ) if isinstance(result, dict) and result.get('ok') is False: return result if not isinstance(result, dict): return self.executor.build_failure_payload( '读取应用环境', TypeError('应用环境结果格式无效'), ) return {'ok': True, 'env': env, 'runtime': result} async def get_app_config(self, env: str) -> dict[str, Any]: result = await self.executor.run_local( self.app_runtime.get_app_config_snapshot, label='读取应用配置', ) if isinstance(result, dict) and result.get('ok') is False: return result return {'ok': True, 'env': env, 'config': result} async def get_app_routes(self, env: str) -> dict[str, Any]: result = await self.executor.run_local( lambda: self.app_runtime.get_app_routes_snapshot(env, group_by='tag'), label='读取应用路由', ) return ( result if isinstance(result, dict) else self.executor.build_failure_payload( '读取应用路由', TypeError('应用路由结果格式无效'), ) ) async def get_doctor(self, env: str) -> dict[str, Any]: database, redis, crypto = await asyncio.gather( self.get_database_ping(), self.get_redis_ping(), self.get_crypto_validation(), ) payload = { 'env': env, 'database': database, 'redis': redis, 'crypto': crypto, } payload['ok'] = all(isinstance(item, dict) and item.get('ok', False) for item in (database, redis, crypto)) if not payload['ok']: payload['exit_code'] = DEPENDENCY_ERROR return payload async def get_completion_doctor(self) -> dict[str, Any]: result = await self.executor.run_local( self.completion_doctor.build_completion_doctor_payload, label='读取补全诊断', ) return ( result if isinstance(result, dict) else self.executor.build_failure_payload( '读取补全诊断', TypeError('补全诊断结果格式无效'), ) ) async def get_completion_preview(self, shell: str) -> TuiGeneratedOutput: def _render() -> str: return self.completion_installer.render_completion_script( self.application_builder.build(), shell, ) result = await self.executor.run_local(_render, label='生成补全脚本') if isinstance(result, dict) and not result.get('ok', False): return TuiGeneratedOutput( stdout='', stderr=str(result.get('message') or result.get('error') or '生成补全脚本失败'), returncode=1, ) return TuiGeneratedOutput(stdout=str(result)) async def get_database_ping(self) -> dict[str, Any]: result = await self.executor.run_async( self.database_runtime.ping_database, label='数据库探活', ) return result if isinstance(result, dict) else {} async def get_database_current(self) -> dict[str, Any]: result = await self.executor.run_async( self.database_runtime.get_current_revision_async, label='读取数据库版本', ) return result if isinstance(result, dict) else {} async def get_database_heads(self) -> dict[str, Any]: result = await self.executor.run_local( self.database_runtime.get_alembic_heads, label='读取 Alembic heads', ) return result if isinstance(result, dict) else {} async def get_database_history(self, *, limit: int = 8) -> dict[str, Any]: result = await self.executor.run_local( lambda: self.database_runtime.get_alembic_history(limit=limit), label='读取 Alembic 历史', ) return result if isinstance(result, dict) else {} async def get_redis_ping(self) -> dict[str, Any]: result = await self.executor.run_async( self.operations_runtime.ping_redis, label='Redis 探活', ) return result if isinstance(result, dict) else {} async def get_dependencies(self) -> dict[str, Any]: result = await self.executor.run_local( self.operations_runtime.get_dependency_versions, label='读取依赖信息', ) return result if isinstance(result, dict) else {} async def get_server_info(self) -> dict[str, Any]: result = await self.executor.run_async( self.operations_runtime.get_server_info, label='读取服务器信息', ) return result if isinstance(result, dict) else {} async def get_crypto_validation(self) -> dict[str, Any]: result = await self.executor.run_local( self.crypto_runtime.validate_crypto_config, label='校验传输加密配置', ) return result if isinstance(result, dict) else {} async def get_crypto_public_key(self) -> dict[str, Any]: result = await self.executor.run_local( self.crypto_runtime.export_public_key, label='读取传输加密公钥', ) return result if isinstance(result, dict) else {} async def get_cache_stats(self) -> dict[str, Any]: result = await self.executor.run_async( self.cache_runtime.get_cache_stats, label='读取缓存统计', ) return result if isinstance(result, dict) else {} async def get_cache_keys(self, cache_name: str) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.cache_runtime.list_cache_keys(cache_name), label='读取缓存键', ) return result if isinstance(result, dict) else {} async def get_cache_value(self, cache_name: str, cache_key: str) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.cache_runtime.get_cache_value(cache_name, cache_key), label='读取缓存值', ) return result if isinstance(result, dict) else {} async def get_cache_ttl(self, cache_name: str, cache_key: str) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.cache_runtime.get_cache_ttl(cache_name, cache_key), label='读取缓存 TTL', ) return result if isinstance(result, dict) else {} async def get_jobs(self) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.job_runtime.list_jobs(paged=True, page_size=8), label='读取定时任务', ) return result if isinstance(result, dict) else {} async def get_job_logs( self, *, job_name: str = '', status: str | None = None, page_size: int = 20, ) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.job_runtime.list_job_logs( job_name=job_name, status=status, paged=True, page_size=page_size, ), label='读取任务日志', ) return result if isinstance(result, dict) else {} async def get_job_detail(self, job_id: int) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.job_runtime.get_job_detail(job_id), label='读取任务详情', ) return result if isinstance(result, dict) else {} async def get_gen_tables(self) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.gen_runtime.list_gen_tables(paged=True, page_size=8), label='读取代码生成业务表', ) return result if isinstance(result, dict) else {} async def get_gen_db_tables(self, *, table_name: str = '', page_size: int = 8) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.gen_runtime.list_gen_db_tables( table_name=table_name, paged=True, page_size=page_size, ), label='读取可导入数据库表', ) return result if isinstance(result, dict) else {} async def get_gen_detail(self, table_id: int) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.gen_runtime.get_gen_table_detail(table_id), label='读取代码生成详情', ) return result if isinstance(result, dict) else {} async def get_gen_preview(self, table_id: int) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.gen_runtime.preview_code(table_id), label='生成代码预览', ) return result if isinstance(result, dict) else {} async def get_gen_export_dry_run(self, table_name: str) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.gen_runtime.export_code([table_name], mode='zip', dry_run=True), label='代码导出演练', ) return result if isinstance(result, dict) else {} async def get_configs(self) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.config_runtime.list_configs(paged=True, page_size=8), label='读取参数配置', ) return result if isinstance(result, dict) else {} async def get_config_diagnostics(self) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.config_runtime.diagnose_config(sample_limit=5), label='诊断参数配置', ) return result if isinstance(result, dict) else {} async def get_config_detail(self, config_key: str) -> dict[str, Any]: result = await self.executor.run_async( lambda: self.config_runtime.get_config(config_key, source='both'), label='读取参数配置详情', ) return result if isinstance(result, dict) else {} TUI_RUNTIME_QUERIES = TuiRuntimeQueryService()