449 lines
20 KiB
Python
449 lines
20 KiB
Python
import asyncio
|
||
import sys
|
||
import json
|
||
from typing import List, Dict, Any
|
||
import aiohttp
|
||
|
||
from loguru import logger
|
||
|
||
from core.config import Config
|
||
from core.database import DatabaseManager
|
||
from core.logger import setup_logger
|
||
|
||
|
||
class AccountUploader:
|
||
def __init__(self):
|
||
self.config = Config.from_yaml()
|
||
self.logger = setup_logger(self.config)
|
||
self.db_manager = DatabaseManager(self.config)
|
||
self.api_url = "https://cursorapi.nosqli.com/admin/api.AutoCursor/commonadd"
|
||
self.batch_size = 100
|
||
# 添加代理设置
|
||
self.proxy = "http://1ddbeae0f7a67106fd58:f72e512b10893a1d@gw.dataimpulse.com:823"
|
||
self.use_proxy = True # 默认使用代理
|
||
|
||
async def initialize(self):
|
||
"""初始化数据库"""
|
||
await self.db_manager.initialize()
|
||
|
||
async def cleanup(self):
|
||
"""清理资源"""
|
||
await self.db_manager.cleanup()
|
||
|
||
async def get_success_accounts(self, limit: int = 100, last_id: int = 0) -> List[Dict[str, Any]]:
|
||
"""获取状态为success且未提取的账号"""
|
||
async with self.db_manager.get_connection() as conn:
|
||
cursor = await conn.execute(
|
||
"""
|
||
SELECT
|
||
id, email, password, cursor_password, cursor_cookie, cursor_token
|
||
FROM email_accounts
|
||
WHERE status = 'success' AND sold = 1 AND extracted = 0
|
||
AND id > ?
|
||
ORDER BY id ASC
|
||
LIMIT ?
|
||
""",
|
||
(last_id, limit)
|
||
)
|
||
|
||
rows = await cursor.fetchall()
|
||
|
||
accounts = []
|
||
for row in rows:
|
||
account = {
|
||
"id": row[0],
|
||
"email": row[1],
|
||
"password": row[2],
|
||
"cursor_password": row[3],
|
||
"cursor_cookie": row[4],
|
||
"cursor_token": row[5]
|
||
}
|
||
accounts.append(account)
|
||
|
||
return accounts
|
||
|
||
async def count_success_accounts(self) -> int:
|
||
"""统计未提取的成功账号数量"""
|
||
async with self.db_manager.get_connection() as conn:
|
||
cursor = await conn.execute(
|
||
"SELECT COUNT(*) FROM email_accounts WHERE status = 'success' AND sold = 1 AND extracted = 0"
|
||
)
|
||
count = (await cursor.fetchone())[0]
|
||
return count
|
||
|
||
async def mark_as_extracted(self, account_ids: List[int]) -> bool:
|
||
"""将账号标记为已提取"""
|
||
if not account_ids:
|
||
return True
|
||
|
||
try:
|
||
placeholders = ", ".join("?" for _ in account_ids)
|
||
|
||
async with self.db_manager.get_connection() as conn:
|
||
await conn.execute(
|
||
f"""
|
||
UPDATE email_accounts
|
||
SET
|
||
extracted = 1,
|
||
updated_at = CURRENT_TIMESTAMP
|
||
WHERE id IN ({placeholders})
|
||
""",
|
||
tuple(account_ids)
|
||
)
|
||
await conn.commit()
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"标记账号为已提取时出错: {e}")
|
||
return False
|
||
|
||
async def upload_accounts(self, accounts: List[Dict[str, Any]]) -> bool:
|
||
"""上传账号到API"""
|
||
if not accounts:
|
||
return True
|
||
|
||
try:
|
||
# 准备上传数据
|
||
upload_data = []
|
||
for account in accounts:
|
||
# 根据API需要的格式构建账号项
|
||
upload_item = {
|
||
"email": account["email"],
|
||
"password": account["password"], # 同时提供password字段
|
||
"email_password": account["password"], # 同时提供email_password字段
|
||
"cursor_email": account["email"],
|
||
"cursor_password": account["cursor_password"],
|
||
"cookie": account["cursor_cookie"] or "", # 确保不为None
|
||
"token": account.get("cursor_token", "")
|
||
}
|
||
# 确保所有必须字段都有值
|
||
for key, value in upload_item.items():
|
||
if value is None:
|
||
upload_item[key] = "" # 将None替换为空字符串
|
||
|
||
upload_data.append(upload_item)
|
||
|
||
# 打印上传数据的结构（去掉长字符串的详细内容）
|
||
debug_data = []
|
||
for item in upload_data[:2]: # 只打印前2个账号作为示例
|
||
debug_item = item.copy()
|
||
if "cookie" in debug_item and debug_item["cookie"]:
|
||
debug_item["cookie"] = debug_item["cookie"][:20] + "..." if len(debug_item["cookie"]) > 20 else debug_item["cookie"]
|
||
if "token" in debug_item and debug_item["token"]:
|
||
debug_item["token"] = debug_item["token"][:20] + "..." if len(debug_item["token"]) > 20 else debug_item["token"]
|
||
debug_data.append(debug_item)
|
||
|
||
self.logger.debug(f"准备上传数据示例: {json.dumps(debug_data, ensure_ascii=False)}")
|
||
self.logger.debug(f"API URL: {self.api_url}")
|
||
|
||
# 发送请求
|
||
# 使用代理创建ClientSession
|
||
connector = aiohttp.TCPConnector(ssl=False) # 禁用SSL验证以防代理问题
|
||
async with aiohttp.ClientSession(connector=connector) as session:
|
||
# 添加超时设置
|
||
timeout = aiohttp.ClientTimeout(total=60) # 增加超时时间
|
||
|
||
# 准备代理配置
|
||
proxy = self.proxy if self.use_proxy else None
|
||
if self.use_proxy:
|
||
self.logger.debug(f"通过代理发送请求: {self.proxy.split('@')[1]}")
|
||
else:
|
||
self.logger.debug("不使用代理，直接连接API")
|
||
|
||
# 根据API错误信息，确保发送的是账号数组
|
||
self.logger.debug(f"发送账号数组，共 {len(upload_data)} 条记录")
|
||
|
||
try:
|
||
# 直接发送数组格式，使用代理
|
||
async with session.post(self.api_url, json=upload_data, timeout=timeout, proxy=proxy) as response:
|
||
response_text = await response.text()
|
||
self.logger.debug(f"API响应状态码: {response.status}")
|
||
self.logger.debug(f"API响应内容: {response_text}")
|
||
|
||
if response.status != 200:
|
||
self.logger.error(f"API响应错误 - 状态码: {response.status}")
|
||
return True # 即使HTTP错误也继续处理下一批
|
||
|
||
# 解析响应
|
||
try:
|
||
result = json.loads(response_text)
|
||
except json.JSONDecodeError:
|
||
self.logger.error(f"响应不是有效的JSON: {response_text}")
|
||
return True # JSON解析错误也继续处理下一批
|
||
|
||
# 判断上传是否成功 - 修改判断逻辑
|
||
# API返回code为0表示成功
|
||
if result.get("code") == 0:
|
||
# 检查data中的成功计数
|
||
success_count = result.get("data", {}).get("success", 0)
|
||
failed_count = result.get("data", {}).get("failed", 0)
|
||
|
||
self.logger.info(f"API返回结果: 成功: {success_count}, 失败: {failed_count}")
|
||
|
||
if success_count > 0:
|
||
self.logger.success(f"成功上传 {success_count} 个账号")
|
||
return True
|
||
else:
|
||
# 检查详细错误信息
|
||
details = result.get("data", {}).get("details", [])
|
||
if details:
|
||
for detail in details[:3]: # 只显示前三个错误
|
||
self.logger.error(f"账号 {detail.get('email')} 上传失败: {detail.get('message')}")
|
||
# 输出更详细的错误信息，帮助诊断
|
||
self.logger.debug(f"错误账号详情: {json.dumps(detail, ensure_ascii=False)}")
|
||
|
||
self.logger.error(f"账号上传全部失败: {result.get('msg', '未知错误')}")
|
||
return True # 即使全部失败也继续处理下一批
|
||
else:
|
||
self.logger.error(f"上传失败: {result.get('msg', '未知错误')}")
|
||
|
||
# 检查是否有详细错误信息
|
||
if 'data' in result:
|
||
self.logger.error(f"错误详情: {json.dumps(result['data'], ensure_ascii=False)}")
|
||
|
||
return True # API返回错误也继续处理下一批
|
||
|
||
except aiohttp.ClientError as e:
|
||
self.logger.error(f"HTTP请求错误: {str(e)}")
|
||
return True # 网络错误也继续处理下一批
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"上传账号时出错: {str(e)}")
|
||
import traceback
|
||
self.logger.error(f"错误堆栈: {traceback.format_exc()}")
|
||
return True # 其他错误也继续处理下一批
|
||
|
||
async def process_accounts(self, max_batches: int = 0) -> int:
|
||
"""处理账号上传，max_batches=0表示处理所有批次"""
|
||
total_count = await self.count_success_accounts()
|
||
|
||
if total_count == 0:
|
||
self.logger.info("没有找到待上传的账号")
|
||
return 0
|
||
|
||
self.logger.info(f"找到 {total_count} 个待上传账号")
|
||
|
||
processed_count = 0
|
||
batch_count = 0
|
||
failed_batches = 0 # 记录失败的批次数
|
||
last_id = 0 # 记录最后处理的ID
|
||
|
||
while True:
|
||
# 检查是否达到最大批次
|
||
if max_batches > 0 and batch_count >= max_batches:
|
||
self.logger.info(f"已达到指定的最大批次数 {max_batches}")
|
||
break
|
||
|
||
# 获取一批账号
|
||
accounts = await self.get_success_accounts(self.batch_size, last_id)
|
||
|
||
if not accounts:
|
||
self.logger.info("没有更多账号可上传")
|
||
break
|
||
|
||
batch_count += 1
|
||
current_batch_size = len(accounts)
|
||
self.logger.info(f"处理第 {batch_count} 批 - {current_batch_size} 个账号")
|
||
|
||
# 上传账号
|
||
account_ids = [account["id"] for account in accounts]
|
||
upload_success = await self.upload_accounts(accounts)
|
||
|
||
if upload_success:
|
||
# 标记为已提取
|
||
mark_success = await self.mark_as_extracted(account_ids)
|
||
|
||
if mark_success:
|
||
processed_count += current_batch_size
|
||
self.logger.success(f"第 {batch_count} 批处理完成，累计处理 {processed_count}/{total_count}")
|
||
failed_batches = 0 # 重置失败计数
|
||
# 更新最后处理的ID
|
||
last_id = max(account_ids)
|
||
else:
|
||
self.logger.error(f"第 {batch_count} 批标记已提取失败")
|
||
failed_batches += 1
|
||
else:
|
||
self.logger.error(f"第 {batch_count} 批上传失败")
|
||
failed_batches += 1
|
||
|
||
# 如果连续失败次数过多，才中断处理
|
||
if failed_batches >= 3:
|
||
self.logger.error(f"连续 {failed_batches} 批处理失败，停止处理")
|
||
break
|
||
|
||
# 简单的延迟，避免请求过快
|
||
await asyncio.sleep(1)
|
||
|
||
return processed_count
|
||
|
||
async def test_api_connection(self) -> bool:
|
||
"""测试API连接是否正常"""
|
||
self.logger.info(f"正在测试API连接: {self.api_url}")
|
||
self.logger.info(f"使用代理: {'使用代理' if self.use_proxy else '不使用代理'}")
|
||
|
||
try:
|
||
# 准备一个简单的测试数据
|
||
test_data = [
|
||
{
|
||
"email": "test@example.com",
|
||
"password": "test_password",
|
||
"email_password": "test_password", # 同时提供两个字段
|
||
"cursor_email": "test@example.com",
|
||
"cursor_password": "test_cursor_password",
|
||
"cookie": "test_cookie",
|
||
"token": "test_token"
|
||
}
|
||
]
|
||
|
||
# 使用代理创建ClientSession
|
||
connector = aiohttp.TCPConnector(ssl=False) # 禁用SSL验证以防代理问题
|
||
async with aiohttp.ClientSession(connector=connector) as session:
|
||
# 准备代理配置
|
||
proxy = self.proxy if self.use_proxy else None
|
||
|
||
# 尝试HEAD请求检查服务器是否可达
|
||
try:
|
||
self.logger.debug("尝试HEAD请求...")
|
||
async with session.head(self.api_url, timeout=aiohttp.ClientTimeout(total=10), proxy=proxy) as response:
|
||
self.logger.debug(f"HEAD响应状态码: {response.status}")
|
||
if response.status >= 400:
|
||
self.logger.error(f"API服务器响应错误: {response.status}")
|
||
return False
|
||
except Exception as e:
|
||
self.logger.warning(f"HEAD请求失败: {str(e)}，尝试GET请求...")
|
||
|
||
# 尝试GET请求
|
||
try:
|
||
base_url = self.api_url.split('/admin')[0]
|
||
ping_url = f"{base_url}/ping"
|
||
self.logger.debug(f"尝试GET请求检查服务健康: {ping_url}")
|
||
async with session.get(ping_url, timeout=aiohttp.ClientTimeout(total=10), proxy=proxy) as response:
|
||
self.logger.debug(f"GET响应状态码: {response.status}")
|
||
if response.status == 200:
|
||
response_text = await response.text()
|
||
self.logger.debug(f"GET响应内容: {response_text}")
|
||
return True
|
||
except Exception as e:
|
||
self.logger.warning(f"GET请求失败: {str(e)}")
|
||
|
||
# 尝试OPTIONS请求获取允许的HTTP方法
|
||
try:
|
||
self.logger.debug("尝试OPTIONS请求...")
|
||
async with session.options(self.api_url, timeout=aiohttp.ClientTimeout(total=10), proxy=proxy) as response:
|
||
self.logger.debug(f"OPTIONS响应状态码: {response.status}")
|
||
if response.status == 200:
|
||
allowed_methods = response.headers.get('Allow', '')
|
||
self.logger.debug(f"允许的HTTP方法: {allowed_methods}")
|
||
return 'POST' in allowed_methods
|
||
except Exception as e:
|
||
self.logger.warning(f"OPTIONS请求失败: {str(e)}")
|
||
|
||
# 最后尝试一个小请求
|
||
self.logger.debug("尝试轻量级POST请求...")
|
||
try:
|
||
# 不同格式的请求体
|
||
formats = [
|
||
{"test": "connection"},
|
||
[{"test": "connection"}],
|
||
"test"
|
||
]
|
||
|
||
for i, payload in enumerate(formats):
|
||
try:
|
||
async with session.post(self.api_url, json=payload, timeout=aiohttp.ClientTimeout(total=10), proxy=proxy) as response:
|
||
self.logger.debug(f"POST格式{i+1}响应状态码: {response.status}")
|
||
response_text = await response.text()
|
||
self.logger.debug(f"POST格式{i+1}响应内容: {response_text[:200]}...")
|
||
|
||
# 即使返回错误也是说明服务器可达
|
||
if response.status != 0:
|
||
return True
|
||
except Exception as e:
|
||
self.logger.debug(f"POST格式{i+1}失败: {str(e)}")
|
||
|
||
return False
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"测试POST请求失败: {str(e)}")
|
||
return False
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"API连接测试失败: {str(e)}")
|
||
return False
|
||
|
||
def set_proxy_usage(self, use_proxy: bool):
|
||
"""设置是否使用代理"""
|
||
self.use_proxy = use_proxy
|
||
if use_proxy:
|
||
self.logger.info(f"已启用HTTP代理: {self.proxy.split('@')[1]}")
|
||
else:
|
||
self.logger.info("已禁用HTTP代理，将直接连接")
|
||
|
||
|
||
async def main():
|
||
# 解析命令行参数
|
||
import argparse
|
||
parser = argparse.ArgumentParser(description="上传成功账号到API并标记为已提取")
|
||
parser.add_argument("--batches", type=int, default=0, help="处理的最大批次数，0表示处理所有批次")
|
||
parser.add_argument("--batch-size", type=int, default=100, help="每批处理的账号数量")
|
||
parser.add_argument("--dry-run", action="store_true", help="预览模式，不实际上传和更新")
|
||
parser.add_argument("--test-api", action="store_true", help="测试API连接")
|
||
parser.add_argument("--no-proxy", action="store_true", help="不使用代理直接连接")
|
||
|
||
args = parser.parse_args()
|
||
|
||
# 初始化
|
||
uploader = AccountUploader()
|
||
await uploader.initialize()
|
||
|
||
if args.batch_size > 0:
|
||
uploader.batch_size = args.batch_size
|
||
|
||
# 设置是否使用代理
|
||
uploader.set_proxy_usage(not args.no_proxy)
|
||
|
||
try:
|
||
# 测试API连接
|
||
if args.test_api:
|
||
logger.info("开始测试API连接...")
|
||
is_connected = await uploader.test_api_connection()
|
||
if is_connected:
|
||
logger.success("API连接测试成功！")
|
||
else:
|
||
logger.error("API连接测试失败！")
|
||
return
|
||
|
||
# 预览模式
|
||
if args.dry_run:
|
||
total_count = await uploader.count_success_accounts()
|
||
if total_count == 0:
|
||
logger.info("没有找到待上传的账号")
|
||
return
|
||
|
||
logger.info(f"预览模式：找到 {total_count} 个待上传账号")
|
||
|
||
# 获取示例账号
|
||
accounts = await uploader.get_success_accounts(min(5, total_count))
|
||
logger.info(f"示例账号 ({len(accounts)}/{total_count}):")
|
||
|
||
for account in accounts:
|
||
logger.info(f"ID: {account['id']}, 邮箱: {account['email']}, Cursor密码: {account['cursor_password']}")
|
||
|
||
batch_count = (total_count + uploader.batch_size - 1) // uploader.batch_size
|
||
logger.info(f"预计分 {batch_count} 批上传，每批 {uploader.batch_size} 个账号")
|
||
return
|
||
|
||
# 实际处理
|
||
processed = await uploader.process_accounts(args.batches)
|
||
logger.success(f"处理完成，共上传 {processed} 个账号")
|
||
|
||
except Exception as e:
|
||
logger.error(f"程序执行出错: {str(e)}")
|
||
finally:
|
||
await uploader.cleanup()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|