import os
import sys
import json
import asyncio
import logging
import uuid
from datetime import datetime, timezone
from io import BytesIO
from typing import List, Optional

if sys.platform == "win32":
    asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks, UploadFile, File, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.responses import Response, StreamingResponse
from urllib.parse import urlparse
import httpx
from settings import CORS_ORIGINS, SERVE_WEB, STATIC_DIR
from help_pages import serve_credential_tool
from web_static import mount_frontend
from pydantic import BaseModel, Field
from sqlalchemy import select, update, delete, text, func
from sqlalchemy.ext.asyncio import AsyncSession

from models.database import engine, Base, get_db, AsyncSessionLocal
from models.db_migrate import (
    migrate_accounts_quota_disabled as _migrate_accounts_quota_disabled,
    migrate_accounts_table as _migrate_accounts_table,
    migrate_account_videos_table as _migrate_account_videos_table,
    migrate_message_logs_table as _migrate_message_logs_table,
    migrate_payment_orders_table as _migrate_payment_orders_table,
    migrate_rules_table as _migrate_rules_table,
    migrate_users_table as _migrate_users_table,
)
from models.db_config import database_config_to_response
from models.models import Account, AccountProfileDetail, AccountVideo, AutoReplyRule, MessageLog, ReceivedMessageLog, SystemLog, User
from auth.router import router as auth_router, users_router
from auth.settings_router import router as settings_router
from payments.router import router as payments_router
from desktop_router import router as desktop_router
from link_cards_router import router as link_cards_router, UPLOAD_DIR as LINK_CARD_UPLOAD_DIR
from auth.dependencies import get_current_user, require_admin, require_write
from auth.account_limits import ensure_can_add_account
from auth.scopes import (
    accounts_for_user,
    get_owned_account,
    get_accessible_rule,
    logs_for_user,
    rules_for_user,
    owned_account_ids,
    received_logs_for_user,
    system_logs_for_user,
)
from auth.roles import is_admin
from auth.passwords import hash_password
from rpa_engine.batch_start import BatchStartQueue
from rpa_engine.playwright_worker import DouyinWorker
from rpa_engine.douyin_im.session import DouyinImSession
from rpa_engine.douyin_im.http_client import DouyinImHttpClient
from rpa_engine.douyin_im.conv_util import build_conversation_id, normalize_conversation_id, resolve_peer_uid
from rpa_engine.douyin_im.message_content import (
    message_preview,
    normalize_outgoing_content,
    parse_stored_content,
)
from rpa_engine.credential import (
    assess_account_credential,
    build_im_session_from_storage,
    build_cookie_credential_detail,
)
from utils.cookie_store import (
    write_cookie_file,
    read_cookie_file,
    clear_cookie_file,
    get_cookie_path,
    cookie_summary,
    validate_cookie_json,
    analyze_cookie,
)
from rpa_engine.device_profiles import list_device_profiles, profile_label_for_ua, resolve_user_agent

logger = logging.getLogger("main")
from utils import system_logger

app = FastAPI(title="抖音多账号自动回复管理系统 API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# RPA 任务管理器
class WorkerManager:
    def __init__(self):
        self.workers = {} # account_id -> DouyinWorker
        self._account_locks: dict[int, asyncio.Lock] = {}
        self._preparation_locks: dict[int, asyncio.Lock] = {}

    def _account_lock(self, account_id: int) -> asyncio.Lock:
        return self._account_locks.setdefault(int(account_id), asyncio.Lock())

    def preparation_lock(self, account_id: int) -> asyncio.Lock:
        """Serialize credential preparation across single and batch starts."""
        return self._preparation_locks.setdefault(int(account_id), asyncio.Lock())

    async def start_worker(self, account_id: int, login_mode: str = "auto"):
        async with self._account_lock(account_id):
            if account_id in self.workers:
                worker = self.workers[account_id]
                if worker.is_running:
                    return False
                del self.workers[account_id]

            worker = DouyinWorker(account_id, login_mode=login_mode)
            self.workers[account_id] = worker
            await worker.start()
            return True

    async def stop_worker(self, account_id: int):
        async with self._account_lock(account_id):
            if account_id in self.workers:
                worker = self.workers[account_id]
                await worker.stop()
                if self.workers.get(account_id) is worker:
                    del self.workers[account_id]
                return True
            return False

    def is_running(self, account_id: int) -> bool:
        worker = self.workers.get(account_id)
        return worker.is_running if worker else False

manager = WorkerManager()

UPLOAD_DIR = os.path.join(os.path.dirname(__file__), "uploads", "messages")
os.makedirs(UPLOAD_DIR, exist_ok=True)
app.mount("/api/media/messages", StaticFiles(directory=UPLOAD_DIR), name="message-media")

os.makedirs(LINK_CARD_UPLOAD_DIR, exist_ok=True)
app.mount("/api/media/link-cards", StaticFiles(directory=LINK_CARD_UPLOAD_DIR), name="link-card-media")

app.include_router(auth_router)
app.include_router(users_router)
app.include_router(settings_router)
app.include_router(payments_router)
app.include_router(desktop_router)
app.include_router(link_cards_router)


@app.get("/help/credential-tool", include_in_schema=False)
async def help_credential_tool_page():
    """凭证采集工具（Playwright storage_state 本地生成页）。"""
    return serve_credential_tool()


@app.get("/api/help/credential-tool", include_in_schema=False)
async def api_help_credential_tool_page():
    """同上，兼容前端开发代理（/api -> 8800）。"""
    return serve_credential_tool()


@app.get("/api/health")
async def health_check():
    """健康检查，供宝塔/Nginx/监控探活。"""
    web_ready = STATIC_DIR.is_dir() and (STATIC_DIR / "index.html").is_file()
    return {
        "status": "ok",
        "web_ui": web_ready,
        "serve_web": SERVE_WEB,
    }


_MEDIA_PROXY_HOSTS = (
    "douyinpic.com",
    "douyin.com",
    "byteimg.com",
    "ibyteimg.com",
    "snssdk.com",
    "amemv.com",
    "douyinstatic.com",
    "bytednsdoc.com",
)
try:
    _MEDIA_PROXY_MAX_BYTES = max(
        1024 * 1024,
        int(os.getenv("KEFU_MEDIA_PROXY_MAX_BYTES", str(20 * 1024 * 1024))),
    )
except (TypeError, ValueError):
    _MEDIA_PROXY_MAX_BYTES = 20 * 1024 * 1024
def _is_allowed_media_host(hostname: str) -> bool:
    host = str(hostname or "").lower().rstrip(".")
    return any(host == allowed or host.endswith(f".{allowed}") for allowed in _MEDIA_PROXY_HOSTS)


@app.get("/api/media/proxy")
async def proxy_media(
    url: str = Query(..., min_length=8, max_length=4096),
):
    """代理抖音 CDN 图片/媒体，解决浏览器 Referer 限制导致无法预览。

    说明：浏览器 <img>/<audio> 标签直接请求该接口时无法携带 JWT，故此处不做登录鉴权；
    安全性由下方的「仅允许抖音 CDN 域名」白名单保证，不会被当作通用代理滥用。
    """
    parsed = urlparse(url.strip())
    if parsed.scheme not in ("http", "https") or not parsed.netloc:
        raise HTTPException(status_code=400, detail="无效媒体地址")
    if not _is_allowed_media_host(parsed.hostname or ""):
        raise HTTPException(status_code=400, detail="不支持的媒体域名")
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": "https://www.douyin.com/",
    }
    try:
        from rpa_engine.douyin_im.traffic_control import get_traffic_controller

        async with get_traffic_controller().media_proxy_slot():
            async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:
                async with client.stream("GET", url, headers=headers) as resp:
                    resp.raise_for_status()
                    if not _is_allowed_media_host(resp.url.host or ""):
                        raise HTTPException(status_code=400, detail="媒体地址跳转到非白名单域名")
                    content_length = int(resp.headers.get("content-length") or 0)
                    if content_length > _MEDIA_PROXY_MAX_BYTES:
                        raise HTTPException(status_code=413, detail="媒体文件过大，无法代理预览")
                    content_type = resp.headers.get("content-type") or ""
                    content_buffer = bytearray()
                    async for chunk in resp.aiter_bytes():
                        content_buffer.extend(chunk)
                        if len(content_buffer) > _MEDIA_PROXY_MAX_BYTES:
                            raise HTTPException(status_code=413, detail="媒体文件过大，无法代理预览")
                    content = bytes(content_buffer)
    except HTTPException:
        raise
    except Exception as exc:
        raise HTTPException(status_code=502, detail=f"媒体加载失败: {exc}") from exc
    # 抖音 IM 图片 CDN 常返回 application/octet-stream，浏览器 <img> 不渲染；
    # 按文件头嗅探真实类型并改正，否则前端只显示破图。
    if not content_type or "octet-stream" in content_type.lower() or content_type.lower().startswith("application/"):
        sniffed = _sniff_media_type(content)
        if sniffed:
            content_type = sniffed
        elif not content_type:
            content_type = "application/octet-stream"
    return Response(content=content, media_type=content_type)


def _sniff_media_type(data: bytes) -> str:
    if not data or len(data) < 12:
        return ""
    if data[:3] == b"\xff\xd8\xff":
        return "image/jpeg"
    if data[:8] == b"\x89PNG\r\n\x1a\n":
        return "image/png"
    if data[:4] in (b"GIF8",):
        return "image/gif"
    if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
        return "image/webp"
    if data[:2] == b"BM":
        return "image/bmp"
    if data[:4] == b"\x00\x00\x01\x00":
        return "image/x-icon"
    # 音频/视频常见
    if data[4:8] == b"ftyp":
        return "video/mp4"
    if data[:3] == b"ID3" or data[:2] == b"\xff\xfb":
        return "audio/mpeg"
    if data[:4] == b"OggS":
        return "audio/ogg"
    return ""


