douyin/backend/main.py

import os
import sys
import json
import asyncio
import logging
import uuid
from datetime import datetime
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, serve_link_card_debug_guide
from web_static import mount_frontend
from pydantic import BaseModel
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, AccountVideo, AutoReplyRule, LinkCardPage, 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.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

    async def start_worker(self, account_id: int, login_mode: str = "auto"):
        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):
        if account_id in self.workers:
            worker = self.workers[account_id]
            await worker.stop()
            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("/help/link-card-debug", include_in_schema=False)
async def help_link_card_debug_page():
    """链接卡片私信本地联调指南。"""
    return serve_link_card_debug_guide()


@app.get("/api/help/link-card-debug", include_in_schema=False)
async def api_help_link_card_debug_page():
    return serve_link_card_debug_guide()


@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",
)


@app.get("/api/media/proxy")
async def proxy_media(
    url: str = Query(..., min_length=8),
):
    """代理抖音 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 any(host in parsed.netloc for host in _MEDIA_PROXY_HOSTS):
        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:
        async with httpx.AsyncClient(timeout=20, follow_redirects=True) as client:
            resp = await client.get(url, headers=headers)
            resp.raise_for_status()
    except Exception as exc:
        raise HTTPException(status_code=502, detail=f"媒体加载失败: {exc}") from exc
    content_type = resp.headers.get("content-type") or ""
    # 抖音 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(resp.content)
        if sniffed:
            content_type = sniffed
        elif not content_type:
            content_type = "application/octet-stream"
    return Response(content=resp.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.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()
    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():
    # 停止所有正在运行的 RPA 任务
    for account_id in list(manager.workers.keys()):
        await manager.stop_worker(account_id)
    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: 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
    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 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 _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 or 0),
        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 | interaction | raw_im
    im_message_type: Optional[int] = None  # protobuf message_type，如互动消息默认 8
    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 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. 账号管理接口

@app.get("/api/accounts", response_model=List[AccountResponse])
async def get_accounts(
    db: AsyncSession = Depends(get_db),
    user: User = Depends(get_current_user),
):
    result = await db.execute(accounts_for_user(user))
    accounts = result.scalars().all()
    # 更新内存中的运行状态与数据库同步，以防异常断开
    quota_stopped = False
    for acc in accounts:
        if acc.quota_disabled and manager.is_running(acc.id):
            await manager.stop_worker(acc.id)
            acc.status = "offline"
            quota_stopped = True
        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"):
            acc.status = "offline"
    if quota_stopped:
        await db.commit()
    return [_build_account_response(acc) for acc in accounts]

@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 body.reply_delay_seconds is not None:
        account.reply_delay_seconds = max(0, int(body.reply_delay_seconds))
    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/accounts/{account_id}/cookie", response_model=AccountCookieResponse)
async def get_account_cookie(
    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)
    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),
):
    account = await get_owned_account(db, user, account_id, write=True)

    try:
        parsed_data = validate_cookie_json(body.cookie_data)
        standard_json_str = json.dumps(parsed_data, ensure_ascii=False, indent=2)
        cookie_path = write_cookie_file(account_id, standard_json_str)
    except (ValueError, Exception) as e:
        raise HTTPException(status_code=400, detail=f"Cookie 格式错误: {e}")

    account.cookie_data = standard_json_str
    account.cookie_path = cookie_path
    account.cookie_updated_at = datetime.utcnow()
    account.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 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),
):
    account = await get_owned_account(db, user, account_id, write=True)

    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.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 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)
    account = await _reset_account_credentials(account_id, db)
    return {
        "message": "已清除 Cookie 与 IM 会话数据，请重新登录",
        "status": account.status,
    }


@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)

    if account.quota_disabled:
        if manager.is_running(account_id):
            await manager.stop_worker(account_id)
            account.status = "offline"
            await db.commit()
        raise HTTPException(status_code=403, detail="该账号因额度不足已被停用，请购买额度或联系管理员")

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

    if assessment.get("should_reset") and login_mode != "im_direct":
        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",
        )

    started = await manager.start_worker(account_id, login_mode=login_mode)
    if started:
        account.status = "starting"
        account.qr_code_base64 = None
        account.error_message = None
        await db.commit()

        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,
        }
    else:
        return {"status": "running", "message": "RPA worker is already running."}

@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)
    
    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", response_model=List[RuleResponse])
async def get_rules(
    account_id: Optional[int] = None,
    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)
        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())
    result = await db.execute(stmt)
    return result.scalars().all()

@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."}


@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", response_model=List[LogResponse])
async def get_logs(
    account_id: Optional[int] = None,
    limit: int = 50,
    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)
    stmt = logs_for_user(user, account_id).order_by(MessageLog.created_at.desc()).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

    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 asyncio.to_thread(
        upload_im_image,
        session,
        raw,
        filename=filename,
        content_type=file.content_type or "image/jpeg",
    )
    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 body.im_message_type is not None and content.startswith("{"):
        try:
            data = json.loads(content)
            if isinstance(data, dict):
                data["im_message_type"] = body.im_message_type
                content = json.dumps(data, ensure_ascii=False, separators=(",", ":"))
        except json.JSONDecodeError:
            pass
    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，无法发送私信")

    risk_notice = ""

    async def _do_send() -> bool:
        nonlocal last_error, risk_notice
        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 ""
            risk_notice = worker._im_service.last_send_risk_notice 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 ""
            risk_notice = http.last_send_risk_notice 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" if not risk_notice else "partial",
            error_message=risk_notice or None,
        )
        db.add(log)
        await db.commit()
        system_logger.record(
            "手动发送私信成功" if not risk_notice else "手动发送私信（内容风控未通过，已自动补发兜底内容）",
            detail=f"会话 {body.conversation_id}：{message_preview(content)}{risk_notice}",
            level="success" if not risk_notice else "warning",
            category="send",
            account_id=account_id,
        )
        reply_msg = "私信发送成功"
        if risk_notice:
            if "已自动补发" in risk_notice:
                reply_msg = (
                    "抖音内容风控未通过（raw_check_code=1），网页链接卡可能显示为空白，"
                    "系统已自动补发【封面图+标题/描述/链接文字】，对方仍能看到有效内容"
                )
            else:
                reply_msg = (
                    "消息已投递，但抖音内容风控未通过（raw_check_code=1），"
                    "对方看到的很可能是空白卡片或纯文字，并非真正的网页链接卡"
                )
        return SendMessageResponse(success=True, message=reply_msg)

    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"
        )