4

telegram_bot.py

@@ -2011,7 +2011,7 @@ class ProvisionerBot:
             pass  # 忽略编辑失败
 
     async def _finalize_import_batch(self, chat_id: int):
-        """完成批量导入，发送最终结果"""
+        """完成批量导入，发送最终结果，并自动验证 account_id"""
         async with self._import_progress_lock:
             if self._import_progress_message is None:
                 return
@@ -2027,10 +2027,187 @@ class ProvisionerBot:
             # 更新最终进度
             await self._update_import_progress(chat_id, is_final=True)
 
+            # 保存统计数据用于后续验证
+            added_count = self._import_batch_stats.get("total_added", 0)
+
             # 重置状态
             self._import_progress_message = None
             self._reset_import_batch_stats()
 
+        # 如果有新增账号，自动验证 account_id 并移除无效账号
+        if added_count > 0:
+            await self._validate_and_cleanup_accounts(chat_id)
+
+    async def _validate_and_cleanup_accounts(self, chat_id: int):
+        """验证新导入账号的 account_id，移除无效账号
+
+        Args:
+            chat_id: Telegram 聊天 ID，用于发送进度消息
+        """
+        import json
+        from pathlib import Path
+        from concurrent.futures import ThreadPoolExecutor, as_completed
+
+        # 发送开始验证的消息
+        progress_msg = await self.app.bot.send_message(
+            chat_id=chat_id,
+            text="<b>🔍 正在验证账号...</b>\n\n⏳ 获取 account_id 中...",
+            parse_mode="HTML"
+        )
+
+        try:
+            # 读取当前 team.json
+            team_json_path = Path(TEAM_JSON_FILE)
+            if not team_json_path.exists():
+                await self.app.bot.edit_message_text(
+                    chat_id=chat_id,
+                    message_id=progress_msg.message_id,
+                    text="❌ team.json 不存在",
+                    parse_mode="HTML"
+                )
+                return
+
+            with open(team_json_path, "r", encoding="utf-8") as f:
+                accounts = json.load(f)
+                if not isinstance(accounts, list):
+                    accounts = [accounts]
+
+            # 筛选需要验证的账号 (有 token 但没有 account_id)
+            accounts_to_verify = []
+            for i, acc in enumerate(accounts):
+                token = acc.get("token", "")
+                account_id = acc.get("account_id", "")
+                if token and not account_id:
+                    accounts_to_verify.append((i, acc))
+
+            if not accounts_to_verify:
+                await self.app.bot.edit_message_text(
+                    chat_id=chat_id,
+                    message_id=progress_msg.message_id,
+                    text="<b>✅ 验证完成</b>\n\n所有账号已有 account_id，无需验证",
+                    parse_mode="HTML"
+                )
+                return
+
+            total = len(accounts_to_verify)
+            await self.app.bot.edit_message_text(
+                chat_id=chat_id,
+                message_id=progress_msg.message_id,
+                text=f"<b>🔍 正在验证账号...</b>\n\n⏳ 验证 {total} 个账号的 account_id...",
+                parse_mode="HTML"
+            )
+
+            # 并行获取 account_id
+            valid_indices = set()  # 验证成功的账号索引
+            failed_accounts = []   # 验证失败的账号信息
+
+            def verify_account(idx_acc_tuple):
+                """验证单个账号"""
+                idx, acc = idx_acc_tuple
+                email = acc.get("account") or acc.get("email", "")
+                token = acc.get("token", "")
+
+                # 构造临时 team 配置用于获取 account_id
+                temp_team = {
+                    "name": email.split("@")[0] if "@" in email else f"Team{idx}",
+                    "auth_token": token,
+                    "account_id": ""
+                }
+
+                from team_service import fetch_account_id
+                account_id = fetch_account_id(temp_team, silent=True)
+
+                return idx, email, account_id
+
+            # 使用线程池并行验证
+            max_workers = min(10, total)
+            completed_count = 0
+
+            with ThreadPoolExecutor(max_workers=max_workers) as executor:
+                futures = {executor.submit(verify_account, item): item for item in accounts_to_verify}
+
+                for future in as_completed(futures):
+                    idx, email, account_id = future.result()
+                    completed_count += 1
+
+                    if account_id:
+                        # 验证成功，更新 account_id
+                        accounts[idx]["account_id"] = account_id
+                        valid_indices.add(idx)
+                    else:
+                        # 验证失败
+                        failed_accounts.append({"idx": idx, "email": email})
+
+                    # 每处理 5 个更新一次进度
+                    if completed_count % 5 == 0 or completed_count == total:
+                        try:
+                            await self.app.bot.edit_message_text(
+                                chat_id=chat_id,
+                                message_id=progress_msg.message_id,
+                                text=f"<b>🔍 正在验证账号...</b>\n\n"
+                                     f"进度: {completed_count}/{total}\n"
+                                     f"✅ 成功: {len(valid_indices)}\n"
+                                     f"❌ 失败: {len(failed_accounts)}",
+                                parse_mode="HTML"
+                            )
+                        except Exception:
+                            pass
+
+            # 移除验证失败的账号
+            if failed_accounts:
+                failed_indices = {item["idx"] for item in failed_accounts}
+                accounts = [acc for i, acc in enumerate(accounts) if i not in failed_indices]
+
+            # 保存更新后的 team.json
+            with open(team_json_path, "w", encoding="utf-8") as f:
+                json.dump(accounts, f, ensure_ascii=False, indent=2)
+
+            # 重载配置
+            reload_config()
+
+            # 发送最终结果
+            lines = [
+                "<b>✅ 账号验证完成</b>",
+                "",
+                f"📊 验证结果:",
+                f"  • 验证总数: {total}",
+                f"  • 成功: {len(valid_indices)}",
+                f"  • 失败并移除: {len(failed_accounts)}",
+                f"  • team.json 剩余: {len(accounts)}",
+            ]
+
+            if failed_accounts:
+                lines.append("")
+                lines.append("❌ 已移除的无效账号:")
+                for item in failed_accounts[:5]:  # 最多显示 5 个
+                    lines.append(f"  • {item['email']}")
+                if len(failed_accounts) > 5:
+                    lines.append(f"  ... 还有 {len(failed_accounts) - 5} 个")
+
+            lines.extend([
+                "",
+                "💡 使用 /run_all 或 /run &lt;n&gt; 开始处理"
+            ])
+
+            await self.app.bot.edit_message_text(
+                chat_id=chat_id,
+                message_id=progress_msg.message_id,
+                text="\n".join(lines),
+                parse_mode="HTML"
+            )
+
+        except Exception as e:
+            log.error(f"验证账号失败: {e}")
+            try:
+                await self.app.bot.edit_message_text(
+                    chat_id=chat_id,
+                    message_id=progress_msg.message_id,
+                    text=f"<b>❌ 验证失败</b>\n\n{str(e)}",
+                    parse_mode="HTML"
+                )
+            except Exception:
+                pass
+
     async def _import_batch_timeout_callback(self, context: ContextTypes.DEFAULT_TYPE):
         """批量导入超时回调 - 由 job_queue 调用"""
         chat_id = context.job.data.get("chat_id")
@@ -2323,10 +2500,15 @@ class ProvisionerBot:
                 f"跳过 (重复): {skipped}\n"
                 f"team.json 总数: {len(existing_accounts)}\n\n"
                 f"✅ 配置已自动刷新\n"
-                f"使用 /run_all 或 /run &lt;n&gt; 开始处理",
+                f"⏳ 正在验证账号 account_id...",
                 parse_mode="HTML"
             )
 
+            # 如果有新增账号，自动验证 account_id 并移除无效账号
+            if added > 0:
+                chat_id = update.effective_chat.id
+                await self._validate_and_cleanup_accounts(chat_id)
+
         except Exception as e:
             await update.message.reply_text(f"❌ 保存到 team.json 失败: {e}")