from __future__ import annotations import os from pathlib import Path from app.core.config import settings from app.core.db import get_db from app.core.files import read_json from app.core.redis_client import get_redis from app.services.cluster_runtime_service import get_cluster_snapshot from app.services.detect_service import get_detect_status from app.services.detect_job_service import get_detect_capacity_plan, get_detect_queue_health from app.services.sync_record_service import append_runtime_projection_if_changed, get_sync_summary from app.services.runtime_settings_service import get_runtime_settings from app.services.worker_control_service import detect_sync_agent_runtime, detect_worker_runtime def _runtime_log_path(filename: str) -> str: path = Path(__file__).resolve().parents[2] / "runtime" / "logs" / filename return str(path) def _domain_cookie_status(filename: str) -> tuple[bool, str]: path = Path(settings.domain_root) / filename return path.exists(), str(path) def _bloom_filter_status() -> tuple[bool, str]: try: redis_client = get_redis() modules = redis_client.execute_command("MODULE", "LIST") for module in modules or []: module_parts = module[::2] module_values = module[1::2] module_info = dict(zip(module_parts, module_values)) module_name = str(module_info.get("name", "")).lower() if module_name in {"bf", "redisbloom"}: return True, "RedisBloom 已安装" return False, "Redis 未安装 RedisBloom，当前将使用普通缓存" except Exception as exc: return False, f"RedisBloom 检查失败: {exc}" def _build_multi_region_readiness( *, cluster_snapshot: dict, sync_summary: dict, worker_runtime: dict, sync_agent_runtime: dict, ) -> dict: nodes = list(cluster_snapshot.get("nodes") or []) summary = cluster_snapshot.get("summary") or {} batch_summary = (sync_summary.get("detect_result_batches") or {}) batch_states = batch_summary.get("state_counts") or {} online_control_nodes = int(summary.get("online_control_nodes", 0) or 0) online_worker_nodes = int(summary.get("online_worker_nodes", 0) or 0) mainland_control_nodes = [ node for node in nodes if str(node.get("region") or "") == "mainland" and str(node.get("role") or "") == "control" and str(node.get("status") or "") in {"online", "busy"} ] mainland_worker_nodes = [ node for node in nodes if str(node.get("region") or "") == "mainland" and bool(node.get("is_effective_worker", False)) and str(node.get("status") or "") in {"online", "busy"} ] blocking_issues: list[str] = [] warning_issues: list[str] = [] info_items: list[str] = [] if online_control_nodes <= 0: blocking_issues.append("当前没有在线控制面节点，无法视为正式可用集群。") if settings.node_region == "mainland" and settings.node_role == "control": if not sync_agent_runtime.get("running", False): blocking_issues.append("当前节点应承载 sync-agent，但服务未运行。") if str(settings.sync_target_api_base_url or "").strip() == "": blocking_issues.append("当前节点未配置 SYNC_TARGET_API_BASE_URL，无法向海外控制面推送。") if not bool(settings.sync_push_enabled): blocking_issues.append("当前节点未启用 SYNC_PUSH_ENABLED，结果同步不会自动执行。") if settings.node_region == "overseas" and online_control_nodes > 0 and not mainland_control_nodes: warning_issues.append("当前尚未观察到在线的大陆 controller 节点，后续自动结果同步仍未进入正式双地域态。") if online_worker_nodes <= 0: warning_issues.append("当前没有在线 Worker 节点，检测任务无法在多机状态下继续推进。") offline_nodes = list(summary.get("offline_nodes") or []) stale_nodes = list(summary.get("stale_nodes") or []) if stale_nodes: warning_issues.append(f"存在失活节点: {'、'.join(stale_nodes)}") if offline_nodes: warning_issues.append(f"存在离线节点: {'、'.join(offline_nodes)}") failed_batches = int(batch_states.get("failed", 0) or 0) projected_batches = int(batch_states.get("projected", 0) or 0) pushing_batches = int(batch_states.get("pushing", 0) or 0) synced_batches = int(batch_states.get("synced", 0) or 0) if failed_batches > 0: warning_issues.append(f"存在 {failed_batches} 个结果批次同步失败，需要检查 sync-agent 或目标接收面。") if projected_batches > 0: warning_issues.append(f"存在 {projected_batches} 个结果批次仍待推送。") if pushing_batches > 0: info_items.append(f"当前有 {pushing_batches} 个结果批次正在推送。") if synced_batches > 0: info_items.append(f"最近已接收 {synced_batches} 个结果批次。") if worker_runtime.get("running", False): info_items.append("当前节点本机 Worker 进程在线。") if settings.node_region == "mainland" and settings.node_role == "control" and sync_agent_runtime.get("running", False): info_items.append("当前节点本机 sync-agent 在线。") if mainland_worker_nodes: info_items.append(f"在线大陆 Worker {len(mainland_worker_nodes)} 台。") if mainland_control_nodes: info_items.append(f"在线大陆 controller {len(mainland_control_nodes)} 台。") if blocking_issues: status = "blocking" summary_text = blocking_issues[0] elif warning_issues: status = "attention" summary_text = warning_issues[0] else: status = "ready" summary_text = "当前多机与跨地域骨架已进入可联调、可持续观察状态。" return { "status": status, "ready": status == "ready", "summary": summary_text, "blocking_issues": blocking_issues, "warnings": warning_issues, "info": info_items, "cluster": { "online_control_nodes": online_control_nodes, "online_worker_nodes": online_worker_nodes, "mainland_control_nodes": len(mainland_control_nodes), "mainland_worker_nodes": len(mainland_worker_nodes), }, "sync": { "enabled": bool(sync_summary.get("enabled", False)), "source_region": sync_summary.get("source_region", ""), "target_region": sync_summary.get("target_region", ""), "projected_batches": projected_batches, "pushing_batches": pushing_batches, "failed_batches": failed_batches, "synced_batches": synced_batches, }, } def _build_participating_detect_nodes(*, cluster_snapshot: dict, detect_snapshot: dict, worker_runtime: dict) -> list[dict]: cluster_nodes = list(cluster_snapshot.get("nodes") or []) active_job = detect_snapshot.get("active_job") or {} queue_nodes = list((detect_snapshot.get("queue_health") or {}).get("nodes") or []) cluster_map = { str(item.get("node_code") or "").strip(): item for item in cluster_nodes if str(item.get("node_code") or "").strip() } queue_map = { str(item.get("node_code") or "").strip(): item for item in queue_nodes if str(item.get("node_code") or "").strip() } merged: list[dict] = [] seen: set[str] = set() for item in list(active_job.get("node_stats") or []): node_code = str(item.get("node_code") or "").strip() if not node_code or node_code == "unassigned": continue seen.add(node_code) cluster_node = cluster_map.get(node_code, {}) queue_node = queue_map.get(node_code, {}) merged.append( { "node_code": node_code, "role": str(cluster_node.get("role") or "worker"), "region": str(cluster_node.get("region") or settings.node_region), "status": str(cluster_node.get("status") or "unknown"), "is_effective_worker": bool(cluster_node.get("is_effective_worker", False) or str(cluster_node.get("role") or "") == "worker"), "detect_participating": True, "current_load": int(cluster_node.get("current_load", 0) or 0), "items_total": int(item.get("items_total", 0) or 0), "items_pending": int(item.get("items_pending", 0) or 0), "items_claimed": int(item.get("items_claimed", 0) or 0), "items_running": int(item.get("items_running", 0) or 0), "items_completed": int(item.get("items_completed", 0) or 0), "items_failed": int(item.get("items_failed", 0) or 0), "processed_recent": int(queue_node.get("processed_recent", 0) or 0), "processed_per_minute": float(queue_node.get("processed_per_minute", 0) or 0), } ) if worker_runtime.get("running", False) and settings.node_code not in seen: cluster_node = cluster_map.get(settings.node_code, {}) if cluster_node: merged.append( { "node_code": settings.node_code, "role": str(cluster_node.get("role") or settings.node_role), "region": str(cluster_node.get("region") or settings.node_region), "status": str(cluster_node.get("status") or "online"), "is_effective_worker": True, "detect_participating": True, "current_load": int(cluster_node.get("current_load", 0) or 0), "items_total": 0, "items_pending": 0, "items_claimed": 0, "items_running": 0, "items_completed": 0, "items_failed": 0, "processed_recent": 0, "processed_per_minute": 0, } ) return sorted( merged, key=lambda item: ( -int(item.get("items_running", 0) or 0), -int(item.get("items_claimed", 0) or 0), -int(item.get("processed_recent", 0) or 0), str(item.get("node_code") or ""), ), ) def get_runtime_status() -> dict: runtime_settings = get_runtime_settings() worker_runtime = detect_worker_runtime() sync_agent_runtime = detect_sync_agent_runtime() api_pid = os.getpid() detect_snapshot = get_detect_status() latest_run = (detect_snapshot.get("runs") or [None])[0] or {} cluster_snapshot = get_cluster_snapshot() queue_health = get_detect_queue_health(window_minutes=15) effective_online_worker_nodes = int((cluster_snapshot.get("summary") or {}).get("online_worker_nodes", 0) or 0) if effective_online_worker_nodes <= 0 and worker_runtime.get("running", False): effective_online_worker_nodes = max(1, worker_runtime.get("process_count", 1) or 1) capacity_plan = get_detect_capacity_plan( queue_health=queue_health, online_worker_nodes=effective_online_worker_nodes, target_finish_hours=6, ) detect_payload = { "phase_label": latest_run.get("phase_label", ""), "phase_detail": latest_run.get("phase_detail", ""), "recent_event": detect_snapshot.get("recent_event", ""), "recent_warning": detect_snapshot.get("recent_warning", ""), "progress_percent": detect_snapshot.get("progress_percent", 0), "progress": detect_snapshot.get("progress", {}), "active_thread_count": detect_snapshot.get("active_thread_count", 0), "max_thread_count": detect_snapshot.get("max_thread_count", 0), "available_proxy_count": detect_snapshot.get("available_proxy_count", 0), "proxy_pool_count": detect_snapshot.get("proxy_pool_count", 0), "proxy_runtime_label": detect_snapshot.get("proxy_runtime_label", ""), "proxy_runtime_detail": detect_snapshot.get("proxy_runtime_detail", ""), "proxy_runtime_reason": detect_snapshot.get("proxy_runtime_reason", ""), "proxy_supplier_empty": detect_snapshot.get("proxy_supplier_empty", False), "proxy_last_refresh_status": detect_snapshot.get("proxy_last_refresh_status", ""), "proxy_last_refresh_time": detect_snapshot.get("proxy_last_refresh_time", ""), "proxy_last_refresh_source_count": detect_snapshot.get("proxy_last_refresh_source_count", 0), "proxy_last_refresh_total_items": detect_snapshot.get("proxy_last_refresh_total_items", 0), "proxy_last_validated_count": detect_snapshot.get("proxy_last_validated_count", 0), "proxy_last_available_count": detect_snapshot.get("proxy_last_available_count", 0), "proxy_source_stats": detect_snapshot.get("proxy_source_stats", []), "dependency_alerts": detect_snapshot.get("dependency_alerts", []), "active_job": detect_snapshot.get("active_job"), "runs_count": len(detect_snapshot.get("runs") or []), "worker_online": worker_runtime.get("running", False), "worker_mode": worker_runtime.get("mode", runtime_settings.get("worker_mode", "windows-local")), "queue_health": queue_health, "capacity_plan": capacity_plan, } detect_payload["participating_nodes"] = _build_participating_detect_nodes( cluster_snapshot=cluster_snapshot, detect_snapshot=detect_payload, worker_runtime=worker_runtime, ) append_runtime_projection_if_changed(detect=detect_payload, cluster=cluster_snapshot) sync_summary = get_sync_summary(record_limit=5) readiness = _build_multi_region_readiness( cluster_snapshot=cluster_snapshot, sync_summary=sync_summary, worker_runtime=worker_runtime, sync_agent_runtime=sync_agent_runtime, ) return { "api": { "service": "domain-api", "version": "0.1.0", "api_prefix": settings.api_prefix, "pid": api_pid, "host": settings.api_host, "port": settings.api_port, "mode": runtime_settings.get("worker_mode", "windows-local"), "service_name": runtime_settings.get("api_service_name", settings.api_service_name), "health_url": f"http://127.0.0.1:{settings.api_port}/health", "stdout_log": _runtime_log_path("domain-api.stdout.log"), "stderr_log": _runtime_log_path("domain-api.stderr.log"), }, "node": { "code": settings.node_code, "region": settings.node_region, "role": settings.node_role, }, "worker": { "mode": worker_runtime.get("mode", runtime_settings.get("worker_mode", "windows-local")), "service_name": runtime_settings.get("worker_service_name", settings.worker_service_name), "running": worker_runtime.get("running", False), "process_count": worker_runtime.get("process_count", 0), "latest_start_time": worker_runtime.get("latest_start_time", ""), "message": worker_runtime.get("message", ""), "log_path": str(Path(settings.domain_root) / "detect_worker.log"), }, "sync_agent": { "mode": sync_agent_runtime.get("mode", runtime_settings.get("worker_mode", "windows-local")), "service_name": runtime_settings.get("sync_agent_service_name", settings.sync_agent_service_name), "running": sync_agent_runtime.get("running", False), "process_count": sync_agent_runtime.get("process_count", 0), "latest_start_time": sync_agent_runtime.get("latest_start_time", ""), "message": sync_agent_runtime.get("message", ""), "expected_on_this_node": settings.node_region == "mainland" and settings.node_role == "control", }, "detect": detect_payload, "cluster": cluster_snapshot, "sync": sync_summary, "readiness": readiness, } def get_runtime_preflight() -> dict: runtime_settings = get_runtime_settings() checks: list[dict[str, object]] = [] detect_options = read_json("detect_options.json", default={}) domain_root = Path(settings.domain_root) checks.append( { "key": "domain_root", "label": "domainCheck 目录", "ok": domain_root.exists(), "message": str(domain_root), } ) try: with get_db() as conn: with conn.cursor() as cur: cur.execute("select 1") cur.fetchone() checks.append({"key": "database", "label": "PostgreSQL", "ok": True, "message": f"{settings.db_host}:{settings.db_port}/{settings.db_database}"}) except Exception as exc: checks.append({"key": "database", "label": "PostgreSQL", "ok": False, "message": str(exc)}) try: redis_client = get_redis() redis_client.ping() checks.append({"key": "redis", "label": "Redis", "ok": True, "message": f"{settings.redis_host}:{settings.redis_port}/{settings.redis_db}"}) except Exception as exc: checks.append({"key": "redis", "label": "Redis", "ok": False, "message": str(exc)}) else: bloom_ok, bloom_message = _bloom_filter_status() checks.append( { "key": "redis_bloom", "label": "RedisBloom", "ok": True, "message": bloom_message, "level": "info" if bloom_ok else "warn", "degraded": not bloom_ok, } ) worker_mode = runtime_settings.get("worker_mode", "windows-local") checks.append({"key": "worker_mode", "label": "运行模式", "ok": True, "message": worker_mode}) jucha_enabled = bool(detect_options.get("detect_jucha")) jucha_cookie_ok, jucha_cookie_path = _domain_cookie_status("jucha_cookies.pkl") checks.append( { "key": "detect_jucha", "label": "聚查检测", "ok": True if not jucha_enabled else jucha_cookie_ok, "message": "已启用" if jucha_enabled else "未启用", } ) checks.append( { "key": "jucha_cookie", "label": "聚查 Cookie", "ok": True if not jucha_enabled else jucha_cookie_ok, "message": jucha_cookie_path if jucha_enabled else "未启用聚查检测，无需本地 Cookie", "level": "info" if (not jucha_enabled or jucha_cookie_ok) else "warn", } ) juziseo_enabled = bool(detect_options.get("detect_juziseo")) juziseo_cookie_ok, juziseo_cookie_path = _domain_cookie_status("juziseo_cookies.pkl") checks.append( { "key": "detect_juziseo", "label": "桔子SEO检测", "ok": True if not juziseo_enabled else juziseo_cookie_ok, "message": "已启用" if juziseo_enabled else "未启用", } ) checks.append( { "key": "juziseo_cookie", "label": "桔子SEO Cookie", "ok": True if not juziseo_enabled else juziseo_cookie_ok, "message": juziseo_cookie_path if juziseo_enabled else "未启用桔子SEO检测，无需本地 Cookie", "level": "info" if (not juziseo_enabled or juziseo_cookie_ok) else "warn", } ) if worker_mode == "linux-systemd": checks.append( { "key": "worker_service_name", "label": "Worker service 名", "ok": bool(runtime_settings.get("worker_service_name")), "message": runtime_settings.get("worker_service_name", ""), } ) checks.append( { "key": "api_service_name", "label": "API service 名", "ok": bool(runtime_settings.get("api_service_name")), "message": runtime_settings.get("api_service_name", ""), } ) checks.append( { "key": "sync_agent_service_name", "label": "Sync agent service 名", "ok": bool(runtime_settings.get("sync_agent_service_name")), "message": runtime_settings.get("sync_agent_service_name", ""), } ) else: checks.append( { "key": "windows_scripts", "label": "Windows 启停脚本", "ok": (Path(settings.domain_root).parent / "start_domain_api.ps1").exists() and (Path(settings.domain_root).parent / "stop_domain_api.ps1").exists(), "message": "start_domain_api.ps1 / stop_domain_api.ps1", } ) overall_ok = all(bool(item["ok"]) for item in checks) return { "ok": overall_ok, "checks": checks, }