import asyncio
import base64
import errno
import hashlib
import json
import logging
import os
import random
import unicodedata
from contextlib import contextmanager
from datetime import datetime, timedelta, timezone
from pathlib import Path
from zoneinfo import ZoneInfo
from core.browser import get_browser, get_persistent_browser_context, sanitize_profile_name
from core.msg_builder import build_message, build_message_candidates
from core.protocol_dispatch import run_protocol_tasks
from core.send_state import parse_sent_at, target_is_strong_confirmed_today
from utils.config import get_config, get_userData, normalize_unique_id, save_userData
from utils.logger import setup_logger
logger = setup_logger(level=logging.DEBUG)
debug_artifacts_dir = Path("logs/debug_artifacts")
debug_artifacts_dir.mkdir(parents=True, exist_ok=True)
CREATOR_HOME_URL = "https://creator.douyin.com/"
CREATOR_CHAT_URL = "https://creator.douyin.com/creator-micro/data/following/chat"
async def retry_operation(name, operation, retries=3, delay=2, *args, **kwargs):
for attempt in range(retries):
try:
return await operation(*args, **kwargs)
except Exception as exc:
if attempt < retries - 1:
logger.warning("%s failed, retry %s/%s: %s", name, attempt + 1, retries, exc)
await asyncio.sleep(delay)
else:
logger.error("%s failed after %s attempts: %s", name, retries, exc)
raise
def _safe_name(value):
return "".join(ch if ch.isalnum() or ch in ("-", "_") else "_" for ch in value)[:80]
def _normalize_target_name(value):
raw = unicodedata.normalize("NFKC", str(value or ""))
for token in ("\u200b", "\u200c", "\u200d", "\ufeff"):
raw = raw.replace(token, "")
raw = raw.replace("\xa0", " ")
return " ".join(raw.split()).strip()
def _current_run_mode():
if _manual_run_unsent_only():
return "manual_unsent_only"
if _manual_run_failed_only():
return "manual_failed_only"
return "manual" if _is_manual_run() else "scheduled"
def _coerce_non_negative_int(value, default):
try:
return max(0, int(value))
except (TypeError, ValueError):
return max(0, int(default))
def _coerce_positive_int(value, default):
try:
return max(1, int(value))
except (TypeError, ValueError):
return max(1, int(default))
def _coerce_positive_float(value, default):
try:
parsed = float(value)
except (TypeError, ValueError):
parsed = float(default)
if parsed <= 0:
parsed = float(default)
return parsed
def _normalize_friend_list_scan_config(active_config):
raw = active_config.get("friendListScan", {}) or {}
return {
"maxScanSeconds": _coerce_positive_int(raw.get("maxScanSeconds", 300), 300),
"idleScanSeconds": _coerce_positive_int(raw.get("idleScanSeconds", 120), 120),
"scrollStepPx": _coerce_positive_int(raw.get("scrollStepPx", 400), 400),
"scrollDelaySeconds": _coerce_positive_float(raw.get("scrollDelaySeconds", 1.5), 1.5),
}
def _normalize_send_strategy(active_config):
raw = active_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
return {
"accountStartDelaySecondsMin": start_min,
"accountStartDelaySecondsMax": start_max,
"messageIntervalSecondsMin": message_min,
"messageIntervalSecondsMax": message_max,
}
def _normalize_persistent_profile_config(active_config):
raw = active_config.get("persistentBrowserProfiles", {}) or {}
return {
"enabled": bool(raw.get("enabled", False)),
"root": str(
os.getenv("SPARKFLOW_BROWSER_PROFILE_ROOT")
or raw.get("root")
or "/opt/douyin-sparkflow/state/browser-profiles"
),
"seedCookiesWhenEmpty": bool(raw.get("seedCookiesWhenEmpty", True)),
"syncStoredCookiesBeforeRun": bool(raw.get("syncStoredCookiesBeforeRun", True)),
"refreshStoredCookiesAfterLogin": bool(raw.get("refreshStoredCookiesAfterLogin", True)),
}
def _account_profile_name(user):
unique_id = normalize_unique_id(user.get("unique_id"))
username = str(user.get("username") or "").strip()
if unique_id:
return f"uid-{unique_id}"
if username:
return f"user-{sanitize_profile_name(username)}"
return "unknown"
def _random_delay_seconds(send_strategy, min_key, max_key):
return random.randint(send_strategy[min_key], send_strategy[max_key])
async def _sleep_with_log(seconds, reason, account_name):
if seconds <= 0:
return
logger.info("%s for %s by %ss", reason, account_name, seconds)
await asyncio.sleep(seconds)
LOGIN_REQUIRED_TEXTS = (
"扫码登录",
"验证码登录",
"密码登录",
"登录/注册",
"登录后免费",
"请登录",
"身份验证",
"安全验证",
"风险提示",
"环境异常",
)
LOGIN_REQUIRED_SELECTORS = (
".login-mask",
".login-guide-container",
".login-img-code-wrapper",
".pc-login-verification-modal",
)
LOGIN_REQUIRED_DIALOG_SELECTORS = (
".semi-modal-content",
'div[role="dialog"]',
)
NON_LOGIN_DIALOG_DISMISS_TEXTS = (
"我知道了",
"知道了",
"好的",
"确定",
"确认",
"稍后再说",
"关闭",
)
NON_LOGIN_DIALOG_CLOSE_SELECTORS = (
".semi-modal-close",
'button[aria-label="Close"]',
'button[aria-label="关闭"]',
'[aria-label="Close"]',
'[aria-label="关闭"]',
)
AUTH_COOKIE_NAMES = {
"sessionid",
"sessionid_ss",
"sid_guard",
"sid_tt",
"uid_tt",
"uid_tt_ss",
}
async def _body_text_sample(page, limit=1200):
try:
return (await page.locator("body").inner_text(timeout=3000)).replace("\n", " ")[:limit]
except Exception as exc:
return f"BODY_TEXT_ERROR={exc!r}"
def _find_login_required_text(text):
sample = text or ""
for marker in LOGIN_REQUIRED_TEXTS:
if marker in sample:
return marker
return None
async def _page_has_login_required_prompt(page):
for selector in LOGIN_REQUIRED_SELECTORS:
locator = page.locator(selector).first
try:
if await locator.count() > 0 and await locator.is_visible(timeout=500):
return True, f"visible selector {selector}"
except Exception:
continue
for selector in LOGIN_REQUIRED_DIALOG_SELECTORS:
locator = page.locator(selector)
try:
count = min(await locator.count(), 5)
except Exception:
continue
for index in range(count):
item = locator.nth(index)
try:
if not await item.is_visible(timeout=500):
continue
dialog_text = (await item.inner_text(timeout=1000)).replace("\n", " ")
except Exception:
continue
matched_text = _find_login_required_text(dialog_text)
if matched_text:
return True, (
f"visible dialog {selector} contains {matched_text!r}: "
f"{dialog_text[:300]}"
)
logger.debug(
"Ignoring visible non-login dialog for selector %s: %s",
selector,
dialog_text[:160],
)
sample = await _body_text_sample(page)
matched_text = _find_login_required_text(sample)
if matched_text:
return True, f"body contains {matched_text!r}: {sample[:300]}"
return False, ""
async def ensure_not_login_required(page, account_name, stage):
is_required, detail = await _page_has_login_required_prompt(page)
if is_required:
raise RuntimeError(f"login_required at {stage} for {account_name}: {detail}")
async def _dismiss_non_login_dialogs(page, account_name, stage):
dismissed = 0
for selector in LOGIN_REQUIRED_DIALOG_SELECTORS:
locator = page.locator(selector)
try:
count = min(await locator.count(), 5)
except Exception:
continue
for index in range(count):
item = locator.nth(index)
try:
if not await item.is_visible(timeout=500):
continue
dialog_text = (await item.inner_text(timeout=1000)).replace("\n", " ")
except Exception:
continue
if _find_login_required_text(dialog_text):
continue
clicked = False
for text in NON_LOGIN_DIALOG_DISMISS_TEXTS:
button = item.get_by_text(text, exact=True).first
try:
if await button.count() > 0 and await button.is_visible(timeout=500):
await button.click(timeout=2000)
clicked = True
break
except Exception:
continue
if not clicked:
for close_selector in NON_LOGIN_DIALOG_CLOSE_SELECTORS:
close_button = item.locator(close_selector).first
try:
if await close_button.count() > 0 and await close_button.is_visible(timeout=500):
await close_button.click(timeout=2000)
clicked = True
break
except Exception:
continue
if clicked:
dismissed += 1
logger.info(
"Dismissed non-login dialog for %s at %s: %s",
account_name,
stage,
dialog_text[:160],
)
await asyncio.sleep(0.5)
else:
logger.debug(
"Visible non-login dialog for %s at %s was not dismissed: %s",
account_name,
stage,
dialog_text[:160],
)
return dismissed
async def apply_stored_cookies_to_profile(context, cookies, account_name, only_when_empty=False):
if not cookies:
return
if only_when_empty:
try:
current_cookies = await context.cookies(
["https://creator.douyin.com/", "https://www.douyin.com/"]
)
except Exception:
current_cookies = []
if any(cookie.get("name") in AUTH_COOKIE_NAMES for cookie in current_cookies):
logger.info(
"Persistent profile for %s already has auth cookies; not seeding stored cookies",
account_name,
)
return
await context.add_cookies(cookies)
logger.info("Applied %s stored cookies to persistent profile for %s", len(cookies), account_name)
async def refresh_stored_cookies_from_profile(context, user, account_name):
try:
cookies = await context.cookies()
except Exception as exc:
logger.warning("Could not read persistent profile cookies for %s: %s", account_name, exc)
return
if not cookies:
return
accounts = get_userData(force_reload=True)
matched_account = _find_matching_account(accounts, user)
if matched_account is None:
logger.warning("Could not find account to refresh cookies for user=%s", account_name)
return
matched_account["cookies"] = list(cookies)
save_userData(accounts)
user["cookies"] = list(cookies)
logger.info("Refreshed stored cookies for %s from persistent profile count=%s", account_name, len(cookies))
def _is_stop_account_category(category):
return _is_account_level_failure_category(category)
async def save_debug_artifacts(page, account_name, target_name, stage):
if not get_config(force_reload=True).get("saveDebugArtifacts", False):
return
timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
stem = f"{timestamp}-{_safe_name(account_name)}-{_safe_name(target_name)}-{stage}"
screenshot_path = debug_artifacts_dir / f"{stem}.png"
html_path = debug_artifacts_dir / f"{stem}.html"
await page.screenshot(path=str(screenshot_path), full_page=True)
html_path.write_text(await page.content(), encoding="utf-8")
logger.info("Saved debug artifacts at stage=%s for %s/%s", stage, account_name, target_name)
async def locate_chat_input(page):
selectors = [
"xpath=//div[contains(@class, 'chat-input-dccKiL')]//div[@contenteditable='true']",
"xpath=//div[@contenteditable='true' and @role='textbox']",
"xpath=(//div[@contenteditable='true'])[last()]",
]
last_error = None
for selector in selectors:
locator = page.locator(selector).first
try:
await locator.wait_for(state="visible", timeout=10000)
await locator.click(timeout=5000)
return locator, selector
except Exception as exc:
last_error = exc
raise RuntimeError(f"Unable to locate chat input, last error: {last_error}")
async def read_chat_input_text(chat_input):
try:
return await chat_input.evaluate(
"""(node) => {
const raw = node.innerText ?? node.textContent ?? "";
return raw.trim();
}"""
)
except Exception:
return ""
def _message_probe_text(message):
for line in str(message or "").splitlines():
probe = line.strip()
if probe:
return probe
return str(message or "").strip()
def _normalize_message_text(value):
return "\n".join(
line.strip()
for line in str(value or "").replace("\r\n", "\n").replace("\r", "\n").split("\n")
if line.strip()
).strip()
async def _detect_send_failure_indicator(page):
"""Return a short string describing a visible send-failure UI element, or ""."""
try:
return await page.evaluate(
"""() => {
// Red exclamation icon commonly used by chat UIs for failed sends.
const failIcons = document.querySelectorAll(
"svg[class*='fail'], svg[class*='Fail'], svg[class*='error'], svg[class*='Error'],"
+ " [class*='exclamation'], [class*='Exclamation'],"
+ " [class*='retry'], [class*='Retry'], [class*='resend'], [class*='Resend']"
);
for (const el of failIcons) {
const r = el.getBoundingClientRect();
if (r.width > 0 && r.height > 0) {
// Climb up to find the associated message row text.
let row = el.closest("li, [role='listitem'], [class*='message'], [class*='Message'], [class*='item']");
let snippet = String(row?.innerText || "").slice(0, 40);
return "fail_icon(" + (el.className || "").slice(0, 60) + ") near=" + snippet;
}
}
// Explicit failure / retry text.
const bodyText = String(document.body?.innerText || "");
const failKeywords = ["发送失败", "重试", "重新发送", "send failed", "retry", "resend"];
for (const kw of failKeywords) {
if (bodyText.includes(kw)) {
return "fail_text=" + kw;
}
}
return "";
}"""
)
except Exception:
return ""
async def start_im_send_observer(page, account_name, target_name):
"""Observe creator IM service calls for one browser send attempt."""
state = {
"enabled": False,
"send_request_seen": False,
"send_response_seen": False,
"send_receipt": {},
"mark_read_calls": [],
"identity_security_token_calls": [],
"events": [],
"error": "",
}
pending = {}
session = None
try:
session = await page.context.new_cdp_session(page)
await session.send("Network.enable")
state["enabled"] = True
except Exception as exc:
state["error"] = f"cdp_unavailable: {exc}"
logger.warning("IM observer unavailable for %s/%s: %s", account_name, target_name, exc)
async def disabled_summary(extra_wait_seconds=0):
if extra_wait_seconds:
await asyncio.sleep(extra_wait_seconds)
return dict(state)
return disabled_summary
def _trim(value, limit=220):
return str(value or "")[:limit]
def _call_kind(url):
value = str(url or "")
if "/v1/message/send" in value:
return "message_send"
if "mark_read" in value:
return "mark_read"
if "identity_security_token" in value:
return "identity_security_token"
if "imapi.douyin.com" in value:
return "imapi_other"
return ""
def _safe_url(url):
return str(url or "").split("?", 1)[0]
def _header_value(headers, name):
target = name.lower()
for key, value in (headers or {}).items():
if str(key).lower() == target:
return str(value)
return ""
def _decode_body(body):
raw = body.get("body") or ""
if body.get("base64Encoded"):
try:
return base64.b64decode(raw).decode("utf-8", errors="replace")
except Exception:
return ""
return raw
def _parse_json_body(text):
try:
return json.loads(text)
except Exception:
return None
def _extract_error_text(data, body_text):
if isinstance(data, dict):
for key in ("status_msg", "message", "msg", "err_msg", "error", "reason"):
value = data.get(key)
if value:
return _trim(value, 300)
nested = data.get("data")
if isinstance(nested, dict):
for key in ("status_msg", "message", "msg", "err_msg", "error", "reason"):
value = nested.get(key)
if value:
return _trim(value, 300)
return _trim(body_text, 300)
def _json_success(data):
if not isinstance(data, dict):
return None
success_values = []
for key in ("status_code", "err_no", "errno", "error_code", "code"):
if key in data:
success_values.append(data.get(key) in (0, "0", None))
message = str(data.get("message") or data.get("status_msg") or "").lower()
if message:
success_values.append(message in ("success", "ok"))
if success_values:
return all(success_values)
return None
def _receipt_body_meta(data, body_text):
meta = {"bodyLen": len(body_text or "")}
if body_text:
meta["bodySha256"] = hashlib.sha256(body_text.encode("utf-8", errors="replace")).hexdigest()
if isinstance(data, dict):
meta["jsonKeys"] = sorted(str(key) for key in data.keys())[:20]
nested = data.get("data")
if isinstance(nested, dict):
meta["dataKeys"] = sorted(str(key) for key in nested.keys())[:20]
for source_key, dest_key in (
("server_message_id", "serverMessageId"),
("message_id", "messageId"),
("msg_id", "messageId"),
("conversation_id", "conversationId"),
("conversation_short_id", "conversationShortId"),
):
value = nested.get(source_key)
if value:
meta[dest_key] = _trim(value, 120)
for source_key, dest_key in (
("server_message_id", "serverMessageId"),
("message_id", "messageId"),
("msg_id", "messageId"),
("conversation_id", "conversationId"),
("conversation_short_id", "conversationShortId"),
):
value = data.get(source_key)
if value:
meta[dest_key] = _trim(value, 120)
return meta
def _record(event):
state["events"].append(event)
if len(state["events"]) > 50:
del state["events"][:-50]
def on_request(params):
request = params.get("request") or {}
url = request.get("url") or ""
kind = _call_kind(url)
if not kind:
return
request_id = params.get("requestId")
post_data = request.get("postData") or ""
method = request.get("method") or ""
is_send_post = kind == "message_send" and method.upper() == "POST"
pending[request_id] = {"url": url, "kind": kind, "request": request, "is_send_post": is_send_post}
_record({"kind": "request", "call": kind, "url": _safe_url(url), "method": method, "postLen": len(post_data)})
if is_send_post:
state["send_request_seen"] = True
logger.info("IM observer saw message send request for %s/%s url=%s postLen=%s", account_name, target_name, _trim(_safe_url(url), 160), len(post_data))
def on_response(params):
request_id = params.get("requestId")
item = pending.get(request_id)
if not item:
return
response = params.get("response") or {}
item["response"] = response
kind = item.get("kind")
headers = response.get("headers") or {}
status = response.get("status")
logid = _header_value(headers, "x-tt-logid") or _header_value(headers, "x-tt-trace-log") or _header_value(headers, "x-tt-trace-id")
event = {"kind": "response", "call": kind, "url": _safe_url(item.get("url")), "status": status}
if logid:
event["logid"] = _trim(logid, 120)
_record(event)
if item.get("is_send_post"):
state["send_response_seen"] = True
state["send_receipt"] = {"call": kind, "httpStatus": status, "ok": False, "pendingBody": True, "logid": _trim(logid, 120), "url": _safe_url(item.get("url"))}
logger.info("IM observer saw message send response for %s/%s status=%s logid=%s", account_name, target_name, status, _trim(logid, 80))
async def fetch_body(request_id):
item = pending.get(request_id) or {}
kind = item.get("kind") or ""
url = item.get("url") or ""
response = item.get("response") or {}
status = response.get("status")
headers = response.get("headers") or {}
logid = _header_value(headers, "x-tt-logid") or _header_value(headers, "x-tt-trace-log") or _header_value(headers, "x-tt-trace-id")
try:
body = await session.send("Network.getResponseBody", {"requestId": request_id})
body_text = _decode_body(body)
data = _parse_json_body(body_text)
http_ok = isinstance(status, (int, float)) and 200 <= status < 300
json_ok = _json_success(data)
receipt = {"call": kind, "httpStatus": status, "ok": bool(http_ok and (json_ok is not False)), "jsonParsed": isinstance(data, dict), "logid": _trim(logid, 120), "url": _safe_url(url)}
receipt.update(_receipt_body_meta(data, body_text))
if json_ok is not None:
receipt["jsonOk"] = bool(json_ok)
if not receipt["ok"]:
receipt["reason"] = _extract_error_text(data, body_text)
if item.get("is_send_post"):
state["send_receipt"] = receipt
_record({"kind": "body", "call": kind, "status": status, "ok": receipt["ok"], "jsonParsed": receipt["jsonParsed"], "logid": receipt.get("logid", ""), "reason": receipt.get("reason", "")})
elif kind == "mark_read":
state["mark_read_calls"].append(receipt)
_record({"kind": "body", "call": kind, "status": status, "ok": receipt["ok"]})
elif kind == "identity_security_token":
has_token = isinstance(data, dict) and bool((data.get("data") or {}).get("identity_security_token"))
receipt["hasToken"] = has_token
receipt.pop("reason", None)
state["identity_security_token_calls"].append(receipt)
_record({"kind": "body", "call": kind, "status": status, "ok": receipt["ok"], "hasToken": has_token})
else:
_record({"kind": "body", "call": kind, "status": status, "ok": receipt["ok"]})
except Exception as exc:
error = _trim(exc, 160)
_record({"kind": "body_error", "call": kind, "url": _safe_url(url), "error": error})
if item.get("is_send_post"):
state["send_receipt"] = {"call": kind, "httpStatus": status, "ok": False, "logid": _trim(logid, 120), "url": _safe_url(url), "reason": f"response_body_unavailable: {error}"}
finally:
pending.pop(request_id, None)
def on_loading_finished(params):
request_id = params.get("requestId")
if request_id in pending:
asyncio.create_task(fetch_body(request_id))
session.on("Network.requestWillBeSent", on_request)
session.on("Network.responseReceived", on_response)
session.on("Network.loadingFinished", on_loading_finished)
async def summary(extra_wait_seconds=3):
if extra_wait_seconds:
await asyncio.sleep(extra_wait_seconds)
try:
await session.detach()
except Exception:
pass
return dict(state)
return summary
async def snapshot_last_own_message(page, chat_input=None):
try:
return await page.evaluate(
r"""() => {
// ---- DOM structure (douyin creator private message) ----
// Each row is ...
.
// - time separator: class contains "time-" (e.g. time-Tl7Z4j)
// - own message: class contains "is-me" (e.g. is-me-cb9NAa)
// - peer message: no "is-me" and no "time-"
// Text is inside .
const allRows = Array.from(
document.querySelectorAll("[class*='box-item-']")
).filter(el => {
const r = el.getBoundingClientRect();
return r.width > 0 && r.height > 0;
});
// Walk backwards to find the boundary: first time separator that
// indicates yesterday or earlier. Everything after it is "today".
let todayStart = 0;
for (let i = allRows.length - 1; i >= 0; i--) {
const cls = String(allRows[i].className || "");
if (!cls.includes("time-")) continue;
const label = String(allRows[i].innerText || "").trim();
// "今天" / "刚刚" / "N分钟前" / pure time like "11:01" => today
// "昨天" / "星期X" / "2026-..." => not today
if (/^(昨天|星期|20\d{2}-)/.test(label)) {
todayStart = i + 1;
break;
}
}
// From bottom, find the last own-message row within today's section.
for (let i = allRows.length - 1; i >= todayStart; i--) {
const cls = String(allRows[i].className || "");
if (!cls.includes("is-me")) continue;
if (cls.includes("time-")) continue;
const pre = allRows[i].querySelector("pre, [class*='text-']");
const text = String(pre?.innerText || allRows[i].innerText || "").trim();
if (!text) continue;
const rect = allRows[i].getBoundingClientRect();
return {
text,
centerX: rect.left + rect.width / 2,
centerY: rect.top + rect.height / 2,
top: rect.top,
bottom: rect.bottom,
right: rect.right,
className: cls.slice(0, 240),
};
}
return null;
}""",
)
except Exception as exc:
logger.debug("Unable to snapshot last own message: %s", exc)
return None
async def count_today_own_message_matches(page, target_text):
"""Count how many of *target_text* appear in today's own-message bubbles."""
try:
return await page.evaluate(
r"""(targetText) => {
const allRows = Array.from(
document.querySelectorAll("[class*='box-item-']")
).filter(el => {
const r = el.getBoundingClientRect();
return r.width > 0 && r.height > 0;
});
let todayStart = 0;
for (let i = allRows.length - 1; i >= 0; i--) {
const cls = String(allRows[i].className || "");
if (!cls.includes("time-")) continue;
const label = String(allRows[i].innerText || "").trim();
if (/^(昨天|星期|20\d{2}-)/.test(label)) {
todayStart = i + 1;
break;
}
}
let count = 0;
for (let i = todayStart; i < allRows.length; i++) {
const cls = String(allRows[i].className || "");
if (!cls.includes("is-me")) continue;
if (cls.includes("time-")) continue;
const pre = allRows[i].querySelector("pre, [class*='text-']");
const text = String(pre?.innerText || allRows[i].innerText || "").trim();
if (text === targetText) count++;
}
return count;
}""",
_normalize_message_text(target_text),
)
except Exception:
return 0
async def count_visible_message_matches(page, message, chat_input=None):
probe = _message_probe_text(message)
if not probe:
return 0
locator = page.locator(f"text={probe}")
try:
count = await locator.count()
except Exception:
return 0
input_box = None
if chat_input is not None:
try:
input_box = await chat_input.bounding_box(timeout=1000)
except Exception:
input_box = None
visible_count = 0
for index in range(min(count, 200)):
try:
item = locator.nth(index)
if not await item.is_visible(timeout=500):
continue
if await item.evaluate("(node) => Boolean(node.closest('[contenteditable=\"true\"]'))"):
continue
if input_box:
box = await item.bounding_box(timeout=500)
if not box:
continue
center_x = box["x"] + box["width"] / 2
center_y = box["y"] + box["height"] / 2
in_chat_column = (
input_box["x"] - 80
<= center_x
<= input_box["x"] + input_box["width"] + 80
)
above_input = center_y < input_box["y"] + input_box["height"] + 80
if not (in_chat_column and above_input):
continue
visible_count += 1
except Exception:
continue
return visible_count
async def confirm_message_sent(page, chat_input, message, before_snapshot=None):
expected_message = _normalize_message_text(message)
before_signature = None
if before_snapshot:
before_signature = (
before_snapshot.get("text"),
round(float(before_snapshot.get("centerY") or 0), 1),
round(float(before_snapshot.get("right") or 0), 1),
)
# Record how many times the target text already appears today so we can
# detect a *new* occurrence even if the last-own-message slot was taken by
# a manual send from the user's phone.
before_count = await count_today_own_message_matches(page, expected_message)
last_input_text = ""
last_snapshot = None
deadline = asyncio.get_running_loop().time() + 8
while True:
await asyncio.sleep(1)
last_input_text = await read_chat_input_text(chat_input)
# Fail fast if the page shows a send-failure indicator (red exclamation,
# retry button, or explicit failure text) before we even check the bubble.
failure_indicator = await _detect_send_failure_indicator(page)
if failure_indicator:
return False, f"send failure indicator detected: {failure_indicator}"
last_snapshot = await snapshot_last_own_message(page, chat_input=chat_input)
last_signature = None
if last_snapshot:
last_signature = (
last_snapshot.get("text"),
round(float(last_snapshot.get("centerY") or 0), 1),
round(float(last_snapshot.get("right") or 0), 1),
)
if (
not last_input_text
and last_snapshot
and _normalize_message_text(last_snapshot.get("text")) == expected_message
and last_signature != before_signature
):
return (
True,
"last own message matches sent text; chat input cleared",
)
# Count fallback: even if the last own bubble is not our message (e.g.
# the user sent something else from their phone), a new occurrence of
# the target text today means our send went through.
if (
not last_input_text
and not last_snapshot
):
after_count = await count_today_own_message_matches(page, expected_message)
if after_count > before_count:
return (
True,
"count fallback: new own message with target text appeared today",
)
if asyncio.get_running_loop().time() >= deadline:
break
if last_input_text:
return False, f"chat input still contains: {last_input_text!r}"
# Final count fallback before giving up.
after_count = await count_today_own_message_matches(page, expected_message)
if after_count > before_count:
return (
True,
"count fallback at deadline: new own message with target text appeared today",
)
if last_snapshot:
return False, (
"last own message did not match sent text: "
f"expected={expected_message!r} actual={last_snapshot.get('text')!r}"
)
return (
False,
"chat input cleared but no new own message bubble was confirmed",
)
async def find_visible_spark_message_candidate(page, chat_input=None, candidates=None):
messages = []
for candidate in candidates or build_message_candidates():
probe = _message_probe_text(candidate)
if probe and probe not in messages:
messages.append(probe)
for candidate in messages:
try:
if await count_visible_message_matches(page, candidate, chat_input=chat_input) > 0:
return candidate
except Exception:
continue
return ""
async def detect_message_already_sent(page, chat_input, message, before_snapshot=None):
try:
if chat_input is not None:
sent_ok, detail = await confirm_message_sent(
page,
chat_input,
message,
before_snapshot=before_snapshot,
)
if sent_ok:
return True, detail
except Exception:
pass
return False, ""
def classify_browser_failure(stage, exc):
detail = str(exc or "")
lowered = detail.lower()
if "friend_list_unavailable" in lowered:
return "friend_list_unavailable"
if "friend_list_incomplete" in lowered:
return "friend_list_incomplete"
if "login_required" in lowered:
return "login_required"
if any(token in detail for token in LOGIN_REQUIRED_TEXTS):
return "login_required"
if "page crashed" in lowered or "target page, context or browser has been closed" in lowered:
return "page_crashed"
if "timeout" in lowered:
if stage in {"open_creator_home", "open_chat_page"}:
return "navigation_timeout"
if stage == "locate_chat_input":
return "chat_input_timeout"
if stage == "friend_list":
return "friend_list_timeout"
return "timeout"
if "unable to locate chat input" in lowered:
return "chat_input_not_found"
if "could not find the friend list scroll container" in lowered:
return "friend_list_container_missing"
if "chat input still contains" in lowered or "visible message count did not increase" in lowered:
return "send_unconfirmed"
if "missing targets" in lowered:
return "friend_not_found"
if stage in {"open_creator_home", "open_chat_page"}:
return "navigation_failed"
return "unknown"
class FriendListUnavailableError(RuntimeError):
pass
class FriendListIncompleteError(RuntimeError):
pass
ACCOUNT_LEVEL_FAILURE_CATEGORIES = {
"login_required",
"friend_list_unavailable",
"friend_list_timeout",
"friend_list_incomplete",
"friend_list_container_missing",
"navigation_timeout",
"navigation_failed",
"page_crashed",
}
TEMPORARY_ACCOUNT_FAILURE_CATEGORIES = {
"friend_list_unavailable",
"friend_list_timeout",
"friend_list_incomplete",
"friend_list_container_missing",
"navigation_timeout",
"navigation_failed",
"page_crashed",
"timeout",
}
def _is_account_level_failure_category(category):
return category in ACCOUNT_LEVEL_FAILURE_CATEGORIES
def _target_display_name(target):
if isinstance(target, dict):
for key in ("name", "displayName", "nickname", "username", "unique_id", "uid"):
value = str(target.get(key) or "").strip()
if value:
return value
return ""
return str(target or "").strip()
def _build_normalized_target_map(targets):
normalized_targets = {}
for target in targets or []:
display_name = _target_display_name(target)
normalized_name = _normalize_target_name(display_name)
if normalized_name:
normalized_targets[normalized_name] = display_name
return normalized_targets
async def _locator_count_with_timeout(locator, account_name, stage, label, timeout_seconds=3):
try:
return await asyncio.wait_for(locator.count(), timeout=timeout_seconds)
except asyncio.TimeoutError:
logger.warning(
"Account %s locator count timed out at %s label=%s after %ss",
account_name,
stage,
label,
timeout_seconds,
)
return None
async def _first_non_empty_locator(page, selectors):
for selector in selectors:
locator = page.locator(selector)
count = await _locator_count_with_timeout(
locator,
"unknown",
"first_non_empty_locator",
selector,
)
if count is None:
continue
for index in range(min(count, 5)):
item = locator.nth(index)
try:
if await item.is_visible(timeout=500):
return selector, locator
except Exception:
continue
return "", None
async def _selector_visible(page, selectors):
for selector in selectors:
locator = page.locator(selector).first
try:
count = await _locator_count_with_timeout(
locator,
"unknown",
"selector_visible",
selector,
)
if count and await locator.is_visible(timeout=500):
return selector
except Exception:
continue
return ""
async def _click_first_visible_locator(candidates, account_name, stage):
last_error = None
for label, locator in candidates:
try:
count = await _locator_count_with_timeout(
locator,
account_name,
stage,
label,
)
count = min(count or 0, 5)
except Exception as exc:
last_error = exc
continue
for index in range(count):
item = locator.nth(index)
try:
if not await item.is_visible(timeout=800):
continue
await item.click(timeout=3000)
logger.info("Account %s clicked %s at %s", account_name, label, stage)
return True
except Exception as exc:
last_error = exc
continue
if last_error:
logger.debug("Account %s could not click a visible locator at %s: %s", account_name, stage, last_error)
return False
async def _open_friends_tab(page, account_name, fallback_selector, target_selectors):
await _dismiss_non_login_dialogs(page, account_name, "before_open_friends_tab")
_, existing_locator = await _first_non_empty_locator(page, target_selectors)
if existing_locator:
logger.info("Account %s friend list is already visible before tab click", account_name)
return "already_visible"
sub_app = page.locator('xpath=//*[@id="sub-app"]')
text_selector = (
'xpath=//*[@id="sub-app"]//*[self::div or self::button or self::span or @role="tab"]'
'[contains(normalize-space(.), "朋友") or contains(normalize-space(.), "好友") '
'or contains(normalize-space(.), "互关")]'
)
page_text_selector = (
'xpath=//*[self::div or self::button or self::span or @role="tab" or @role="button"]'
'[contains(normalize-space(.), "朋友私信") or contains(normalize-space(.), "好友私信") '
'or contains(normalize-space(.), "朋友") or contains(normalize-space(.), "好友") '
'or contains(normalize-space(.), "互关")]'
)
candidates = [
("page tab text 朋友私信", page.get_by_text("朋友私信", exact=True)),
("page tab text 好友私信", page.get_by_text("好友私信", exact=True)),
("tab text 朋友", sub_app.get_by_text("朋友", exact=True)),
("tab text 好友", sub_app.get_by_text("好友", exact=True)),
("tab text 互关", sub_app.get_by_text("互关", exact=True)),
("tab text 朋友私信", sub_app.get_by_text("朋友私信", exact=True)),
("tab text 好友私信", sub_app.get_by_text("好友私信", exact=True)),
("sub-app friend tab text", page.locator(text_selector)),
("page friend tab text", page.locator(page_text_selector)),
("fallback friends tab xpath", page.locator(fallback_selector)),
]
if await _click_first_visible_locator(candidates, account_name, "open_friends_tab"):
await asyncio.sleep(2)
return "clicked"
await page.wait_for_selector(fallback_selector, timeout=30000)
await page.locator(fallback_selector).click(timeout=5000)
logger.info("Account %s clicked fallback friends tab after explicit wait", account_name)
await asyncio.sleep(1)
return "fallback_wait_clicked"
async def _reopen_friend_chat_page(page, account_name):
logger.warning(
"Account %s friend list did not become visible within 60s; reopening chat page once",
account_name,
)
await _dismiss_non_login_dialogs(page, account_name, "before_reopen_chat_page")
try:
await page.reload(wait_until="load", timeout=120000)
await asyncio.sleep(2)
except Exception as exc:
logger.warning("Account %s reload before reopen chat page failed: %s", account_name, exc)
await retry_operation(
"reopen chat page after friend list wait",
page.goto,
retries=2,
delay=5,
url=CREATOR_CHAT_URL,
)
await asyncio.sleep(3)
await ensure_not_login_required(page, account_name, "reopen_chat_page")
await _dismiss_non_login_dialogs(page, account_name, "after_reopen_chat_page")
async def _wait_for_friend_list_ready(
page,
account_name,
target_selectors,
no_more_selectors,
loading_selectors,
timeout_seconds=120,
empty_grace_seconds=45,
):
started_at = asyncio.get_running_loop().time()
last_loading_seen_at = None
last_dialog_dismissed_at = None
while True:
dismissed_count = await _dismiss_non_login_dialogs(page, account_name, "friend_list_ready")
if dismissed_count:
last_dialog_dismissed_at = asyncio.get_running_loop().time()
selector, locator = await _first_non_empty_locator(page, target_selectors)
if locator:
logger.debug("Account %s friend list ready via selector %s", account_name, selector)
return selector, locator
is_login_required, login_detail = await _page_has_login_required_prompt(page)
if is_login_required:
raise RuntimeError(f"login_required at friend_list for {account_name}: {login_detail}")
no_more_selector = await _selector_visible(page, no_more_selectors)
if no_more_selector:
elapsed = asyncio.get_running_loop().time() - started_at
if elapsed > empty_grace_seconds:
raise FriendListUnavailableError(
f"friend_list_unavailable for {account_name}: no friend entries visible; "
f"empty/end selector {no_more_selector} is visible after {empty_grace_seconds}s"
)
logger.debug(
"Account %s saw empty/end selector %s after %.1fs; waiting within grace window",
account_name,
no_more_selector,
elapsed,
)
loading_selector = await _selector_visible(page, loading_selectors)
if loading_selector:
last_loading_seen_at = asyncio.get_running_loop().time()
logger.debug("Account %s waiting for friend list loading selector %s", account_name, loading_selector)
elapsed = asyncio.get_running_loop().time() - started_at
if elapsed > timeout_seconds:
loading_note = ""
if last_loading_seen_at:
loading_note = f"; loading was visible {elapsed - last_loading_seen_at:.1f}s ago"
if last_dialog_dismissed_at:
loading_note += f"; dialog was dismissed {elapsed - (last_dialog_dismissed_at - started_at):.1f}s ago"
raise FriendListUnavailableError(
f"friend_list_unavailable for {account_name}: no friend entries became visible "
f"within {timeout_seconds}s{loading_note}"
)
await asyncio.sleep(1)
async def _extract_friend_display_name(element):
name_selectors = (
"""xpath=.//span[contains(@class, "item-header-name-")]""",
"""xpath=.//*[contains(@class, "item-header-name-")]""",
)
for selector in name_selectors:
locator = element.locator(selector).first
try:
if await locator.count() > 0:
text = (await locator.inner_text(timeout=1000)).strip()
if text:
return text
except Exception:
continue
try:
text = (await element.inner_text(timeout=1000)).strip()
except Exception:
return ""
for line in text.splitlines():
line = line.strip()
if line:
return line
return ""
async def _extract_friend_stable_keys(element):
stable_keys = set()
attrs = (
"data-e2e",
"data-id",
"data-user-id",
"data-conversation-id",
"data-sec-uid",
"href",
"aria-label",
"title",
)
for attr in attrs:
try:
value = await element.get_attribute(attr)
except Exception:
value = None
if value:
stable_keys.add(f"{attr}:{str(value).strip()[:240]}")
try:
links = await element.locator("xpath=.//a[@href]").all()
except Exception:
links = []
for link in links[:3]:
try:
href = await link.get_attribute("href")
except Exception:
href = None
if href:
stable_keys.add(f"href:{str(href).strip()[:240]}")
return sorted(stable_keys)
async def _extract_friend_record(element):
display_name = await _extract_friend_display_name(element)
normalized_name = _normalize_target_name(display_name)
if not normalized_name:
return None
return {
"visibleName": display_name,
"normalizedName": normalized_name,
"stableKeys": await _extract_friend_stable_keys(element),
}
async def _first_scrollable_friends_element(page, selectors):
for selector in selectors:
try:
handle = await page.locator(selector).first.element_handle()
except Exception:
handle = None
if handle:
return selector, handle
return "", None
async def scroll_and_select_user(page, user, account_name, targets, friend_scan_config=None, index_targets=None):
friends_tab_selector = 'xpath=//*[@id="sub-app"]/div/div/div[1]/div[2]'
target_selectors = (
'xpath=//*[@id="sub-app"]/div/div[1]/div[2]/div[2]'
'//div[contains(@class, "semi-list-item-body")]',
'xpath=//*[@id="sub-app"]//div[contains(@class, "semi-list-item-body") and .//span[contains(@class, "item-header-name-")]]',
'xpath=//*[@id="sub-app"]//li[.//span[contains(@class, "item-header-name-")]]',
)
scrollable_friends_selectors = (
'xpath=//*[@id="sub-app"]/div/div[1]/div[2]/div[2]/div/div/div[3]/div/div/div/ul/div',
'xpath=//*[@id="sub-app"]//ul/div',
'xpath=//*[@id="sub-app"]//div[contains(@class, "semi-list")]//ul/..',
)
no_more_selectors = (
'xpath=//div[contains(@class, "no-more-tip-ftdJnu")]',
'xpath=//*[@id="sub-app"]//*[contains(text(), "没有更多") or contains(text(), "暂无")]',
)
loading_selectors = (
'xpath=//div[contains(@class, "semi-spin")]',
'xpath=//*[@id="sub-app"]//*[contains(@class, "loading") or contains(@class, "Loading")]',
)
friend_scan_config = friend_scan_config or _normalize_friend_list_scan_config({})
max_scan_seconds = friend_scan_config["maxScanSeconds"]
idle_scan_seconds = friend_scan_config["idleScanSeconds"]
scroll_step_px = friend_scan_config["scrollStepPx"]
scroll_delay_seconds = friend_scan_config["scrollDelaySeconds"]
ready_timeout_seconds = 60
empty_grace_seconds = 30
ready_errors = []
active_target_selector = ""
for ready_attempt in range(2):
logger.debug("Account %s is opening the friends tab attempt=%s", account_name, ready_attempt + 1)
await _open_friends_tab(page, account_name, friends_tab_selector, target_selectors)
try:
active_target_selector, _ = await _wait_for_friend_list_ready(
page,
account_name,
target_selectors,
no_more_selectors,
loading_selectors,
timeout_seconds=ready_timeout_seconds,
empty_grace_seconds=empty_grace_seconds,
)
break
except FriendListUnavailableError as exc:
ready_errors.append(str(exc))
if ready_attempt == 0:
await _reopen_friend_chat_page(page, account_name)
continue
raise FriendListUnavailableError(
f"friend_list_unavailable for {account_name}: no friend entries after "
f"two {ready_timeout_seconds}s waits; first={ready_errors[0]}; second={ready_errors[-1]}"
) from exc
await asyncio.sleep(2)
normalized_targets = _build_normalized_target_map(targets)
normalized_index_targets = _build_normalized_target_map(index_targets or targets)
found_usernames = set()
remaining_targets = set(normalized_targets)
remaining_index_targets = set(normalized_index_targets)
friend_index = {}
scan_started_at = asyncio.get_running_loop().time()
last_new_friend_at = scan_started_at
logger.debug(
"Account %s friend list scan config maxScanSeconds=%s idleScanSeconds=%s scrollStepPx=%s scrollDelaySeconds=%s readyTimeoutSeconds=%s emptyGraceSeconds=%s",
account_name,
max_scan_seconds,
idle_scan_seconds,
scroll_step_px,
scroll_delay_seconds,
ready_timeout_seconds,
empty_grace_seconds,
)
def missing_target_names():
return sorted(normalized_targets[item] for item in remaining_targets)
def missing_index_target_names():
return sorted(normalized_index_targets[item] for item in remaining_index_targets)
def persist_index(scan_complete):
_persist_friend_index(
user,
friend_index,
datetime.now(timezone.utc).isoformat(timespec="seconds"),
scan_complete=scan_complete,
missing_targets=missing_index_target_names(),
)
while True:
now_monotonic = asyncio.get_running_loop().time()
if now_monotonic - scan_started_at > max_scan_seconds:
persist_index(False)
logger.warning(
"Account %s friend list scan timed out after %ss. Missing delivery targets: %s; missing indexed targets=%s; scannedFriends=%s",
account_name,
max_scan_seconds,
missing_target_names(),
missing_index_target_names(),
len(found_usernames),
)
raise FriendListIncompleteError(
f"friend_list_incomplete for {account_name}: scan timed out after {max_scan_seconds}s; "
f"missingTargets={missing_target_names()}; scannedFriends={len(found_usernames)}"
)
await _dismiss_non_login_dialogs(page, account_name, "friend_list_scan")
selector, target_locator = await _first_non_empty_locator(page, (active_target_selector,) + tuple(target_selectors))
if not target_locator:
active_target_selector, _ = await _wait_for_friend_list_ready(
page,
account_name,
target_selectors,
no_more_selectors,
loading_selectors,
timeout_seconds=30,
empty_grace_seconds=15,
)
continue
target_elements = await target_locator.all()
clicked_delivery_target = False
for element in target_elements:
friend_record = await _extract_friend_record(element)
if not friend_record:
continue
target_name = friend_record["visibleName"]
normalized_target_name = friend_record["normalizedName"]
if not normalized_target_name:
continue
if normalized_target_name in found_usernames:
continue
found_usernames.add(normalized_target_name)
friend_index[normalized_target_name] = friend_record
remaining_index_targets.discard(normalized_target_name)
last_new_friend_at = asyncio.get_running_loop().time()
logger.debug("Account %s found friend entry %s", account_name, target_name)
matched_target_name = normalized_targets.get(normalized_target_name)
if matched_target_name and normalized_target_name in remaining_targets:
await element.click()
logger.info("Account %s selected target friend %s", account_name, target_name)
if matched_target_name != target_name:
logger.info(
"Account %s normalized target %r matched visible friend %r",
account_name,
matched_target_name,
target_name,
)
scan_paused_at = asyncio.get_running_loop().time()
yield matched_target_name
scan_resumed_at = asyncio.get_running_loop().time()
scan_pause_seconds = max(0, scan_resumed_at - scan_paused_at)
if scan_pause_seconds:
scan_started_at += scan_pause_seconds
last_new_friend_at += scan_pause_seconds
logger.debug(
"Account %s paused friend list scan timer for %.1fs while delivering %s",
account_name,
scan_pause_seconds,
matched_target_name,
)
remaining_targets.discard(normalized_target_name)
clicked_delivery_target = True
if not remaining_targets and not remaining_index_targets:
logger.info("Account %s found all delivery and indexed target friends", account_name)
persist_index(True)
return
break
if clicked_delivery_target:
continue
if not remaining_targets and not remaining_index_targets:
logger.info("Account %s found all delivery and indexed target friends", account_name)
persist_index(True)
return
else:
if await _selector_visible(page, no_more_selectors):
persist_index(True)
logger.warning(
"Account %s reached the end of the friend list. Missing delivery targets: %s; missing indexed targets=%s",
account_name,
missing_target_names(),
missing_index_target_names(),
)
return
now_monotonic = asyncio.get_running_loop().time()
if found_usernames and now_monotonic - last_new_friend_at > idle_scan_seconds:
persist_index(False)
logger.warning(
"Account %s friend list scan made no progress for %ss. Missing delivery targets: %s; missing indexed targets=%s; scannedFriends=%s",
account_name,
idle_scan_seconds,
missing_target_names(),
missing_index_target_names(),
len(found_usernames),
)
raise FriendListIncompleteError(
f"friend_list_incomplete for {account_name}: no new friends for {idle_scan_seconds}s; "
f"missingTargets={missing_target_names()}; scannedFriends={len(found_usernames)}"
)
if await _selector_visible(page, loading_selectors):
logger.debug("Account %s is waiting for more friends to load", account_name)
await asyncio.sleep(scroll_delay_seconds)
scrollable_selector, scrollable_element = await _first_scrollable_friends_element(page, scrollable_friends_selectors)
if not scrollable_element:
raise FriendListUnavailableError(
f"friend_list_unavailable for {account_name}: could not find the friend list scroll container"
)
logger.debug("Account %s scrolls friend list via selector %s", account_name, scrollable_selector)
await page.evaluate(f"(element) => element.scrollTop += {scroll_step_px}", scrollable_element)
await asyncio.sleep(scroll_delay_seconds)
def _is_manual_run():
return os.getenv("SPARKFLOW_MANUAL_RUN") == "1"
def _schedule_timezone():
timezone_name = (
str(os.getenv("SPARKFLOW_TIMEZONE") or "").strip()
or str(os.getenv("TZ") or "").strip()
or "Asia/Shanghai"
)
try:
return ZoneInfo(timezone_name)
except Exception:
if timezone_name == "Asia/Shanghai":
logger.warning("Falling back to fixed UTC+8 because %r is unavailable", timezone_name)
return timezone(timedelta(hours=8), name="Asia/Shanghai")
logger.warning("Falling back to system timezone because %r is unavailable", timezone_name)
return datetime.now().astimezone().tzinfo
def _normalize_send_window(config):
raw = config.get("dailySendWindow", {}) or {}
normalized = {
"enabled": bool(raw.get("enabled", False)),
"startHour": int(raw.get("startHour", 10)),
"endHour": int(raw.get("endHour", 18)),
"scheduleIntervalMinutes": max(1, int(raw.get("scheduleIntervalMinutes", 10))),
}
if normalized["startHour"] < 0 or normalized["startHour"] > 23:
normalized["enabled"] = False
if normalized["endHour"] < 1 or normalized["endHour"] > 24:
normalized["enabled"] = False
if normalized["endHour"] <= normalized["startHour"]:
normalized["enabled"] = False
if bool(raw.get("enabled", False)) and not normalized["enabled"]:
logger.warning("Invalid dailySendWindow=%s, disabling windowed sending for this run", raw)
return normalized
def _account_identity(user):
return str(user.get("unique_id") or user.get("username") or "unknown").strip()
def _parse_sent_at(raw_value, local_tz):
return parse_sent_at(raw_value, local_tz)
def _manual_run_failed_only():
return _is_manual_run() and os.getenv("SPARKFLOW_MANUAL_FAILED_ONLY") == "1"
def _manual_run_unsent_only():
return _is_manual_run() and os.getenv("SPARKFLOW_MANUAL_UNSENT_ONLY") == "1"
def _requested_account_refs():
raw = os.getenv("SPARKFLOW_ACCOUNT_REFS")
if raw is None:
return None
return {item.strip() for item in raw.split(",") if item.strip()}
def _unsent_retry_max_attempts():
raw_value = str(os.getenv("SPARKFLOW_UNSENT_RETRY_MAX_ATTEMPTS") or "3").strip()
try:
return max(1, int(raw_value))
except ValueError:
logger.warning("Invalid SPARKFLOW_UNSENT_RETRY_MAX_ATTEMPTS=%r, using 3", raw_value)
return 3
def _target_sent_today(user, target_name, now):
return target_is_strong_confirmed_today(user, target_name, now)
def _target_unconfirmed_today(user, target_name, now):
history = dict(user.get("message_history") or {})
entry = dict(history.get(target_name) or {})
sent_at = _parse_sent_at(entry.get("sentAt"), now.tzinfo)
return bool(
sent_at
and sent_at.date() == now.date()
and not _target_sent_today(user, target_name, now)
)
def _target_failed_today(user, target_name, now):
queue = dict(user.get("failure_queue") or {})
entry = queue.get(target_name) or {}
last_attempt_at = _parse_sent_at(entry.get("lastAttemptAt"), now.tzinfo)
return bool(last_attempt_at and last_attempt_at.date() == now.date())
def _account_failure_entry_today(user, now):
entry = dict(user.get("account_failure") or {})
last_attempt_at = _parse_sent_at(entry.get("lastAttemptAt"), now.tzinfo)
if last_attempt_at and last_attempt_at.date() == now.date():
return entry
return {}
def _account_failure_attempts_today(user, now):
return _coerce_attempt_count(_account_failure_entry_today(user, now))
def _account_failure_pause_after_attempts():
raw_value = str(os.getenv("SPARKFLOW_ACCOUNT_FAILURE_PAUSE_AFTER_ATTEMPTS") or "2").strip()
try:
return max(1, int(raw_value))
except ValueError:
logger.warning("Invalid SPARKFLOW_ACCOUNT_FAILURE_PAUSE_AFTER_ATTEMPTS=%r, using 2", raw_value)
return 2
def _temporary_account_failure_cooldown_minutes():
raw_value = str(os.getenv("SPARKFLOW_TEMP_ACCOUNT_FAILURE_COOLDOWN_MINUTES") or "60").strip()
try:
return max(1, int(raw_value))
except ValueError:
logger.warning("Invalid SPARKFLOW_TEMP_ACCOUNT_FAILURE_COOLDOWN_MINUTES=%r, using 60", raw_value)
return 60
def _account_paused_by_failure_today(user, now):
entry = _account_failure_entry_today(user, now)
if _coerce_attempt_count(entry) < _account_failure_pause_after_attempts():
return False
category = str(entry.get("category") or "")
if category not in TEMPORARY_ACCOUNT_FAILURE_CATEGORIES:
return True
last_attempt_at = _parse_sent_at(entry.get("lastAttemptAt"), now.tzinfo)
if not last_attempt_at:
return True
cooldown_seconds = _temporary_account_failure_cooldown_minutes() * 60
elapsed_seconds = (now - last_attempt_at).total_seconds()
if elapsed_seconds < cooldown_seconds:
remaining_seconds = int(cooldown_seconds - elapsed_seconds)
logger.info(
"Account %s temporary account failure category=%s cooling down for %ss",
user.get("username") or user.get("unique_id") or "unknown",
category,
remaining_seconds,
)
return True
logger.info(
"Account %s temporary account failure category=%s cooldown expired; allowing retry",
user.get("username") or user.get("unique_id") or "unknown",
category,
)
return False
def _friend_index_complete_today(user, now):
meta = dict(user.get("friend_index_meta") or {})
last_scan_at = _parse_sent_at(meta.get("lastScanAt"), now.tzinfo)
return bool(last_scan_at and last_scan_at.date() == now.date() and meta.get("lastScanComplete"))
def _target_failure_attempts_today(user, target_name, now):
queue = dict(user.get("failure_queue") or {})
entry = queue.get(target_name) or {}
last_attempt_at = _parse_sent_at(entry.get("lastAttemptAt"), now.tzinfo)
if not last_attempt_at or last_attempt_at.date() != now.date():
return 0
try:
return int(entry.get("attemptCount") or 0)
except (TypeError, ValueError):
return 0
def _target_failure_category_today(user, target_name, now):
queue = dict(user.get("failure_queue") or {})
entry = queue.get(target_name) or {}
last_attempt_at = _parse_sent_at(entry.get("lastAttemptAt"), now.tzinfo)
if not last_attempt_at or last_attempt_at.date() != now.date():
return ""
return str(entry.get("category") or "").strip()
def _target_has_non_retryable_failure_today(user, target_name, now):
return _target_failure_category_today(user, target_name, now) in {
"protocol_check_message_not_pass",
"protocol_check_message_self_visible",
"protocol_user_blocked",
"protocol_user_not_in_conversation",
"protocol_check_conversation_not_pass",
}
def _pending_failed_targets(user, now):
queue = dict(user.get("failure_queue") or {})
targets = []
account_failure = _account_failure_entry_today(user, now)
account_failure_targets = list(account_failure.get("affectedTargets") or [])
if account_failure_targets:
for target_name in account_failure_targets:
if target_name in (user.get("targets") or []) and not _target_sent_today(user, target_name, now):
targets.append(target_name)
if targets:
return targets
for target_name in user.get("targets") or []:
if _target_sent_today(user, target_name, now):
continue
if _target_unconfirmed_today(user, target_name, now):
targets.append(target_name)
continue
if target_name in queue and _target_failed_today(user, target_name, now):
targets.append(target_name)
return targets
def _pending_unsent_targets(user, now):
retry_targets = []
skipped_targets = []
max_attempts = _unsent_retry_max_attempts()
for target_name in user.get("targets") or []:
if _target_sent_today(user, target_name, now):
continue
if _target_has_non_retryable_failure_today(user, target_name, now):
skipped_targets.append(f"{target_name}(non_retryable)")
continue
attempts_today = _target_failure_attempts_today(user, target_name, now)
if attempts_today >= max_attempts:
skipped_targets.append(f"{target_name}({attempts_today})")
continue
retry_targets.append(target_name)
return retry_targets, skipped_targets
def _scheduled_send_time(user, target_name, send_window, now):
window_minutes = (send_window["endHour"] - send_window["startHour"]) * 60
start_of_window = now.replace(
hour=send_window["startHour"],
minute=0,
second=0,
microsecond=0,
)
seed = f"{now.date().isoformat()}|{_account_identity(user)}|{target_name}"
digest = hashlib.sha256(seed.encode("utf-8")).digest()
offset_minutes = int.from_bytes(digest[:8], "big") % window_minutes
return start_of_window + timedelta(minutes=offset_minutes)
def _select_due_targets(user, send_window, now):
targets = list(user.get("targets") or [])
if not send_window.get("enabled") or _is_manual_run():
return targets, [], [], []
window_start = now.replace(
hour=send_window["startHour"],
minute=0,
second=0,
microsecond=0,
)
window_end = now.replace(
hour=send_window["endHour"],
minute=0,
second=0,
microsecond=0,
)
window_grace_end = window_end + timedelta(minutes=send_window["scheduleIntervalMinutes"])
if now < window_start or now > window_grace_end:
already_sent = []
pending_targets = []
queued_failures = []
for target_name in targets:
if _target_sent_today(user, target_name, now):
already_sent.append(target_name)
continue
if _target_failed_today(user, target_name, now):
queued_failures.append(target_name)
continue
pending_targets.append((target_name, _scheduled_send_time(user, target_name, send_window, now)))
return [], already_sent, pending_targets, queued_failures
due_targets = []
already_sent = []
pending_targets = []
queued_failures = []
account_paused = _account_paused_by_failure_today(user, now)
for target_name in targets:
if _target_sent_today(user, target_name, now):
already_sent.append(target_name)
continue
if _target_failed_today(user, target_name, now):
queued_failures.append(target_name)
continue
scheduled_at = _scheduled_send_time(user, target_name, send_window, now)
if account_paused:
pending_targets.append((target_name, scheduled_at))
elif now >= scheduled_at:
due_targets.append(target_name)
else:
pending_targets.append((target_name, scheduled_at))
if account_paused and (pending_targets or queued_failures):
logger.info(
"Account %s is paused by account-level failure attempts=%s threshold=%s",
user.get("username") or user.get("unique_id") or "unknown",
_account_failure_attempts_today(user, now),
_account_failure_pause_after_attempts(),
)
return due_targets, already_sent, pending_targets, queued_failures
def _prepare_active_users_for_run(active_config, active_user_data):
schedule_tz = _schedule_timezone()
now = datetime.now(schedule_tz)
if _manual_run_failed_only():
logger.info("SPARKFLOW_MANUAL_RUN=1, retrying queued failures only")
runnable_users = []
for user in active_user_data:
retry_targets = _pending_failed_targets(user, now)
already_sent = [target for target in user.get("targets") or [] if _target_sent_today(user, target, now)]
logger.info(
"manual-retry user=%s retryTargetCount=%s strongConfirmedToday=%s",
user.get("username", "unknown"),
len(retry_targets),
len(already_sent),
)
if retry_targets:
runnable_user = dict(user)
runnable_user["targets"] = retry_targets
runnable_users.append(runnable_user)
if not runnable_users:
logger.info("No queued failures are pending for manual retry")
return runnable_users
if _manual_run_unsent_only():
logger.info(
"SPARKFLOW_MANUAL_RUN=1 and SPARKFLOW_MANUAL_UNSENT_ONLY=1, retrying today's unsent targets only"
)
runnable_users = []
for user in active_user_data:
retry_targets, skipped_targets = _pending_unsent_targets(user, now)
already_sent = [target for target in user.get("targets") or [] if _target_sent_today(user, target, now)]
logger.info(
"manual-unsent user=%s retryTargetCount=%s strongConfirmedToday=%s skippedCount=%s",
user.get("username", "unknown"),
len(retry_targets),
len(already_sent),
len(skipped_targets),
)
if retry_targets:
runnable_user = dict(user)
runnable_user["targets"] = retry_targets
runnable_users.append(runnable_user)
if not runnable_users:
logger.info("No unsent targets are pending for manual retry")
return runnable_users
if _is_manual_run():
logger.info("SPARKFLOW_MANUAL_RUN=1, bypassing daily send window")
return [dict(user, targets=list(user.get("targets") or [])) for user in active_user_data]
send_window = _normalize_send_window(active_config)
if not send_window.get("enabled"):
return [dict(user, targets=list(user.get("targets") or [])) for user in active_user_data]
logger.info(
"dailySendWindow enabled startHour=%s endHour=%s intervalMinutes=%s timezone=%s now=%s",
send_window["startHour"],
send_window["endHour"],
send_window["scheduleIntervalMinutes"],
getattr(schedule_tz, "key", str(schedule_tz)),
now.isoformat(timespec="seconds"),
)
runnable_users = []
for user in active_user_data:
due_targets, already_sent, pending_targets, queued_failures = _select_due_targets(user, send_window, now)
logger.info(
"windowed user=%s due=%s strongConfirmed=%s pending=%s queuedFailures=%s",
user.get("username", "unknown"),
len(due_targets),
len(already_sent),
len(pending_targets),
len(queued_failures),
)
if due_targets:
runnable_user = dict(user)
runnable_user["targets"] = due_targets
runnable_users.append(runnable_user)
if not runnable_users:
logger.info("No targets are due for the current windowed run")
return runnable_users
def _account_match_tokens(user):
tokens = set()
username = str(user.get("username") or "").strip()
unique_id = str(user.get("unique_id") or "").strip()
normalized_unique_id = normalize_unique_id(unique_id)
if username:
tokens.add(username.lower())
if unique_id:
tokens.add(unique_id.lower())
if normalized_unique_id:
tokens.add(normalized_unique_id.lower())
return tokens
def _find_matching_account(accounts, user):
target_username = str(user.get("username") or "").strip()
target_unique_id = normalize_unique_id(user.get("unique_id"))
if not target_username and not target_unique_id:
return None
for account in accounts:
account_username = str(account.get("username") or "").strip()
account_unique_id = normalize_unique_id(account.get("unique_id"))
if target_unique_id and account_unique_id == target_unique_id:
return account
if target_username and account_username == target_username:
return account
return None
def _coerce_attempt_count(entry):
try:
return int(dict(entry or {}).get("attemptCount") or 0)
except (TypeError, ValueError):
return 0
def _persist_account_send_failure(user, category, reason, attempted_at, affected_targets=None):
accounts = get_userData(force_reload=True)
matched_account = _find_matching_account(accounts, user)
if matched_account is None:
logger.warning(
"Could not find account to persist account-level browser failure for user=%s",
user.get("username", "unknown"),
)
return
affected_targets = list(affected_targets or [])
existing_entry = dict(matched_account.get("account_failure") or {})
entry = {
"category": category,
"reason": reason,
"firstAttemptAt": existing_entry.get("firstAttemptAt") or attempted_at,
"lastAttemptAt": attempted_at,
"attemptCount": _coerce_attempt_count(existing_entry) + 1,
"lastRunMode": _current_run_mode(),
"affectedTargets": affected_targets,
}
failure_queue = dict(matched_account.get("failure_queue") or {})
for target_name in affected_targets:
failure_queue.pop(target_name, None)
if failure_queue:
matched_account["failure_queue"] = failure_queue
else:
matched_account.pop("failure_queue", None)
matched_account["account_failure"] = entry
save_userData(accounts)
user["account_failure"] = dict(entry)
user_queue = dict(user.get("failure_queue") or {})
for target_name in affected_targets:
user_queue.pop(target_name, None)
if user_queue:
user["failure_queue"] = user_queue
else:
user.pop("failure_queue", None)
logger.warning(
"Paused browser sends for account %s category=%s affectedTargets=%s reason=%s",
matched_account.get("username", "unknown"),
category,
affected_targets,
reason,
)
def _clear_account_send_failure(user):
accounts = get_userData(force_reload=True)
matched_account = _find_matching_account(accounts, user)
changed = False
if matched_account is not None and matched_account.pop("account_failure", None) is not None:
changed = True
if changed:
save_userData(accounts)
user.pop("account_failure", None)
def _persist_friend_index(user, friend_records, scanned_at, *, scan_complete, missing_targets=None):
accounts = get_userData(force_reload=True)
matched_account = _find_matching_account(accounts, user)
if matched_account is None:
logger.warning(
"Could not find account to persist friend index for user=%s",
user.get("username", "unknown"),
)
return
existing_index = dict(matched_account.get("friend_index") or {})
for normalized_name, record in (friend_records or {}).items():
entry = dict(existing_index.get(normalized_name) or {})
entry.update(
{
"visibleName": record.get("visibleName") or "",
"normalizedName": normalized_name,
"stableKeys": list(record.get("stableKeys") or []),
"lastSeenAt": scanned_at,
}
)
existing_index[normalized_name] = entry
meta = {
"lastScanAt": scanned_at,
"lastScanComplete": bool(scan_complete),
"scannedCount": len(friend_records or {}),
"missingTargets": list(missing_targets or []),
}
matched_account["friend_index"] = existing_index
matched_account["friend_index_meta"] = meta
if scan_complete and friend_records:
matched_account.pop("account_failure", None)
save_userData(accounts)
user["friend_index"] = dict(existing_index)
user["friend_index_meta"] = dict(meta)
if scan_complete and friend_records:
user.pop("account_failure", None)
logger.info(
"Persisted friend index for %s scanned=%s complete=%s missingTargets=%s",
matched_account.get("username", "unknown"),
len(friend_records or {}),
bool(scan_complete),
list(missing_targets or []),
)
def _persist_browser_send_failure(user, target_name, message, category, reason, attempted_at, server_receipt=None):
accounts = get_userData(force_reload=True)
matched_account = _find_matching_account(accounts, user)
if matched_account is None:
logger.warning(
"Could not find account to persist browser send failure for user=%s target=%s",
user.get("username", "unknown"),
target_name,
)
return
queue = dict(matched_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 attempted_at,
"lastAttemptAt": attempted_at,
"attemptCount": _coerce_attempt_count(existing_entry) + 1,
"lastRunMode": _current_run_mode(),
}
if server_receipt:
queue[target_name]["serverReceipt"] = server_receipt
matched_account["failure_queue"] = queue
save_userData(accounts)
user_queue = dict(user.get("failure_queue") or {})
user_queue[target_name] = dict(queue[target_name])
user["failure_queue"] = user_queue
logger.warning(
"Queued failed browser send for %s/%s category=%s reason=%s",
matched_account.get("username", "unknown"),
target_name,
category,
reason,
)
def _persist_browser_send_success(user, target_name, message, sent_at, server_receipt=None):
accounts = get_userData(force_reload=True)
matched_account = _find_matching_account(accounts, user)
if matched_account is None:
logger.warning(
"Could not find account to persist browser send history for user=%s target=%s",
user.get("username", "unknown"),
target_name,
)
return
receipt_summary = ""
if isinstance(server_receipt, dict):
receipt_summary = "message_send http={} logid={}".format(
server_receipt.get("httpStatus"),
server_receipt.get("logid") or "",
)
strong_entry = {
"message": message,
"sentAt": sent_at,
"status": "confirmed",
"confirmationLevel": "strong",
"confirmationSource": "cdp_message_send_receipt" if server_receipt else "browser_visible_count_increased",
"confirmationDetail": receipt_summary,
"needsVerification": False,
}
if server_receipt:
strong_entry["serverReceipt"] = server_receipt
history = dict(matched_account.get("message_history") or {})
history[target_name] = strong_entry
matched_account["message_history"] = history
queue = dict(matched_account.get("failure_queue") or {})
queue.pop(target_name, None)
if queue:
matched_account["failure_queue"] = queue
else:
matched_account.pop("failure_queue", None)
matched_account.pop("account_failure", None)
save_userData(accounts)
user_history = dict(user.get("message_history") or {})
user_history[target_name] = dict(strong_entry)
user["message_history"] = user_history
user_queue = dict(user.get("failure_queue") or {})
user_queue.pop(target_name, None)
if user_queue:
user["failure_queue"] = user_queue
else:
user.pop("failure_queue", None)
user.pop("account_failure", None)
logger.info(
"Persisted browser send history for %s/%s at %s",
matched_account.get("username", "unknown"),
target_name,
sent_at,
)
def _split_sender_modes(active_config, runnable_user_data):
if not active_config.get("useProtocolSender", True):
return [], runnable_user_data
browser_sender_accounts = {
str(item).strip().lower()
for item in (active_config.get("browserSenderAccounts") or [])
if str(item).strip()
}
if not browser_sender_accounts:
return runnable_user_data, []
protocol_users = []
browser_users = []
for user in runnable_user_data:
if _account_match_tokens(user) & browser_sender_accounts:
browser_users.append(user)
else:
protocol_users.append(user)
return protocol_users, browser_users
def _pid_is_alive(pid):
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
except OSError as exc:
if getattr(exc, "winerror", None) == 87 or exc.errno == errno.ESRCH:
return False
if exc.errno in (errno.EPERM, errno.EACCES):
return True
raise
return True
def _extract_lock_pid(raw):
for line in str(raw or "").splitlines():
if line.startswith("pid="):
try:
return int(line.split("=", 1)[1].strip())
except (TypeError, ValueError):
return None
try:
return int(str(raw or "").strip())
except (TypeError, ValueError):
return None
def _browser_account_lock_is_stale(lock_path, raw):
pid = _extract_lock_pid(raw)
if pid is not None and not _pid_is_alive(pid):
return True, f"missing pid={pid}"
try:
age_seconds = datetime.now(timezone.utc).timestamp() - lock_path.stat().st_mtime
except OSError:
return True, "missing lock file"
if age_seconds > 7200:
return True, f"older than 7200s pid={pid}"
return False, ""
async def _acquire_browser_account_lock(user, account_name):
lock_dir = Path("logs/browser-account-locks")
lock_dir.mkdir(parents=True, exist_ok=True)
identity = _account_identity(user) or account_name
lock_path = lock_dir / f"{_safe_name(identity)}.lock"
started_at = asyncio.get_running_loop().time()
last_logged_at = 0
while True:
try:
handle = lock_path.open("x", encoding="utf-8")
handle.write(
f"pid={os.getpid()}\n"
f"account={account_name}\n"
f"createdAt={datetime.now(timezone.utc).isoformat(timespec='seconds')}\n"
)
handle.flush()
logger.debug("Acquired browser account lock for %s at %s", account_name, lock_path)
return handle, lock_path
except FileExistsError:
raw = lock_path.read_text(encoding="utf-8", errors="ignore")
is_stale, stale_reason = _browser_account_lock_is_stale(lock_path, raw)
if is_stale:
logger.warning(
"Removing stale browser account lock for %s at %s: %s",
account_name,
lock_path,
stale_reason,
)
try:
lock_path.unlink()
except FileNotFoundError:
pass
continue
now = asyncio.get_running_loop().time()
if now - started_at > 7200:
raise RuntimeError(f"timed out waiting for browser account lock for {account_name}")
if now - last_logged_at >= 30:
logger.info(
"Waiting for existing browser account lock for %s at %s",
account_name,
lock_path,
)
last_logged_at = now
await asyncio.sleep(5)
def _release_browser_account_lock(handle, lock_path, account_name):
try:
handle.close()
finally:
try:
lock_path.unlink()
logger.debug("Released browser account lock for %s at %s", account_name, lock_path)
except FileNotFoundError:
pass
def _browser_account_timeout_seconds(friend_scan_config, target_count):
raw_value = str(os.getenv("SPARKFLOW_BROWSER_ACCOUNT_TIMEOUT_SECONDS") or "").strip()
if raw_value:
try:
return max(300, int(raw_value))
except ValueError:
logger.warning(
"Invalid SPARKFLOW_BROWSER_ACCOUNT_TIMEOUT_SECONDS=%r, using calculated timeout",
raw_value,
)
scan_seconds = int((friend_scan_config or {}).get("maxScanSeconds") or 300)
# Bound one account run even if Playwright or the page wedges below our selector timeouts.
return max(900, scan_seconds + 420 + max(1, target_count) * 240)
async def run_browser_tasks(active_config, browser_user_data):
if not browser_user_data:
return
send_strategy = _normalize_send_strategy(active_config)
friend_scan_config = _normalize_friend_list_scan_config(active_config)
profile_config = _normalize_persistent_profile_config(active_config)
semaphore = asyncio.Semaphore(active_config["taskCount"] if active_config["multiTask"] else 1)
tasks = []
if profile_config["enabled"]:
logger.info(
"Browser sender persistent profiles enabled root=%s seedCookiesWhenEmpty=%s syncStoredCookiesBeforeRun=%s refreshStoredCookiesAfterLogin=%s",
profile_config["root"],
profile_config["seedCookiesWhenEmpty"],
profile_config["syncStoredCookiesBeforeRun"],
profile_config["refreshStoredCookiesAfterLogin"],
)
for user in browser_user_data:
logger.info(
"Using persistent browser sender for user=%s targetCount=%s",
user.get("username", "unknown"),
len(user["targets"]),
)
tasks.append(do_user_task(None, user, semaphore, send_strategy, profile_config, friend_scan_config))
await asyncio.gather(*tasks)
return
playwright, browser = await get_browser()
try:
for user in browser_user_data:
logger.info(
"Using browser sender for user=%s targetCount=%s",
user.get("username", "unknown"),
len(user["targets"]),
)
tasks.append(do_user_task(browser, user, semaphore, send_strategy, profile_config, friend_scan_config))
await asyncio.gather(*tasks)
finally:
await playwright.stop()
await browser.close()
async def do_user_task(browser, user, semaphore, send_strategy, profile_config, friend_scan_config):
async with semaphore:
account_name = user.get("username", "unknown")
account_lock_handle = None
account_lock_path = None
try:
account_lock_handle, account_lock_path = await _acquire_browser_account_lock(user, account_name)
timeout_seconds = _browser_account_timeout_seconds(
friend_scan_config,
len(user.get("targets") or []),
)
logger.info(
"Account %s browser sender timeout guard is %ss",
account_name,
timeout_seconds,
)
try:
await asyncio.wait_for(
_do_user_task_locked(
browser,
user,
send_strategy,
profile_config,
friend_scan_config,
account_name,
),
timeout=timeout_seconds,
)
except asyncio.TimeoutError:
attempted_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
reason = f"browser sender exceeded {timeout_seconds}s timeout guard"
logger.exception("Account %s browser sender timed out", account_name)
for target_name in user.get("targets") or []:
_persist_browser_send_failure(
user,
target_name,
"",
"timeout",
reason,
attempted_at,
)
finally:
if account_lock_handle is not None:
_release_browser_account_lock(account_lock_handle, account_lock_path, account_name)
async def _do_user_task_locked(browser, user, send_strategy, profile_config, friend_scan_config, account_name):
cookies = user["cookies"]
targets = user["targets"]
start_delay = _random_delay_seconds(
send_strategy,
"accountStartDelaySecondsMin",
"accountStartDelaySecondsMax",
)
await _sleep_with_log(start_delay, "Delaying browser sender start", account_name)
owned_playwright = None
profile_dir = None
if profile_config["enabled"]:
owned_playwright, context, profile_dir = await get_persistent_browser_context(
_account_profile_name(user),
root=profile_config["root"],
)
logger.info("Opened persistent browser profile for %s at %s", account_name, profile_dir)
if profile_config["syncStoredCookiesBeforeRun"]:
await apply_stored_cookies_to_profile(context, cookies, account_name)
elif profile_config["seedCookiesWhenEmpty"]:
await apply_stored_cookies_to_profile(context, cookies, account_name, only_when_empty=True)
else:
context = await browser.new_context()
context.set_default_navigation_timeout(120000)
context.set_default_timeout(120000)
yielded_targets = set()
page = None
try:
page = await context.new_page()
try:
await retry_operation(
"open creator home",
page.goto,
retries=3,
delay=5,
url=CREATOR_HOME_URL,
)
if not profile_config["enabled"]:
await context.add_cookies(cookies)
await retry_operation(
"open chat page",
page.goto,
retries=3,
delay=5,
url=CREATOR_CHAT_URL,
)
await asyncio.sleep(3)
await ensure_not_login_required(page, account_name, "open_chat_page")
if profile_config["enabled"] and profile_config["refreshStoredCookiesAfterLogin"]:
await refresh_stored_cookies_from_profile(context, user, account_name)
except Exception as exc:
attempted_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
category = classify_browser_failure("open_chat_page", exc)
reason = str(exc)
logger.exception("Account %s failed before target delivery", account_name)
if _is_account_level_failure_category(category):
_persist_account_send_failure(user, category, reason, attempted_at, targets)
else:
for target_name in targets:
_persist_browser_send_failure(user, target_name, "", category, reason, attempted_at)
return
logger.info("Account %s started the message flow", account_name)
try:
index_targets = targets
last_message = ""
schedule_now = datetime.now(_schedule_timezone())
if not _friend_index_complete_today(user, schedule_now):
index_targets = list(user.get("targets") or targets)
logger.info(
"Account %s will refresh today's friend index while delivering targetCount=%s",
account_name,
len(targets),
)
async for target_name in scroll_and_select_user(
page,
user,
account_name,
targets,
friend_scan_config,
index_targets=index_targets,
):
yielded_targets.add(target_name)
message = ""
chat_input = None
last_own_message_before = None
send_receipt = {}
try:
await save_debug_artifacts(page, account_name, target_name, "selected-friend")
chat_input, selector_used = await locate_chat_input(page)
logger.info("Using chat input selector %s for %s/%s", selector_used, account_name, target_name)
previous_entry = dict(user.get("message_history") or {}).get(target_name) or {}
previous_message = str(previous_entry.get("message") or "")
message = build_message(previous_message=previous_message, last_message=last_message)
last_message = message
logger.info(
"Prepared message for %s/%s length=%s previousMatch=%s",
account_name,
target_name,
len(message),
bool(previous_message and previous_message == message),
)
last_own_message_before = await snapshot_last_own_message(
page,
chat_input=chat_input,
)
logger.info(
"Last own message before send for %s/%s length=%s",
account_name,
target_name,
len((last_own_message_before or {}).get("text", "")),
)
lines = message.split("\n")
for index, line in enumerate(lines):
await chat_input.type(line, delay=50)
if index < len(lines) - 1:
await chat_input.press("Shift+Enter")
await save_debug_artifacts(page, account_name, target_name, "typed-message")
im_observer_summary = await start_im_send_observer(page, account_name, target_name)
logger.info("Pressing Enter to send message for %s/%s", account_name, target_name)
await chat_input.press("Enter")
sent_ok, detail = await confirm_message_sent(
page,
chat_input,
message,
before_snapshot=last_own_message_before,
)
im_summary = await im_observer_summary()
logger.info(
"IM send observer summary for %s/%s: request=%s response=%s events=%s error=%s",
account_name,
target_name,
im_summary.get("send_request_seen"),
im_summary.get("send_response_seen"),
im_summary.get("events"),
im_summary.get("error", ""),
)
send_receipt = im_summary.get("send_receipt") or {}
server_ok = bool(send_receipt.get("ok"))
detail = (
f"{detail}; im_observer request={im_summary.get('send_request_seen')} "
f"response={im_summary.get('send_response_seen')} serverOk={server_ok} "
f"logid={send_receipt.get('logid', '')} reason={send_receipt.get('reason', '')}"
)
await save_debug_artifacts(page, account_name, target_name, "after-send")
if im_summary.get("enabled"):
if not im_summary.get("send_request_seen"):
raise RuntimeError(f"server send request was not observed; {detail}")
if not im_summary.get("send_response_seen"):
raise RuntimeError(f"server send response was not observed; {detail}")
if not server_ok:
raise RuntimeError(f"server send receipt rejected; {detail}")
elif not sent_ok:
raise RuntimeError(detail)
logger.info("Message send confirmed for %s/%s by server receipt: %s", account_name, target_name, detail)
_persist_browser_send_success(
user,
target_name,
message,
datetime.now(timezone.utc).isoformat(timespec="seconds"),
server_receipt=send_receipt,
)
interval = _random_delay_seconds(
send_strategy,
"messageIntervalSecondsMin",
"messageIntervalSecondsMax",
)
await _sleep_with_log(interval, "Delaying next browser message", account_name)
except Exception as exc:
sent_ok = False
detail = ""
if message:
sent_ok, detail = await detect_message_already_sent(
page,
chat_input,
message,
before_snapshot=last_own_message_before,
)
if sent_ok and not str(exc).startswith("server send"):
logger.warning(
"Recovered send outcome for %s/%s after failure: %s",
account_name,
target_name,
detail,
)
_persist_browser_send_success(
user,
target_name,
message,
datetime.now(timezone.utc).isoformat(timespec="seconds"),
server_receipt=send_receipt,
)
continue
logger.exception("Send flow failed for %s/%s", account_name, target_name)
await save_debug_artifacts(page, account_name, target_name, "send-error")
category = classify_browser_failure("send_flow", exc)
reason = str(exc)
is_login_required, login_detail = await _page_has_login_required_prompt(page)
if is_login_required:
category = "login_required"
reason = f"{reason}\n{login_detail}"
logger.warning(
"Account %s hit login_required during send flow: %s",
account_name,
login_detail,
)
if _is_stop_account_category(category):
affected_targets = [
target_name,
*[target for target in targets if target not in yielded_targets],
]
attempted_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
_persist_account_send_failure(
user,
category,
reason,
attempted_at,
affected_targets,
)
return
_persist_browser_send_failure(
user,
target_name,
message,
category,
reason,
datetime.now(timezone.utc).isoformat(timespec="seconds"),
server_receipt=send_receipt,
)
interval = _random_delay_seconds(
send_strategy,
"messageIntervalSecondsMin",
"messageIntervalSecondsMax",
)
await _sleep_with_log(interval, "Delaying next browser message after failure", account_name)
except Exception as exc:
remaining_targets = [target for target in targets if target not in yielded_targets]
attempted_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
category = classify_browser_failure("friend_list", exc)
reason = str(exc)
is_login_required, login_detail = await _page_has_login_required_prompt(page)
if is_login_required:
category = "login_required"
reason = f"{reason}\n{login_detail}"
logger.exception("Target selection failed for %s", account_name)
if _is_account_level_failure_category(category):
_persist_account_send_failure(user, category, reason, attempted_at, remaining_targets)
else:
for target_name in remaining_targets:
_persist_browser_send_failure(user, target_name, "", category, reason, attempted_at)
return
missing_targets = [target for target in targets if target not in yielded_targets]
if missing_targets:
attempted_at = datetime.now(timezone.utc).isoformat(timespec="seconds")
for target_name in missing_targets:
_persist_browser_send_failure(
user,
target_name,
"",
"friend_not_found",
"target not selected from friend list",
attempted_at,
)
finally:
if page is not None:
try:
await page.close()
except Exception:
pass
await context.close()
if owned_playwright is not None:
await owned_playwright.stop()
async def runTasks():
active_config = get_config(force_reload=True)
all_user_data = get_userData(force_reload=True)
requested_refs = _requested_account_refs()
if requested_refs is not None:
all_user_data = [user for user in all_user_data if user.get("account_ref") in requested_refs]
active_user_data = [user for user in all_user_data if user.get("enabled", True)]
disabled_user_data = [user for user in all_user_data if not user.get("enabled", True)]
logger.info("Starting tasks with config")
logger.info("multiTask=%s taskCount=%s", active_config["multiTask"], active_config["taskCount"])
send_strategy = active_config.get("sendStrategy", {}) or {}
logger.info(
"messageConfig templateConfigured=%s variantCount=%s shuffleTargets=%s",
bool(str(active_config.get("messageTemplate") or "").strip()),
len(send_strategy.get("messageVariants") or []),
bool(send_strategy.get("shuffleTargets", True)),
)
logger.info("hitokotoTypeCount=%s", len(active_config.get("hitokotoTypes") or []))
logger.info("enabledUsers=%s disabledUsers=%s", len(active_user_data), len(disabled_user_data))
for user in active_user_data:
logger.info("user=%s targetCount=%s", user.get("username", "unknown"), len(user["targets"]))
for user in disabled_user_data:
logger.info("skipping disabled user=%s", user.get("username", "unknown"))
if not active_user_data:
logger.warning("No enabled accounts are available for the task run")
return
runnable_user_data = _prepare_active_users_for_run(active_config, active_user_data)
if not runnable_user_data:
return
try:
with task_run_lock():
protocol_user_data, browser_user_data = _split_sender_modes(active_config, runnable_user_data)
if protocol_user_data:
await run_protocol_tasks(active_config, protocol_user_data, build_message)
await run_browser_tasks(active_config, browser_user_data)
except TaskRunAlreadyInProgress:
logger.warning("Skipping task run because another task run is already in progress")
class TaskRunAlreadyInProgress(RuntimeError):
"""Raised when a live task process already owns the global run lock."""
@contextmanager
def task_run_lock():
lock_path = Path("logs/task.run.lock")
lock_path.parent.mkdir(parents=True, exist_ok=True)
def _lock_owner_is_alive(pid):
return _pid_is_alive(pid)
while True:
try:
handle = lock_path.open("x", encoding="utf-8")
break
except FileExistsError as exc:
raw_pid = lock_path.read_text(encoding="utf-8", errors="ignore").strip()
stale_pid = None
try:
stale_pid = int(raw_pid)
except (TypeError, ValueError):
stale_pid = None
if stale_pid is not None and not _lock_owner_is_alive(stale_pid):
logger.warning("Removing stale task lock owned by missing pid=%s", stale_pid)
try:
lock_path.unlink()
except FileNotFoundError:
pass
continue
if stale_pid is None:
logger.warning("Removing unreadable stale task lock with contents=%r", raw_pid)
try:
lock_path.unlink()
except FileNotFoundError:
pass
continue
raise TaskRunAlreadyInProgress("another task run is already in progress") from exc
try:
handle.write(f"{os.getpid()}\n")
handle.flush()
yield
finally:
handle.close()
try:
lock_path.unlink()
except FileNotFoundError:
pass