# 系统诊断日志：内存缓冲区 -> 数据库的后台落库任务
_system_log_flush_task: Optional[asyncio.Task] = None


async def _seed_system_logs():
    """启动时把最近的系统日志读回内存缓冲区。"""
    try:
        async with AsyncSessionLocal() as db:
            stmt = select(SystemLog).order_by(SystemLog.id.desc()).limit(500)
            rows = (await db.execute(stmt)).scalars().all()
            system_logger.seed([
                {
                    "id": r.id,
                    "account_id": r.account_id,
                    "level": r.level,
                    "category": r.category,
                    "event": r.event,
                    "detail": r.detail or "",
                    "created_at": (r.created_at or datetime.utcnow()).isoformat(),
                }
                for r in rows
            ])
    except Exception as e:
        print(f"Seed system logs failed: {e}")


async def _flush_system_logs_loop():
    """周期性地把内存中的诊断日志落库以便持久化。"""
    while True:
        try:
            await asyncio.sleep(2)
            pending = system_logger.drain_pending()
            if not pending:
                continue
            async with AsyncSessionLocal() as db:
                for e in pending:
                    try:
                        created = datetime.fromisoformat(e["created_at"])
                    except Exception:
                        created = datetime.utcnow()
                    db.add(SystemLog(
                        account_id=e.get("account_id"),
                        level=e.get("level") or "info",
                        category=e.get("category") or "system",
                        event=(e.get("event") or "")[:255],
                        detail=e.get("detail") or "",
                        created_at=created,
                    ))
                await db.commit()
        except asyncio.CancelledError:
            break
        except Exception as exc:
            print(f"Flush system logs failed: {exc}")


async def _sync_legacy_cookie_files():
    """将旧版仅保存在文件的 Cookie 同步到数据库"""
    async with AsyncSessionLocal() as db:
        result = await db.execute(select(Account))
        accounts = result.scalars().all()
        changed = False
        for acc in accounts:
            if acc.cookie_data:
                continue
            file_data = read_cookie_file(acc.id)
            if file_data:
                acc.cookie_data = file_data
                acc.cookie_path = get_cookie_path(acc.id)
                acc.cookie_updated_at = datetime.utcnow()
                changed = True
        if changed:
            await db.commit()


def _account_has_cookie(account: Account) -> bool:
    if account.cookie_data:
        return True
    if account.cookie_path and os.path.exists(account.cookie_path):
        return True
    return os.path.exists(get_cookie_path(account.id))


def _build_account_im_session(account: Account) -> DouyinImSession:
    cookie_data = _get_account_cookie_data(account)
    storage = json.loads(cookie_data) if cookie_data else {}
    session = build_im_session_from_storage(storage, account.im_session_data)
    session.user_agent = resolve_user_agent(account.user_agent or session.user_agent)
    return session


def _resolve_log_peer_id(session: DouyinImSession, conversation_id: str) -> str:
    """手动发送日志统一存粉丝 UID，便于前端按 peer 合并会话。"""
    my_uid = int(session.my_uid or 0)
    if not my_uid:
        try:
            from rpa_engine.douyin_im.auth import DouyinAuth

            auth = DouyinAuth.from_im_session(session)
            my_uid = int(auth.get_uid() or 0)
        except Exception:
            my_uid = 0
    peer_uid = resolve_peer_uid(conversation_id, my_uid)
    if peer_uid:
        return str(peer_uid)
    return (conversation_id or "").strip()


async def _conversations_from_logs(
    db: AsyncSession,
    account_id: int,
    my_uid: int = 0,
) -> list[dict]:
    """从消息日志聚合会话列表（兜底）。"""
    stmt = (
        select(MessageLog)
        .where(MessageLog.account_id == account_id)
        .order_by(MessageLog.created_at.desc())
        .limit(500)
    )
    result = await db.execute(stmt)
    logs = result.scalars().all()
    seen: dict[str, dict] = {}
    for log in logs:
        if log.sender_name == "[系统发送]":
            continue
        raw_id = (log.sender_id or "").strip()
        name = (log.sender_name or "").strip()
        conv_id = normalize_conversation_id(raw_id, my_uid) if my_uid else raw_id
        if not conv_id and name.isdigit() and my_uid:
            conv_id = build_conversation_id(my_uid, int(name))
        if not conv_id and name.isdigit():
            conv_id = name
        key = conv_id or name
        peer_uid = raw_id if raw_id.isdigit() else ""
        if not peer_uid and conv_id and my_uid:
            from rpa_engine.douyin_im.conv_util import resolve_peer_uid

            resolved = resolve_peer_uid(conv_id, int(my_uid))
            if resolved:
                peer_uid = str(resolved)
        if not key or key in seen:
            continue
        seen[key] = {
            "conversation_id": conv_id,
            "sender_name": name or (f"用户{conv_id[-6:]}" if conv_id.isdigit() else "未知用户"),
            "sender_id": peer_uid or conv_id or raw_id or None,
            "peer_uid": peer_uid,
            "sender_avatar": (log.sender_avatar or "").strip() or None,
            "content": message_preview(log.message_content or ""),
            "unread_count": 0,
        }
    return list(seen.values())


async def _reset_account_credentials(account_id: int, db: AsyncSession) -> Account:
    """清除账号 Cookie、IM 会话等登录数据，并停止托管。"""
    await manager.stop_worker(account_id)
    result = await db.execute(select(Account).where(Account.id == account_id))
    account = result.scalar_one_or_none()
    if not account:
        raise HTTPException(status_code=404, detail="Account not found")

    clear_cookie_file(account_id)
    account.cookie_data = None
    account.cookie_path = None
    account.cookie_updated_at = None
    account.im_session_data = None
    account.qr_code_base64 = None
    account.error_message = None
    account.status = "offline"
    account.updated_at = datetime.utcnow()
    await db.execute(
        update(AccountProfileDetail)
        .where(AccountProfileDetail.account_id == account_id)
        .values(sec_user_id=None, synced_at=None)
    )
    await db.commit()
    await db.refresh(account)
    return account


async def _persist_im_session_data(account_id: int, session: DouyinImSession, db: AsyncSession) -> None:
    payload = json.dumps(session.to_dict(), ensure_ascii=False)
    await db.execute(
        update(Account)
        .where(Account.id == account_id)
        .values(im_session_data=payload, updated_at=datetime.utcnow())
    )
    await db.commit()


async def _seed_app_config():
    from auth.system_settings import ensure_default_settings

    async with AsyncSessionLocal() as db:
        await ensure_default_settings(db)


async def _seed_admin_user():
    """首次启动创建默认管理员，并将无归属抖音账号划归管理员。"""
    admin_password = os.getenv("KEFU_ADMIN_PASSWORD", "admin123")
    async with AsyncSessionLocal() as db:
        result = await db.execute(select(User).where(User.username == "admin"))
        admin = result.scalar_one_or_none()
        if not admin:
            admin = User(
                username="admin",
                password_hash=hash_password(admin_password),
                display_name="系统管理员",
                role="admin",
                is_active=True,
                email_verified=True,
                max_accounts=-1,
            )
            db.add(admin)
            await db.commit()
            await db.refresh(admin)
            print(f"Default admin created: username=admin password={admin_password}")

        await db.execute(
            update(Account).where(Account.owner_id.is_(None)).values(owner_id=admin.id)
        )
        await db.commit()


# 初始化数据库
@app.on_event("startup")
async def startup():
    # 放大默认线程池：每个托管账号的 a_bogus/web_protect/ts_sign 签名都是阻塞的 Node
    # 子进程调用，经 asyncio.to_thread 跑在默认线程池里。Python 默认池大小仅
    # min(32, cpu+4)，在 1 核云服务器上只有 5 个线程，导致超过 5 个账号并发时第 6 个
    # 账号的签名/取信息/发送会一直排队阻塞直至超时失败。这里显式放大线程池消除该瓶颈。
    import concurrent.futures as _futures

    try:
        _pool_size = int(os.getenv("KEFU_THREAD_POOL_SIZE", "0") or 0)
    except ValueError:
        _pool_size = 0
    if _pool_size <= 0:
        _pool_size = max(64, ((os.cpu_count() or 1) * 8))
    loop = asyncio.get_running_loop()
    loop.set_default_executor(
        _futures.ThreadPoolExecutor(
            max_workers=_pool_size,
            thread_name_prefix="kefu-sign",
        )
    )
    print(f"[kefu] thread pool size = {_pool_size}")

    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
        await conn.run_sync(_migrate_accounts_table)
        await conn.run_sync(_migrate_rules_table)
        await conn.run_sync(_migrate_message_logs_table)
        await conn.run_sync(_migrate_account_videos_table)
        await conn.run_sync(_migrate_users_table)
        await conn.run_sync(_migrate_payment_orders_table)
        await conn.run_sync(_migrate_accounts_quota_disabled)
    await _seed_app_config()
    await _seed_admin_user()
    # 进程启动时没有任何内存 Worker；复位异常退出遗留的运行状态。
    # 同时，账号数量/并发限制已移除，清理历史“额度停用”标记。
    async with AsyncSessionLocal() as db:
        await db.execute(
            update(Account)
            .where(Account.status.in_(("online", "logging_in", "starting")))
            .values(status="offline")
        )
        await db.execute(
            update(Account)
            .where(Account.quota_disabled.is_(True))
            .values(quota_disabled=False)
        )
        await db.commit()
    await _sync_legacy_cookie_files()
    await _seed_system_logs()
    global _system_log_flush_task
    _system_log_flush_task = asyncio.create_task(_flush_system_logs_loop())
    db_info = database_config_to_response()
    print(f"Database tables created successfully ({db_info['db_type']}: {db_info['database_url_display']}).")
    print("Douyin RPA backend ready (login: wait-for-scan + auto-save cookie).")

