from __future__ import annotations

import json
import socket
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
from app.services.cluster_runtime_service import register_node_heartbeat
from app.services.sync_record_service import _decode_json, _normalize_region


def _format_time(value: datetime | None) -> str:
    return value.isoformat(sep=" ", timespec="seconds") if value else ""


def _ingest_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/sync-ingest"
    if text.endswith("/api/v1/runtime"):
        return f"{text}/sync-ingest"
    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"


def _refresh_remote_runtime_node(*, source_region: str, projection: dict, received_at: datetime | None = None) -> None:
    node_info = projection.get("node") or {}
    node_code = str(node_info.get("node_code") or "").strip()
    region = _normalize_region(node_info.get("region"), source_region)
    role = str(node_info.get("role") or "control").strip() or "control"
    hostname = str(node_info.get("hostname") or "").strip() or socket.gethostname()
    ip = str(node_info.get("ip") or "").strip()
    if not node_code:
        node_code = f"{region}-{role}-imported"

    metadata = {
        "service": "runtime-ingest",
        "projection_source_region": source_region,
        "worker_mode": projection.get("worker_mode", ""),
        "phase_label": projection.get("phase_label", ""),
        "phase_detail": projection.get("phase_detail", ""),
        "proxy_runtime_label": projection.get("proxy_runtime_label", ""),
        "proxy_runtime_reason": projection.get("proxy_runtime_reason", ""),
        "updated_at": _format_time(received_at or datetime.now()),
    }
    register_node_heartbeat(
        node_code=node_code,
        region=region,
        role=role,
        status="online",
        current_load=int(((projection.get("progress") or {}).get("running", 0) or 0)),
        metadata=metadata,
        hostname_override=hostname,
        ip_override=ip,
    )


def _load_latest_projection(sync_type: str) -> dict | None:
    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, source_region, target_region, status, payload_json, created_at, updated_at
                FROM detect_sync_records
                WHERE sync_type = %s
                  AND source_region = %s
                  AND target_region = %s
                ORDER BY created_at DESC, id DESC
                LIMIT 1
                """,
                (sync_type, source_region, target_region),
            )
            row = cur.fetchone()
    if not row:
        return None
    return {
        "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],
    }


def _load_pushable_projections(sync_type: str, limit: int) -> list[dict]:
    safe_limit = max(1, min(int(limit or 1), max(1, int(settings.sync_batch_size or 200))))
    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, source_region, target_region, status, payload_json, created_at, updated_at
                FROM detect_sync_records
                WHERE sync_type = %s
                  AND source_region = %s
                  AND target_region = %s
                ORDER BY created_at ASC, id ASC
                LIMIT %s
                """,
                (sync_type, source_region, target_region, safe_limit * 5),
            )
            rows = cur.fetchall()

    selected: list[dict] = []
    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_attempt = _latest_push_attempt(projection["id"], projection["target_region"], sync_type)
        if latest_attempt and latest_attempt["status"] == "success":
            continue
        if latest_attempt and latest_attempt["status"] == "pending":
            continue
        if latest_attempt and latest_attempt["status"] == "failed":
            last_created_at = latest_attempt.get("created_at")
            if isinstance(last_created_at, datetime):
                now = datetime.now(last_created_at.tzinfo) if last_created_at.tzinfo else datetime.now()
                if now - last_created_at < timedelta(seconds=max(10, int(settings.sync_poll_interval_seconds or 30))):
                    continue
        selected.append(projection)
        if len(selected) >= safe_limit:
            break
    return selected


def _latest_push_attempt(source_record_id: int, target_region: str, sync_type: str) -> dict | None:
    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 = 'runtime_push'
                  AND source_region = %s
                  AND target_region = %s
                  AND (payload_json->>'sync_type') = %s
                  AND (payload_json->>'source_record_id') = %s
                ORDER BY created_at DESC, id DESC
                LIMIT 1
                """,
                (
                    _normalize_region(settings.sync_source_region, settings.node_region),
                    target_region,
                    sync_type,
                    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 _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,
        "source_record_id": source_record["id"],
        "projection_hash": (source_record.get("payload") or {}).get("projection_hash", ""),
        "ingest_url": ingest_url,
    }
    with get_db() as conn:
        with conn.cursor() as cur:
            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
                """,
                (
                    "runtime_push",
                    source_record["source_region"],
                    source_record["target_region"],
                    "pending",
                    json.dumps(payload, ensure_ascii=False),
                    "",
                ),
            )
            record_id = int(cur.fetchone()[0])
        conn.commit()
    return record_id


