from __future__ import annotations import json import re import subprocess import threading import time from datetime import datetime, timedelta, timezone from app.core.config import settings from app.core.db import get_db from app.core.files import resolve_domain_path, tail_lines from app.core.redis_client import get_redis from app.services.debug_event_service import list_debug_events from app.services.cluster_runtime_service import ensure_runtime_schema from app.services.runtime_settings_service import get_runtime_settings from app.services.detect_run_service import sync_detect_runs from app.services.detect_job_service import get_active_detect_job_summary, get_detect_queue_health from app.services.settings_service import get_settings_payload, resolve_process_count, resolve_thread_count from app.services.sync_record_service import append_detect_result_projection_if_changed from app.services.worker_control_service import detect_worker_runtime _PROXY_COUNT_RE = re.compile(r"当前可用代理数[:：]\s*(\d+)") _PROXY_REFRESH_COUNT_RE = re.compile(r"代理池刷新完成，共\s*(\d+)\s*个可用代理") _PROXY_CACHE_COUNT_RE = re.compile(r"继续沿用缓存\s*(\d+)\s*个") _PROXY_SHARED_SNAPSHOT_COUNT_RE = re.compile(r"(?:复用共享代理快照|共享代理快照)\s*(\d+)\s*个") _THREAD_COUNT_RE = re.compile(r"当前实际线程数量[:：]\s*(\d+)\s*/\s*(\d+)") _STEP_TRACE_DOMAIN_RE = re.compile(r"domain=([^\s|]+)") _REGISTER_DOMAIN_RE = re.compile(r"检测注册状态[:：]\s*([^\s]+)") _RUNTIME_STATE_KEY = "domain_tool:detect_runtime_state" _TIMESTAMP_FORMATS = ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S") _SYSLOG_TIMESTAMP_FORMAT = "%b %d %H:%M:%S" _REMOTE_LOG_MAX_CHARS = 500 _REMOTE_DEBUG_EVENT_TYPES = { "worker_log", "active_job_snapshot", "domain_started", "domain_completed", "domain_failed", "domain_blacklisted", "task_pull_success", "task_pull_partial", "task_pull_failed", "queue_overdue_leases", } _DETECT_STATUS_CACHE_LOCK = threading.Lock() _DETECT_STATUS_CACHE_TTL_SECONDS = 3.0 _DETECT_STATUS_CACHE_VALUE: dict | None = None _DETECT_STATUS_CACHE_EXPIRES_AT = 0.0 _AGGREGATE_RUNTIME_NODE_STALE_AFTER = timedelta(seconds=90) _DISABLED_MANAGED_NODE_CACHE: set[str] = set() _DISABLED_MANAGED_NODE_CACHE_EXPIRES_AT = 0.0 def _clone_detect_status_payload(value: dict | None) -> dict: try: return json.loads(json.dumps(dict(value or {}), ensure_ascii=False)) except Exception: return dict(value or {}) def _extract_debug_event_job_identity(payload: dict | None) -> dict: normalized_payload = dict(payload or {}) if isinstance(payload, dict) else {} nested_job = normalized_payload.get("job") if isinstance(normalized_payload.get("job"), dict) else {} raw_job_id = normalized_payload.get("job_id") if raw_job_id in (None, "", 0, "0"): raw_job_id = normalized_payload.get("target_job_id") if raw_job_id in (None, "", 0, "0"): raw_job_id = nested_job.get("job_id") try: job_id = int(raw_job_id or 0) except Exception: job_id = 0 return { "job_id": job_id, "job_code": str( normalized_payload.get("job_code") or normalized_payload.get("target_job_code") or nested_job.get("job_code") or "" ).strip(), "cycle_token": str(normalized_payload.get("cycle_token") or nested_job.get("cycle_token") or "").strip(), "has_identity": bool(job_id > 0 or str( normalized_payload.get("job_code") or normalized_payload.get("target_job_code") or nested_job.get("job_code") or "" ).strip()), } def _debug_event_matches_active_job(record: dict, active_job: dict | None) -> bool: normalized_active_job = dict(active_job or {}) active_job_id = int(normalized_active_job.get("job_id") or 0) active_job_code = str( normalized_active_job.get("runtime_job_code") or normalized_active_job.get("job_code") or "" ).strip() active_cycle_token = str(normalized_active_job.get("current_cycle_token") or "").strip() if active_job_id <= 0 and not active_job_code and not active_cycle_token: return True identity = _extract_debug_event_job_identity(record.get("payload")) if not identity["has_identity"]: return False event_job_id = int(identity["job_id"] or 0) event_job_code = str(identity["job_code"] or "").strip() event_cycle_token = str(identity["cycle_token"] or "").strip() if active_cycle_token and event_cycle_token and event_cycle_token != active_cycle_token: return False if active_job_id > 0 and event_job_id > 0 and event_job_id != active_job_id: return False if active_job_code and event_job_code and event_job_code != active_job_code: return False if active_job_id > 0 and event_job_id == active_job_id: return True if active_job_code and event_job_code and event_job_code == active_job_code: return True if active_cycle_token and event_cycle_token and event_cycle_token == active_cycle_token: return True return False def _extract_remote_log_node_code(line: str) -> str: text = str(line or "").strip() if not text.startswith("["): return "" first_close = text.find("]") if first_close < 0: return "" second_open = text.find("[", first_close + 1) second_close = text.find("]", second_open + 1) if second_open >= 0 else -1 if second_open < 0 or second_close < 0: return "" return text[second_open + 1:second_close].strip() def _slice_remote_log_lines_fairly(lines: list[str], *, limit: int = 240, min_per_node: int = 12) -> list[str]: safe_limit = max(1, int(limit or 240)) if len(lines) <= safe_limit: return list(lines or []) normalized_lines = [str(line or "").strip() for line in list(lines or []) if str(line or "").strip()] if len(normalized_lines) <= safe_limit: return normalized_lines if min_per_node <= 0: return normalized_lines[-safe_limit:] kept_indexes: set[int] = set() per_node_counts: dict[str, int] = {} for index in range(len(normalized_lines) - 1, -1, -1): node_code = _extract_remote_log_node_code(normalized_lines[index]) if not node_code: continue current_count = int(per_node_counts.get(node_code, 0) or 0) if current_count >= min_per_node: continue kept_indexes.add(index) per_node_counts[node_code] = current_count + 1 if len(kept_indexes) >= safe_limit: break for index in range(len(normalized_lines) - 1, -1, -1): if len(kept_indexes) >= safe_limit: break kept_indexes.add(index) return [normalized_lines[index] for index in sorted(kept_indexes)] def _runtime_state_key(node_code: str | None = None) -> str: normalized_node_code = str(node_code or settings.node_code or "").strip() if not normalized_node_code: return _RUNTIME_STATE_KEY return f"{_RUNTIME_STATE_KEY}:{normalized_node_code}" def _local_worker_expected_on_this_node() -> bool: return not ( str(settings.node_region or "").strip() == "overseas" and str(settings.node_role or "").strip() == "control" ) def _int_value(value: object) -> int: try: return int(value or 0) except Exception: return 0 def _max_runtime_metric(*values: object) -> int: return max((_int_value(value) for value in values), default=0) def _resolve_capacity_node_code(node_code: str, settings_payload: dict) -> tuple[str, bool]: normalized_node_code = str(node_code or "").strip() if not normalized_node_code: return "", False node_thread_counts = dict(settings_payload.get("node_thread_counts") or {}) node_process_counts = dict(settings_payload.get("node_process_counts") or {}) parent_node_code, separator, suffix = normalized_node_code.rpartition("-") if ( separator and parent_node_code and suffix.isalpha() and len(suffix) <= 3 and ( parent_node_code in node_thread_counts or parent_node_code in node_process_counts or bool(re.search(r"\d$", parent_node_code)) ) ): return parent_node_code, True if normalized_node_code in node_thread_counts or normalized_node_code in node_process_counts: return normalized_node_code, False return normalized_node_code, False def _is_current_participant_bucket(item: dict | None) -> bool: payload = dict(item or {}) node_code = str(payload.get("node_code") or "").strip() if not node_code or node_code == "unassigned": return False return any( _int_value(payload.get(field)) > 0 for field in ("items_claimed", "items_running", "display_running", "current_load", "active_threads") ) def _aggregate_runtime_node_is_live(item: dict | None) -> bool: payload = dict(item or {}) node_code = str(payload.get("node_code") or "").strip() if not node_code: return False if node_code == "unassigned": return True status = str(payload.get("status") or "").strip().lower() if status in {"stale", "offline"}: return False last_heartbeat_at = _parse_time(payload.get("last_heartbeat_at")) if last_heartbeat_at is None: return True reference_now = datetime.now(last_heartbeat_at.tzinfo) if last_heartbeat_at.tzinfo else datetime.now() return (reference_now - last_heartbeat_at) <= _AGGREGATE_RUNTIME_NODE_STALE_AFTER def _filter_live_aggregate_runtime_nodes(node_rows: list[dict] | None) -> list[dict]: disabled_node_codes = _load_disabled_managed_node_codes( [ str(item.get("node_code") or "").strip() for item in list(node_rows or []) if isinstance(item, dict) ] ) return [ dict(item) for item in list(node_rows or []) if isinstance(item, dict) and str(item.get("node_code") or "").strip() not in disabled_node_codes and _aggregate_runtime_node_is_live(item) ] def _build_aggregate_detect_capacity(*, active_job: dict | None, settings_payload: dict) -> dict: node_rows = _filter_live_aggregate_runtime_nodes( list((active_job or {}).get("distributed_node_stats") or (active_job or {}).get("node_stats") or []) ) node_thread_counts = dict(settings_payload.get("node_thread_counts") or {}) node_process_counts = dict(settings_payload.get("node_process_counts") or {}) participant_node_codes: list[str] = [] process_count_total = 0 max_threads_total = 0 representative_thread_count = 0 child_parent_codes: set[str] = set() for raw_item in node_rows: if not isinstance(raw_item, dict) or not _is_current_participant_bucket(raw_item): continue node_code = str(raw_item.get("node_code") or "").strip() capacity_node_code, is_child_instance = _resolve_capacity_node_code(node_code, settings_payload) if is_child_instance and capacity_node_code: child_parent_codes.add(capacity_node_code) for raw_item in node_rows: if not isinstance(raw_item, dict) or not _is_current_participant_bucket(raw_item): continue node_code = str(raw_item.get("node_code") or "").strip() capacity_node_code, is_child_instance = _resolve_capacity_node_code(node_code, settings_payload) if not capacity_node_code: continue if not is_child_instance and node_code in child_parent_codes: continue if node_code not in participant_node_codes: participant_node_codes.append(node_code) thread_resolution_node_code = node_code if node_code in node_thread_counts else capacity_node_code thread_resolution = resolve_thread_count(node_code=thread_resolution_node_code, settings_payload=settings_payload) per_process_thread_count = max(1, int(thread_resolution["effective_thread_count"] or 1)) if representative_thread_count <= 0: representative_thread_count = per_process_thread_count if is_child_instance: process_count = 1 else: if node_code in node_process_counts: process_resolution = resolve_process_count(node_code=node_code, settings_payload=settings_payload) process_count = max(1, int(process_resolution["effective_process_count"] or 1)) else: process_count = 1 process_count_total += process_count max_threads_total += process_count * per_process_thread_count return { "participant_node_codes": participant_node_codes, "participant_node_count": len(participant_node_codes), "process_count": process_count_total, "max_threads": max_threads_total, "per_process_thread_count": representative_thread_count, } def _merge_aggregate_active_job_with_queue_health(active_job: dict | None, queue_health: dict | None) -> dict | None: normalized_active_job = dict(active_job or {}) normalized_queue_health = dict(queue_health or {}) if not normalized_queue_health.get("has_active_job"): return normalized_active_job or active_job queue_payload = dict(normalized_queue_health.get("queue") or {}) queue_job = dict(normalized_queue_health.get("job") or {}) raw_queue_nodes = [dict(item) for item in list(normalized_queue_health.get("nodes") or []) if isinstance(item, dict)] queue_nodes = _filter_live_aggregate_runtime_nodes(raw_queue_nodes) if not queue_payload and not queue_nodes and not queue_job: return normalized_active_job or active_job queue_display_running = sum( _max_runtime_metric( item.get("display_running"), item.get("current_load"), item.get("active_threads"), item.get("items_running"), ) for item in queue_nodes if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned" ) if raw_queue_nodes: display_max_threads = sum(_int_value(item.get("max_threads")) for item in queue_nodes) display_items_running = queue_display_running display_active_threads = queue_display_running else: display_max_threads = _max_runtime_metric( sum(_int_value(item.get("max_threads")) for item in queue_nodes), normalized_active_job.get("display_max_threads"), ) display_items_running = _max_runtime_metric( queue_display_running, queue_payload.get("display_running"), normalized_active_job.get("display_items_running"), normalized_active_job.get("display_active_threads"), ) display_active_threads = _max_runtime_metric( queue_display_running, queue_payload.get("display_running"), normalized_active_job.get("display_active_threads"), normalized_active_job.get("display_items_running"), ) merged = dict(normalized_active_job) merged.update( { "job_id": queue_job.get("job_id", merged.get("job_id")), "job_code": queue_job.get("job_code", merged.get("job_code")), "status": queue_job.get("status", merged.get("status")), "progress_percent": queue_job.get("progress_percent", merged.get("progress_percent", 0)), "items_total": _int_value(queue_payload.get("items_total", merged.get("items_total"))), "items_pending": _int_value(queue_payload.get("pending", merged.get("items_pending"))), "items_claimed": _int_value(queue_payload.get("claimed", merged.get("items_claimed"))), "items_running": _int_value(queue_payload.get("running", merged.get("items_running"))), "items_completed": _int_value(queue_payload.get("completed", merged.get("items_completed"))), "items_blacklisted": _int_value(queue_payload.get("blacklisted", merged.get("items_blacklisted"))), "items_failed": _int_value(queue_payload.get("failed", merged.get("items_failed"))), "display_items_running": display_items_running, "display_active_threads": display_active_threads, "display_max_threads": display_max_threads, } ) if raw_queue_nodes: merged["node_stats"] = list(queue_nodes) merged["distributed_node_stats"] = list(queue_nodes) return merged def _extract_dependency_alerts(lines: list[str]) -> list[dict]: alerts: list[dict] = [] recent_lines = lines[-120:] if lines else [] degraded_line = "" for line in reversed(recent_lines): if "外部依赖异常，步骤降级继续执行" in line: degraded_line = line alerts.append( { "kind": "dependency_degraded", "level": "warning", "title": "外部依赖降级继续", "detail": line, } ) break if degraded_line: return alerts for line in reversed(recent_lines): if "WaybackDetector" in line or "web.archive.org" in line: alerts.append( { "kind": "wayback", "level": "warning", "title": "时光机依赖异常", "detail": line, } ) break for line in reversed(recent_lines): if any(keyword in line for keyword in ("HTTPSConnectionPool", "Connection refused", "Read timed out", "ConnectTimeout")): alerts.append( { "kind": "network", "level": "warning", "title": "外部网络波动", "detail": line, } ) break return alerts def _extract_available_proxy_count(lines: list[str]) -> int: for line in reversed(lines): count = _extract_available_proxy_count_from_text(line) if count > 0: return count return 0 def _extract_available_proxy_count_from_text(text: str) -> int: normalized_text = str(text or "").strip() if not normalized_text: return 0 for pattern in ( _PROXY_COUNT_RE, _PROXY_REFRESH_COUNT_RE, _PROXY_CACHE_COUNT_RE, _PROXY_SHARED_SNAPSHOT_COUNT_RE, ): match = pattern.search(normalized_text) if not match: continue try: return int(match.group(1) or 0) except Exception: continue return 0 def _extract_active_thread_snapshot(lines: list[str]) -> dict: for line in reversed(lines): match = _THREAD_COUNT_RE.search(line) if match: return { "active": int(match.group(1)), "max": int(match.group(2)), } return {"active": 0, "max": 0} def _estimate_active_threads_from_recent_lines(lines: list[str], *, limit: int) -> int: if not lines: return 0 active_domains: list[str] = [] seen_domains: set[str] = set() for line in reversed(lines[-80:]): match = _STEP_TRACE_DOMAIN_RE.search(line) or _REGISTER_DOMAIN_RE.search(line) if not match: continue domain = str(match.group(1) or "").strip() if not domain or domain in seen_domains: continue seen_domains.add(domain) active_domains.append(domain) if len(active_domains) >= int(limit): break return len(active_domains) def _parse_time(raw: str | None) -> datetime | None: if not raw: return None text = str(raw).strip() if not text: return None try: return datetime.fromisoformat(text) except ValueError: pass for fmt in _TIMESTAMP_FORMATS: try: return datetime.strptime(text, fmt) except ValueError: continue return None def _extract_log_time(line: str, *, reference_year: int | None = None) -> datetime | None: if len(line) < 19: text = str(line or "").strip() else: text = str(line or "") candidates = [text[:26], text[:19]] for candidate in candidates: for fmt in _TIMESTAMP_FORMATS: if len(candidate) != len(datetime.now().strftime(fmt)): continue try: return datetime.strptime(candidate, fmt) except ValueError: continue syslog_candidate = str(text[:15] or "").strip() if syslog_candidate: try: parsed = datetime.strptime(syslog_candidate, _SYSLOG_TIMESTAMP_FORMAT) return parsed.replace(year=int(reference_year or datetime.now().year)) except ValueError: pass return None def _read_worker_journal_lines(service_name: str, *, max_lines: int) -> tuple[list[str], str | None]: normalized_service_name = str(service_name or "").strip() if not normalized_service_name: return [], None try: completed = subprocess.run( ["journalctl", "-u", normalized_service_name, "-n", str(max_lines), "--no-pager"], capture_output=True, text=True, timeout=12, ) except Exception: return [], None output = str(completed.stdout or "").strip() if completed.returncode != 0 or not output: return [], None lines = [str(line or "").rstrip() for line in output.splitlines() if str(line or "").strip()] if not lines: return [], None return lines[-max_lines:], datetime.now(timezone.utc).isoformat() def _load_recent_worker_lines(runtime_settings: dict, *, max_lines: int = 160) -> tuple[bool, str | None, list[str]]: worker_log = resolve_domain_path("detect_worker.log", "logs/detect_worker.log") worker_online = False last_log_time: str | None = None recent_lines = tail_lines("detect_worker.log", max_lines=max_lines) if worker_log and worker_log.exists(): modified = datetime.fromtimestamp(worker_log.stat().st_mtime, tz=timezone.utc) last_log_time = modified.isoformat() worker_online = (datetime.now(timezone.utc) - modified).total_seconds() < 180 if str(runtime_settings.get("worker_mode") or "").strip() == "linux-systemd": service_name = str(runtime_settings.get("worker_service_name") or "").strip() or "domaincheck-worker" journal_lines, journal_last_time = _read_worker_journal_lines(service_name, max_lines=max_lines) if journal_lines: recent_lines = journal_lines worker_online = True if journal_last_time: last_log_time = journal_last_time return worker_online, last_log_time, recent_lines def _filter_lines_since(lines: list[str], started_at: str | None) -> list[str]: started_time = _parse_time(started_at) if not started_time: return lines filtered: list[str] = [] parsed_any = False for line in lines: line_time = _extract_log_time(line, reference_year=started_time.year) if line_time is None: continue parsed_any = True if line_time >= started_time: filtered.append(line) if not parsed_any: return lines return filtered or lines def _recent_event(lines: list[str]) -> str: interesting_keywords = ( "开始检测", "获取到", "当前批次检测完成", "域名检测任务完成", "代理池刷新完成", "没有需要检测的域名", "检测已停止", ) for line in reversed(lines): if any(keyword in line for keyword in interesting_keywords): return line return "" def _build_remote_log_lines( active_job: dict | None, *, enabled: bool, mode: str, limit: int = 240, ) -> list[str]: return _build_remote_log_snapshot(active_job, enabled=enabled, mode=mode, limit=limit)["lines"] def _build_remote_log_snapshot( active_job: dict | None, *, enabled: bool, mode: str, limit: int = 240, ) -> dict: if not enabled: return { "lines": [], "line_count": 0, "last_at": "", "last_line": "", "source_nodes": [], "source_node_count": 0, } if not active_job: return { "lines": [], "line_count": 0, "last_at": "", "last_line": "", "source_nodes": [], "source_node_count": 0, } events = list(active_job.get("current_cycle_events") or active_job.get("recent_events") or []) if not events: return { "lines": [], "line_count": 0, "last_at": "", "last_line": "", "source_nodes": [], "source_node_count": 0, } lines: list[str] = [] source_nodes: set[str] = set() source_node_summaries: dict[str, dict] = {} last_at = "" last_line = "" normalized_mode = str(mode or "key").strip().lower() if normalized_mode not in {"key", "full"}: normalized_mode = "key" current_cycle_token = str(active_job.get("current_cycle_token") or "").strip() for event in reversed(events): event_type = str(event.get("event_type") or "").strip() if event_type != "worker_log": continue created_at = str(event.get("created_at") or "").strip() node_code = str(event.get("node_code") or "").strip() or "unknown" message = str(event.get("message") or "").strip() if not message: continue payload = event.get("payload") if isinstance(event.get("payload"), dict) else {} event_cycle_token = str(payload.get("cycle_token") or "").strip() if current_cycle_token and event_cycle_token and event_cycle_token != current_cycle_token: continue event_mode = str(payload.get("log_mode") or "key").strip().lower() if event_mode not in {"key", "full"}: event_mode = "key" if normalized_mode != "full" and event_mode == "full": continue if len(message) > _REMOTE_LOG_MAX_CHARS: message = f"{message[:_REMOTE_LOG_MAX_CHARS]}..." formatted_line = f"[{created_at}] [{node_code}] {message}" lines.append(formatted_line) source_nodes.add(node_code) node_summary = source_node_summaries.setdefault( node_code, { "node_code": node_code, "line_count": 0, "key_line_count": 0, "full_line_count": 0, "last_at": "", "last_line": "", }, ) node_summary["line_count"] += 1 if event_mode == "full": node_summary["full_line_count"] += 1 else: node_summary["key_line_count"] += 1 node_summary["last_at"] = created_at node_summary["last_line"] = formatted_line last_at = created_at last_line = formatted_line sliced_lines = lines[-max(1, int(limit or 240)) :] sorted_source_node_summaries = sorted( source_node_summaries.values(), key=lambda item: ( str(item.get("last_at") or ""), str(item.get("node_code") or ""), ), reverse=True, ) return { "lines": sliced_lines, "line_count": len(sliced_lines), "last_at": last_at, "last_line": last_line, "source_nodes": sorted(source_nodes), "source_node_count": len(source_nodes), "source_node_summaries": sorted_source_node_summaries, } def _build_remote_log_snapshot_from_debug_events( active_job: dict | None, *, enabled: bool, mode: str, limit: int = 240, ) -> dict: if not enabled: return { "lines": [], "line_count": 0, "last_at": "", "last_line": "", "source_nodes": [], "source_node_count": 0, "source_node_summaries": [], } normalized_mode = str(mode or "key").strip().lower() if normalized_mode not in {"key", "full"}: normalized_mode = "key" participating_node_codes = { str(item.get("node_code") or "").strip() for item in list((active_job or {}).get("node_stats") or []) if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned" } created_after = (datetime.now() - timedelta(hours=6)).strftime("%Y-%m-%d %H:%M:%S") safe_limit = max(1, int(limit or 240)) node_limit = max(40, min(200, safe_limit)) records: list[dict] = [] if participating_node_codes: for node_code in sorted(participating_node_codes): payload = list_debug_events( limit=node_limit, created_after=created_after, node_code=node_code, ) records.extend(list(payload.get("records") or [])) records.sort( key=lambda item: ( str(item.get("created_at") or ""), int(item.get("id") or 0), ), reverse=True, ) else: payload = list_debug_events(limit=max(safe_limit * 4, 240), created_after=created_after) records = list(payload.get("records") or []) if not records: return { "lines": [], "line_count": 0, "last_at": "", "last_line": "", "source_nodes": [], "source_node_count": 0, "source_node_summaries": [], } lines: list[str] = [] source_nodes: set[str] = set() source_node_summaries: dict[str, dict] = {} last_at = "" last_line = "" for record in reversed(records): event_type = str(record.get("event_type") or "").strip() if event_type not in _REMOTE_DEBUG_EVENT_TYPES: continue node_code = str(record.get("node_code") or "").strip() or "unknown" if participating_node_codes and node_code not in participating_node_codes: continue if not _debug_event_matches_active_job(record, active_job): continue message = str(record.get("message") or "").strip() if not message: continue created_at = str(record.get("created_at") or "").strip() payload = record.get("payload") if isinstance(record.get("payload"), dict) else {} event_mode = str(payload.get("log_mode") or "key").strip().lower() if event_mode not in {"key", "full"}: event_mode = "key" if normalized_mode != "full" and event_mode == "full": continue if len(message) > _REMOTE_LOG_MAX_CHARS: message = f"{message[:_REMOTE_LOG_MAX_CHARS]}..." formatted_line = f"[{created_at}] [{node_code}] {message}" lines.append(formatted_line) source_nodes.add(node_code) node_summary = source_node_summaries.setdefault( node_code, { "node_code": node_code, "line_count": 0, "key_line_count": 0, "full_line_count": 0, "last_at": "", "last_line": "", }, ) node_summary["line_count"] += 1 if event_mode == "full": node_summary["full_line_count"] += 1 else: node_summary["key_line_count"] += 1 node_summary["last_at"] = created_at node_summary["last_line"] = formatted_line last_at = created_at last_line = formatted_line sliced_lines = _slice_remote_log_lines_fairly(lines, limit=safe_limit) sorted_source_node_summaries = sorted( source_node_summaries.values(), key=lambda item: ( str(item.get("last_at") or ""), str(item.get("node_code") or ""), ), reverse=True, ) return { "lines": sliced_lines, "line_count": len(sliced_lines), "last_at": last_at, "last_line": last_line, "source_nodes": sorted(source_nodes), "source_node_count": len(source_nodes), "source_node_summaries": sorted_source_node_summaries, } def _merge_remote_log_snapshots(primary: dict, secondary: dict, *, limit: int = 240) -> dict: merged_lines: list[str] = [] seen_lines: set[str] = set() for raw_line in list(primary.get("lines") or []) + list(secondary.get("lines") or []): line = str(raw_line or "").strip() if not line or line in seen_lines: continue seen_lines.add(line) merged_lines.append(line) if limit > 0: merged_lines = _slice_remote_log_lines_fairly(merged_lines, limit=limit) summaries: dict[str, dict] = {} for snapshot in (primary, secondary): for raw_summary in list(snapshot.get("source_node_summaries") or []): if not isinstance(raw_summary, dict): continue node_code = str(raw_summary.get("node_code") or "").strip() if not node_code: continue summary = summaries.setdefault( node_code, { "node_code": node_code, "line_count": 0, "key_line_count": 0, "full_line_count": 0, "last_at": "", "last_line": "", }, ) summary["line_count"] = max(int(summary.get("line_count", 0) or 0), int(raw_summary.get("line_count", 0) or 0)) summary["key_line_count"] = max(int(summary.get("key_line_count", 0) or 0), int(raw_summary.get("key_line_count", 0) or 0)) summary["full_line_count"] = max(int(summary.get("full_line_count", 0) or 0), int(raw_summary.get("full_line_count", 0) or 0)) raw_last_at = str(raw_summary.get("last_at") or "") if raw_last_at >= str(summary.get("last_at") or ""): summary["last_at"] = raw_last_at summary["last_line"] = str(raw_summary.get("last_line") or "") source_nodes = sorted( { str(node_code or "").strip() for node_code in list(primary.get("source_nodes") or []) + list(secondary.get("source_nodes") or []) if str(node_code or "").strip() } ) last_at = max(str(primary.get("last_at") or ""), str(secondary.get("last_at") or "")) last_line = str(primary.get("last_line") or "") if str(secondary.get("last_at") or "") >= str(primary.get("last_at") or ""): last_line = str(secondary.get("last_line") or last_line) return { "lines": merged_lines, "line_count": len(merged_lines), "last_at": last_at, "last_line": last_line, "source_nodes": source_nodes, "source_node_count": len(source_nodes), "source_node_summaries": sorted( summaries.values(), key=lambda item: ( str(item.get("last_at") or ""), str(item.get("node_code") or ""), ), reverse=True, ), } def _resolve_remote_log_lines( active_job: dict | None, runs: list[dict], *, enabled: bool, mode: str, limit: int = 240, ) -> list[str]: return _build_remote_log_lines(active_job, enabled=enabled, mode=mode, limit=limit) def _resolve_remote_log_snapshot( active_job: dict | None, runs: list[dict], *, enabled: bool, mode: str, limit: int = 240, ) -> dict: primary_snapshot = _build_remote_log_snapshot(active_job, enabled=enabled, mode=mode, limit=limit) debug_snapshot = _build_remote_log_snapshot_from_debug_events(active_job, enabled=enabled, mode=mode, limit=limit) if int(primary_snapshot.get("line_count", 0) or 0) <= 0: return debug_snapshot if int(debug_snapshot.get("line_count", 0) or 0) <= 0: return primary_snapshot return _merge_remote_log_snapshots(primary_snapshot, debug_snapshot, limit=limit) def _load_runtime_state() -> dict: try: redis_client = get_redis() for key in (_runtime_state_key(), _RUNTIME_STATE_KEY): raw = redis_client.get(key) if not raw: continue data = json.loads(raw) if not isinstance(data, dict): continue if key == _RUNTIME_STATE_KEY: payload_node_code = str(data.get("node_code") or "").strip() if payload_node_code and payload_node_code != str(settings.node_code or "").strip(): continue return data return {} except Exception: return {} def _load_disabled_managed_node_codes(node_codes: list[str] | None = None) -> set[str]: global _DISABLED_MANAGED_NODE_CACHE, _DISABLED_MANAGED_NODE_CACHE_EXPIRES_AT normalized_codes = [ str(item or "").strip() for item in list(node_codes or []) if str(item or "").strip() ] now_ts = time.time() if not normalized_codes and now_ts < _DISABLED_MANAGED_NODE_CACHE_EXPIRES_AT: return set(_DISABLED_MANAGED_NODE_CACHE) try: with get_db() as conn: with conn.cursor() as cur: if normalized_codes: cur.execute( """ SELECT node_code FROM ops_managed_nodes WHERE is_enabled = FALSE AND node_code = ANY(%s) """, (normalized_codes,), ) else: cur.execute( """ SELECT node_code FROM ops_managed_nodes WHERE is_enabled = FALSE """ ) rows = list(cur.fetchall() or []) except Exception: if normalized_codes: return {code for code in normalized_codes if code in _DISABLED_MANAGED_NODE_CACHE} return set(_DISABLED_MANAGED_NODE_CACHE) disabled_codes = { str(row[0] or "").strip() for row in rows if str(row[0] or "").strip() } if normalized_codes: return disabled_codes _DISABLED_MANAGED_NODE_CACHE = disabled_codes _DISABLED_MANAGED_NODE_CACHE_EXPIRES_AT = now_ts + 5.0 return set(disabled_codes) def _load_runtime_state_from_cluster_node() -> dict: try: with get_db() as conn: with conn.cursor() as cur: cur.execute( """ SELECT current_load, metadata_json, last_heartbeat_at FROM detect_worker_nodes WHERE node_code = %s LIMIT 1 """, (settings.node_code,), ) row = cur.fetchone() if not row: return {} current_load, metadata_json, last_heartbeat_at = row metadata = metadata_json if isinstance(metadata_json, dict) else {} if not metadata: return {} runtime_state = { "node_code": settings.node_code, "phase": str(metadata.get("phase") or metadata.get("phase_label") or "").strip(), "detail": str(metadata.get("detail") or metadata.get("phase_detail") or "").strip(), "service_running": True, "detecting": bool(metadata.get("detecting", False) or int(current_load or 0) > 0), "stop_requested": False, "available_proxy_count": int(metadata.get("available_proxy_count", 0) or 0), "active_threads": int(metadata.get("active_threads", 0) or 0), "max_threads": int(metadata.get("max_threads", 0) or 0), "job_id": metadata.get("job_id"), "job_code": str(metadata.get("job_code") or metadata.get("active_job_code") or "").strip(), "updated_at": str(metadata.get("updated_at") or (_format_time(last_heartbeat_at) if last_heartbeat_at else "")).strip(), } if runtime_state["detail"] or runtime_state["active_threads"] > 0 or runtime_state["max_threads"] > 0: return runtime_state return {} except Exception: return {} def _load_runtime_states_from_cluster_nodes(node_codes: list[str] | tuple[str, ...]) -> dict[str, dict]: normalized_node_codes = [ str(item or "").strip() for item in list(node_codes or []) if str(item or "").strip() ] if not normalized_node_codes: return {} try: with get_db() as conn: with conn.cursor() as cur: cur.execute( """ SELECT node_code, current_load, metadata_json, last_heartbeat_at FROM detect_worker_nodes WHERE node_code = ANY(%s) """, (normalized_node_codes,), ) rows = list(cur.fetchall() or []) except Exception: return {} payload: dict[str, dict] = {} for row in rows: node_code = str(row[0] or "").strip() if not node_code: continue current_load = int(row[1] or 0) metadata_json = row[2] last_heartbeat_at = row[3] metadata = metadata_json if isinstance(metadata_json, dict) else {} payload[node_code] = { "node_code": node_code, "current_load": current_load, "last_heartbeat_at": ( last_heartbeat_at.isoformat(sep=" ", timespec="seconds") if hasattr(last_heartbeat_at, "isoformat") else str(last_heartbeat_at or "").strip() ), "available_proxy_count": int( metadata.get("available_proxy_count", metadata.get("proxy_last_available_count", 0)) or 0 ), "proxy_runtime_label": str(metadata.get("proxy_runtime_label") or "").strip(), "proxy_runtime_reason": str(metadata.get("proxy_runtime_reason") or "").strip(), "proxy_last_refresh_status": str(metadata.get("proxy_last_refresh_status") or "").strip(), "proxy_last_refresh_time": str(metadata.get("proxy_last_refresh_time") or "").strip(), "proxy_last_refresh_source_count": int(metadata.get("proxy_last_refresh_source_count", 0) or 0), "proxy_last_refresh_total_items": int(metadata.get("proxy_last_refresh_total_items", 0) or 0), "proxy_last_validated_count": int(metadata.get("proxy_last_validated_count", 0) or 0), "active_threads": int(metadata.get("active_threads", 0) or 0), "max_threads": int(metadata.get("max_threads", 0) or 0), "detect_participating": bool(metadata.get("detect_participating", False) or current_load > 0), } return payload def _load_recent_proxy_debug_events(node_codes: list[str] | tuple[str, ...], *, window_minutes: int = 20) -> list[dict]: normalized_node_codes = [ str(item or "").strip() for item in list(node_codes or []) if str(item or "").strip() and str(item or "").strip() != "unassigned" ] if not normalized_node_codes: return [] safe_window_minutes = max(5, min(int(window_minutes or 20), 120)) created_after = datetime.now() - timedelta(minutes=safe_window_minutes) safe_limit = max(80, min(len(normalized_node_codes) * 20, 800)) try: with get_db() as conn: with conn.cursor() as cur: cur.execute( """ SELECT node_code, message, payload_json, created_at FROM detect_debug_events WHERE event_type = 'worker_log' AND node_code = ANY(%s) AND created_at >= %s ORDER BY created_at DESC, id DESC LIMIT %s """, (normalized_node_codes, created_after, safe_limit), ) rows = list(cur.fetchall() or []) except Exception: return [] return [ { "event_type": "worker_log", "node_code": str(row[0] or "").strip(), "message": str(row[1] or "").strip(), "payload": row[2] if isinstance(row[2], dict) else {}, "created_at": ( row[3].isoformat(sep=" ", timespec="seconds") if hasattr(row[3], "isoformat") else str(row[3] or "").strip() ), } for row in rows if str(row[0] or "").strip() and str(row[1] or "").strip() ] def _is_proxy_runtime_message(message: str) -> bool: normalized_message = str(message or "").strip() if not normalized_message: return False if _extract_available_proxy_count_from_text(normalized_message) > 0: return True lowered_message = normalized_message.lower() return any( keyword in normalized_message or keyword in lowered_message for keyword in ( "代理", "proxy", "cooldown", "rate limited", ) ) def _infer_proxy_runtime_label_from_message(message: str, *, available_proxy_count: int) -> tuple[str, str]: normalized_message = str(message or "").strip() lowered_message = normalized_message.lower() if available_proxy_count > 0: return "代理正常", "aggregate_log_healthy" if "rate limited" in lowered_message or "cooldown" in lowered_message or "冷却" in normalized_message: return "代理源暂时冷却中", "aggregate_log_cooldown" if any(keyword in normalized_message for keyword in ("未取到新代理", "未取到可用代理数据", "无可用代理", "未返回可用代理数据")): return "代理待补货", "aggregate_log_empty" if "等待首刷" in normalized_message: return "等待首刷", "aggregate_log_waiting" return "", "" def _build_aggregate_proxy_runtime_rows_from_events( *, active_job: dict | None, settings_payload: dict, participant_node_codes: list[str], participant_server_codes: list[str], ) -> dict[str, dict]: normalized_active_job = dict(active_job or {}) events = list(normalized_active_job.get("current_cycle_events") or normalized_active_job.get("recent_events") or []) events.extend(_load_recent_proxy_debug_events(participant_node_codes)) if not events: return {} current_cycle_token = str(normalized_active_job.get("current_cycle_token") or "").strip() allowed_server_codes = {str(item or "").strip() for item in list(participant_server_codes or []) if str(item or "").strip()} rows: dict[str, dict] = {} for event in events: if not isinstance(event, dict): continue if str(event.get("event_type") or "").strip() != "worker_log": continue payload = event.get("payload") if isinstance(event.get("payload"), dict) else {} event_cycle_token = str(payload.get("cycle_token") or "").strip() if current_cycle_token and event_cycle_token and event_cycle_token != current_cycle_token: continue node_code = str(event.get("node_code") or "").strip() if not node_code or node_code == "unassigned": continue capacity_node_code, _ = _resolve_capacity_node_code(node_code, settings_payload) server_code = str(capacity_node_code or node_code or "").strip() if not server_code or (allowed_server_codes and server_code not in allowed_server_codes): continue message = str(event.get("message") or "").strip() if not _is_proxy_runtime_message(message): continue created_at = str(event.get("created_at") or "").strip() available_proxy_count = _extract_available_proxy_count_from_text(message) label, reason = _infer_proxy_runtime_label_from_message( message, available_proxy_count=available_proxy_count, ) row = rows.setdefault( server_code, { "node_code": server_code, "available_proxy_count": 0, "proxy_runtime_label": "", "proxy_runtime_reason": "", "proxy_last_refresh_status": "", "proxy_last_refresh_time": "", "proxy_last_refresh_source_count": 0, "proxy_last_refresh_total_items": 0, "proxy_last_validated_count": 0, "_count_seen_at": "", "_status_seen_at": "", }, ) if available_proxy_count > 0 and created_at >= str(row.get("_count_seen_at") or ""): row["available_proxy_count"] = int(available_proxy_count or 0) row["_count_seen_at"] = created_at if created_at >= str(row.get("_status_seen_at") or ""): row["proxy_last_refresh_status"] = message row["proxy_last_refresh_time"] = created_at row["_status_seen_at"] = created_at if label: row["proxy_runtime_label"] = label if reason: row["proxy_runtime_reason"] = reason return { server_code: { key: value for key, value in row.items() if not str(key).startswith("_") } for server_code, row in rows.items() } def _build_aggregate_proxy_runtime( *, active_job: dict | None, settings_payload: dict, fallback_available_proxy_count: int, fallback_proxy_runtime: dict, ) -> tuple[int, dict]: node_rows = list((active_job or {}).get("distributed_node_stats") or (active_job or {}).get("node_stats") or []) participant_node_codes: list[str] = [] participant_server_codes: list[str] = [] for raw_item in node_rows: if not isinstance(raw_item, dict) or not _is_current_participant_bucket(raw_item): continue node_code = str(raw_item.get("node_code") or "").strip() if node_code and node_code != "unassigned" and node_code not in participant_node_codes: participant_node_codes.append(node_code) capacity_node_code, _ = _resolve_capacity_node_code(node_code, settings_payload) normalized_server_code = str(capacity_node_code or node_code or "").strip() if not normalized_server_code or normalized_server_code == "unassigned": continue if normalized_server_code not in participant_server_codes: participant_server_codes.append(normalized_server_code) if not participant_server_codes: return fallback_available_proxy_count, fallback_proxy_runtime event_runtime = _build_aggregate_proxy_runtime_rows_from_events( active_job=active_job, settings_payload=settings_payload, participant_node_codes=participant_node_codes, participant_server_codes=participant_server_codes, ) cluster_runtime = _load_runtime_states_from_cluster_nodes(participant_server_codes) rows: list[dict] = [] for code in participant_server_codes: runtime_row = dict(cluster_runtime.get(code) or {}) event_row = dict(event_runtime.get(code) or {}) if not runtime_row and not event_row: continue merged_row = { "node_code": code, "available_proxy_count": int(runtime_row.get("available_proxy_count", 0) or 0), "proxy_runtime_label": str(runtime_row.get("proxy_runtime_label") or "").strip(), "proxy_runtime_reason": str(runtime_row.get("proxy_runtime_reason") or "").strip(), "proxy_last_refresh_status": str(runtime_row.get("proxy_last_refresh_status") or "").strip(), "proxy_last_refresh_time": str(runtime_row.get("proxy_last_refresh_time") or runtime_row.get("last_heartbeat_at") or "").strip(), "proxy_last_refresh_source_count": int(runtime_row.get("proxy_last_refresh_source_count", 0) or 0), "proxy_last_refresh_total_items": int(runtime_row.get("proxy_last_refresh_total_items", 0) or 0), "proxy_last_validated_count": int(runtime_row.get("proxy_last_validated_count", 0) or 0), } if int(merged_row.get("available_proxy_count", 0) or 0) <= 0 and int(event_row.get("available_proxy_count", 0) or 0) > 0: merged_row["available_proxy_count"] = int(event_row.get("available_proxy_count", 0) or 0) if not str(merged_row.get("proxy_runtime_label") or "").strip(): merged_row["proxy_runtime_label"] = str(event_row.get("proxy_runtime_label") or "").strip() if not str(merged_row.get("proxy_runtime_reason") or "").strip(): merged_row["proxy_runtime_reason"] = str(event_row.get("proxy_runtime_reason") or "").strip() if not str(merged_row.get("proxy_last_refresh_status") or "").strip(): merged_row["proxy_last_refresh_status"] = str(event_row.get("proxy_last_refresh_status") or "").strip() if not str(merged_row.get("proxy_last_refresh_time") or "").strip(): merged_row["proxy_last_refresh_time"] = str(event_row.get("proxy_last_refresh_time") or "").strip() if int(merged_row.get("proxy_last_refresh_source_count", 0) or 0) <= 0: merged_row["proxy_last_refresh_source_count"] = int(event_row.get("proxy_last_refresh_source_count", 0) or 0) if int(merged_row.get("proxy_last_refresh_total_items", 0) or 0) <= 0: merged_row["proxy_last_refresh_total_items"] = int(event_row.get("proxy_last_refresh_total_items", 0) or 0) if int(merged_row.get("proxy_last_validated_count", 0) or 0) <= 0: merged_row["proxy_last_validated_count"] = int(event_row.get("proxy_last_validated_count", 0) or 0) rows.append(merged_row) if not rows: return fallback_available_proxy_count, fallback_proxy_runtime total_available_proxy_count = sum(max(0, int(item.get("available_proxy_count", 0) or 0)) for item in rows) latest_refresh_time = max((str(item.get("proxy_last_refresh_time") or "") for item in rows), default="") source_count = sum(int(item.get("proxy_last_refresh_source_count", 0) or 0) for item in rows) raw_items = sum(int(item.get("proxy_last_refresh_total_items", 0) or 0) for item in rows) validated_count = sum(int(item.get("proxy_last_validated_count", 0) or 0) for item in rows) refresh_status_parts = [ f"{str(item.get('node_code') or '')}:{str(item.get('proxy_last_refresh_status') or '').strip()}" for item in rows if str(item.get("proxy_last_refresh_status") or "").strip() ] refresh_status = "；".join(refresh_status_parts[:6]) if len(refresh_status_parts) > 6: refresh_status = f"{refresh_status} 等 {len(refresh_status_parts)} 台" if total_available_proxy_count > 0: return total_available_proxy_count, { "state": "healthy", "label": "集群代理正常", "detail": ( f"参与服务器 {len(rows)} 台，共可用 {total_available_proxy_count} 个代理" + (f"；最近状态：{refresh_status}" if refresh_status else "") ), "direct_fallback_active": False, "reason": "aggregate_healthy", "last_refresh_status": refresh_status, "last_refresh_time": latest_refresh_time, "source_count": source_count, "raw_items": raw_items, "validated_count": validated_count, "available_count": total_available_proxy_count, "source_stats": [], "supplier_empty": False, } fallback_label = next((str(item.get("proxy_runtime_label") or "").strip() for item in rows if str(item.get("proxy_runtime_label") or "").strip()), "") fallback_reason = next((str(item.get("proxy_runtime_reason") or "").strip() for item in rows if str(item.get("proxy_runtime_reason") or "").strip()), "") if fallback_label: return 0, { "state": "warming_up", "label": fallback_label, "detail": ( f"参与服务器 {len(rows)} 台，当前尚未汇总到可用代理" + (f"；最近状态：{refresh_status}" if refresh_status else "") ), "direct_fallback_active": bool(fallback_proxy_runtime.get("direct_fallback_active", False)), "reason": fallback_reason or "aggregate_proxy_unavailable", "last_refresh_status": refresh_status, "last_refresh_time": latest_refresh_time, "source_count": source_count, "raw_items": raw_items, "validated_count": validated_count, "available_count": 0, "source_stats": [], "supplier_empty": bool(fallback_proxy_runtime.get("supplier_empty", False)), } return fallback_available_proxy_count, fallback_proxy_runtime def _normalize_recent_warning(runtime_state: dict, recent_lines: list[str], available_proxy_count: int) -> str: runtime_warning = str(runtime_state.get("recent_warning", "") or "").strip() if runtime_warning: if runtime_warning in {"未刷新", "代理未启用", "未配置代理池链接"}: return "" if available_proxy_count > 0 and ("无可用代理" in runtime_warning or "未取到可用代理数据" in runtime_warning): return "" return runtime_warning recent_proxy_warning = next( (line for line in reversed(recent_lines) if "代理池刷新失败" in line or "无可用代理" in line or "Redis订阅失败" in line), "", ) if available_proxy_count > 0 and "无可用代理" in recent_proxy_warning: return "" return recent_proxy_warning def _build_proxy_runtime_snapshot(settings_payload: dict, runtime_state: dict, available_proxy_count: int) -> dict: proxy_config = settings_payload.get("proxy_config") or {} proxy_enable = bool(proxy_config.get("proxy_enable", False)) allow_direct = bool(proxy_config.get("allow_direct", False)) refresh_status = str(runtime_state.get("proxy_last_refresh_status", "") or "").strip() refresh_time = str(runtime_state.get("proxy_last_refresh_time", "") or "").strip() source_count = len(proxy_config.get("proxy_urls", [])) source_stats = runtime_state.get("proxy_last_source_stats") or [] raw_items = int(runtime_state.get("proxy_last_refresh_total_items", 0) or 0) validated = int(runtime_state.get("proxy_last_validated_count", 0) or 0) available = int(runtime_state.get("proxy_last_available_count", available_proxy_count) or available_proxy_count) source_ok_count = sum(1 for item in source_stats if str(item.get("status", "") or "").strip() == "ok") supplier_empty = bool(source_stats) and source_ok_count == len(source_stats) and raw_items <= 0 if not proxy_enable: return { "state": "disabled", "label": "未启用代理", "detail": "当前使用直连模式，未启用代理池", "direct_fallback_active": True, "reason": "proxy_disabled", "last_refresh_status": refresh_status or "代理未启用", "last_refresh_time": refresh_time, "source_count": source_count, "raw_items": raw_items, "validated_count": validated, "available_count": available, "source_stats": source_stats, "supplier_empty": False, } if available_proxy_count > 0: detail = f"代理池当前可用 {available_proxy_count} 个代理，配置来源 {source_count} 个" if refresh_status: detail = f"{detail}；最近状态：{refresh_status}" return { "state": "healthy", "label": "代理正常", "detail": detail, "direct_fallback_active": False, "reason": "healthy", "last_refresh_status": refresh_status, "last_refresh_time": refresh_time, "source_count": source_count, "raw_items": raw_items, "validated_count": validated, "available_count": available, "source_stats": source_stats, "supplier_empty": False, } if refresh_status in {"", "未刷新"} and source_count > 0: if allow_direct: return { "state": "degraded_direct", "label": "等待首刷", "detail": f"代理配置已下发，但代理池尚未完成首轮刷新；当前先按直连继续执行；最近状态：{refresh_status or '未刷新'}", "direct_fallback_active": True, "reason": "proxy_not_refreshed_yet", "last_refresh_status": refresh_status or "未刷新", "last_refresh_time": refresh_time, "source_count": source_count, "raw_items": raw_items, "validated_count": validated, "available_count": available, "source_stats": source_stats, "supplier_empty": False, } return { "state": "warming_up", "label": "等待首刷", "detail": "代理配置已下发，但代理池尚未完成首轮刷新；由于未允许直连，检测链路会等待代理刷新完成", "direct_fallback_active": False, "reason": "proxy_not_refreshed_yet", "last_refresh_status": refresh_status or "未刷新", "last_refresh_time": refresh_time, "source_count": source_count, "raw_items": raw_items, "validated_count": validated, "available_count": available, "source_stats": source_stats, "supplier_empty": False, } if allow_direct: detail = "代理池当前无可用代理，已自动降级为直连继续执行" reason = "no_available_proxy" if supplier_empty: reason = "supplier_empty_pool" detail = "代理源最近都返回正常响应，但原始代理数为 0，当前判断为供应池为空；系统已自动降级为直连继续执行" elif raw_items > 0 and validated > 0: reason = "proxy_validation_zero" detail = ( f"代理源最近返回了 {raw_items} 个代理，已验证 {validated} 个，但当前 0 个可用；" "系统已自动降级为直连继续执行" ) if refresh_status: detail = f"{detail}；最近状态：{refresh_status}" return { "state": "degraded_direct", "label": "降级直连", "detail": detail, "direct_fallback_active": True, "reason": reason, "last_refresh_status": refresh_status or "当前无可用代理", "last_refresh_time": refresh_time, "source_count": source_count, "raw_items": raw_items, "validated_count": validated, "available_count": available, "source_stats": source_stats, "supplier_empty": supplier_empty, } detail = "代理池当前无可用代理，且未允许直连，检测链路会等待代理恢复" reason = "no_available_proxy" if supplier_empty: reason = "supplier_empty_pool" detail = "代理源最近都返回正常响应，但原始代理数为 0，当前判断为供应池为空；由于未允许直连，检测链路会等待代理恢复" elif raw_items > 0 and validated > 0: reason = "proxy_validation_zero" detail = ( f"代理源最近返回了 {raw_items} 个代理，已验证 {validated} 个，但当前 0 个可用；" "由于未允许直连，检测链路会等待代理恢复" ) if refresh_status: detail = f"{detail}；最近状态：{refresh_status}" return { "state": "blocked_no_proxy", "label": "等待代理", "detail": detail, "direct_fallback_active": False, "reason": reason, "last_refresh_status": refresh_status or "当前无可用代理", "last_refresh_time": refresh_time, "source_count": source_count, "raw_items": raw_items, "validated_count": validated, "available_count": available, "source_stats": source_stats, "supplier_empty": supplier_empty, } def get_detect_status() -> dict: global _DETECT_STATUS_CACHE_EXPIRES_AT, _DETECT_STATUS_CACHE_VALUE now_ts = time.monotonic() with _DETECT_STATUS_CACHE_LOCK: if _DETECT_STATUS_CACHE_VALUE is not None and now_ts < _DETECT_STATUS_CACHE_EXPIRES_AT: return _clone_detect_status_payload(_DETECT_STATUS_CACHE_VALUE) try: ensure_runtime_schema() except Exception: # Node agent heartbeats should degrade gracefully even if runtime schema # initialization is temporarily unavailable. pass queries = { "pending": "select count(*) from domains where detect_status = 0", "completed": "select count(*) from domains where detect_status = 1", "running": "select count(*) from domains where detect_status = 2", "blacklisted": "select count(*) from domains where detect_status = 3", "failed": "select count(*) from domains where detect_status = 4", "registerable": "select count(*) from domains where detect_status = 1 and register_status = 2", } progress: dict[str, int] = {key: 0 for key in queries} try: with get_db() as conn: with conn.cursor() as cur: for key, query in queries.items(): try: cur.execute(query) progress[key] = cur.fetchone()[0] except Exception: progress[key] = 0 except Exception: # Worker runtime/status pages should still render using runtime-state and # cluster fallbacks even when the local DB endpoint is temporarily wrong # or unreachable (for example remote worker nodes without direct DB access). pass settings_payload = get_settings_payload() worker_expected_on_this_node = _local_worker_expected_on_this_node() runtime_settings = get_runtime_settings() worker_online, last_log_time, recent_lines = _load_recent_worker_lines(runtime_settings, max_lines=160) runtime = detect_worker_runtime() runtime_state = _load_runtime_state() if not runtime_state: runtime_state = _load_runtime_state_from_cluster_node() if not worker_expected_on_this_node: worker_online = False last_log_time = "" recent_lines = [] runtime_state = {} runtime_started_at = runtime.get("latest_start_time", "") recent_lines = _filter_lines_since(recent_lines, runtime_started_at) available_proxy_count = _extract_available_proxy_count(recent_lines) active_thread_snapshot = _extract_active_thread_snapshot(recent_lines) if runtime_state: available_proxy_count = int(runtime_state.get("available_proxy_count", available_proxy_count) or available_proxy_count) active_thread_snapshot = { "active": int(runtime_state.get("active_threads", active_thread_snapshot["active"]) or active_thread_snapshot["active"]), "max": int(runtime_state.get("max_threads", active_thread_snapshot["max"]) or active_thread_snapshot["max"]), } registerable_count = int(progress.get("registerable", 0) or 0) progress_total = sum( int(progress.get(key, 0) or 0) for key in ("pending", "completed", "running", "blacklisted", "failed") ) progress_done = progress.get("completed", 0) + progress.get("blacklisted", 0) + progress.get("failed", 0) progress_percent = round((progress_done / progress_total) * 100, 2) if progress_total > 0 else 0 runtime_started_at = runtime.get("latest_start_time", "") recent_lines = _filter_lines_since(recent_lines, runtime_started_at) recent_proxy_warning = _normalize_recent_warning(runtime_state, recent_lines, available_proxy_count) proxy_runtime = _build_proxy_runtime_snapshot(settings_payload, runtime_state, available_proxy_count) thread_count_resolution = resolve_thread_count(settings_payload=settings_payload) effective_thread_count = int(thread_count_resolution["effective_thread_count"]) worker_online = worker_online or runtime.get("running", False) if runtime_state.get("service_running") is True: worker_online = True if active_thread_snapshot["active"] <= 0 and (runtime_state.get("detecting", False) or runtime.get("running", False)): estimated_active_threads = _estimate_active_threads_from_recent_lines( recent_lines, limit=max(1, effective_thread_count), ) if estimated_active_threads > 0: active_thread_snapshot["active"] = estimated_active_threads inferred_detecting = bool( runtime_state.get("detecting", False) or int(progress.get("running", 0) or 0) > 0 or int(active_thread_snapshot.get("active", 0) or 0) > 0 ) if not inferred_detecting: active_thread_snapshot = {"active": 0, "max": active_thread_snapshot["max"] or effective_thread_count} settings_summary = { "thread_count": effective_thread_count, "thread_count_default": int(thread_count_resolution["default_thread_count"]), "thread_count_source": str(thread_count_resolution["source"]), "thread_count_override": thread_count_resolution["override_thread_count"], "thread_count_node_code": str(thread_count_resolution["node_code"]), "proxy_enable": settings_payload["proxy_config"].get("proxy_enable", False), "allow_direct": settings_payload["proxy_config"].get("allow_direct", False), "proxy_pool_count": len(settings_payload["proxy_config"].get("proxy_urls", [])), } try: active_job = get_active_detect_job_summary(event_limit=240) except Exception: active_job = None aggregate_detect_view = bool(settings.node_region == "overseas" and settings.node_role == "control" and active_job) aggregate_queue_health = {} if aggregate_detect_view: try: aggregate_queue_health = get_detect_queue_health(window_minutes=15) except Exception: aggregate_queue_health = {} active_job = _merge_aggregate_active_job_with_queue_health(active_job, aggregate_queue_health) aggregate_capacity = ( _build_aggregate_detect_capacity(active_job=active_job, settings_payload=settings_payload) if aggregate_detect_view else { "participant_node_codes": [], "participant_node_count": 0, "process_count": 0, "max_threads": 0, "per_process_thread_count": 0, } ) if aggregate_detect_view: raw_aggregate_node_rows = list(active_job.get("distributed_node_stats") or active_job.get("node_stats") or []) aggregate_node_rows = _filter_live_aggregate_runtime_nodes(raw_aggregate_node_rows) aggregate_display_running = sum( _max_runtime_metric( item.get("display_running"), item.get("current_load"), item.get("active_threads"), item.get("items_running"), ) for item in aggregate_node_rows if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned" ) if raw_aggregate_node_rows: display_running = aggregate_display_running else: display_running = _max_runtime_metric( active_job.get("display_active_threads"), active_job.get("display_items_running"), (aggregate_queue_health.get("queue") or {}).get("display_running"), ) progress = { "pending": int(active_job.get("items_pending", 0) or 0), "running": display_running, "completed": int(active_job.get("items_completed", 0) or 0), "failed": int(active_job.get("items_failed", 0) or 0), "blacklisted": int(active_job.get("items_blacklisted", 0) or 0), "registerable": registerable_count, } progress_percent = float(active_job.get("progress_percent", 0) or 0) local_node_bucket = {} for item in list((active_job or {}).get("node_stats") or []): if str(item.get("node_code") or "").strip() == str(settings.node_code or "").strip(): local_node_bucket = item break local_runtime_load = int( local_node_bucket.get("active_threads") or local_node_bucket.get("items_running") or 0 ) local_runtime_max_threads = int(local_node_bucket.get("max_threads", 0) or 0) if active_thread_snapshot["active"] <= 0 and local_runtime_load > 0: active_thread_snapshot["active"] = local_runtime_load if active_thread_snapshot["max"] <= 0: active_thread_snapshot["max"] = local_runtime_max_threads or effective_thread_count if aggregate_detect_view: raw_distributed_node_stats = list(active_job.get("distributed_node_stats") or active_job.get("node_stats") or []) distributed_node_stats = _filter_live_aggregate_runtime_nodes(raw_distributed_node_stats) raw_participant_count = sum( 1 for item in raw_distributed_node_stats if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned" ) live_participant_count = sum( 1 for item in distributed_node_stats if str(item.get("node_code") or "").strip() and str(item.get("node_code") or "").strip() != "unassigned" ) dropped_aggregate_node_count = max(0, raw_participant_count - live_participant_count) aggregated_active_threads = 0 aggregated_max_threads = 0 for item in distributed_node_stats: node_code = str(item.get("node_code") or "").strip() if not node_code or node_code == "unassigned": continue aggregated_active_threads += int( item.get("active_threads") or item.get("items_running") or 0 ) aggregated_max_threads += int(item.get("max_threads", 0) or 0) if aggregated_active_threads > 0: active_thread_snapshot["active"] = aggregated_active_threads elif dropped_aggregate_node_count <= 0 and _max_runtime_metric( active_job.get("display_active_threads"), active_job.get("display_items_running"), progress.get("running"), ) > 0: active_thread_snapshot["active"] = _max_runtime_metric( active_job.get("display_active_threads"), active_job.get("display_items_running"), progress.get("running"), ) if aggregated_max_threads > 0: active_thread_snapshot["max"] = aggregated_max_threads elif dropped_aggregate_node_count <= 0 and _max_runtime_metric( active_job.get("display_max_threads"), aggregate_capacity.get("max_threads"), ) > 0: active_thread_snapshot["max"] = _max_runtime_metric( active_job.get("display_max_threads"), aggregate_capacity.get("max_threads"), ) configured_max_threads = int(aggregate_capacity.get("max_threads", 0) or 0) if configured_max_threads > 0: active_thread_snapshot["max"] = max(active_thread_snapshot["max"], configured_max_threads) available_proxy_count, proxy_runtime = _build_aggregate_proxy_runtime( active_job=active_job, settings_payload=settings_payload, fallback_available_proxy_count=available_proxy_count, fallback_proxy_runtime=proxy_runtime, ) display_worker_process_count = int(runtime.get("process_count", 0) or 0) if aggregate_detect_view and int(aggregate_capacity.get("process_count", 0) or 0) > 0: display_worker_process_count = int(aggregate_capacity.get("process_count", 0) or 0) runtime_snapshot = { **runtime, "detecting": inferred_detecting, "proxy_runtime_state": proxy_runtime["state"], "proxy_runtime_label": proxy_runtime["label"], "proxy_runtime_detail": proxy_runtime["detail"], "proxy_direct_fallback_active": proxy_runtime["direct_fallback_active"], "proxy_runtime_reason": proxy_runtime["reason"], "proxy_supplier_empty": proxy_runtime["supplier_empty"], } runs = sync_detect_runs(runtime_snapshot, progress, settings_summary, active_job=active_job) worker_log_sync_enabled = bool(runtime_settings.get("worker_log_sync_enabled", False)) worker_log_sync_mode = str(runtime_settings.get("worker_log_sync_mode", "key") or "key") remote_log_snapshot = _resolve_remote_log_snapshot( active_job, runs, enabled=worker_log_sync_enabled, mode=worker_log_sync_mode, limit=240, ) remote_log_lines = list(remote_log_snapshot.get("lines") or []) dependency_alerts = _extract_dependency_alerts(recent_lines) if _local_worker_expected_on_this_node(): append_detect_result_projection_if_changed( detect={ "active_job": active_job, "progress": progress, "phase_label": runtime_state.get("phase", ""), "phase_detail": runtime_state.get("detail", ""), } ) result = { "worker_online": worker_online, "worker_mode": runtime.get("mode", "windows-local"), "worker_service_name": runtime_settings.get("worker_service_name", ""), "api_service_name": runtime_settings.get("api_service_name", ""), "worker_process_count": display_worker_process_count, "worker_latest_start_time": runtime.get("latest_start_time", ""), "worker_runtime_message": runtime.get("message", ""), "runtime_state": runtime_state, "phase_label": runtime_state.get("phase", ""), "phase_detail": runtime_state.get("detail", ""), "detecting": inferred_detecting, "thread_count": effective_thread_count, "thread_count_default": int(thread_count_resolution["default_thread_count"]), "thread_count_source": str(thread_count_resolution["source"]), "thread_count_override": thread_count_resolution["override_thread_count"], "thread_count_node_code": str(thread_count_resolution["node_code"]), "active_thread_count": active_thread_snapshot["active"], "max_thread_count": active_thread_snapshot["max"] or effective_thread_count, "aggregate_process_count": int(aggregate_capacity.get("process_count", 0) or 0), "aggregate_participating_node_count": int(aggregate_capacity.get("participant_node_count", 0) or 0), "aggregate_participating_node_codes": list(aggregate_capacity.get("participant_node_codes") or []), "aggregate_max_thread_count": int(aggregate_capacity.get("max_threads", 0) or 0), "aggregate_thread_count_per_process": int(aggregate_capacity.get("per_process_thread_count", 0) or 0), "proxy_enable": settings_payload["proxy_config"].get("proxy_enable", False), "allow_direct": settings_payload["proxy_config"].get("allow_direct", False), "proxy_pool_count": len(settings_payload["proxy_config"].get("proxy_urls", [])), "available_proxy_count": available_proxy_count, "proxy_runtime_state": proxy_runtime["state"], "proxy_runtime_label": proxy_runtime["label"], "proxy_runtime_detail": proxy_runtime["detail"], "proxy_direct_fallback_active": proxy_runtime["direct_fallback_active"], "proxy_runtime_reason": proxy_runtime["reason"], "proxy_supplier_empty": proxy_runtime["supplier_empty"], "proxy_last_refresh_status": proxy_runtime["last_refresh_status"], "proxy_last_refresh_time": proxy_runtime["last_refresh_time"], "proxy_last_refresh_source_count": proxy_runtime["source_count"], "proxy_last_refresh_total_items": proxy_runtime["raw_items"], "proxy_last_validated_count": proxy_runtime["validated_count"], "proxy_last_available_count": proxy_runtime["available_count"], "proxy_source_stats": proxy_runtime["source_stats"], "dependency_alerts": dependency_alerts, "last_worker_log_time": last_log_time, "progress": progress, "progress_percent": progress_percent, "recent_event": runtime_state.get("detail") or _recent_event(recent_lines), "recent_warning": recent_proxy_warning, "aggregate_detect_view": aggregate_detect_view, "log_lines": recent_lines, "remote_log_lines": remote_log_lines, "remote_log_line_count": int(remote_log_snapshot.get("line_count", 0) or 0), "remote_log_last_at": str(remote_log_snapshot.get("last_at") or ""), "remote_log_last_line": str(remote_log_snapshot.get("last_line") or ""), "remote_log_nodes": list(remote_log_snapshot.get("source_nodes") or []), "remote_log_node_count": int(remote_log_snapshot.get("source_node_count", 0) or 0), "remote_log_node_summaries": list(remote_log_snapshot.get("source_node_summaries") or []), "runs": runs, "active_job": active_job, "worker_log_sync_enabled": worker_log_sync_enabled, "worker_log_sync_mode": worker_log_sync_mode, } with _DETECT_STATUS_CACHE_LOCK: _DETECT_STATUS_CACHE_VALUE = _clone_detect_status_payload(result) _DETECT_STATUS_CACHE_EXPIRES_AT = time.monotonic() + _DETECT_STATUS_CACHE_TTL_SECONDS return result