From d3223a75a499a21918f329cf3b6b2b6ba1d36bf3 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 17 Apr 2026 03:36:03 +0800 Subject: [PATCH] debug --- domain-api/app/api/routes/runtime.py | 18 +- domain-api/app/services/detect_job_service.py | 21 +- .../app/services/runtime_control_service.py | 10 +- domain-api/app/services/sync_push_service.py | 476 ++++++++++++++++++ domain-api/app/sync_agent.py | 4 +- 5 files changed, 525 insertions(+), 4 deletions(-) diff --git a/domain-api/app/api/routes/runtime.py b/domain-api/app/api/routes/runtime.py index ecb7c73..722780d 100644 --- a/domain-api/app/api/routes/runtime.py +++ b/domain-api/app/api/routes/runtime.py @@ -6,7 +6,11 @@ from app.schemas.common import ApiResponse from app.services.cluster_runtime_service import get_cluster_snapshot from app.services.runtime_control_service import runtime_action from app.services.runtime_status_service import get_runtime_preflight, get_runtime_status -from app.services.sync_push_service import ingest_runtime_projection +from app.services.sync_push_service import ( + ack_detect_task_projection, + export_detect_task_projection, + ingest_runtime_projection, +) from app.services.sync_record_service import get_sync_summary, list_sync_records router = APIRouter(tags=["runtime"]) @@ -48,6 +52,18 @@ def runtime_sync_ingest(payload: dict, x_domaincheck_sync_token: Optional[str] = return ApiResponse(code=0 if ok else 1, message=message, data=data) +@router.get("/runtime/task-export", response_model=ApiResponse) +def runtime_task_export(limit: int = 200, x_domaincheck_sync_token: Optional[str] = Header(default=None)) -> ApiResponse: + ok, message, data = export_detect_task_projection(limit=limit, shared_token=x_domaincheck_sync_token) + return ApiResponse(code=0 if ok else 1, message=message, data=data) + + +@router.post("/runtime/task-ack", response_model=ApiResponse) +def runtime_task_ack(payload: dict, x_domaincheck_sync_token: Optional[str] = Header(default=None)) -> ApiResponse: + ok, message, data = ack_detect_task_projection(payload, shared_token=x_domaincheck_sync_token) + return ApiResponse(code=0 if ok else 1, message=message, data=data) + + @router.post("/runtime/actions/{action}", response_model=ApiResponse) def runtime_action_trigger(action: str) -> ApiResponse: ok, message, data = runtime_action(action) diff --git a/domain-api/app/services/detect_job_service.py b/domain-api/app/services/detect_job_service.py index d0c3eff..4f1026f 100644 --- a/domain-api/app/services/detect_job_service.py +++ b/domain-api/app/services/detect_job_service.py @@ -512,8 +512,27 @@ def create_detect_job_if_needed(limit: int = 1000, created_by: str = "system") - domain_ids = [row[0] for row in cur.fetchall()] if not domain_ids: conn.rollback() - return None + if not domain_ids: + try: + from app.services.sync_push_service import pull_detect_task_batch_now + + pull_detect_task_batch_now(limit=max(1, int(limit or 1000))) + except Exception: + pass + + with get_db() as conn: + conn.autocommit = False + with conn.cursor() as cur: + cur.execute(_selection_sql(), (max(1, int(limit or 1000)),)) + domain_ids = [row[0] for row in cur.fetchall()] + if not domain_ids: + conn.rollback() + return None + + with get_db() as conn: + conn.autocommit = False + with conn.cursor() as cur: job_code = f"detect-{datetime.now().strftime('%Y%m%d%H%M%S')}-{uuid4().hex[:6]}" plan_hash = uuid4().hex cur.execute( diff --git a/domain-api/app/services/runtime_control_service.py b/domain-api/app/services/runtime_control_service.py index 508a05c..5c14968 100644 --- a/domain-api/app/services/runtime_control_service.py +++ b/domain-api/app/services/runtime_control_service.py @@ -5,7 +5,7 @@ import subprocess from pathlib import Path from app.core.config import settings -from app.services.sync_push_service import push_runtime_projection_now +from app.services.sync_push_service import pull_detect_task_batch_now, push_runtime_projection_now from app.services.runtime_settings_service import get_runtime_settings from app.services.worker_control_service import _run_systemctl, start_worker, stop_worker @@ -104,6 +104,14 @@ def runtime_action(action: str) -> tuple[bool, str, dict]: "refresh_runtime": True, **(data or {}), } + if action == "pull_tasks": + ok, message, data = pull_detect_task_batch_now() + return ok, message, { + "action": action, + "poll_after_seconds": 2, + "refresh_runtime": True, + **(data or {}), + } return False, f"不支持的运行时动作: {action}", { "action": action, "poll_after_seconds": 0, diff --git a/domain-api/app/services/sync_push_service.py b/domain-api/app/services/sync_push_service.py index 64ce824..55ac0bf 100644 --- a/domain-api/app/services/sync_push_service.py +++ b/domain-api/app/services/sync_push_service.py @@ -6,6 +6,7 @@ import urllib.error import urllib.parse import urllib.request from datetime import datetime, timedelta +from uuid import uuid4 from app.core.config import settings from app.core.db import get_db @@ -28,11 +29,35 @@ def _ingest_url(base_url: str) -> str: return f"{text}/api/v1/runtime/sync-ingest" +def _task_export_url(base_url: str) -> str: + text = str(base_url or "").strip().rstrip("/") + if not text: + return "" + if text.endswith("/api/v1"): + return f"{text}/runtime/task-export" + if text.endswith("/api/v1/runtime"): + return f"{text}/task-export" + return f"{text}/api/v1/runtime/task-export" + + +def _task_ack_url(base_url: str) -> str: + text = str(base_url or "").strip().rstrip("/") + if not text: + return "" + if text.endswith("/api/v1"): + return f"{text}/runtime/task-ack" + if text.endswith("/api/v1/runtime"): + return f"{text}/task-ack" + return f"{text}/api/v1/runtime/task-ack" + + def _projection_ingest_type(sync_type: str) -> str: if sync_type == "runtime_projection": return "runtime_ingest" if sync_type == "detect_result_projection": return "detect_result_ingest" + if sync_type == "detect_task_projection": + return "detect_task_ingest" return "sync_ingest" @@ -182,6 +207,366 @@ def _latest_push_attempt(source_record_id: int, target_region: str, sync_type: s } +def _latest_ingest_attempt(source_record_id: int, target_region: str, sync_type: str) -> dict | None: + ingest_sync_type = _projection_ingest_type(sync_type) + with get_db() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT id, status, payload_json, error_message, created_at, updated_at + FROM detect_sync_records + WHERE sync_type = %s + AND source_region = %s + AND target_region = %s + AND (payload_json->>'source_record_id') = %s + ORDER BY created_at DESC, id DESC + LIMIT 1 + """, + ( + ingest_sync_type, + _normalize_region(settings.sync_source_region, settings.node_region), + target_region, + str(int(source_record_id)), + ), + ) + row = cur.fetchone() + if not row: + return None + return { + "id": row[0], + "status": row[1], + "payload": _decode_json(row[2]), + "error_message": row[3] or "", + "created_at": row[4], + "updated_at": row[5], + } + + +def _task_selection_sql() -> str: + return """ + SELECT id, domain, tld, source_type, use_status, detect_status, register_status, expire_date, jucha_status, juziseo_status + FROM domains + WHERE + detect_status IN (0, 4) + OR (use_status = 0 AND detect_status = 1 AND register_status = 3 AND expire_date < CURRENT_DATE) + ORDER BY id ASC + LIMIT %s + """ + + +def _load_pending_task_projection(limit: int) -> dict | None: + source_region = _normalize_region(settings.sync_source_region, settings.node_region) + target_region = _normalize_region(settings.sync_target_region, "overseas") + safe_limit = max(1, min(int(limit or 1000), max(1, int(settings.sync_batch_size or 200)))) + with get_db() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT id, source_region, target_region, status, payload_json, created_at, updated_at + FROM detect_sync_records + WHERE sync_type = 'detect_task_projection' + AND source_region = %s + AND target_region = %s + ORDER BY created_at ASC, id ASC + LIMIT %s + """, + (source_region, target_region, safe_limit * 3), + ) + rows = cur.fetchall() + for row in rows: + projection = { + "id": row[0], + "source_region": row[1], + "target_region": row[2], + "status": row[3], + "payload": _decode_json(row[4]), + "created_at": row[5], + "updated_at": row[6], + } + latest_ingest = _latest_ingest_attempt(projection["id"], projection["target_region"], "detect_task_projection") + if latest_ingest and latest_ingest["status"] == "received": + continue + return projection + return None + + +def export_detect_task_projection(limit: int = 1000, *, shared_token: str | None = None) -> tuple[bool, str, dict]: + configured_token = str(settings.sync_shared_token or "").strip() + incoming_token = str(shared_token or "").strip() + if configured_token and incoming_token != configured_token: + return False, "同步 token 校验失败", {} + + existing = _load_pending_task_projection(limit) + if existing: + return True, "已返回待确认的任务批次", { + "source_record_id": existing["id"], + "projection_hash": (existing.get("payload") or {}).get("projection_hash", ""), + "projection": (existing.get("payload") or {}).get("projection", {}), + "deduplicated": True, + } + + source_region = _normalize_region(settings.sync_source_region, settings.node_region) + target_region = _normalize_region(settings.sync_target_region, "overseas") + safe_limit = max(1, min(int(limit or 1000), max(1, int(settings.sync_batch_size or 200)))) + with get_db() as conn: + with conn.cursor() as cur: + cur.execute(_task_selection_sql(), (safe_limit,)) + rows = cur.fetchall() + if not rows: + return False, "当前没有可下发的待检测任务", {"batch_size": 0} + + items = [] + for row in rows: + items.append( + { + "source_domain_id": int(row[0]), + "domain": str(row[1] or "").strip(), + "tld": str(row[2] or "").strip(), + "source_type": int(row[3] or 0), + "use_status": int(row[4] or 0), + "detect_status": int(row[5] or 0), + "register_status": int(row[6] or 0), + "expire_date": row[7].isoformat(sep=" ", timespec="seconds") if row[7] else "", + "jucha_status": int(row[8] or 0), + "juziseo_status": int(row[9] or 0), + } + ) + + projection = { + "batch_code": f"task-{datetime.now().strftime('%Y%m%d%H%M%S')}-{uuid4().hex[:6]}", + "items_total": len(items), + "selection_limit": safe_limit, + "items": items, + } + payload = { + "projection": projection, + "projection_hash": f"{uuid4().hex}-{len(items)}", + } + cur.execute( + """ + INSERT INTO detect_sync_records ( + sync_type, source_region, target_region, status, payload_json, error_message, created_at, updated_at + ) VALUES (%s, %s, %s, %s, %s::jsonb, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, + ( + "detect_task_projection", + source_region, + target_region, + "projected", + json.dumps(payload, ensure_ascii=False), + "", + ), + ) + record_id = int(cur.fetchone()[0]) + conn.commit() + return True, "待检测任务批次已生成", { + "source_record_id": record_id, + "projection_hash": payload["projection_hash"], + "projection": projection, + "deduplicated": False, + } + + +def ack_detect_task_projection(payload: dict, *, shared_token: str | None = None) -> tuple[bool, str, dict]: + configured_token = str(settings.sync_shared_token or "").strip() + incoming_token = str(shared_token or "").strip() + if configured_token and incoming_token != configured_token: + return False, "同步 token 校验失败", {} + + source_record_id = int(payload.get("source_record_id") or 0) + projection_hash = str(payload.get("projection_hash") or "").strip() + if source_record_id <= 0 or not projection_hash: + return False, "任务确认参数不完整", {} + + source_region = _normalize_region(settings.sync_source_region, settings.node_region) + target_region = _normalize_region(settings.sync_target_region, "overseas") + with get_db() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT id + FROM detect_sync_records + WHERE sync_type = 'detect_task_ingest' + AND source_region = %s + AND target_region = %s + AND (payload_json->>'source_record_id') = %s + AND (payload_json->>'projection_hash') = %s + ORDER BY created_at DESC, id DESC + LIMIT 1 + """, + (source_region, target_region, str(source_record_id), projection_hash), + ) + existing = cur.fetchone() + if existing: + return True, "任务批次确认已存在", {"record_id": int(existing[0]), "deduplicated": True} + + cur.execute( + """ + INSERT INTO detect_sync_records ( + sync_type, source_region, target_region, status, payload_json, error_message, created_at, updated_at + ) VALUES (%s, %s, %s, %s, %s::jsonb, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, + ( + "detect_task_ingest", + source_region, + target_region, + "received", + json.dumps( + { + "source_record_id": source_record_id, + "projection_hash": projection_hash, + "received_at": _format_time(datetime.now()), + }, + ensure_ascii=False, + ), + "", + ), + ) + record_id = int(cur.fetchone()[0]) + conn.commit() + return True, "任务批次确认成功", {"record_id": record_id, "deduplicated": False} + + +def ingest_detect_task_projection(payload: dict, *, shared_token: str | None = None) -> tuple[bool, str, dict]: + configured_token = str(settings.sync_shared_token or "").strip() + incoming_token = str(shared_token or "").strip() + if configured_token and incoming_token != configured_token: + return False, "同步 token 校验失败", {} + + sync_type = str(payload.get("sync_type") or "detect_task_projection").strip() or "detect_task_projection" + if sync_type != "detect_task_projection": + return False, "同步类型不匹配", {"sync_type": sync_type} + + source_region = _normalize_region(payload.get("source_region"), "unknown") + source_record_id = int(payload.get("source_record_id") or 0) + projection_hash = str(payload.get("projection_hash") or "").strip() + projection = payload.get("projection") or {} + items = list(projection.get("items") or []) + target_region = _normalize_region(settings.node_region, "overseas") + received_at = datetime.now() + + if source_record_id <= 0 or not projection_hash: + return False, "任务批次参数不完整", {} + if not items: + return False, "任务批次为空", {"source_record_id": source_record_id} + + with get_db() as conn: + with conn.cursor() as cur: + cur.execute( + """ + SELECT id + FROM detect_sync_records + WHERE sync_type = 'detect_task_ingest' + AND source_region = %s + AND target_region = %s + AND (payload_json->>'source_record_id') = %s + AND (payload_json->>'projection_hash') = %s + ORDER BY created_at DESC, id DESC + LIMIT 1 + """, + (source_region, target_region, str(source_record_id), projection_hash), + ) + existing = cur.fetchone() + if existing: + return True, "任务批次已接收，已按幂等处理", {"record_id": int(existing[0]), "deduplicated": True} + + inserted_count = 0 + updated_count = 0 + for item in items: + domain = str(item.get("domain") or "").strip().lower() + if not domain: + continue + tld = str(item.get("tld") or "").strip() + if not tld and "." in domain: + tld = domain.rsplit(".", 1)[-1] + expire_date_text = str(item.get("expire_date") or "").strip() + cur.execute( + """ + INSERT INTO domains ( + domain, tld, source_type, use_status, detect_status, register_status, + has_beian, company_type, website_url, beian_year, snapshot_years, + expire_date, create_time, update_time, review_status, detect_time, + backlink_count, jucha_status, juziseo_status + ) VALUES ( + %s, %s, %s, %s, %s, %s, + 1, NULL, NULL, NULL, NULL, + NULLIF(%s, '')::timestamp, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, 0, NULL, + 0, %s, %s + ) + ON CONFLICT (domain) DO UPDATE SET + tld = EXCLUDED.tld, + source_type = EXCLUDED.source_type, + use_status = EXCLUDED.use_status, + register_status = EXCLUDED.register_status, + expire_date = COALESCE(EXCLUDED.expire_date, domains.expire_date), + jucha_status = EXCLUDED.jucha_status, + juziseo_status = EXCLUDED.juziseo_status, + detect_status = CASE + WHEN domains.detect_status IN (1, 2, 3, 4) THEN domains.detect_status + ELSE EXCLUDED.detect_status + END, + update_time = CURRENT_TIMESTAMP + RETURNING (xmax = 0) AS inserted + """, + ( + domain, + tld, + int(item.get("source_type") or 0), + int(item.get("use_status") or 0), + int(item.get("detect_status") or 0), + int(item.get("register_status") or 0), + expire_date_text, + int(item.get("jucha_status") or 0), + int(item.get("juziseo_status") or 0), + ), + ) + inserted = bool((cur.fetchone() or [False])[0]) + if inserted: + inserted_count += 1 + else: + updated_count += 1 + + cur.execute( + """ + INSERT INTO detect_sync_records ( + sync_type, source_region, target_region, status, payload_json, error_message, created_at, updated_at + ) VALUES (%s, %s, %s, %s, %s::jsonb, %s, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, + ( + "detect_task_ingest", + source_region, + target_region, + "received", + json.dumps( + { + "source_record_id": source_record_id, + "projection_hash": projection_hash, + "batch_code": str(projection.get("batch_code") or "").strip(), + "items_total": len(items), + "inserted_count": inserted_count, + "updated_count": updated_count, + "received_at": _format_time(received_at), + }, + ensure_ascii=False, + ), + "", + ), + ) + record_id = int(cur.fetchone()[0]) + conn.commit() + return True, "任务批次接收成功", { + "record_id": record_id, + "source_record_id": source_record_id, + "items_total": len(items), + "inserted_count": inserted_count, + "updated_count": updated_count, + "deduplicated": False, + } + + def _create_push_attempt(*, source_record: dict, ingest_url: str, sync_type: str) -> int: payload = { "sync_type": sync_type, @@ -479,3 +864,94 @@ def push_runtime_projection_now() -> tuple[bool, str, dict]: if success_count == 0: return False, "当前没有成功推送的同步投影", {"action": "push_sync", "results": results} return True, f"同步推送完成，成功 {success_count}/{len(results)}", {"action": "push_sync", "results": results} + + +def pull_detect_task_batch_now(limit: int | None = None) -> tuple[bool, str, dict]: + if settings.node_region != "mainland" or settings.node_role != "control": + return False, "当前节点无需拉取待检测任务批次", {"action": "pull_tasks"} + + export_url = _task_export_url(settings.sync_target_api_base_url) + ack_url = _task_ack_url(settings.sync_target_api_base_url) + if not export_url or not ack_url: + return False, "未配置任务拉取目标地址", {"action": "pull_tasks"} + + safe_limit = max(1, min(int(limit or settings.sync_batch_size or 200), max(1, int(settings.sync_batch_size or 200)))) + request_url = f"{export_url}?limit={safe_limit}" + request = urllib.request.Request( + request_url, + headers={**({"X-Domaincheck-Sync-Token": settings.sync_shared_token} if settings.sync_shared_token else {})}, + method="GET", + ) + try: + with urllib.request.urlopen(request, timeout=20) as response: + raw = response.read().decode("utf-8") + data = json.loads(raw) if raw else {} + except urllib.error.HTTPError as exc: + error_body = exc.read().decode("utf-8", errors="ignore") if hasattr(exc, "read") else "" + return False, f"拉取待检测任务失败: HTTP {getattr(exc, 'code', 500)}", { + "action": "pull_tasks", + "http_status": getattr(exc, "code", 500), + "response_text": error_body[:500], + } + except Exception as exc: + return False, f"拉取待检测任务失败: {exc}", {"action": "pull_tasks"} + + if int(data.get("code", 1) or 1) != 0: + return False, str(data.get("message") or "拉取待检测任务失败"), {"action": "pull_tasks", "response": data} + + payload = data.get("data") or {} + source_record_id = int(payload.get("source_record_id") or 0) + projection_hash = str(payload.get("projection_hash") or "").strip() + projection = payload.get("projection") or {} + if source_record_id <= 0 or not projection_hash or not projection: + return False, "远端当前没有可拉取的待检测任务批次", {"action": "pull_tasks", "batch_size": 0} + + ingest_ok, ingest_message, ingest_data = ingest_detect_task_projection( + { + "sync_type": "detect_task_projection", + "source_region": _normalize_region(settings.sync_target_region, "overseas"), + "source_record_id": source_record_id, + "projection_hash": projection_hash, + "projection": projection, + }, + shared_token=settings.sync_shared_token, + ) + if not ingest_ok: + return False, ingest_message, {"action": "pull_tasks", **(ingest_data or {})} + + ack_request = urllib.request.Request( + ack_url, + data=json.dumps( + { + "source_record_id": source_record_id, + "projection_hash": projection_hash, + }, + ensure_ascii=False, + ).encode("utf-8"), + headers={ + "Content-Type": "application/json", + **({"X-Domaincheck-Sync-Token": settings.sync_shared_token} if settings.sync_shared_token else {}), + }, + method="POST", + ) + ack_data = {} + try: + with urllib.request.urlopen(ack_request, timeout=15) as response: + raw = response.read().decode("utf-8") + ack_response = json.loads(raw) if raw else {} + ack_data = ack_response.get("data") or {} + except Exception as exc: + return True, f"{ingest_message}；但远端确认失败: {exc}", { + "action": "pull_tasks", + "source_record_id": source_record_id, + **(ingest_data or {}), + } + + return True, "待检测任务批次拉取并入库成功", { + "action": "pull_tasks", + "source_record_id": source_record_id, + "projection_hash": projection_hash, + "batch_code": str(projection.get("batch_code") or "").strip(), + **(ingest_data or {}), + "ack": ack_data, + } diff --git a/domain-api/app/sync_agent.py b/domain-api/app/sync_agent.py index 3a00a31..3fa8029 100644 --- a/domain-api/app/sync_agent.py +++ b/domain-api/app/sync_agent.py @@ -4,7 +4,7 @@ import logging import time from app.core.config import settings -from app.services.sync_push_service import push_runtime_projection_now +from app.services.sync_push_service import pull_detect_task_batch_now, push_runtime_projection_now logger = logging.getLogger("domaincheck.sync_agent") @@ -28,6 +28,8 @@ def main() -> None: try: ok, message, data = push_runtime_projection_now() logger.info("sync tick: ok=%s message=%s data=%s", ok, message, 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) except Exception as exc: logger.exception("sync tick failed: %s", exc) time.sleep(interval)