1929 lines
81 KiB
Python
1929 lines
81 KiB
Python
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
|