douyin-sparkflow/DouYinSparkFlow/core/protocol_dispatch.py

import asyncio
import json
import os
import random
import shutil
import subprocess
from datetime import datetime, timezone
from pathlib import Path

from core.msg_builder import build_messages_for_targets
from utils.config import get_userData, normalize_unique_id, repo_root, save_userData
from utils.logger import setup_logger


logger = setup_logger()
PROTOCOL_SCRIPT = repo_root() / "core" / "protocol_sender.mjs"
NODE_HELPER_IMAGE = "node:22-alpine"


def _coerce_non_negative_int(value, default):
    try:
        return max(0, int(value))
    except (TypeError, ValueError):
        return max(0, int(default))


def _normalize_send_strategy(config):
    raw = config.get("sendStrategy", {}) or {}
    start_min = _coerce_non_negative_int(raw.get("accountStartDelaySecondsMin", 0), 0)
    start_max = _coerce_non_negative_int(raw.get("accountStartDelaySecondsMax", start_min), start_min)
    if start_max < start_min:
        start_max = start_min

    message_min = _coerce_non_negative_int(raw.get("messageIntervalSecondsMin", 0), 0)
    message_max = _coerce_non_negative_int(raw.get("messageIntervalSecondsMax", message_min), message_min)
    if message_max < message_min:
        message_max = message_min

    strategy = {
        "shuffleTargets": bool(raw.get("shuffleTargets", True)),
        "accountStartDelaySecondsMin": start_min,
        "accountStartDelaySecondsMax": start_max,
        "messageIntervalSecondsMin": message_min,
        "messageIntervalSecondsMax": message_max,
        "messageVariants": [str(item).strip() for item in raw.get("messageVariants", []) if str(item).strip()],
    }
    if os.getenv("SPARKFLOW_MANUAL_RUN") == "1":
        strategy["accountStartDelaySecondsMin"] = 0
        strategy["accountStartDelaySecondsMax"] = 0
        strategy["messageIntervalSecondsMin"] = min(strategy["messageIntervalSecondsMin"], 3)
        strategy["messageIntervalSecondsMax"] = min(strategy["messageIntervalSecondsMax"], 6)
    return strategy


def _account_identity_key(account):
    normalized_unique_id = normalize_unique_id(account.get("unique_id"))
    if normalized_unique_id:
        return f"uid:{normalized_unique_id}"

    username = str(account.get("username", "")).strip()
    if username:
        return f"user:{username}"

    return ""


def _coerce_attempt_count(entry):
    try:
        return max(0, int(dict(entry or {}).get("attemptCount") or 0))
    except (TypeError, ValueError):
        return 0


def _protocol_failure_category(entry):
    status_name = str(entry.get("statusName") or "").strip()
    status_code = entry.get("statusCode")
    if status_name == "CheckMessageNotPass" or status_code == 3:
        return "protocol_check_message_not_pass"
    if status_name == "CheckMessageNotPassButSelfVisible" or status_code == 4:
        return "protocol_check_message_self_visible"
    if status_name == "UserNotInConversation" or status_code == 1:
        return "protocol_user_not_in_conversation"
    if status_name == "CheckConversationNotPass" or status_code == 2:
        return "protocol_check_conversation_not_pass"
    if status_name == "UserHasBeenBlock" or status_code == 5:
        return "protocol_user_blocked"
    return "protocol_send_failed"


def _protocol_failure_reason(entry):
    bits = [
        f"statusCode={entry.get('statusCode')}",
        f"statusName={entry.get('statusName') or ''}",
        f"statusMsg={entry.get('statusMsg') or ''}",
    ]
    summary = entry.get("sendResultSummary") or {}
    raw_keys = summary.get("rawKeys") or []
    if raw_keys:
        bits.append(f"rawKeys={','.join(map(str, raw_keys))}")
    return " ".join(bits)


def _persist_protocol_account_failure(account, category, reason, affected_targets=None):
    now_iso = datetime.now(timezone.utc).isoformat(timespec="seconds")
    all_accounts = get_userData(force_reload=True)
    accounts_by_identity = {
        identity: item
        for item in all_accounts
        for identity in [_account_identity_key(item)]
        if identity
    }
    target_account = accounts_by_identity.get(_account_identity_key(account))
    if not target_account:
        return

    affected_targets = list(affected_targets or [])
    existing_entry = dict(target_account.get("account_failure") or {})
    target_account["account_failure"] = {
        "category": category,
        "reason": reason,
        "firstAttemptAt": existing_entry.get("firstAttemptAt") or now_iso,
        "lastAttemptAt": now_iso,
        "attemptCount": _coerce_attempt_count(existing_entry) + 1,
        "lastRunMode": "protocol",
        "affectedTargets": affected_targets,
    }
    save_userData(all_accounts)


