from __future__ import annotations import logging import time from app.core.config import settings from app.services.debug_event_service import push_debug_event from app.services.detect_job_service import get_active_detect_job_summary, get_detect_queue_health, list_recent_detect_run_events from app.services.sync_record_service import append_detect_result_projection_if_changed from app.services.sync_push_service import pull_detect_task_batch_now, push_runtime_projection_now logger = logging.getLogger("domaincheck.sync_agent") def _append_detect_result_projection_snapshot(active_job: dict) -> None: if not active_job: return append_detect_result_projection_if_changed( detect={ "active_job": active_job, "progress": { "pending": int(active_job.get("items_pending", 0) or 0), "running": int(active_job.get("items_running", 0) or 0), "completed": int(active_job.get("items_completed", 0) or 0), "blacklisted": int(active_job.get("items_blacklisted", 0) or 0), "failed": int(active_job.get("items_failed", 0) or 0), }, "phase_label": str(active_job.get("status") or "").strip(), "phase_detail": f"sync-agent snapshot for {active_job.get('job_code', '')}", } ) def _emit_structured_tick( *, base_event_type: str, ok: bool, message: str, data: dict | None = None, ) -> None: payload = {"ok": ok, "data": data or {}} event_type = f"{base_event_type}_failed" level = "warning" if ok: event_type = f"{base_event_type}_success" level = "info" if "但远端确认失败" in str(message or ""): event_type = f"{base_event_type}_partial" level = "warning" results = list((data or {}).get("results") or []) if results and any(not bool(item.get("ok")) for item in results): event_type = f"{base_event_type}_partial" level = "warning" push_debug_event( service="sync-agent", event_type=event_type, level=level, message=message, payload=payload, ) def _emit_sync_result_breakdown(data: dict | None) -> None: results = list((data or {}).get("results") or []) for item in results: sync_type = str(item.get("sync_type") or "").strip() or "unknown" ok = bool(item.get("ok")) message = str(item.get("message") or "").strip() or f"{sync_type} sync result" result_data = item.get("data") or {} event_type = f"{sync_type}_sync_failed" level = "warning" if ok: event_type = f"{sync_type}_sync_success" level = "info" if isinstance(result_data, dict) and result_data.get("success_count") is not None: batch_count = int(result_data.get("batch_count") or 0) success_count = int(result_data.get("success_count") or 0) if batch_count > 0 and success_count < batch_count: event_type = f"{sync_type}_sync_partial" level = "warning" if "成功 1/" in message or "但" in message: event_type = f"{sync_type}_sync_partial" level = "warning" push_debug_event( service="sync-agent", event_type=event_type, level=level, message=message, payload={ "sync_type": sync_type, "ok": ok, "data": result_data, }, ) def main() -> None: logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) interval = max(10, int(settings.sync_poll_interval_seconds or 30)) logger.info( "sync agent started: node=%s source=%s target=%s interval=%ss enabled=%s", settings.node_code, settings.sync_source_region, settings.sync_target_region, interval, settings.sync_push_enabled, ) while True: try: active_job = get_active_detect_job_summary(event_limit=10) if active_job: _append_detect_result_projection_snapshot(active_job) ok, message, data = push_runtime_projection_now() logger.info("sync tick: ok=%s message=%s data=%s", ok, message, data) push_debug_event( service="sync-agent", event_type="sync_tick", level="info" if ok else "warning", message=message, payload={"ok": ok, "data": data}, ) _emit_structured_tick(base_event_type="sync_push", ok=ok, message=message, data=data) _emit_sync_result_breakdown(data) pull_ok, pull_message, pull_data = pull_detect_task_batch_now() logger.info("task pull tick: ok=%s message=%s data=%s", pull_ok, pull_message, pull_data) push_debug_event( service="sync-agent", event_type="task_pull_tick", level="info" if pull_ok else "warning", message=pull_message, payload={"ok": pull_ok, "data": pull_data}, ) _emit_structured_tick(base_event_type="task_pull", ok=pull_ok, message=pull_message, data=pull_data) if active_job: queue_health = get_detect_queue_health(window_minutes=15) recent_events = list_recent_detect_run_events(limit=8) push_debug_event( service="detect-runtime", event_type="active_job_snapshot", level="info", message=f"active job {active_job.get('job_code', '')} status={active_job.get('status', '')}", payload={ "job": { "job_id": active_job.get("job_id"), "job_code": active_job.get("job_code", ""), "status": active_job.get("status", ""), "items_total": active_job.get("items_total", 0), "items_pending": active_job.get("items_pending", 0), "items_claimed": active_job.get("items_claimed", 0), "items_running": active_job.get("items_running", 0), "items_completed": active_job.get("items_completed", 0), "items_failed": active_job.get("items_failed", 0), "progress_percent": active_job.get("progress_percent", 0), "node_stats": list(active_job.get("node_stats") or []), }, "queue_health": queue_health, "recent_events": recent_events, }, ) if queue_health.get("queue", {}).get("overdue_leases", 0): push_debug_event( service="detect-runtime", event_type="queue_overdue_leases", level="warning", message=f"检测队列存在过期租约 {queue_health.get('queue', {}).get('overdue_leases', 0)} 个", payload=queue_health, ) for event in recent_events: event_type = str(event.get("event_type") or "").strip() if event_type not in {"domain_started", "domain_completed", "domain_failed", "domain_blacklisted"}: continue push_debug_event( service="worker-event", event_type=event_type, level=str(event.get("level") or "info"), message=str(event.get("message") or "").strip(), payload={ "job_id": event.get("job_id"), "node_code": event.get("node_code", ""), "created_at": event.get("created_at", ""), **(event.get("payload") or {}), }, ) except Exception as exc: logger.exception("sync tick failed: %s", exc) push_debug_event( service="sync-agent", event_type="sync_tick_failed", level="error", message=str(exc), payload={}, ) time.sleep(interval) if __name__ == "__main__": main()