@app.on_event("shutdown")
async def shutdown():
    # Stop queued preparations first so no new workers appear while the
    # existing workers are being drained below.
    await batch_start_queue.stop()
    # 停止所有正在运行的 RPA 任务
    for account_id in list(manager.workers.keys()):
        await manager.stop_worker(account_id)
    from rpa_engine.douyin_im.traffic_control import shutdown_traffic_controller
    await shutdown_traffic_controller()
    if _system_log_flush_task:
        _system_log_flush_task.cancel()
        try:
            await _system_log_flush_task
        except asyncio.CancelledError:
            pass
        # 退出前把剩余日志落库
        pending = system_logger.drain_pending()
        if pending:
            async with AsyncSessionLocal() as db:
                for e in pending:
                    try:
                        created = datetime.fromisoformat(e["created_at"])
                    except Exception:
                        created = datetime.utcnow()
                    db.add(SystemLog(
                        account_id=e.get("account_id"),
                        level=e.get("level") or "info",
                        category=e.get("category") or "system",
                        event=(e.get("event") or "")[:255],
                        detail=e.get("detail") or "",
                        created_at=created,
                    ))
                await db.commit()
    print("All RPA workers stopped.")

# --- Pydantic 模型的定义 ---
class AccountCreate(BaseModel):
    phone: Optional[str] = None
    cookie_data: Optional[str] = None

class AccountResponse(BaseModel):
    id: int
    username: Optional[str] = None
    avatar_url: Optional[str] = None
    douyin_uid: Optional[str] = None
    phone: Optional[str] = None
    status: str
    error_message: Optional[str] = None
    has_cookie: bool = False
    cookie_valid: bool = False
    cookie_expired: bool = False
    cookie_status: Optional[str] = None
    cookie_updated_at: Optional[datetime] = None
    cookie_count: int = 0
    has_sessionid: bool = False
    reply_delay_seconds: Optional[int] = None
    reply_delay_effective: int = 0
    reply_cooldown_seconds: Optional[int] = None
    reply_cooldown_effective: int = 0
    follow_welcome_enabled: bool = False
    follow_welcome_content: Optional[str] = None
    user_agent: Optional[str] = None
    user_agent_label: Optional[str] = None
    quota_disabled: bool = False

    class Config:
        from_attributes = True


class AccountUpdate(BaseModel):
    phone: Optional[str] = None
    username: Optional[str] = None
    # 正数=账号专属排队间隔；0/null=未单独配置，继承系统默认
    reply_delay_seconds: Optional[int] = None
    # >=0 设为该账号专属冷却秒数；<0（如 -1）恢复为继承全局设置
    reply_cooldown_seconds: Optional[int] = None
    follow_welcome_enabled: Optional[bool] = None
    follow_welcome_content: Optional[str] = None
    user_agent: Optional[str] = None


class ReplyQueueItemResponse(BaseModel):
    job_id: str
    account_id: int
    position: int
    status: str
    expedited: bool = False
    description: str = ""
    sender_name: str = "未知用户"
    sender_id: Optional[str] = None
    sender_avatar: Optional[str] = None
    conversation_id: Optional[str] = None
    incoming_content: str = ""
    incoming_contents: List[str] = []
    message_count: int = 1
    replies: List[str] = []
    interval_seconds: int = 0
    enqueued_at: datetime
    scheduled_at: datetime
    remaining_seconds: int = 0


class ReplyQueueResponse(BaseModel):
    account_id: int
    running: bool = False
    pending_count: int = 0
    interval_seconds: int = 0
    global_send_interval_seconds: float = 1.0
    global_send_pending_count: int = 0
    global_account_pending_count: int = 0
    global_active_account_id: Optional[int] = None
    server_time: datetime
    items: List[ReplyQueueItemResponse] = []


class ReplyQueueSummaryItem(BaseModel):
    account_id: int
    running: bool = False
    pending_count: int = 0
    next_scheduled_at: Optional[datetime] = None
    next_remaining_seconds: int = 0


class ReplyQueueSummaryResponse(BaseModel):
    total_pending: int = 0
    server_time: datetime
    items: List[ReplyQueueSummaryItem] = []


class ReplyQueueActionResponse(BaseModel):
    job_id: str
    status: str
    message: str
    shifted_count: int = 0


class AccountCookieUpdate(BaseModel):
    cookie_data: str


class AccountCookieResponse(BaseModel):
    account_id: int
    cookie_data: Optional[str] = None
    cookie_updated_at: Optional[datetime] = None
    cookie_count: int = 0
    cookie_valid: bool = False
    cookie_expired: bool = False
    cookie_status: Optional[str] = None
    expires_at: Optional[str] = None
    key_names: List[str] = []
    has_sessionid: bool = False
    sessionid: Optional[str] = None
    sessionid_ss: Optional[str] = None
    im_ready: bool = False
    im_status: Optional[str] = None
    can_skip_browser: bool = False
    should_reset: bool = False


class AccountVideoItem(BaseModel):
    id: int
    aweme_id: str
    title: str = ""
    cover_url: Optional[str] = None
    video_url: Optional[str] = None
    share_url: Optional[str] = None
    play_url: Optional[str] = None
    create_time: Optional[str] = None
    digg_count: Optional[int] = None
    comment_count: Optional[int] = None
    play_count: Optional[int] = None
    media_type: Optional[str] = None


class AccountProfileDetailResponse(BaseModel):
    account_id: int
    uid: Optional[str] = None
    nickname: Optional[str] = None
    avatar_url: Optional[str] = None
    unique_id: Optional[str] = None
    signature: Optional[str] = None
    sec_user_id: Optional[str] = None
    profile_url: Optional[str] = None
    video_count: Optional[int] = None
    video_count_douyin: Optional[int] = None
    cached_work_count: int = 0
    playable_video_count: int = 0
    video_work_count: int = 0
    image_work_count: int = 0
    follower_count: Optional[int] = None
    following_count: Optional[int] = None
    total_favorited: Optional[int] = None
    favoriting_count: Optional[int] = None
    fetched: bool = False
    message: Optional[str] = None
    synced_at: Optional[str] = None
    videos: List[AccountVideoItem] = []


def _get_account_cookie_data(account: Account) -> Optional[str]:
    if account.cookie_data:
        return account.cookie_data
    return read_cookie_file(account.id)


async def _build_cookie_response(account: Account, cookie_data: Optional[str] = None) -> AccountCookieResponse:
    cookie_data = cookie_data if cookie_data is not None else _get_account_cookie_data(account)
    summary = cookie_summary(cookie_data)
    im_detail = await build_cookie_credential_detail(
        cookie_data,
        account.im_session_data,
        runtime_check=True,
    )
    return AccountCookieResponse(
        account_id=account.id,
        cookie_data=cookie_data,
        cookie_updated_at=account.cookie_updated_at,
        cookie_count=summary["cookie_count"],
        cookie_valid=summary["cookie_valid"],
        cookie_expired=summary["cookie_expired"],
        cookie_status=summary["reason"],
        expires_at=summary["expires_at"],
        key_names=summary["key_names"],
        has_sessionid=im_detail["has_sessionid"],
        sessionid=im_detail["sessionid"] or None,
        sessionid_ss=im_detail["sessionid_ss"] or None,
        im_ready=im_detail["im_ready"],
        im_status=im_detail["im_status"],
        can_skip_browser=im_detail["can_skip_browser"],
        should_reset=im_detail.get("should_reset", False),
    )


def _global_cooldown_seconds() -> int:
    try:
        from auth.system_settings import get_cached_settings

        return max(0, int(get_cached_settings().auto_reply_cooldown_seconds or 0))
    except Exception:
        return 0


def _global_reply_delay_seconds() -> int:
    try:
        from auth.system_settings import get_cached_settings

        return max(0, int(get_cached_settings().auto_reply_delay_seconds or 0))
    except Exception:
        return 0


def _resolve_effective_reply_delay(account: Account) -> int:
    account_value = max(0, int(account.reply_delay_seconds or 0))
    if account_value > 0:
        return account_value
    return _global_reply_delay_seconds()


def _resolve_effective_cooldown(account: Account) -> int:
    if account.reply_cooldown_seconds is not None:
        return max(0, int(account.reply_cooldown_seconds))
    return _global_cooldown_seconds()


def _build_account_response(account: Account) -> AccountResponse:
    cookie_data = _get_account_cookie_data(account)
    summary = cookie_summary(cookie_data)
    return AccountResponse(
        id=account.id,
        username=account.username,
        avatar_url=account.avatar_url,
        douyin_uid=account.douyin_uid,
        phone=account.phone,
        status=account.status,
        error_message=account.error_message,
        has_cookie=_account_has_cookie(account),
        cookie_valid=summary["cookie_valid"],
        cookie_expired=summary["cookie_expired"],
        cookie_status=summary["reason"],
        cookie_updated_at=account.cookie_updated_at,
        cookie_count=summary["cookie_count"],
        has_sessionid=summary.get("has_sessionid", False),
        reply_delay_seconds=(
            int(account.reply_delay_seconds)
            if int(account.reply_delay_seconds or 0) > 0
            else None
        ),
        reply_delay_effective=_resolve_effective_reply_delay(account),
        reply_cooldown_seconds=(
            int(account.reply_cooldown_seconds)
            if account.reply_cooldown_seconds is not None
            else None
        ),
        reply_cooldown_effective=_resolve_effective_cooldown(account),
        follow_welcome_enabled=bool(account.follow_welcome_enabled),
        follow_welcome_content=account.follow_welcome_content or None,
        user_agent=account.user_agent or None,
        user_agent_label=profile_label_for_ua(account.user_agent),
        quota_disabled=bool(account.quota_disabled),
    )