def _record_protocol_target_failure(target_account, target_name, message, category, reason):
    now_iso = datetime.now(timezone.utc).isoformat(timespec="seconds")
    queue = dict(target_account.get("failure_queue") or {})
    existing_entry = dict(queue.get(target_name) or {})
    queue[target_name] = {
        "category": category,
        "reason": reason,
        "message": message,
        "firstAttemptAt": existing_entry.get("firstAttemptAt") or now_iso,
        "lastAttemptAt": now_iso,
        "attemptCount": _coerce_attempt_count(existing_entry) + 1,
        "lastRunMode": "protocol",
    }
    target_account["failure_queue"] = queue


def _merge_protocol_runtime_state(accounts, result_by_username):
    changed = False
    now_iso = datetime.now(timezone.utc).isoformat(timespec="seconds")
    all_accounts = get_userData(force_reload=True)
    accounts_by_identity = {
        identity: account
        for account in all_accounts
        for identity in [_account_identity_key(account)]
        if identity
    }

    for account in accounts:
        target_account = accounts_by_identity.get(_account_identity_key(account))
        if not target_account:
            continue

        result = result_by_username.get(account.get("username"))
        if not result:
            continue

        protocol_cache = result.get("protocol_targets_cache")
        if protocol_cache is not None:
            target_account["protocol_targets_cache"] = protocol_cache
            target_account["protocol_user_id"] = result.get("userId", "")
            changed = True

        history = dict(target_account.get("message_history") or {})
        for entry in result.get("sent", []):
            if entry.get("dryRun") or not entry.get("success", True):
                continue

            target = str(entry.get("target", "")).strip()
            message = str(entry.get("message", "")).strip()
            if not target or not message:
                continue

            history[target] = {
                "message": message,
                "sentAt": str(entry.get("sentAt", now_iso)),
            }
            changed = True

        if history:
            target_account["message_history"] = history

        for entry in result.get("sent", []):
            if entry.get("dryRun") or entry.get("success", True):
                continue
            target = str(entry.get("target", "")).strip()
            if not target:
                continue
            _record_protocol_target_failure(
                target_account,
                target,
                str(entry.get("message", "")).strip(),
                _protocol_failure_category(entry),
                _protocol_failure_reason(entry),
            )
            changed = True

        unresolved = result.get("unresolved", []) or []
        for entry in unresolved:
            target = str(entry.get("target", "")).strip()
            if not target:
                continue
            _record_protocol_target_failure(
                target_account,
                target,
                "",
                str(entry.get("reason") or "protocol_unresolved"),
                str(entry.get("reason") or "protocol could not resolve target"),
            )
            changed = True

    if changed:
        save_userData(all_accounts)


def _host_repo_root():
    candidates = [
        Path("/opt/douyin-sparkflow/DouYinSparkFlow"),
        repo_root(),
    ]
    for candidate in candidates:
        if (candidate / "core" / "protocol_sender.mjs").exists():
            return candidate
    return repo_root()


def _build_protocol_command():
    node_path = shutil.which("node")
    if node_path:
        return [node_path, str(PROTOCOL_SCRIPT)], repo_root(), "local-node", str(repo_root())

    docker_path = shutil.which("docker")
    if docker_path:
        host_repo = _host_repo_root()
        return (
            [
                docker_path,
                "run",
                "--rm",
                "-i",
                "--network",
                "host",
                "-v",
                f"{host_repo}:/workspace",
                "-w",
                "/workspace",
                NODE_HELPER_IMAGE,
                "node",
                "core/protocol_sender.mjs",
            ],
            repo_root(),
            "docker-node-helper",
            "/workspace",
        )

    raise RuntimeError("Neither node nor docker is available for the protocol sender")


