feat(weibo): 增加强制分配权限与账号详情

tests/test_weibo_accounts.py

@@ -31,6 +31,7 @@ from web.backend.routers.weibo import (
     create_bind_qrcodes,
     disable_account,
     enable_account,
+    force_assign_accounts,
     get_account,
     import_cookies,
     import_workbench_accounts_by_text,
@@ -52,6 +53,7 @@ from web.backend.schemas import (
     WeiboBatchAssignRequest,
     WeiboCookieImport,
     WeiboExchangeRequest,
+    WeiboForceAssignRequest,
     WeiboPaymentStatusRequest,
     WeiboRechargeRequest,
     WeiboWorkbenchAccountsRequest,
@@ -999,6 +1001,11 @@ def test_weibo_batch_assign_cleans_previous_workbench(db):
     )
     db.add_all([operator, old_support, new_support, admin])
     db.commit()
+    operator.custom_permissions = [
+        *ROLE_PERMISSIONS["operation"],
+        "weibo:assign:force",
+    ]
+    db.commit()
     account = WeiboAccount(
         uid="4014000311",
         nickname="待改分配",
@@ -1021,10 +1028,11 @@ def test_weibo_batch_assign_cleans_previous_workbench(db):
     )
     db.commit()
 
-    result = batch_assign_accounts(
-        WeiboBatchAssignRequest(
+    result = force_assign_accounts(
+        WeiboForceAssignRequest(
             account_ids=[account.id],
             assigned_to=new_support.id,
+            expected_count=1,
         ),
         db=db,
         current=operator,
@@ -1054,7 +1062,7 @@ def test_weibo_batch_assign_cleans_previous_workbench(db):
     )
     audit = (
         db.query(AuditLog)
-        .filter(AuditLog.action == "weibo:account:assign")
+        .filter(AuditLog.action == "weibo:account:force_assign")
         .order_by(AuditLog.id.desc())
         .first()
     )
@@ -1063,6 +1071,148 @@ def test_weibo_batch_assign_cleans_previous_workbench(db):
     assert "SCF=a" not in audit.detail
 
 
+def test_weibo_force_assign_requires_dedicated_permission(db):
+    old_support = User(
+        username="weibo-force-old-support",
+        password_hash="hash",
+        role="support",
+        is_active=True,
+    )
+    support = User(
+        username="weibo-force-support",
+        password_hash="hash",
+        role="support",
+        is_active=True,
+    )
+    operator = User(
+        username="weibo-force-operator",
+        password_hash="hash",
+        role="operation",
+        is_active=True,
+    )
+    db.add_all([old_support, support, operator])
+    db.commit()
+    account = WeiboAccount(
+        uid="4014000312",
+        nickname="强制分配权限",
+        cookie="SCF=a; SUB=force-permission",
+        assigned_to=old_support.id,
+    )
+    db.add(account)
+    db.commit()
+
+    assert "weibo:assign:force" not in ROLE_PERMISSIONS["operation"]
+    with pytest.raises(HTTPException) as exc_info:
+        batch_assign_accounts(
+            WeiboBatchAssignRequest(
+                account_ids=[account.id],
+                assigned_to=support.id,
+            ),
+            db=db,
+            current=operator,
+        )
+    assert exc_info.value.status_code == 403
+
+    with pytest.raises(HTTPException) as exc_info:
+        force_assign_accounts(
+            WeiboForceAssignRequest(
+                account_ids=[account.id],
+                assigned_to=support.id,
+                expected_count=1,
+            ),
+            db=db,
+            current=operator,
+        )
+    assert exc_info.value.status_code == 403
+
+    operator.custom_permissions = [
+        *ROLE_PERMISSIONS["operation"],
+        "weibo:assign:force",
+    ]
+    db.commit()
+    result = force_assign_accounts(
+        WeiboForceAssignRequest(
+            account_ids=[account.id],
+            assigned_to=support.id,
+            expected_count=1,
+        ),
+        db=db,
+        current=operator,
+    )
+
+    assert result["updated"] == 1
+    db.refresh(account)
+    assert account.assigned_to == support.id
+
+
+def test_weibo_force_assign_by_tag_is_count_guarded(db, current_user):
+    support = User(
+        username="weibo-force-tag-support",
+        password_hash="hash",
+        role="support",
+        is_active=True,
+    )
+    db.add(support)
+    db.commit()
+    tagged = [
+        WeiboAccount(
+            uid=f"40140004{index:02d}",
+            nickname=f"标签账号{index}",
+            cookie=f"SCF=a; SUB=tag-{index}",
+            tag="强制组",
+        )
+        for index in range(2)
+    ]
+    other = WeiboAccount(
+        uid="4014000499",
+        nickname="其他标签",
+        cookie="SCF=a; SUB=other-tag",
+        tag="其他组",
+    )
+    db.add_all([*tagged, other])
+    db.commit()
+
+    with pytest.raises(HTTPException) as exc_info:
+        force_assign_accounts(
+            WeiboForceAssignRequest(
+                tag="强制组",
+                assigned_to=support.id,
+                expected_count=1,
+            ),
+            db=db,
+            current=current_user,
+        )
+    assert exc_info.value.status_code == 409
+
+    result = force_assign_accounts(
+        WeiboForceAssignRequest(
+            tag="强制组",
+            assigned_to=support.id,
+            expected_count=2,
+        ),
+        db=db,
+        current=current_user,
+    )
+
+    assert result["source"] == "tag"
+    assert result["updated"] == 2
+    assigned = db.query(WeiboAccount).filter(
+        WeiboAccount.id.in_([account.id for account in tagged])
+    )
+    assert {account.assigned_to for account in assigned} == {support.id}
+    db.refresh(other)
+    assert other.assigned_to is None
+    audit = (
+        db.query(AuditLog)
+        .filter(AuditLog.action == "weibo:account:force_assign")
+        .order_by(AuditLog.id.desc())
+        .first()
+    )
+    assert audit is not None
+    assert '"source":"tag"' in audit.detail
+    assert '"tag":"强制组"' in audit.detail
+
+
 def test_support_cannot_update_another_users_account_by_reimport(db):
     support_a = User(
         username="weibo-reimport-a",