diff --git a/backend/main.py b/backend/main.py index 2d1187c..2958c5c 100644 --- a/backend/main.py +++ b/backend/main.py @@ -20,7 +20,7 @@ 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 import select, update, delete, text, func, case from sqlalchemy.ext.asyncio import AsyncSession from models.database import engine, Base, get_db, AsyncSessionLocal @@ -625,6 +625,15 @@ class AccountResponse(BaseModel): from_attributes = True +class DashboardAccountStatsResponse(BaseModel): + """Safe account totals shown to every authenticated dashboard user.""" + + total_accounts: int = 0 + online_accounts: int = 0 + my_accounts: int = 0 + my_online_accounts: int = 0 + + class AccountUpdate(BaseModel): phone: Optional[str] = None username: Optional[str] = None @@ -1010,6 +1019,51 @@ def _account_matches_keyword(account: Account, keyword: str) -> bool: ) +@app.get( + "/api/dashboard/account-stats", + response_model=DashboardAccountStatsResponse, +) +async def get_dashboard_account_stats( + db: AsyncSession = Depends(get_db), + user: User = Depends(get_current_user), +): + """Return public platform totals without exposing other users' accounts.""" + + result = await db.execute( + select( + func.count(Account.id).label("total_accounts"), + func.coalesce( + func.sum(case((Account.status == "online", 1), else_=0)), + 0, + ).label("online_accounts"), + func.coalesce( + func.sum(case((Account.owner_id == user.id, 1), else_=0)), + 0, + ).label("my_accounts"), + func.coalesce( + func.sum( + case( + ( + (Account.owner_id == user.id) + & (Account.status == "online"), + 1, + ), + else_=0, + ) + ), + 0, + ).label("my_online_accounts"), + ) + ) + row = result.one() + return DashboardAccountStatsResponse( + total_accounts=int(row.total_accounts or 0), + online_accounts=int(row.online_accounts or 0), + my_accounts=int(row.my_accounts or 0), + my_online_accounts=int(row.my_online_accounts or 0), + ) + + @app.get("/api/accounts") async def get_accounts( page: Optional[int] = Query(None, ge=1), diff --git a/backend/rpa_engine/douyin_im/reply_queue.py b/backend/rpa_engine/douyin_im/reply_queue.py index 60fd86b..c2b4426 100644 --- a/backend/rpa_engine/douyin_im/reply_queue.py +++ b/backend/rpa_engine/douyin_im/reply_queue.py @@ -87,20 +87,29 @@ class AccountReplyQueue: details: Optional[dict[str, Any]] = None, merge_key: str = "", merge_keys: Optional[Iterable[str]] = None, + immediate_if_idle: bool = False, ) -> int: - """Append one reply job and return its current 1-based queue position.""" + """Append one reply job and return its current 1-based queue position. + + When ``immediate_if_idle`` is enabled, the first job in a completely + idle account queue reserves a zero-second slot. Jobs arriving behind + it still reserve the configured interval, so the normal per-account + pacing resumes from the second job onward. + """ interval = max(0.0, float(delay_seconds or 0)) loop = asyncio.get_running_loop() async with self._state_lock: if not self._running or not self._task or self._task.done(): raise RuntimeError("reply queue is not running") - due_at = max(loop.time(), self._tail_due_at) + interval + queue_is_idle = self.pending_count == 0 + slot_seconds = 0.0 if immediate_if_idle and queue_is_idle else interval + due_at = max(loop.time(), self._tail_due_at) + slot_seconds self._tail_due_at = due_at self._waiting.append( _QueueItem( job_id=uuid.uuid4().hex, due_at=due_at, - slot_seconds=interval, + slot_seconds=slot_seconds, callback=callback, description=description, queued_at=time.time(), @@ -256,9 +265,10 @@ class AccountReplyQueue: item = self._waiting.pop(selected_index) shift_seconds = max(0.0, item.slot_seconds) shifted_count = 0 - for later in self._waiting[selected_index:]: - later.due_at -= shift_seconds - shifted_count += 1 + if shift_seconds > 0: + for later in self._waiting[selected_index:]: + later.due_at -= shift_seconds + shifted_count += 1 item.due_at = asyncio.get_running_loop().time() item.expedited = True diff --git a/backend/rpa_engine/douyin_im/service.py b/backend/rpa_engine/douyin_im/service.py index c2bb54d..9e06e44 100644 --- a/backend/rpa_engine/douyin_im/service.py +++ b/backend/rpa_engine/douyin_im/service.py @@ -578,19 +578,32 @@ class DouyinImService: "replies": list(replies), }, merge_keys=queue_merge_keys, + immediate_if_idle=True, ) + scheduled_wait = 0 if position == 1 else delay_seconds logger.info( - "Queued reply to %s for account %s: position=%s interval=%ss", + "Queued reply to %s for account %s: position=%s wait=%ss interval=%ss", sender, self.account_id, position, + scheduled_wait, delay_seconds, ) + if position == 1: + queue_detail = ( + f"{sender} 是当前账号队列的首条任务,等待时间为 0 秒;" + f"后续任务仍按 {delay_seconds} 秒间隔排队。" + ) + else: + queue_detail = ( + f"{sender} 当前排在第 {position} 位;账号生效间隔为 {delay_seconds} 秒," + "后续任务继续依次排队。" + ) system_logger.record( "自动回复已进入账号队列", detail=( - f"{sender} 当前排在第 {position} 位;账号生效间隔为 {delay_seconds} 秒," - "账号内计时与排位独立;到点后再进入全局带宽队列逐条投递。" + f"{queue_detail} 账号内计时与排位独立;" + "发送时仍进入全局带宽队列逐条投递。" ), level="info", category="send", diff --git a/backend/tests/test_dashboard_account_stats.py b/backend/tests/test_dashboard_account_stats.py new file mode 100644 index 0000000..50a1cff --- /dev/null +++ b/backend/tests/test_dashboard_account_stats.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import os +import sys +import unittest +from collections import namedtuple +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import AsyncMock + + +BACKEND_DIR = Path(__file__).resolve().parents[1] +os.environ["KEFU_DB_TYPE"] = "sqlite" +os.environ["KEFU_DATABASE_URL"] = "" +os.environ["KEFU_DB_PATH"] = str(BACKEND_DIR / "kefu.db") +if str(BACKEND_DIR) not in sys.path: + sys.path.insert(0, str(BACKEND_DIR)) + +import main + + +_AggregateRow = namedtuple( + "_AggregateRow", + ( + "total_accounts", + "online_accounts", + "my_accounts", + "my_online_accounts", + ), +) + + +class _AggregateResult: + def __init__(self, row): + self._row = row + + def one(self): + return self._row + + +class DashboardAccountStatsTests(unittest.IsolatedAsyncioTestCase): + async def test_conditional_aggregate_maps_global_and_personal_counts(self): + row = _AggregateRow( + total_accounts=12, + online_accounts=5, + my_accounts=3, + my_online_accounts=2, + ) + db = SimpleNamespace( + execute=AsyncMock(return_value=_AggregateResult(row)) + ) + user = SimpleNamespace(id=42, role="operator") + + response = await main.get_dashboard_account_stats(db=db, user=user) + + self.assertIsInstance(response, main.DashboardAccountStatsResponse) + self.assertEqual(response.total_accounts, 12) + self.assertEqual(response.online_accounts, 5) + self.assertEqual(response.my_accounts, 3) + self.assertEqual(response.my_online_accounts, 2) + db.execute.assert_awaited_once() + + async def test_owner_scope_is_only_inside_personal_aggregates(self): + row = _AggregateRow( + total_accounts=8, + online_accounts=4, + my_accounts=2, + my_online_accounts=1, + ) + db = SimpleNamespace( + execute=AsyncMock(return_value=_AggregateResult(row)) + ) + user = SimpleNamespace(id=73, role="viewer") + + await main.get_dashboard_account_stats(db=db, user=user) + + statement = db.execute.await_args.args[0] + sql = " ".join(str(statement).lower().split()) + compiled_params = list(statement.compile().params.values()) + + # All roles receive the same global totals. The current user id may + # appear in CASE expressions for the two personal counters, but must + # never filter the entire aggregate query through a global WHERE. + self.assertIn("owner_id", sql) + self.assertGreaterEqual(sql.count("case when"), 3) + self.assertEqual(sql.count("accounts.owner_id"), 2) + self.assertIn(73, compiled_params) + self.assertNotIn(" where ", f" {sql} ") + self.assertEqual(db.execute.await_count, 1) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_reply_queue.py b/backend/tests/test_reply_queue.py index fd5f1d4..0d6b590 100644 --- a/backend/tests/test_reply_queue.py +++ b/backend/tests/test_reply_queue.py @@ -30,36 +30,127 @@ class AccountReplyQueueTests(unittest.IsolatedAsyncioTestCase): await asyncio.sleep(0.002) self.assertEqual(queue.pending_count, 0) - async def test_one_account_runs_three_jobs_at_successive_fifo_slots(self): + async def test_immediate_if_idle_runs_first_now_then_successive_fifo_slots(self): queue = await self._start_queue(account_id=101) interval = 0.05 loop = asyncio.get_running_loop() started_at = loop.time() calls: list[tuple[int, float]] = [] finished = asyncio.Event() + first_started = asyncio.Event() + release_first = asyncio.Event() def callback_for(index: int): async def callback() -> None: calls.append((index, loop.time() - started_at)) + if index == 0: + first_started.set() + await release_first.wait() if len(calls) == 3: finished.set() return callback - for index in range(3): - await queue.enqueue(interval, callback_for(index), description=str(index)) + await queue.enqueue( + interval, + callback_for(0), + description="0", + immediate_if_idle=True, + ) + await asyncio.wait_for(first_started.wait(), timeout=0.1) + for index in (1, 2): + await queue.enqueue( + interval, + callback_for(index), + description=str(index), + immediate_if_idle=True, + ) + release_first.set() await asyncio.wait_for(finished.wait(), timeout=0.75) await self._wait_until_idle(queue) self.assertEqual([index for index, _ in calls], [0, 1, 2]) elapsed = [timestamp for _, timestamp in calls] - for timestamp, expected in zip(elapsed, (interval, interval * 2, interval * 3)): + for timestamp, expected in zip(elapsed, (0, interval, interval * 2)): self.assertGreaterEqual(timestamp, expected - 0.015) self.assertLess(timestamp, expected + 0.15) self.assertGreaterEqual(elapsed[1] - elapsed[0], interval - 0.02) self.assertGreaterEqual(elapsed[2] - elapsed[1], interval - 0.02) + async def test_immediate_if_idle_does_not_bypass_active_send(self): + queue = await self._start_queue(account_id=102) + interval = 1.0 + first_started = asyncio.Event() + release_first = asyncio.Event() + second_started = asyncio.Event() + + async def first_callback() -> None: + first_started.set() + await release_first.wait() + + async def second_callback() -> None: + second_started.set() + + await queue.enqueue( + interval, + first_callback, + description="first", + immediate_if_idle=True, + ) + await asyncio.wait_for(first_started.wait(), timeout=0.1) + await queue.enqueue( + interval, + second_callback, + description="second", + immediate_if_idle=True, + ) + + snapshot = await queue.snapshot() + self.assertEqual([item["status"] for item in snapshot], ["sending", "waiting"]) + self.assertEqual(snapshot[0]["interval_seconds"], 0) + self.assertEqual(snapshot[1]["interval_seconds"], int(interval)) + + release_first.set() + await asyncio.sleep(0.02) + self.assertFalse(second_started.is_set()) + + async def test_immediate_if_idle_resets_after_queue_drains(self): + queue = await self._start_queue(account_id=103) + interval = 1.0 + first_finished = asyncio.Event() + second_started = asyncio.Event() + release_second = asyncio.Event() + + async def first_callback() -> None: + first_finished.set() + + async def second_callback() -> None: + second_started.set() + await release_second.wait() + + await queue.enqueue( + interval, + first_callback, + description="first wave", + immediate_if_idle=True, + ) + await asyncio.wait_for(first_finished.wait(), timeout=0.1) + await self._wait_until_idle(queue) + + await queue.enqueue( + interval, + second_callback, + description="second wave", + immediate_if_idle=True, + ) + await asyncio.wait_for(second_started.wait(), timeout=0.1) + snapshot = await queue.snapshot() + self.assertEqual(len(snapshot), 1) + self.assertEqual(snapshot[0]["status"], "sending") + self.assertEqual(snapshot[0]["interval_seconds"], 0) + release_second.set() + async def test_separate_account_queues_reach_first_slot_without_blocking(self): first_queue = await self._start_queue(account_id=201) second_queue = await self._start_queue(account_id=202) @@ -194,6 +285,47 @@ class AccountReplyQueueTests(unittest.IsolatedAsyncioTestCase): new_due = datetime.fromisoformat(new["scheduled_at"]).timestamp() self.assertAlmostEqual(old_due - new_due, interval, delta=0.05) + async def test_send_now_on_zero_slot_does_not_claim_later_jobs_shifted(self): + queue = await self._start_queue(account_id=512) + interval = 1.0 + + async def noop() -> None: + pass + + # Both enqueues complete without yielding to the consumer, preserving + # the narrow management-API window where the zero-slot first job is + # still waiting and can be selected by send-now. + await queue.enqueue( + interval, + noop, + description="immediate first", + immediate_if_idle=True, + ) + await queue.enqueue( + interval, + noop, + description="scheduled second", + immediate_if_idle=True, + ) + before = await queue.snapshot() + second_before = next( + item for item in before if item["description"] == "scheduled second" + ) + + result = await queue.send_now(before[0]["job_id"]) + after = await queue.snapshot() + second_after = next( + item for item in after if item["description"] == "scheduled second" + ) + + self.assertEqual(result["status"], "accepted") + self.assertEqual(result["shifted_count"], 0) + second_due_before = datetime.fromisoformat( + second_before["scheduled_at"] + ).timestamp() + second_due_after = datetime.fromisoformat(second_after["scheduled_at"]).timestamp() + self.assertAlmostEqual(second_due_after, second_due_before, delta=0.01) + async def test_send_now_middle_runs_first_and_only_shifts_jobs_behind_it(self): queue = await self._start_queue(account_id=503) interval = 0.12 diff --git a/backend/tests/test_reply_queue_integration.py b/backend/tests/test_reply_queue_integration.py index 9ce0808..de7bf90 100644 --- a/backend/tests/test_reply_queue_integration.py +++ b/backend/tests/test_reply_queue_integration.py @@ -23,7 +23,9 @@ from rpa_engine.playwright_worker import DouyinWorker class _RecordingQueue: def __init__(self) -> None: - self.jobs: list[tuple[float, object, str, dict, frozenset[str]]] = [] + self.jobs: list[ + tuple[float, object, str, dict, frozenset[str], bool] + ] = [] async def enqueue( self, @@ -33,11 +35,19 @@ class _RecordingQueue: details=None, merge_key="", merge_keys=None, + immediate_if_idle=False, ) -> int: keys = merge_keys if merge_keys is not None else [merge_key] normalized_keys = frozenset(str(key) for key in keys if str(key or "").strip()) self.jobs.append( - (delay_seconds, callback, description, dict(details or {}), normalized_keys) + ( + delay_seconds, + callback, + description, + dict(details or {}), + normalized_keys, + bool(immediate_if_idle), + ) ) return len(self.jobs) @@ -75,6 +85,7 @@ class _RecordingQueue: job[2], merged_details, frozenset(job[4] | incoming_keys), + job[5], ) return { "status": "merged", @@ -123,6 +134,8 @@ class ReplyQueueIntegrationTests(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(service._reply_queue.jobs), 1) self.assertEqual(match_reply.await_count, 1) + self.assertEqual(service._reply_queue.jobs[0][0], 60) + self.assertTrue(service._reply_queue.jobs[0][5]) details = service._reply_queue.jobs[0][3] self.assertEqual(details["sender_name"], "张三") self.assertEqual(details["conversation_id"], "conv-1") diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 6647265..e6d485f 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -63,7 +63,7 @@ const handleLogout = () => { - 账号管理 + {{ auth.isAdmin ? '账号管理' : '我的账号' }} diff --git a/frontend/src/views/Accounts.vue b/frontend/src/views/Accounts.vue index 56d536b..5550a4e 100644 --- a/frontend/src/views/Accounts.vue +++ b/frontend/src/views/Accounts.vue @@ -1,6 +1,6 @@