def _update_push_attempt(record_id: int, *, status: str, payload: dict | None = None, error_message: str = "") -> None:
    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                UPDATE detect_sync_records
                SET status = %s,
                    payload_json = %s::jsonb,
                    error_message = %s,
                    updated_at = CURRENT_TIMESTAMP
                WHERE id = %s
                """,
                (
                    str(status or "").strip() or "pending",
                    json.dumps(payload or {}, ensure_ascii=False),
                    str(error_message or "").strip(),
                    int(record_id),
                ),
            )
        conn.commit()


def ingest_runtime_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 "runtime_projection").strip() or "runtime_projection"
    ingest_sync_type = _projection_ingest_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 {}
    target_region = _normalize_region(settings.node_region, "overseas")
    received_at = datetime.now()

    with get_db() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT id
                FROM detect_sync_records
                WHERE sync_type = %s
                  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
                """,
                (ingest_sync_type, source_region, target_region, str(source_record_id), projection_hash),
            )
            existing = cur.fetchone()
            if existing:
                if sync_type == "runtime_projection":
                    _refresh_remote_runtime_node(source_region=source_region, projection=projection, received_at=received_at)
                cur.execute(
                    """
                    UPDATE detect_sync_records
                    SET updated_at = CURRENT_TIMESTAMP
                    WHERE id = %s
                    """,
                    (int(existing[0]),),
                )
                conn.commit()
                return True, "同步投影已存在，已按幂等处理", {"record_id": int(existing[0]), "deduplicated": True}

            stored_payload = {
                "sync_type": sync_type,
                "source_record_id": source_record_id,
                "projection_hash": projection_hash,
                "projection": projection,
                "received_at": _format_time(received_at),
            }
            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
                """,
                (
                    ingest_sync_type,
                    source_region,
                    target_region,
                    "received",
                    json.dumps(stored_payload, ensure_ascii=False),
                    "",
                ),
            )
            record_id = int(cur.fetchone()[0])
        conn.commit()
    if sync_type == "runtime_projection":
        _refresh_remote_runtime_node(source_region=source_region, projection=projection, received_at=received_at)
    return True, "同步投影接收成功", {"record_id": record_id, "deduplicated": False}


def _push_projection_now(sync_type: str, ingest_url: str) -> tuple[bool, str, dict]:
    source_record = _load_latest_projection(sync_type)
    if not source_record:
        return False, f"当前没有可推送的{sync_type}", {"action": "push_sync", "sync_type": sync_type}
    return _push_projection_record(source_record, sync_type, ingest_url)


def _push_projection_batch(sync_type: str, ingest_url: str) -> tuple[bool, str, dict]:
    pending_records = _load_pushable_projections(sync_type, limit=max(1, int(settings.sync_batch_size or 200)))
    if not pending_records:
        latest = _load_latest_projection(sync_type)
        if latest:
            latest_attempt = _latest_push_attempt(latest["id"], latest["target_region"], sync_type)
            if latest_attempt and latest_attempt["status"] == "success":
                return True, "当前批次已全部同步完成", {
                    "action": "push_sync",
                    "sync_type": sync_type,
                    "deduplicated": True,
                    "batch_count": 0,
                }
        return False, f"当前没有可推送的{sync_type}", {"action": "push_sync", "sync_type": sync_type, "batch_count": 0}

    results: list[dict] = []
    success_count = 0
    for record in pending_records:
        ok, message, data = _push_projection_record(record, sync_type, ingest_url)
        results.append(
            {
                "source_record_id": record["id"],
                "ok": ok,
                "message": message,
                "data": data,
            }
        )
        if ok:
            success_count += 1

    overall_ok = success_count > 0
    message = f"{sync_type} 批量推送完成，成功 {success_count}/{len(results)}"
    return overall_ok, message, {
        "action": "push_sync",
        "sync_type": sync_type,
        "batch_count": len(results),
        "success_count": success_count,
        "results": results,
    }


def _push_projection_record(source_record: dict, sync_type: str, ingest_url: str) -> tuple[bool, str, dict]:
    latest_attempt = _latest_push_attempt(source_record["id"], source_record["target_region"], sync_type)
    if latest_attempt and latest_attempt["status"] == "success":
        last_created_at = latest_attempt.get("created_at")
        if sync_type != "runtime_projection" or not isinstance(last_created_at, datetime):
            return True, "该投影已推送，无需重复发送", {
                "action": "push_sync",
                "sync_type": sync_type,
                "source_record_id": source_record["id"],
                "deduplicated": True,
            }
        now = datetime.now(last_created_at.tzinfo) if last_created_at.tzinfo else datetime.now()
        if now - last_created_at < timedelta(seconds=max(20, int(settings.sync_poll_interval_seconds or 30))):
            return True, "该投影已推送，无需重复发送", {
                "action": "push_sync",
                "sync_type": sync_type,
                "source_record_id": source_record["id"],
                "deduplicated": True,
            }
    if latest_attempt and latest_attempt["status"] == "pending":
        return True, "该投影已有同步推送进行中，暂不重复发送", {
            "action": "push_sync",
            "sync_type": sync_type,
            "source_record_id": source_record["id"],
            "deduplicated": True,
        }
    if latest_attempt and latest_attempt["status"] == "failed":
        last_created_at = latest_attempt.get("created_at")
        if isinstance(last_created_at, datetime):
            now = datetime.now(last_created_at.tzinfo) if last_created_at.tzinfo else datetime.now()
            if now - last_created_at < timedelta(seconds=max(10, int(settings.sync_poll_interval_seconds or 30))):
                return False, "最近一次同步推送刚失败，等待下个重试窗口", {
                    "action": "push_sync",
                    "sync_type": sync_type,
                    "source_record_id": source_record["id"],
                    "deduplicated": True,
                }

    attempt_id = _create_push_attempt(source_record=source_record, ingest_url=ingest_url, sync_type=sync_type)
    request_payload = {
        "sync_type": sync_type,
        "source_region": source_record["source_region"],
        "source_record_id": source_record["id"],
        "projection_hash": (source_record.get("payload") or {}).get("projection_hash", ""),
        "projection": (source_record.get("payload") or {}).get("projection", {}),
        "created_at": _format_time(source_record.get("created_at")),
    }
    request_body = json.dumps(request_payload, ensure_ascii=False).encode("utf-8")
    request = urllib.request.Request(
        ingest_url,
        data=request_body,
        headers={
            "Content-Type": "application/json",
            **({"X-Domaincheck-Sync-Token": settings.sync_shared_token} if settings.sync_shared_token else {}),
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(request, timeout=15) as response:
            raw = response.read().decode("utf-8")
            data = json.loads(raw) if raw else {}
            status_code = getattr(response, "status", 200)
    except urllib.error.HTTPError as exc:
        error_body = exc.read().decode("utf-8", errors="ignore") if hasattr(exc, "read") else ""
        payload = {
            "sync_type": sync_type,
            "source_record_id": source_record["id"],
            "projection_hash": request_payload["projection_hash"],
            "ingest_url": ingest_url,
            "http_status": getattr(exc, "code", 500),
            "response_text": error_body[:1000],
        }
        _update_push_attempt(attempt_id, status="failed", payload=payload, error_message=f"HTTP {getattr(exc, 'code', 500)}")
        return False, f"同步推送失败: HTTP {getattr(exc, 'code', 500)}", {"action": "push_sync", "sync_type": sync_type, "attempt_id": attempt_id}
    except Exception as exc:
        payload = {
            "sync_type": sync_type,
            "source_record_id": source_record["id"],
            "projection_hash": request_payload["projection_hash"],
            "ingest_url": ingest_url,
        }
        _update_push_attempt(attempt_id, status="failed", payload=payload, error_message=str(exc))
        return False, f"同步推送失败: {exc}", {"action": "push_sync", "sync_type": sync_type, "attempt_id": attempt_id}

    payload = {
        "sync_type": sync_type,
        "source_record_id": source_record["id"],
        "projection_hash": request_payload["projection_hash"],
        "ingest_url": ingest_url,
        "http_status": status_code,
        "response": data,
    }
    _update_push_attempt(attempt_id, status="success", payload=payload, error_message="")
    return True, "投影推送成功", {
        "action": "push_sync",
        "sync_type": sync_type,
        "attempt_id": attempt_id,
        "source_record_id": source_record["id"],
        "response": data,
    }


def push_runtime_projection_now() -> tuple[bool, str, dict]:
    if not settings.sync_push_enabled:
        return False, "未启用同步推送", {"action": "push_sync"}

    ingest_url = _ingest_url(settings.sync_target_api_base_url)
    if not ingest_url:
        return False, "未配置同步目标地址", {"action": "push_sync"}

    results = []
    ok, message, data = _push_projection_now("runtime_projection", ingest_url)
    results.append({"sync_type": "runtime_projection", "ok": ok, "message": message, "data": data})

    ok, message, data = _push_projection_batch("detect_result_projection", ingest_url)
    results.append({"sync_type": "detect_result_projection", "ok": ok, "message": message, "data": data})

    success_count = sum(1 for item in results if item["ok"])
    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,
    }