def _run_protocol_for_user(user, messages_by_target, dry_run, send_strategy):
    command, cwd, runner_label, runtime_repo_root = _build_protocol_command()
    payload = {
        "repoRoot": runtime_repo_root,
        "dryRun": dry_run,
        "account": user,
        "messagesByTarget": messages_by_target,
        "sendStrategy": send_strategy,
    }
    process = subprocess.run(
        command,
        input=json.dumps(payload, ensure_ascii=False),
        text=True,
        capture_output=True,
        cwd=str(cwd),
        check=False,
    )

    stdout = (process.stdout or "").strip()
    if not stdout:
        raise RuntimeError(
            f"protocol sender returned no output for {user.get('username', 'unknown')}: {process.stderr}"
        )

    try:
        data = json.loads(stdout)
    except json.JSONDecodeError as exc:
        raise RuntimeError(
            f"protocol sender produced invalid JSON for {user.get('username', 'unknown')}: {stdout}"
        ) from exc

    if process.returncode != 0 or not data.get("ok"):
        error_message = data.get("error") or process.stderr or "protocol sender failed"
        raise RuntimeError(
            f"{user.get('username', 'unknown')} protocol sender failed: {error_message}"
        )

    data["runner"] = runner_label

    return data


async def run_protocol_tasks(config, accounts, message_builder):
    del message_builder

    dry_run = bool(config.get("protocolDryRun", False))
    multi_task = bool(config.get("multiTask", True))
    concurrency = int(config.get("taskCount", 1)) if multi_task else 1
    semaphore = asyncio.Semaphore(max(concurrency, 1))
    send_strategy = _normalize_send_strategy(config)

    async def _worker(user):
        async with semaphore:
            start_delay = random.randint(
                send_strategy["accountStartDelaySecondsMin"],
                send_strategy["accountStartDelaySecondsMax"],
            )
            if start_delay > 0:
                logger.info(
                    "Delaying protocol sender for %s by %ss to avoid synchronized bursts",
                    user.get("username", "unknown"),
                    start_delay,
                )
                await asyncio.sleep(start_delay)

            logger.info("Starting protocol sender for %s", user.get("username", "unknown"))
            messages_by_target = build_messages_for_targets(
                user.get("targets", []),
                previous_messages=user.get("message_history", {}),
                config=config,
            )
            logger.info(
                "Prepared %s protocol messages for %s with shuffleTargets=%s interval=%s-%ss manual_run=%s",
                len(messages_by_target),
                user.get("username", "unknown"),
                send_strategy["shuffleTargets"],
                send_strategy["messageIntervalSecondsMin"],
                send_strategy["messageIntervalSecondsMax"],
                os.getenv("SPARKFLOW_MANUAL_RUN") == "1",
            )
            result = await asyncio.to_thread(
                _run_protocol_for_user,
                user,
                messages_by_target,
                dry_run,
                send_strategy,
            )
            sent_entries = result.get("sent", [])
            succeeded_count = len([
                entry for entry in sent_entries
                if not entry.get("dryRun") and entry.get("success", True)
            ])
            failed_count = len([
                entry for entry in sent_entries
                if not entry.get("dryRun") and not entry.get("success", True)
            ])
            logger.info(
                "Protocol sender finished for %s resolved=%s unresolved=%s attempted=%s succeeded=%s failed=%s dryRun=%s",
                user.get("username", "unknown"),
                len(result.get("resolved", [])),
                len(result.get("unresolved", [])),
                len(sent_entries),
                succeeded_count,
                failed_count,
                bool(result.get("dryRun")),
            )
            return result

    gathered = await asyncio.gather(*(_worker(user) for user in accounts), return_exceptions=True)

    result_by_username = {}
    failures = []
    for user, item in zip(accounts, gathered):
        if isinstance(item, Exception):
            reason = str(item)
            failures.append(reason)
            logger.error("Protocol sender failed for %s: %s", user.get("username", "unknown"), item)
            _persist_protocol_account_failure(
                user,
                "protocol_sender_failed",
                reason,
                user.get("targets", []),
            )
            continue
        result_by_username[user.get("username")] = item
        unresolved = item.get("unresolved", [])
        if unresolved:
            logger.warning(
                "Protocol sender could not resolve %s targets for %s: %s",
                len(unresolved),
                user.get("username", "unknown"),
                [entry.get("target") for entry in unresolved],
            )

    _merge_protocol_runtime_state(accounts, result_by_username)

    if failures and not result_by_username:
        raise RuntimeError("; ".join(failures))

    return [result_by_username[user.get("username")] for user in accounts if user.get("username") in result_by_username]