class RuleCreate(BaseModel):
    account_id: Optional[int] = None
    keyword: str
    reply_content: str
    match_type: str = "contains" # exact, contains, regex, default
    is_active: Optional[bool] = None  # 创建/更新时可指定启用状态；None=默认启用/不修改

class RuleResponse(BaseModel):
    id: int
    account_id: Optional[int] = None
    keyword: str
    reply_content: str
    match_type: str
    sort_order: int = 0
    is_active: bool

    class Config:
        from_attributes = True


class RuleReorder(BaseModel):
    rule_ids: List[int]

class LogResponse(BaseModel):
    id: int
    account_id: int
    sender_name: str
    sender_id: Optional[str] = None
    sender_avatar: Optional[str] = None
    message_content: str
    reply_content: Optional[str] = None
    status: str
    error_message: Optional[str] = None
    created_at: datetime

    class Config:
        from_attributes = True


class ReceivedMessageLogResponse(BaseModel):
    id: int
    account_id: int
    conversation_id: Optional[str] = None
    sender_id: Optional[str] = None
    sender_name: Optional[str] = None
    sender_avatar: Optional[str] = None
    message_type: Optional[int] = None
    server_message_id: Optional[str] = None
    raw_content: str
    created_at: datetime

    class Config:
        from_attributes = True


class SystemLogResponse(BaseModel):
    id: int
    account_id: Optional[int] = None
    level: str
    category: str
    event: str
    detail: str = ""
    created_at: str


class ConversationResponse(BaseModel):
    conversation_id: str
    sender_name: str
    sender_id: Optional[str] = None
    sender_avatar: Optional[str] = None
    content: str = ""
    unread_count: int = 0


class SendMessageRequest(BaseModel):
    conversation_id: str
    content: str = ""
    message_type: Optional[str] = None  # text | image | sticker
    media_url: Optional[str] = None
    sticker_url: Optional[str] = None
    width: Optional[int] = None
    height: Optional[int] = None
    sticker_id: Optional[str] = None


class UploadImageResponse(BaseModel):
    url: str
    width: Optional[int] = None
    height: Optional[int] = None
    payload: str


class SendMessageResponse(BaseModel):
    success: bool
    message: str
    need_browser_login: bool = False


class CredentialValidateResponse(BaseModel):
    has_cookie: bool = False
    cookie_valid: bool = False
    cookie_status: Optional[str] = None
    has_sessionid: bool = False
    im_ready: bool = False
    can_skip_browser: bool = False
    should_reset: bool = False
    login_mode: str = "browser"
    message: str = ""


class StartAccountRequest(BaseModel):
    login_mode: Optional[str] = None  # im_direct | browser | auto


class BatchStartRequest(BaseModel):
    account_ids: List[int] = Field(default_factory=list)
    all_accounts: bool = False


class DeviceProfileItem(BaseModel):
    id: str
    label: str
    platform: str
    user_agent: str


@app.get("/api/device-profiles", response_model=List[DeviceProfileItem])
async def get_device_profiles(user: User = Depends(get_current_user)):
    """可选的伪装设备头（User-Agent）预设列表。"""
    return list_device_profiles()


# --- API 路由接口 ---

# 1. 账号管理接口

def _account_matches_keyword(account: Account, keyword: str) -> bool:
    if not keyword:
        return True
    haystack = [
        account.username,
        account.douyin_uid,
        account.phone,
        profile_label_for_ua(account.user_agent),
        account.id,
    ]
    return any(
        keyword in str(item).lower()
        for item in haystack
        if item is not None and item != ""
    )


