473 lines
20 KiB
Python
473 lines
20 KiB
Python
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
from datetime import datetime, timezone
|
||
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.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
|
||
from app.services.settings_service import get_settings_payload, 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+)")
|
||
_THREAD_COUNT_RE = re.compile(r"当前实际线程数量[:：]\s*(\d+)\s*/\s*(\d+)")
|
||
_RUNTIME_STATE_KEY = "domain_tool:detect_runtime_state"
|
||
_TIMESTAMP_FORMATS = ("%Y-%m-%d %H:%M:%S.%f", "%Y-%m-%d %H:%M:%S")
|
||
_REMOTE_LOG_MAX_CHARS = 500
|
||
|
||
|
||
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):
|
||
match = _PROXY_COUNT_RE.search(line)
|
||
if match:
|
||
return int(match.group(1))
|
||
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 _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) -> datetime | None:
|
||
if len(line) < 19:
|
||
return None
|
||
candidates = [line[:26], line[: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
|
||
return None
|
||
|
||
|
||
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 = [line for line in lines if (_extract_log_time(line) or started_time) >= started_time]
|
||
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]:
|
||
if not enabled:
|
||
return []
|
||
if not active_job:
|
||
return []
|
||
events = list(active_job.get("current_cycle_events") or active_job.get("recent_events") or [])
|
||
if not events:
|
||
return []
|
||
|
||
lines: list[str] = []
|
||
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]}..."
|
||
lines.append(f"[{created_at}] [{node_code}] {message}")
|
||
return lines[-max(1, int(limit or 240)) :]
|
||
|
||
|
||
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 _load_runtime_state() -> dict:
|
||
try:
|
||
redis_client = get_redis()
|
||
raw = redis_client.get(_RUNTIME_STATE_KEY)
|
||
if not raw:
|
||
return {}
|
||
data = json.loads(raw)
|
||
return data if isinstance(data, dict) else {}
|
||
except Exception:
|
||
return {}
|
||
|
||
|
||
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 allow_direct:
|
||
detail = "代理池当前无可用代理，已自动降级为直连继续执行"
|
||
reason = "no_available_proxy"
|
||
if supplier_empty:
|
||
reason = "supplier_empty_pool"
|
||
detail = "代理源最近都返回正常响应，但原始代理数为 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，当前判断为供应池为空；由于未允许直连，检测链路会等待代理恢复"
|
||
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:
|
||
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",
|
||
}
|
||
progress: dict[str, int] = {}
|
||
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
|
||
|
||
settings_payload = get_settings_payload()
|
||
worker_log = resolve_domain_path("detect_worker.log", "logs/detect_worker.log")
|
||
|
||
worker_online = False
|
||
last_log_time = None
|
||
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
|
||
|
||
recent_lines = tail_lines("detect_worker.log", max_lines=160)
|
||
runtime_state = _load_runtime_state()
|
||
runtime_started_at = runtime.get("latest_start_time", "") if 'runtime' in locals() else ""
|
||
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"]),
|
||
}
|
||
progress_total = sum(progress.values())
|
||
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 = detect_worker_runtime()
|
||
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"])
|
||
runtime_settings = get_runtime_settings()
|
||
worker_online = worker_online or runtime.get("running", False)
|
||
if runtime_state.get("service_running") is True:
|
||
worker_online = True
|
||
if not runtime_state.get("detecting", False) and not progress.get("running", 0):
|
||
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", [])),
|
||
}
|
||
active_job = get_active_detect_job_summary(event_limit=240)
|
||
runtime_snapshot = {
|
||
**runtime,
|
||
"detecting": runtime_state.get("detecting", False),
|
||
"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_lines = _resolve_remote_log_lines(
|
||
active_job,
|
||
runs,
|
||
enabled=worker_log_sync_enabled,
|
||
mode=worker_log_sync_mode,
|
||
limit=240,
|
||
)
|
||
dependency_alerts = _extract_dependency_alerts(recent_lines)
|
||
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", ""),
|
||
}
|
||
)
|
||
|
||
return {
|
||
"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": runtime.get("process_count", 0),
|
||
"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": runtime_state.get("detecting", False),
|
||
"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,
|
||
"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,
|
||
"log_lines": recent_lines,
|
||
"remote_log_lines": remote_log_lines,
|
||
"runs": runs,
|
||
"active_job": active_job,
|
||
"worker_log_sync_enabled": worker_log_sync_enabled,
|
||
"worker_log_sync_mode": worker_log_sync_mode,
|
||
}
|