feat: 实现清理失效账号安全方案（包含预览、SSE测试、安全删除以及本地标记清空）
This commit is contained in:
liumangmang
@@ -38,6 +38,9 @@ from app.schemas.website import (
|
||||
WebsiteBatchSyncResponse,
|
||||
OrganizeGroupsItem,
|
||||
OrganizeGroupsResponse,
|
||||
CleanupInvalidAccountsItem,
|
||||
CleanupInvalidAccountsPreviewResponse,
|
||||
CleanupInvalidAccountsExecuteResponse,
|
||||
)
|
||||
|
||||
from app.services.website_client import Sub2ApiWebsiteClient, _extract_id
|
||||
@@ -865,6 +868,230 @@ def import_upstream_keys_as_accounts(
|
||||
)
|
||||
|
||||
|
||||
def _get_cleanup_candidates(db: Session, website: Website, c: Sub2ApiWebsiteClient) -> dict[str, dict[str, Any]] | None:
|
||||
import re
|
||||
# 1. Fetch remote accounts
|
||||
remote_accounts = c.list_accounts()
|
||||
if remote_accounts is None:
|
||||
return None
|
||||
|
||||
remote_accounts_map = {}
|
||||
for acc in remote_accounts:
|
||||
aid = c.extract_id(acc)
|
||||
if aid:
|
||||
remote_accounts_map[aid] = acc
|
||||
|
||||
candidates = {}
|
||||
|
||||
# 2. Condition A: Local orphaned keys
|
||||
orphaned_keys = db.query(UpstreamGeneratedKey).filter(
|
||||
UpstreamGeneratedKey.status == "orphaned",
|
||||
UpstreamGeneratedKey.imported_website_id == website.id,
|
||||
UpstreamGeneratedKey.imported_account_id.isnot(None),
|
||||
).all()
|
||||
|
||||
for key in orphaned_keys:
|
||||
acc_id = key.imported_account_id
|
||||
remote_acc = remote_accounts_map.get(acc_id)
|
||||
acc_name = remote_acc.get("name") if remote_acc else key.key_name
|
||||
upstream = db.query(Upstream).filter(Upstream.id == key.upstream_id).first()
|
||||
upstream_name = upstream.name if upstream else None
|
||||
|
||||
candidates[acc_id] = {
|
||||
"account_id": acc_id,
|
||||
"account_name": acc_name,
|
||||
"upstream_key_id": key.id,
|
||||
"upstream_name": upstream_name,
|
||||
"source_group_name": key.group_name,
|
||||
"reason": "本地上游 Key 已被标记为 orphaned",
|
||||
"is_local_orphan": True,
|
||||
"db_key": key,
|
||||
}
|
||||
|
||||
# 3. Condition B: Remote SmartUp-marked orphan accounts
|
||||
pattern = re.compile(r"Imported by SmartUp from upstream key #(\d+)")
|
||||
for aid, acc in remote_accounts_map.items():
|
||||
if aid in candidates:
|
||||
continue
|
||||
|
||||
key_id = None
|
||||
for field in ("notes", "description", "remark"):
|
||||
val = acc.get(field)
|
||||
if val and isinstance(val, str):
|
||||
match = pattern.search(val)
|
||||
if match:
|
||||
key_id = int(match.group(1))
|
||||
break
|
||||
if key_id is None:
|
||||
continue
|
||||
|
||||
key = db.query(UpstreamGeneratedKey).filter(
|
||||
UpstreamGeneratedKey.id == key_id
|
||||
).first()
|
||||
|
||||
is_orphan = False
|
||||
reason = ""
|
||||
if not key:
|
||||
is_orphan = True
|
||||
reason = "本地找不到对应的上游 Key 记录"
|
||||
elif key.imported_website_id != website.id or key.imported_account_id != aid:
|
||||
is_orphan = True
|
||||
reason = "本地 Key 记录存在，但未关联到当前目标站的当前账号"
|
||||
elif key.status == 'orphaned':
|
||||
is_orphan = True
|
||||
reason = "本地上游 Key 已被标记为 orphaned"
|
||||
|
||||
if is_orphan:
|
||||
upstream_name = None
|
||||
source_group_name = None
|
||||
if key:
|
||||
upstream = db.query(Upstream).filter(Upstream.id == key.upstream_id).first()
|
||||
upstream_name = upstream.name if upstream else None
|
||||
source_group_name = key.group_name
|
||||
|
||||
candidates[aid] = {
|
||||
"account_id": aid,
|
||||
"account_name": acc.get("name") or aid,
|
||||
"upstream_key_id": key_id,
|
||||
"upstream_name": upstream_name,
|
||||
"source_group_name": source_group_name,
|
||||
"reason": reason,
|
||||
"is_local_orphan": False,
|
||||
"db_key": key,
|
||||
}
|
||||
|
||||
return candidates
|
||||
|
||||
|
||||
def _test_candidate_account(c: Sub2ApiWebsiteClient, account_id: str) -> tuple[str, bool, str]:
|
||||
"""返回 (test_status, cleanup_allowed, message)"""
|
||||
try:
|
||||
res = c.test_account(account_id)
|
||||
status = res.get("status")
|
||||
msg = res.get("message", "")
|
||||
if status == "404":
|
||||
return "404", True, msg
|
||||
elif status == "success":
|
||||
return "test_passed", False, msg
|
||||
elif status == "failed":
|
||||
return "failed", True, msg
|
||||
else:
|
||||
return "test_unknown", False, msg
|
||||
except Exception as e:
|
||||
return "test_unknown", False, f"测试时发生异常: {e}"
|
||||
|
||||
|
||||
@router.post("/api/websites/{wid}/accounts/cleanup-invalid/preview", response_model=CleanupInvalidAccountsPreviewResponse)
|
||||
def preview_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||||
website = db.query(Website).filter(Website.id == wid).first()
|
||||
if not website:
|
||||
raise HTTPException(404, "website not found")
|
||||
|
||||
with _client(website) as c:
|
||||
candidates = _get_cleanup_candidates(db, website, c)
|
||||
if candidates is None:
|
||||
return CleanupInvalidAccountsPreviewResponse(
|
||||
success=False,
|
||||
message="拉取远端账号列表失败，无法执行清理",
|
||||
items=[]
|
||||
)
|
||||
|
||||
items = []
|
||||
for aid, cand in candidates.items():
|
||||
test_status, cleanup_allowed, test_msg = _test_candidate_account(c, aid)
|
||||
items.append(CleanupInvalidAccountsItem(
|
||||
account_id=cand["account_id"],
|
||||
account_name=cand["account_name"],
|
||||
upstream_key_id=cand["upstream_key_id"],
|
||||
upstream_name=cand["upstream_name"],
|
||||
source_group_name=cand["source_group_name"],
|
||||
reason=cand["reason"],
|
||||
test_status=test_status,
|
||||
cleanup_allowed=cleanup_allowed,
|
||||
action_status="previewed",
|
||||
message=test_msg,
|
||||
))
|
||||
|
||||
return CleanupInvalidAccountsPreviewResponse(
|
||||
success=True,
|
||||
message=f"已成功加载 {len(items)} 个失效账号候选对象",
|
||||
items=items
|
||||
)
|
||||
|
||||
|
||||
@router.post("/api/websites/{wid}/accounts/cleanup-invalid/execute", response_model=CleanupInvalidAccountsExecuteResponse)
|
||||
def execute_cleanup_invalid_accounts(wid: int, db: Session = Depends(get_db), _=Depends(get_current_user)):
|
||||
website = db.query(Website).filter(Website.id == wid).first()
|
||||
if not website:
|
||||
raise HTTPException(404, "website not found")
|
||||
|
||||
with _client(website) as c:
|
||||
candidates = _get_cleanup_candidates(db, website, c)
|
||||
if candidates is None:
|
||||
return CleanupInvalidAccountsExecuteResponse(
|
||||
success=False,
|
||||
message="拉取远端账号列表失败，无法执行清理",
|
||||
items=[]
|
||||
)
|
||||
|
||||
items = []
|
||||
for aid, cand in candidates.items():
|
||||
test_status, cleanup_allowed, test_msg = _test_candidate_account(c, aid)
|
||||
|
||||
action_status = "skipped"
|
||||
msg = "测试通过或未知，已跳过删除"
|
||||
|
||||
if cleanup_allowed:
|
||||
try:
|
||||
del_res = c.delete_account(aid)
|
||||
status = del_res.get("status")
|
||||
del_msg = del_res.get("message", "")
|
||||
|
||||
if status in ("deleted", "already_deleted"):
|
||||
action_status = status
|
||||
msg = del_msg
|
||||
|
||||
# Clear database flags if it's a local orphan key
|
||||
if cand["is_local_orphan"] and cand["db_key"]:
|
||||
key = cand["db_key"]
|
||||
key.imported_website_id = None
|
||||
key.imported_account_id = None
|
||||
key.imported_at = None
|
||||
key.imported_target_group_id = None
|
||||
key.imported_target_group_name = None
|
||||
db.add(key)
|
||||
else:
|
||||
action_status = "failed"
|
||||
msg = f"删除失败: {del_msg}"
|
||||
except Exception as e:
|
||||
action_status = "failed"
|
||||
msg = f"删除失败: {e}"
|
||||
else:
|
||||
msg = f"账号状态正常，无需清理 ({test_msg})"
|
||||
|
||||
items.append(CleanupInvalidAccountsItem(
|
||||
account_id=cand["account_id"],
|
||||
account_name=cand["account_name"],
|
||||
upstream_key_id=cand["upstream_key_id"],
|
||||
upstream_name=cand["upstream_name"],
|
||||
source_group_name=cand["source_group_name"],
|
||||
reason=cand["reason"],
|
||||
test_status=test_status,
|
||||
cleanup_allowed=cleanup_allowed,
|
||||
action_status=action_status,
|
||||
message=msg,
|
||||
))
|
||||
|
||||
db.commit()
|
||||
|
||||
deleted_count = sum(1 for item in items if item.action_status in ("deleted", "already_deleted"))
|
||||
return CleanupInvalidAccountsExecuteResponse(
|
||||
success=True,
|
||||
message=f"清理执行完毕，成功处理了 {deleted_count} 个失效账号",
|
||||
items=items
|
||||
)
|
||||
|
||||
|
||||
@router.post("/api/websites/{wid}/groups/organize", response_model=OrganizeGroupsResponse)
|
||||
def organize_website_groups(
|
||||
wid: int,
|
||||
@@ -1306,12 +1533,12 @@ def sync_website_group_bindings(
|
||||
success_count = sum(1 for r in results if r.status == "success")
|
||||
failed_count = sum(1 for r in results if r.status == "failed")
|
||||
skipped_count = sum(1 for r in results if r.status == "skipped")
|
||||
|
||||
|
||||
msg_parts = []
|
||||
if success_count: msg_parts.append(f"成功 {success_count}")
|
||||
if failed_count: msg_parts.append(f"失败 {failed_count}")
|
||||
if skipped_count: msg_parts.append(f"跳过 {skipped_count}")
|
||||
|
||||
|
||||
return WebsiteBatchSyncResponse(
|
||||
total=len(bindings),
|
||||
success=success_count,
|
||||
|
||||
@@ -243,3 +243,28 @@ class OrganizeGroupsResponse(BaseModel):
|
||||
success: bool
|
||||
message: str
|
||||
items: list[OrganizeGroupsItem]
|
||||
|
||||
|
||||
class CleanupInvalidAccountsItem(BaseModel):
|
||||
account_id: str
|
||||
account_name: Optional[str] = None
|
||||
upstream_key_id: Optional[int] = None
|
||||
upstream_name: Optional[str] = None
|
||||
source_group_name: Optional[str] = None
|
||||
reason: str
|
||||
test_status: str
|
||||
cleanup_allowed: bool
|
||||
action_status: Optional[str] = None
|
||||
message: str = ""
|
||||
|
||||
|
||||
class CleanupInvalidAccountsPreviewResponse(BaseModel):
|
||||
success: bool
|
||||
message: str
|
||||
items: list[CleanupInvalidAccountsItem]
|
||||
|
||||
|
||||
class CleanupInvalidAccountsExecuteResponse(BaseModel):
|
||||
success: bool
|
||||
message: str
|
||||
items: list[CleanupInvalidAccountsItem]
|
||||
|
||||
@@ -345,7 +345,7 @@ class Sub2ApiWebsiteClient:
|
||||
|
||||
def account_exists(self, account_id: str, endpoint: str = "/accounts") -> bool | None:
|
||||
"""检查目标账号是否存在。
|
||||
|
||||
|
||||
优先拉取账号列表判断：
|
||||
- 列表成功取到 → return account_id in ids（True=存在，False=已删除）
|
||||
- 列表取不到（None）→ return None（校验失败，不清本地）
|
||||
@@ -357,6 +357,82 @@ class Sub2ApiWebsiteClient:
|
||||
return None
|
||||
return account_id in ids
|
||||
|
||||
def test_account(self, account_id: str, endpoint: str = "/accounts") -> dict[str, Any]:
|
||||
"""测试远端账号可用性。利用 SSE 监听 test_complete 事件的 success 状态。
|
||||
若 404，返回 {"status": "404", "cleanup_allowed": True, "message": "账号不存在"}
|
||||
若 timeout / 无法建立连接，返回 {"status": "timeout"/"error", "cleanup_allowed": False}
|
||||
"""
|
||||
import json
|
||||
url = self._url(f"{endpoint}/{account_id}/test")
|
||||
headers = self._headers()
|
||||
# Accept text/event-stream since it's an SSE stream
|
||||
headers["Accept"] = "text/event-stream"
|
||||
|
||||
try:
|
||||
with self._client.stream("POST", url, headers=headers) as response:
|
||||
if response.status_code == 404:
|
||||
return {"status": "404", "cleanup_allowed": True, "message": "账号在远端不存在 (404)"}
|
||||
if response.status_code >= 400:
|
||||
return {"status": "http_error", "cleanup_allowed": False, "message": f"测试接口返回 HTTP {response.status_code}"}
|
||||
|
||||
# Read SSE stream line by line
|
||||
current_event = None
|
||||
success_value = None
|
||||
has_error_event = False
|
||||
|
||||
for line in response.iter_lines():
|
||||
if not line:
|
||||
continue
|
||||
if line.startswith("event:"):
|
||||
current_event = line[6:].strip()
|
||||
elif line.startswith("data:"):
|
||||
data_str = line[5:].strip()
|
||||
if current_event == "error":
|
||||
has_error_event = True
|
||||
if current_event == "test_complete" or not current_event:
|
||||
try:
|
||||
parsed = json.loads(data_str)
|
||||
if isinstance(parsed, dict):
|
||||
if "success" in parsed:
|
||||
success_value = parsed["success"]
|
||||
elif parsed.get("event") == "test_complete" and "success" in parsed:
|
||||
success_value = parsed["success"]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
if success_value is True:
|
||||
return {"status": "success", "cleanup_allowed": False, "message": "测试成功，账号有效"}
|
||||
elif success_value is False:
|
||||
return {"status": "failed", "cleanup_allowed": True, "message": "测试失败，账号失效"}
|
||||
elif has_error_event:
|
||||
return {"status": "error", "cleanup_allowed": False, "message": "测试接口返回了错误事件"}
|
||||
else:
|
||||
return {"status": "invalid_sse", "cleanup_allowed": False, "message": "未收到有效的测试完成事件"}
|
||||
|
||||
except httpx.TimeoutException as exc:
|
||||
return {"status": "timeout", "cleanup_allowed": False, "message": f"连接超时: {exc}"}
|
||||
except Exception as exc:
|
||||
return {"status": "error", "cleanup_allowed": False, "message": f"请求异常: {exc}"}
|
||||
|
||||
def delete_account(self, account_id: str, endpoint: str = "/accounts") -> dict[str, Any]:
|
||||
"""删除目标账号。
|
||||
若 404 视为 already_deleted。
|
||||
"""
|
||||
url = self._url(f"{endpoint}/{account_id}")
|
||||
headers = self._headers()
|
||||
try:
|
||||
resp = self._client.request("DELETE", url, headers=headers)
|
||||
if resp.status_code == 404:
|
||||
return {"status": "already_deleted", "message": "账号不存在或已在远端被删除"}
|
||||
resp.raise_for_status()
|
||||
return {"status": "deleted", "message": "账号已删除"}
|
||||
except httpx.HTTPStatusError as exc:
|
||||
if exc.response.status_code == 404:
|
||||
return {"status": "already_deleted", "message": "账号不存在或已在远端被删除"}
|
||||
raise WebsiteError(_friendly_http_error(exc)) from exc
|
||||
except Exception as exc:
|
||||
raise WebsiteError(f"删除账号失败: {exc}") from exc
|
||||
|
||||
@staticmethod
|
||||
def extract_id(value: Any) -> str:
|
||||
return _extract_id(value)
|
||||
|
||||
Reference in New Issue
Block a user