@app.get("/api/accounts")
async def get_accounts(
    page: Optional[int] = Query(None, ge=1),
    page_size: int = Query(9, ge=1, le=100),
    q: Optional[str] = None,
    status: Optional[str] = None,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """账号列表。

    - 不带 page 参数：返回全量数组（旧行为，供下拉选择等场景）。
    - 带 page 参数：返回 {items, total, page, page_size} 分页结构，
      支持 q（昵称/抖音ID/手机号/账号ID 搜索）与 status 筛选。
      Cookie 解析等重逻辑只对当前页执行。
    """
    result = await db.execute(accounts_for_user(user))
    accounts = result.scalars().all()
    # 更新内存中的运行状态与数据库同步，以防异常断开
    for acc in accounts:
        is_running = manager.is_running(acc.id)
        if is_running and acc.status == "offline":
            acc.status = "online"
        elif not is_running and acc.status in ("online", "logging_in", "starting"):
            acc.status = "offline"

    if page is None:
        return [_build_account_response(acc) for acc in accounts]

    keyword = (q or "").strip().lower()
    status_filter = (status or "").strip()
    filtered = [
        acc
        for acc in accounts
        if _account_matches_keyword(acc, keyword)
        and (
            not status_filter
            or status_filter == "all"
            or (
                status_filter == "quota_disabled"
                and bool(acc.quota_disabled)
            )
            or (
                status_filter != "quota_disabled"
                and not acc.quota_disabled
                and acc.status == status_filter
            )
        )
    ]
    total = len(filtered)
    start = (page - 1) * page_size
    items = filtered[start : start + page_size]
    return {
        "items": [_build_account_response(acc) for acc in items],
        "total": total,
        "page": page,
        "page_size": page_size,
    }

@app.get("/api/accounts/{account_id}", response_model=AccountResponse)
async def get_account(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    account = await get_owned_account(db, user, account_id)
    return _build_account_response(account)


def _douyin_profile_url(sec_user_id: Optional[str]) -> Optional[str]:
    sec = (sec_user_id or "").strip()
    if not sec:
        return None
    return f"https://www.douyin.com/user/{sec}"


def _build_profile_response(data: dict) -> AccountProfileDetailResponse:
    account_id = int(data["account_id"])
    videos = []
    for item in data.get("videos") or []:
        aweme_id = str(item.get("aweme_id") or "")
        videos.append(
            AccountVideoItem(
                id=int(item["id"]),
                aweme_id=aweme_id,
                title=str(item.get("title") or ""),
                cover_url=item.get("cover_url"),
                video_url=item.get("video_url"),
                share_url=item.get("share_url"),
                play_url=f"/api/accounts/{account_id}/videos/{aweme_id}/play" if aweme_id else None,
                create_time=item.get("create_time"),
                digg_count=item.get("digg_count"),
                comment_count=item.get("comment_count"),
                play_count=item.get("play_count"),
                media_type=item.get("media_type"),
            )
        )
    return AccountProfileDetailResponse(
        account_id=account_id,
        uid=data.get("uid"),
        nickname=data.get("nickname"),
        avatar_url=data.get("avatar_url"),
        unique_id=data.get("unique_id"),
        signature=data.get("signature"),
        sec_user_id=data.get("sec_user_id"),
        profile_url=data.get("profile_url") or _douyin_profile_url(data.get("sec_user_id")),
        video_count=data.get("video_count"),
        video_count_douyin=data.get("video_count_douyin"),
        cached_work_count=int(data.get("cached_work_count") or 0),
        playable_video_count=int(data.get("playable_video_count") or 0),
        video_work_count=int(data.get("video_work_count") or 0),
        image_work_count=int(data.get("image_work_count") or 0),
        follower_count=data.get("follower_count"),
        following_count=data.get("following_count"),
        total_favorited=data.get("total_favorited"),
        favoriting_count=data.get("favoriting_count"),
        fetched=bool(data.get("fetched")),
        message=data.get("message"),
        synced_at=data.get("synced_at"),
        videos=videos,
    )


@app.get("/api/accounts/{account_id}/profile", response_model=AccountProfileDetailResponse)
async def get_account_profile(
    account_id: int,
    sync: bool = False,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """获取托管账号详细资料与作品列表。sync=true 时从抖音同步并写入本地数据库。"""
    account = await get_owned_account(db, user, account_id)
    cookie_data = _get_account_cookie_data(account)

    from rpa_engine.account_profile import load_account_profile_from_db, sync_account_profile_to_db

    if sync:
        if not cookie_data:
            return AccountProfileDetailResponse(
                account_id=account_id,
                uid=account.douyin_uid,
                nickname=account.username,
                avatar_url=account.avatar_url,
                fetched=False,
                message="未保存 Cookie，无法从抖音同步资料",
            )
        try:
            data = await asyncio.wait_for(
                sync_account_profile_to_db(db, account, cookie_data),
                timeout=35,
            )
            return _build_profile_response(data)
        except asyncio.TimeoutError:
            # 抖音接口在云服务器上偶发响应缓慢，超过 nginx 上游超时会直接 504。
            # 这里限定整体时长，超时则回滚未完成的事务并返回本地缓存资料。
            try:
                await db.rollback()
            except Exception:
                pass
            data = await load_account_profile_from_db(db, account)
            data["message"] = (
                data.get("message")
                or "从抖音同步资料超时，已返回本地缓存，请稍后重试"
            )
            return _build_profile_response(data)

    data = await load_account_profile_from_db(db, account)
    if not data.get("fetched") and not cookie_data:
        data["message"] = data.get("message") or "未保存 Cookie，无法拉取抖音资料"
    return _build_profile_response(data)


@app.get("/api/accounts/{account_id}/videos/{aweme_id}/play")
async def stream_account_video(
    account_id: int,
    aweme_id: str,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """代理播放托管账号作品视频（带 Cookie / Referer）。"""
    account = await get_owned_account(db, user, account_id)
    video = (
        await db.execute(
            select(AccountVideo).where(
                AccountVideo.account_id == account_id,
                AccountVideo.aweme_id == aweme_id,
            )
        )
    ).scalar_one_or_none()
    if not video or not video.video_url:
        raise HTTPException(status_code=404, detail="视频不存在或未缓存播放地址")

    cookie_data = _get_account_cookie_data(account)
    headers = {
        "User-Agent": account.user_agent or "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Referer": video.share_url or "https://www.douyin.com/",
        "Accept": "*/*",
    }
    cookies: dict[str, str] = {}
    if cookie_data:
        try:
            from rpa_engine.douyin_im.session import DouyinImSession

            session = DouyinImSession.from_storage_state(json.loads(cookie_data) or {})
            cookies = dict(session.cookies or {})
        except Exception:
            pass

    async with httpx.AsyncClient(follow_redirects=True, timeout=120.0) as client:
        req = client.build_request(
            "GET",
            video.video_url,
            headers=headers,
            cookies=cookies,
        )
        resp = await client.send(req, stream=True)
        if resp.status_code >= 400:
            await resp.aclose()
            raise HTTPException(status_code=502, detail=f"视频拉取失败 ({resp.status_code})")
        media_type = resp.headers.get("content-type") or "video/mp4"

        async def body_iter():
            try:
                async for chunk in resp.aiter_bytes():
                    yield chunk
            finally:
                await resp.aclose()

        return StreamingResponse(body_iter(), media_type=media_type)

@app.put("/api/accounts/{account_id}", response_model=AccountResponse)
async def update_account(
    account_id: int,
    body: AccountUpdate,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    account = await get_owned_account(db, user, account_id, write=True)

    if body.phone is not None:
        account.phone = body.phone
    if body.username is not None:
        account.username = body.username
    if "reply_delay_seconds" in body.model_fields_set:
        # 兼容旧列：0 作为“未设置”，由系统默认值兜底；显式 null 同样清除账号值。
        account.reply_delay_seconds = max(0, int(body.reply_delay_seconds or 0))
    if body.reply_cooldown_seconds is not None:
        # 负值表示恢复继承全局设置
        account.reply_cooldown_seconds = (
            None if int(body.reply_cooldown_seconds) < 0 else max(0, int(body.reply_cooldown_seconds))
        )
    if body.follow_welcome_enabled is not None:
        account.follow_welcome_enabled = bool(body.follow_welcome_enabled)
    if body.follow_welcome_content is not None:
        account.follow_welcome_content = (body.follow_welcome_content or "").strip() or None
    if body.user_agent is not None:
        ua = (body.user_agent or "").strip()
        account.user_agent = ua or None
    account.updated_at = datetime.utcnow()
    await db.commit()
    await db.refresh(account)
    return _build_account_response(account)


@app.get("/api/reply-queues", response_model=ReplyQueueSummaryResponse)
async def get_reply_queue_summaries(
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """聚合当前用户可见账号的排队数量，供账号页低频轮询。"""
    allowed_ids = await owned_account_ids(db, user)
    summaries: list[ReplyQueueSummaryItem] = []
    total_pending = 0
    for account_id, worker in list(manager.workers.items()):
        if allowed_ids is not None and account_id not in allowed_ids:
            continue
        service = worker._im_service if worker else None
        if not service:
            continue
        items = await service.get_reply_queue_snapshot()
        if not items:
            continue
        pending_count = len(items)
        total_pending += pending_count
        first = items[0]
        summaries.append(
            ReplyQueueSummaryItem(
                account_id=account_id,
                running=bool(worker.is_running and service._running),
                pending_count=pending_count,
                next_scheduled_at=first.get("scheduled_at"),
                next_remaining_seconds=int(first.get("remaining_seconds") or 0),
            )
        )
    return ReplyQueueSummaryResponse(
        total_pending=total_pending,
        server_time=datetime.now(timezone.utc),
        items=summaries,
    )


@app.get(
    "/api/accounts/{account_id}/reply-queue",
    response_model=ReplyQueueResponse,
)
async def get_account_reply_queue(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """查看一个账号的完整自动回复排队内容。"""
    account = await get_owned_account(db, user, account_id)
    worker = manager.workers.get(account_id)
    service = worker._im_service if worker else None
    items = await service.get_reply_queue_snapshot() if service else []
    from rpa_engine.douyin_im.traffic_control import get_traffic_controller
    global_send = await get_traffic_controller().send_queue.snapshot()
    queued_for_account = int(global_send.get("per_account", {}).get(account_id, 0) or 0)
    if global_send.get("active_account_id") == account_id:
        queued_for_account += 1
    return ReplyQueueResponse(
        account_id=account_id,
        running=bool(worker and worker.is_running and service and service._running),
        pending_count=len(items),
        interval_seconds=_resolve_effective_reply_delay(account),
        global_send_interval_seconds=float(global_send.get("interval_seconds") or 0),
        global_send_pending_count=int(global_send.get("pending_count") or 0),
        global_account_pending_count=queued_for_account,
        global_active_account_id=global_send.get("active_account_id"),
        server_time=datetime.now(timezone.utc),
        items=items,
    )


@app.post(
    "/api/accounts/{account_id}/reply-queue/{job_id}/send-now",
    response_model=ReplyQueueActionResponse,
    status_code=202,
)
async def send_account_queued_reply_now(
    account_id: int,
    job_id: str,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    """将指定任务原子移入紧急队列，并把它后面的普通任务前移一槽。"""
    await get_owned_account(db, user, account_id, write=True)
    worker = manager.workers.get(account_id)
    service = worker._im_service if worker else None
    if not worker or not worker.is_running or not service or not service._running:
        raise HTTPException(status_code=409, detail="账号托管未运行，无法发送队列任务")

    result = await service.send_queued_reply_now(job_id)
    status = result.get("status")
    if status == "not_running":
        raise HTTPException(status_code=409, detail="账号回复队列已停止")
    if status == "not_found":
        raise HTTPException(status_code=404, detail="队列任务不存在，可能已发送或已取消")
    if status == "already_sending":
        return ReplyQueueActionResponse(
            job_id=job_id,
            status=status,
            message="该任务正在发送中",
        )
    if status == "already_requested":
        return ReplyQueueActionResponse(
            job_id=job_id,
            status=status,
            message="该任务已排入立即发送队列",
        )

    shifted_count = int(result.get("shifted_count") or 0)
    system_logger.record(
        "队列任务已手动设为立即发送",
        detail=f"任务 {job_id} 已进入紧急发送队列，后续 {shifted_count} 项已自动前移。",
        level="info",
        category="send",
        account_id=account_id,
    )
    return ReplyQueueActionResponse(
        job_id=job_id,
        status="accepted",
        message=(
            f"已安排立即发送，后续 {shifted_count} 项已自动前移"
            if shifted_count
            else "已安排立即发送"
        ),
        shifted_count=shifted_count,
    )

DESKTOP_LOGIN_SEC_USER_ID_MISSING = (
    "该账号缺少 sec_user_id，或账号身份已过期，暂不允许软件端登录。"
    "请先在后台账号管理中同步详细资料，或重新扫码刷新账号身份后再试。"
)
DESKTOP_LOGIN_SEC_USER_ID_UNKNOWN = (
    "暂时无法核验 sec_user_id，暂不允许软件端登录，请稍后重试。"
)


async def _require_desktop_login_sec_user_id(
    account: Account,
    db: AsyncSession,
) -> None:
    """Fail closed before a desktop client can receive an account Cookie."""
    try:
        result = await db.execute(
            select(AccountProfileDetail).where(
                AccountProfileDetail.account_id == account.id
            )
        )
        profile = result.scalar_one_or_none()
    except Exception as exc:
        logger.exception(
            "Desktop login sec_user_id check failed for account %s", account.id
        )
        raise HTTPException(
            status_code=503,
            detail=DESKTOP_LOGIN_SEC_USER_ID_UNKNOWN,
        ) from exc

    sec_user_id = str(getattr(profile, "sec_user_id", "") or "").strip()
    identity_stale = bool(
        account.cookie_updated_at
        and (
            profile is None
            or profile.synced_at is None
            or account.cookie_updated_at > profile.synced_at
        )
    )
    if not sec_user_id or identity_stale:
        raise HTTPException(
            status_code=409,
            detail=DESKTOP_LOGIN_SEC_USER_ID_MISSING,
        )


@app.get("/api/accounts/{account_id}/cookie", response_model=AccountCookieResponse)
async def get_account_cookie(
    account_id: int,
    purpose: Optional[str] = None,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """Read a Cookie for management or for legacy desktop login clients.

    Older released desktop clients call this route without a purpose.  Keep
    those clients protected immediately by applying the desktop identity guard
    by default.  The web console explicitly uses ``purpose=management`` so a
    missing identity can still be repaired there.
    """
    account = await get_owned_account(db, user, account_id)
    cookie_data = _get_account_cookie_data(account)
    if cookie_data and purpose != "management":
        await _require_desktop_login_sec_user_id(account, db)
    return await _build_cookie_response(account, cookie_data)


@app.get(
    "/api/accounts/{account_id}/desktop-login-credential",
    response_model=AccountCookieResponse,
)
async def get_desktop_login_credential(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """Return credentials only when the selected Douyin identity is verified.

    This endpoint is intentionally separate from the normal Cookie management
    endpoint: an account missing ``sec_user_id`` must still be editable in the
    web console so the user can repair or replace its credentials.
    """
    account = await get_owned_account(db, user, account_id)
    cookie_data = _get_account_cookie_data(account)
    if not cookie_data:
        # Preserve the clients' existing, more specific "no login state"
        # message; without Cookie data there is no credential to disclose.
        return await _build_cookie_response(account, cookie_data)

    await _require_desktop_login_sec_user_id(account, db)
    return await _build_cookie_response(account, cookie_data)


@app.put("/api/accounts/{account_id}/cookie", response_model=AccountCookieResponse)
async def update_account_cookie(
    account_id: int,
    body: AccountCookieUpdate,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    # Validate first: malformed input must not take a healthy hosted account
    # offline.  Filesystem and database mutations happen only after the worker
    # preparation lock has been acquired below.
    try:
        parsed_data = validate_cookie_json(body.cookie_data)
        standard_json_str = json.dumps(parsed_data, ensure_ascii=False, indent=2)
    except (ValueError, Exception) as e:
        raise HTTPException(status_code=400, detail=f"Cookie 格式错误: {e}")

    # Serialize credential replacement against both single-account and batch
    # preparations.  Keep the lock until the new Cookie and cleared identity
    # are committed so no worker can start in the stop/commit gap.
    async with manager.preparation_lock(account_id):
        account = await get_owned_account(db, user, account_id, write=True)
        await batch_start_queue.cancel_account(account_id)
        await manager.stop_worker(account_id)

        try:
            cookie_path = write_cookie_file(account_id, standard_json_str)
        except Exception as exc:
            raise HTTPException(status_code=400, detail=f"Cookie 保存失败: {exc}") from exc

        account.cookie_data = standard_json_str
        account.cookie_path = cookie_path
        account.cookie_updated_at = datetime.utcnow()
        account.updated_at = datetime.utcnow()
        await db.execute(
            update(AccountProfileDetail)
            .where(AccountProfileDetail.account_id == account_id)
            .values(sec_user_id=None, synced_at=None)
        )
        try:
            from rpa_engine.account_profile import apply_douyin_profile

            await apply_douyin_profile(db, account, standard_json_str)
        except Exception as exc:
            logger.warning(f"Sync profile after cookie update failed: {exc}")
        await db.commit()
        await db.refresh(account)

    return await _build_cookie_response(account, standard_json_str)

@app.delete("/api/accounts/{account_id}/cookie")
async def delete_account_cookie(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    # Deleting credentials uses the same preparation lock as starting a
    # worker, preventing a new worker from appearing after stop_worker but
    # before the cleared credentials are committed.
    async with manager.preparation_lock(account_id):
        account = await get_owned_account(db, user, account_id, write=True)
        await batch_start_queue.cancel_account(account_id)
        await manager.stop_worker(account_id)

        clear_cookie_file(account_id)
        account.cookie_data = None
        account.cookie_path = None
        account.cookie_updated_at = None
        account.im_session_data = None
        account.updated_at = datetime.utcnow()
        await db.execute(
            update(AccountProfileDetail)
            .where(AccountProfileDetail.account_id == account_id)
            .values(sec_user_id=None, synced_at=None)
        )
        await db.commit()
    return {"message": "Cookie cleared successfully."}

@app.post("/api/accounts", response_model=AccountResponse)
async def create_account(
    account_in: AccountCreate,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    cookie_data = (account_in.cookie_data or "").strip()
    standard_json_str = None
    if cookie_data:
        try:
            parsed_data = validate_cookie_json(cookie_data)
            standard_json_str = json.dumps(parsed_data, ensure_ascii=False, indent=2)
        except (ValueError, Exception) as e:
            raise HTTPException(status_code=400, detail=f"Cookie 格式错误: {e}")

    await ensure_can_add_account(db, user)

    account = Account(phone=account_in.phone, status="offline", owner_id=user.id)
    db.add(account)
    await db.commit()
    await db.refresh(account)

    if cookie_data:
        cookie_path = write_cookie_file(account.id, standard_json_str)
        account.cookie_data = standard_json_str
        account.cookie_path = cookie_path
        account.cookie_updated_at = datetime.utcnow()
        try:
            from rpa_engine.account_profile import apply_douyin_profile

            await apply_douyin_profile(db, account, standard_json_str)
        except Exception as exc:
            logger.warning(f"Sync profile after account create failed: {exc}")
        await db.commit()
        await db.refresh(account)

    return _build_account_response(account)

@app.delete("/api/accounts/{account_id}")
async def delete_account(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    await get_owned_account(db, user, account_id, write=True)
    # 停止运行中的任务
    await batch_start_queue.cancel_account(account_id)
    await manager.stop_worker(account_id)
    clear_cookie_file(account_id)

    # 从数据库删除
    await db.execute(delete(Account).where(Account.id == account_id))
    await db.commit()
    return {"message": f"Account {account_id} deleted successfully."}

@app.post("/api/accounts/{account_id}/validate-credential", response_model=CredentialValidateResponse)
async def validate_account_credential(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    account = await get_owned_account(db, user, account_id)

    cookie_data = _get_account_cookie_data(account)
    assessment = await assess_account_credential(cookie_data, account.im_session_data)
    return CredentialValidateResponse(**assessment)


@app.post("/api/accounts/{account_id}/reset-credentials")
async def reset_account_credentials(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    await get_owned_account(db, user, account_id, write=True)
    await batch_start_queue.cancel_account(account_id)
    account = await _reset_account_credentials(account_id, db)
    return {
        "message": "已清除 Cookie 与 IM 会话数据，请重新登录",
        "status": account.status,
    }


async def _start_account_rpa_impl(
    account: Account,
    db: AsyncSession,
    requested_login_mode: Optional[str] = None,
) -> dict:
    """Validate once, persist the starting state, then spawn one worker."""
    account_id = int(account.id)
    if manager.is_running(account_id):
        return {"status": "running", "message": "RPA worker is already running."}

    cookie_data = _get_account_cookie_data(account)
    assessment = await assess_account_credential(cookie_data, account.im_session_data)
    login_mode = requested_login_mode or assessment["login_mode"]
    reset_performed = False

    if assessment.get("should_reset") and login_mode != "im_direct":
        account = await _reset_account_credentials(account_id, db)
        reset_performed = True
        cookie_data = None
        assessment = await assess_account_credential(None, None)
        login_mode = "browser"
    elif assessment.get("should_reset") and login_mode == "im_direct":
        raise HTTPException(
            status_code=400,
            detail=assessment["message"] or "凭证已失效，请先清除旧数据后重新登录",
        )

    if login_mode == "im_direct" and not assessment["can_skip_browser"]:
        raise HTTPException(
            status_code=400,
            detail=assessment["message"] or "凭证未通过验证，无法直连 IM",
        )

    # Write before spawning the task.  This gives both single and batch calls
    # an immediate authoritative status and avoids racing the worker's first
    # database update.  Batch submission normally wrote this state in bulk, so
    # no per-account commit is needed in that path.
    state_changed = (
        account.status != "starting"
        or account.qr_code_base64 is not None
        or account.error_message is not None
    )
    if state_changed:
        account.status = "starting"
        account.qr_code_base64 = None
        account.error_message = None
        await db.commit()

    try:
        started = await manager.start_worker(account_id, login_mode=login_mode)
    except Exception as exc:
        account.status = "error"
        account.error_message = str(exc) or "启动托管失败"
        await db.commit()
        raise

    if started:
        if login_mode == "im_direct":
            msg = assessment["message"] or "凭证有效，正在直连 IM 托管（无需浏览器）"
        elif reset_performed:
            msg = "凭证已失效，已清除旧数据，正在打开浏览器重新登录..."
        elif cookie_data:
            msg = "凭证需刷新，将打开浏览器登录/采集 IM 会话..."
        else:
            msg = "未保存 Cookie，将打开浏览器扫码登录..."

        return {
            "status": "starting",
            "login_mode": login_mode,
            "cookie_valid": assessment["cookie_valid"],
            "im_ready": assessment["im_ready"],
            "skip_qr": login_mode == "im_direct",
            "skip_browser": login_mode == "im_direct",
            "message": msg,
        }
    return {"status": "running", "message": "RPA worker is already running."}


async def _start_queued_account(account_id: int) -> dict:
    """Own-session handler used by BatchStartQueue background workers."""
    async with manager.preparation_lock(account_id):
        async with AsyncSessionLocal() as db:
            result = await db.execute(select(Account).where(Account.id == account_id))
            account = result.scalar_one_or_none()
            if not account:
                raise RuntimeError("账号不存在或已删除")
            if account.quota_disabled:
                raise RuntimeError("账号已停用，无法启动托管")
            try:
                return await _start_account_rpa_impl(account, db)
            except asyncio.CancelledError:
                raise
            except Exception as exc:
                if not manager.is_running(account_id):
                    account.status = "error"
                    account.error_message = str(getattr(exc, "detail", None) or exc or "启动失败")
                    await db.commit()
                raise


batch_start_queue = BatchStartQueue(_start_queued_account)


@app.post("/api/accounts/{account_id}/start")
async def start_account_rpa(
    account_id: int,
    body: StartAccountRequest = StartAccountRequest(),
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    account = await get_owned_account(db, user, account_id, write=True)
    await batch_start_queue.cancel_account(account_id)
    async with manager.preparation_lock(account_id):
        return await _start_account_rpa_impl(account, db, body.login_mode)


@app.post("/api/account-start-batches", status_code=202)
async def submit_account_start_batch(
    body: BatchStartRequest,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    requested_ids = list(
        dict.fromkeys(int(value) for value in body.account_ids if int(value) > 0)
    )
    if not body.all_accounts and not requested_ids:
        raise HTTPException(status_code=400, detail="请选择要启动的账号")
    if len(requested_ids) > 1000:
        raise HTTPException(status_code=400, detail="单次最多提交 1000 个账号")

    # The submit path needs only ids and the disabled flag.  Do not hydrate
    # every account's large cookie/session/QR columns just to enqueue ids.
    candidate_stmt = select(Account.id, Account.quota_disabled)
    if not is_admin(user.role):
        candidate_stmt = candidate_stmt.where(Account.owner_id == user.id)
    if not body.all_accounts:
        candidate_stmt = candidate_stmt.where(Account.id.in_(requested_ids))
    result = await db.execute(candidate_stmt)
    candidates = result.all()

    eligible_ids: list[int] = []
    skipped_running = 0
    skipped_disabled = 0
    for account_id, quota_disabled in candidates:
        if quota_disabled:
            skipped_disabled += 1
        elif manager.is_running(account_id):
            skipped_running += 1
        else:
            eligible_ids.append(int(account_id))

    batch = await batch_start_queue.submit(
        eligible_ids,
        owner_id=int(user.id),
        metadata={
            "requested_count": (
                len(candidates) if body.all_accounts else len(requested_ids)
            ),
            "accessible_count": len(candidates),
            "skipped_running_count": skipped_running,
            "skipped_disabled_count": skipped_disabled,
        },
    )
    return batch


@app.get("/api/account-start-batches/{batch_id}")
async def get_account_start_batch(
    batch_id: str,
    user: User = Depends(get_current_user),
):
    batch = await batch_start_queue.get_batch(
        batch_id,
        owner_id=int(user.id),
        include_items=False,
    )
    if not batch:
        raise HTTPException(status_code=404, detail="启动批次不存在或已过期")
    return batch

@app.post("/api/accounts/{account_id}/stop")
async def stop_account_rpa(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    account = await get_owned_account(db, user, account_id, write=True)

    await batch_start_queue.cancel_account(account_id)
    stopped = await manager.stop_worker(account_id)
    # 强制将数据库中的状态重置为 offline
    account.status = "offline"
    await db.commit()
    
    if stopped:
        return {"status": "stopped", "message": "RPA worker stopped successfully."}
    else:
        return {"status": "offline", "message": "RPA worker was not running."}

@app.get("/api/accounts/{account_id}/qr")
async def get_account_qr(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    account = await get_owned_account(db, user, account_id)
    
    return {
        "status": account.status,
        "qr_code_base64": account.qr_code_base64,
        "error_message": account.error_message
    }

# 2. 自动回复规则接口

@app.get("/api/rules")
async def get_rules(
    account_id: Optional[int] = None,
    page: Optional[int] = Query(None, ge=1),
    page_size: int = Query(10, ge=1, le=100),
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """规则列表。

    - 不带 page 参数：返回全量数组（旧行为）。
    - 带 page 参数：数据库层 LIMIT/OFFSET 分页，返回 {items, total, page, page_size}。
    """
    if account_id is not None:
        await get_owned_account(db, user, account_id)
        stmt = select(AutoReplyRule).where(AutoReplyRule.account_id == account_id)
    else:
        stmt = rules_for_user(user)
    stmt = stmt.order_by(AutoReplyRule.sort_order.asc(), AutoReplyRule.id.asc())

    if page is None:
        result = await db.execute(stmt)
        return [RuleResponse.model_validate(r) for r in result.scalars().all()]

    total = int(
        (
            await db.execute(
                select(func.count()).select_from(stmt.subquery())
            )
        ).scalar()
        or 0
    )
    result = await db.execute(stmt.offset((page - 1) * page_size).limit(page_size))
    return {
        "items": [RuleResponse.model_validate(r) for r in result.scalars().all()],
        "total": total,
        "page": page,
        "page_size": page_size,
    }

@app.post("/api/rules", response_model=RuleResponse)
async def create_rule(
    rule_in: RuleCreate,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    if rule_in.match_type == "default":
        rule_in.keyword = ""
    if rule_in.account_id is None:
        raise HTTPException(status_code=400, detail="请选择适用账号，每条规则必须绑定一个托管账号")
    await get_owned_account(db, user, rule_in.account_id, write=True)

    sort_stmt = select(func.max(AutoReplyRule.sort_order)).where(
        AutoReplyRule.account_id == rule_in.account_id
    )
    max_sort = (await db.execute(sort_stmt)).scalar()
    next_sort = int(max_sort or 0) + 1

    rule = AutoReplyRule(
        owner_id=user.id,
        account_id=rule_in.account_id,
        keyword=rule_in.keyword,
        reply_content=rule_in.reply_content,
        match_type=rule_in.match_type,
        sort_order=next_sort,
        is_active=True if rule_in.is_active is None else bool(rule_in.is_active),
    )
    db.add(rule)
    await db.commit()
    await db.refresh(rule)
    return rule

@app.put("/api/rules/{rule_id}", response_model=RuleResponse)
async def update_rule(
    rule_id: int,
    rule_in: RuleCreate,
    is_active: Optional[bool] = None,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    rule = await get_accessible_rule(db, user, rule_id, write=True)

    if rule_in.match_type == "default":
        rule_in.keyword = ""
    if rule_in.account_id is None:
        raise HTTPException(status_code=400, detail="请选择适用账号，每条规则必须绑定一个托管账号")
    await get_owned_account(db, user, rule_in.account_id, write=True)

    rule.account_id = rule_in.account_id
    rule.keyword = rule_in.keyword
    rule.reply_content = rule_in.reply_content
    rule.match_type = rule_in.match_type
    # 优先使用请求体中的 is_active，其次兼容旧的查询参数
    if rule_in.is_active is not None:
        rule.is_active = bool(rule_in.is_active)
    elif is_active is not None:
        rule.is_active = is_active

    await db.commit()
    await db.refresh(rule)
    return rule

@app.post("/api/rules/{rule_id}/toggle", response_model=RuleResponse)
async def toggle_rule(
    rule_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    rule = await get_accessible_rule(db, user, rule_id, write=True)
    
    rule.is_active = not rule.is_active
    await db.commit()
    await db.refresh(rule)
    return rule

@app.delete("/api/rules/{rule_id}")
async def delete_rule(
    rule_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    await get_accessible_rule(db, user, rule_id, write=True)
    await db.execute(delete(AutoReplyRule).where(AutoReplyRule.id == rule_id))
    await db.commit()
    return {"message": "Rule deleted successfully."}


class RuleMove(BaseModel):
    direction: str  # up | down


@app.post("/api/rules/{rule_id}/move")
async def move_rule(
    rule_id: int,
    body: RuleMove,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    """在同账号规则内上移/下移一位（服务端交换排序，适配前端分页）。"""
    rule = await get_accessible_rule(db, user, rule_id, write=True)
    stmt = (
        select(AutoReplyRule)
        .where(AutoReplyRule.account_id == rule.account_id)
        .order_by(AutoReplyRule.sort_order.asc(), AutoReplyRule.id.asc())
    )
    siblings = list((await db.execute(stmt)).scalars().all())
    index = next((i for i, r in enumerate(siblings) if r.id == rule.id), -1)
    target = index - 1 if body.direction == "up" else index + 1
    if index < 0 or target < 0 or target >= len(siblings):
        return {"message": "已在边界，无需移动"}
    siblings[index], siblings[target] = siblings[target], siblings[index]
    for i, r in enumerate(siblings):
        r.sort_order = i
    await db.commit()
    return {"message": "Rule moved successfully."}


@app.post("/api/rules/reorder")
async def reorder_rules(
    body: RuleReorder,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    if not body.rule_ids:
        return {"message": "No rules to reorder."}
    for index, rule_id in enumerate(body.rule_ids):
        rule = await get_accessible_rule(db, user, rule_id, write=True)
        rule.sort_order = index
    await db.commit()
    return {"message": "Rules reordered successfully."}

# 3. 消息日志接口

@app.get("/api/logs/stats")
async def get_logs_stats(
    account_id: Optional[int] = None,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """消息日志全量统计（数据库计数，不受列表 limit 限制）。"""
    if account_id is not None:
        await get_owned_account(db, user, account_id)
    base = logs_for_user(user, account_id).subquery()
    total = int(
        (await db.execute(select(func.count()).select_from(base))).scalar() or 0
    )
    replied = int(
        (
            await db.execute(
                select(func.count())
                .select_from(base)
                .where(base.c.status == "replied")
            )
        ).scalar()
        or 0
    )
    return {"total": total, "replied": replied}


@app.get("/api/logs", response_model=List[LogResponse])
async def get_logs(
    account_id: Optional[int] = None,
    limit: int = 50,
    offset: int = 0,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    if account_id is not None:
        await get_owned_account(db, user, account_id)
    limit = max(1, min(int(limit or 50), 500))
    stmt = (
        logs_for_user(user, account_id)
        .order_by(MessageLog.created_at.desc())
        .offset(max(0, int(offset or 0)))
        .limit(limit)
    )
    result = await db.execute(stmt)
    return result.scalars().all()


@app.get("/api/received-messages", response_model=List[ReceivedMessageLogResponse])
async def get_received_messages(
    account_id: Optional[int] = None,
    limit: int = 100,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """接收消息原始日志：仅包含收到的消息，内容为接口/通道原样记录。"""
    if account_id is not None:
        await get_owned_account(db, user, account_id)
    limit = max(1, min(int(limit or 100), 500))
    stmt = (
        received_logs_for_user(user, account_id)
        .order_by(ReceivedMessageLog.created_at.desc())
        .limit(limit)
    )
    result = await db.execute(stmt)
    return result.scalars().all()


@app.get("/api/system-logs", response_model=List[SystemLogResponse])
async def get_system_logs(
    account_id: Optional[int] = None,
    level: Optional[str] = None,
    category: Optional[str] = None,
    limit: int = 200,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    """系统诊断日志：私信收发 / 实时连接 / 鉴权 等链路事件，用于排查失败原因。"""
    if account_id is not None:
        await get_owned_account(db, user, account_id)
    entries = system_logger.get_logs(
        account_id=account_id,
        level=level,
        category=category,
        limit=max(1, min(int(limit or 200), 1000)),
    )
    if not is_admin(user.role):
        allowed = await owned_account_ids(db, user)
        entries = [e for e in entries if e.get("account_id") in allowed]
    return [SystemLogResponse(**e) for e in entries]


@app.delete("/api/system-logs")
async def clear_system_logs(
    db: AsyncSession = Depends(get_db),
    _: User = Depends(require_admin),
):
    """清空系统诊断日志（内存缓冲区 + 数据库历史）。"""
    system_logger.clear()
    await db.execute(delete(SystemLog))
    await db.commit()
    return {"message": "系统诊断日志已清空"}


@app.get("/api/accounts/{account_id}/conversations", response_model=List[ConversationResponse])
async def get_account_conversations(
    account_id: int,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    account = await get_owned_account(db, user, account_id)
    if not _account_has_cookie(account):
        raise HTTPException(status_code=400, detail="账号未保存 Cookie，无法拉取私信会话")

    session = _build_account_im_session(account)
    if not session.can_direct_im():
        raise HTTPException(status_code=400, detail="Cookie 无效或缺少 sessionid，无法访问 IM")

    conversations: list[dict] = []
    worker = manager.workers.get(account_id)
    if worker and worker._im_service:
        conversations = worker._im_service.get_cached_conversations()

    if not conversations:
        async with DouyinImHttpClient(session, account_id=account_id) as http:
            conversations = await http.get_conversations()

    if not conversations:
        my_uid = session.my_uid or 0
        if not my_uid:
            from rpa_engine.douyin_im.auth import DouyinAuth
            auth = DouyinAuth()
            auth.perepare_auth(
                session.cookie_header(),
                session.web_protect_str,
                session.keys_str,
            )
            my_uid = auth.get_uid() or 0
        conversations = await _conversations_from_logs(db, account_id, my_uid)

    return [
        ConversationResponse(
            conversation_id=str(c.get("conversation_id") or ""),
            sender_name=str(c.get("sender_name") or "未知用户"),
            sender_id=str(c.get("sender_id") or c.get("peer_uid") or "") or None,
            sender_avatar=c.get("sender_avatar") or None,
            content=str(c.get("content") or ""),
            unread_count=int(c.get("unread_count") or 0),
        )
        for c in conversations
        if c.get("conversation_id") or c.get("sender_name")
    ]


@app.post("/api/accounts/{account_id}/messages/upload-image", response_model=UploadImageResponse)
async def upload_message_image(
    account_id: int,
    file: UploadFile = File(...),
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    account = await get_owned_account(db, user, account_id, write=True)
    if not file.content_type or not file.content_type.startswith("image/"):
        raise HTTPException(status_code=400, detail="仅支持上传图片文件")

    raw = await file.read()
    if not raw:
        raise HTTPException(status_code=400, detail="图片为空")
    if len(raw) > 8 * 1024 * 1024:
        raise HTTPException(status_code=400, detail="图片大小不能超过 8MB")

    ext = ".jpg"
    if file.content_type == "image/png":
        ext = ".png"
    elif file.content_type == "image/gif":
        ext = ".gif"
    elif file.content_type == "image/webp":
        ext = ".webp"

    account_dir = os.path.join(UPLOAD_DIR, str(account_id))
    os.makedirs(account_dir, exist_ok=True)
    filename = f"{uuid.uuid4().hex}{ext}"
    path = os.path.join(account_dir, filename)
    with open(path, "wb") as f:
        f.write(raw)

    width = None
    height = None
    try:
        from PIL import Image

        with Image.open(BytesIO(raw)) as img:
            width, height = img.size
    except Exception:
        pass

    public_url = f"/api/media/messages/{account_id}/{filename}"
    from rpa_engine.douyin_im.message_content import serialize_message_content
    from rpa_engine.douyin_im.image_upload import upload_im_image
    from rpa_engine.douyin_im.traffic_control import submit_outbound

    session = _build_account_im_session(account)
    worker = manager.workers.get(account_id)
    if worker and worker._im_service:
        session = worker._im_service.session

    image_spec: dict = {
        "type": "image",
        "text": "[图片]",
        "url": public_url,
    }
    if width:
        image_spec["width"] = width
    if height:
        image_spec["height"] = height

    uploaded = await submit_outbound(
        account_id,
        lambda: asyncio.to_thread(
            upload_im_image,
            session,
            raw,
            filename=filename,
            content_type=file.content_type or "image/jpeg",
        ),
        description=f"IM image upload {filename}",
    )
    if uploaded.get("uri"):
        for key in ("uri", "url_list", "md5"):
            if uploaded.get(key):
                image_spec[key] = uploaded[key]
        if uploaded.get("url"):
            image_spec["douyin_url"] = uploaded["url"]
        if uploaded.get("width") and not image_spec.get("width"):
            image_spec["width"] = uploaded["width"]
        if uploaded.get("height") and not image_spec.get("height"):
            image_spec["height"] = uploaded["height"]
    elif uploaded.get("error"):
        logger.warning("Douyin image pre-upload failed account=%s: %s", account_id, uploaded["error"])
        raise HTTPException(status_code=502, detail=f"图片上传失败：{uploaded['error']}")

    payload = serialize_message_content(image_spec)
    return UploadImageResponse(url=public_url, width=width, height=height, payload=payload)


@app.post("/api/accounts/{account_id}/messages/send", response_model=SendMessageResponse)
async def send_account_message(
    account_id: int,
    body: SendMessageRequest,
    db: AsyncSession = Depends(get_db),
    user: User = Depends(require_write),
):
    account = await get_owned_account(db, user, account_id, write=True)

    content = normalize_outgoing_content(
        content=body.content or "",
        message_type=body.message_type,
        media_url=body.media_url,
        sticker_url=body.sticker_url,
        width=body.width,
        height=body.height,
        sticker_id=body.sticker_id,
    )
    if not content:
        raise HTTPException(status_code=400, detail="消息内容不能为空")
    if not body.conversation_id:
        raise HTTPException(status_code=400, detail="conversation_id 不能为空")

    worker = manager.workers.get(account_id)
    session = _build_account_im_session(account)
    last_error = ""

    if not (worker and worker._im_service) and not _account_has_cookie(account):
        raise HTTPException(status_code=400, detail="账号未保存 Cookie，无法发送私信")

    async def _do_send() -> bool:
        nonlocal last_error
        if worker and worker._im_service:
            ok = await worker._im_service.send_message(body.conversation_id, content)
            last_error = worker._im_service.last_error or ""
            if ok:
                await _persist_im_session_data(account_id, worker._im_service.session, db)
            return ok
        async with DouyinImHttpClient(session, account_id=account_id) as http:
            ok = await http.send_text_message(body.conversation_id, content)
            last_error = http.last_error or ""
            if ok:
                session.conv_meta = http.session.conv_meta
                await _persist_im_session_data(account_id, session, db)
            return ok

    try:
        sent = await asyncio.wait_for(_do_send(), timeout=45)
    except asyncio.TimeoutError:
        # 云服务器到抖音的网络偶发缓慢，整条发送链路（解析 ticket + 签名 + 发送）
        # 可能超过 nginx 上游超时而直接 504。这里限定整体时长，超时返回受控错误。
        sent = False
        last_error = last_error or "发送超时"
        try:
            await db.rollback()
        except Exception:
            pass
        failed_log = MessageLog(
            account_id=account_id,
            sender_name="[系统发送]",
            sender_id=_resolve_log_peer_id(session, body.conversation_id),
            message_content=content,
            reply_content=content,
            status="failed",
            error_message="发送超时（抖音接口响应缓慢），请稍后重试",
        )
        db.add(failed_log)
        try:
            await db.commit()
        except Exception:
            await db.rollback()
        return SendMessageResponse(
            success=False,
            message="发送超时（抖音接口响应缓慢），请稍后重试",
        )

    if sent:
        log = MessageLog(
            account_id=account_id,
            sender_name="[系统发送]",
            sender_id=_resolve_log_peer_id(session, body.conversation_id),
            message_content=content,
            reply_content=content,
            status="sent",
        )
        db.add(log)
        await db.commit()
        system_logger.record(
            "手动发送私信成功",
            detail=f"会话 {body.conversation_id}：{message_preview(content)}",
            level="success",
            category="send",
            account_id=account_id,
        )
        return SendMessageResponse(success=True, message="私信发送成功")

    failed_log = MessageLog(
        account_id=account_id,
        sender_name="[系统发送]",
        sender_id=_resolve_log_peer_id(session, body.conversation_id),
        message_content=content,
        reply_content=content,
        status="failed",
        error_message=last_error or "私信发送失败",
    )
    db.add(failed_log)
    await db.commit()

    need_browser = (
        not session.keys_str
        or not session.web_protect_str
        or not (session.conv_meta or {})
        or "ticket" in (last_error or "")
        or "签名密钥" in (last_error or "")
        or "web_protect" in (last_error or "")
        or last_error == "INVALID_REQUEST"
    )
    if need_browser:
        msg = last_error or "缺少 IM 签名密钥"
        if last_error == "INVALID_REQUEST":
            msg = "IM 会话创建失败(INVALID_REQUEST)，请停止托管后用浏览器模式重新登录并打开私信页"
        return SendMessageResponse(
            success=False,
            need_browser_login=True,
            message=msg,
        )
    return SendMessageResponse(
        success=False,
        message=last_error or "私信发送失败，请确认 Cookie/IM 会话仍有效",
    )


if SERVE_WEB:
    if not mount_frontend(app, STATIC_DIR):
        print(
            f"[kefu] 未找到前端构建目录 {STATIC_DIR}，当前仅提供 API。"
            " 请执行 install.sh / install.bat 或 cd frontend && npm run build"
        )