📺 演示视频
✨ 核心功能
Chrome CDP 连接
连接已启动的浏览器，无需重新登录账户，保持登录状态，支持多账户同时操作。
异步并发处理
使用 asyncio 实现多渠道并发操作，效率提升10倍，大幅缩短投放时间。
智能账户选择
从 CSV 文件按渠道随机选择账户，避免重复，支持账户池管理。
实时状态监控
可视化展示各渠道运行状态，支持全局控制，一键暂停/继续/停止。
🛠️ 技术栈
本项目采用 Python 作为主要开发语言，结合 Playwright 实现浏览器自动化，使用 Asyncio 实现异步并发处理，通过 Tkinter 构建 GUI 界面。
Python 3.10+
Playwright
Asyncio
Tkinter
Chrome CDP
Threading
💻 核心代码展示
async_event_loop.py
class AsyncioEventLoopThread(threading.Thread):
"""在单独线程中运行 asyncio 事件循环"""
def __init__(self):
super().__init__(daemon=True)
self.loop = None
self.started = threading.Event()
def run(self):
self.loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.loop)
self.started.set()
self.loop.run_forever()
def run_coro(self, coro):
if not self.started.is_set():
self.started.wait(timeout=5)
future = asyncio.run_coroutine_threadsafe(
coro, self.loop
)
return future.result()
chrome_cdp_connection.py
async def async_connect_to_chrome(self):
"""异步连接到Chrome浏览器"""
global browser, playwright_instance
if browser is not None:
if browser.is_connected():
return browser
browser = None
if playwright_instance is None:
playwright_instance = await async_playwright().start()
browser = await playwright_instance.chromium.connect_over_cdp(
CHROME_REMOTE_DEBUGGING_URL
)
return browser
account_selector.py
class AccountSelector:
"""账户选择器 - 从CSV按渠道随机选择"""
def __init__(self, csv_path: str = "zhanghu.csv"):
self.csv_path = csv_path
self.channel_accounts = {}
self.load_csv()
def get_accounts_by_channel(
self, channel_name: str, count: int = 1
) -> List[str]:
if channel_name not in self.channel_accounts:
return []
available = self.channel_accounts[channel_name]
return random.sample(available, min(count, len(available)))
channels.json
{
"channels": [
{
"id": "10001598760",
"name": "番茄2.9系数",
"description": "漫剧iaa"
},
{
"id": "10001598809",
"name": "番茄2.9直投",
"description": "漫剧iaa"
},
{
"id": "10001599003",
"name": "番茄9.9系数",
"description": "漫剧iaa"
}
]
}