三个人的工作室,每月工资两万四,串号事故不断。现在:一台电脑,点一下启动,零串号,一个人看仪表盘就够了。
去年秋天,一个做TikTok美区的老客户半夜给我发来一段语音。
背景音里有人在骂骂咧咧,他声音压得很低:“林工,运营又把A店的货用B店的库存发出去了,亏了三千多。我实在受不了了,你能不能给我写套真的能隔离环境的系统?钱不是问题。”
我说:“你先别急,明天把你们现在的操作流程发给我看看。”
第二天他发来一个录屏。视频里,运营小哥同时开着二十多个Chrome窗口,每个窗口标签页写着不同的店铺名。切号的时候,鼠标在几个窗口间反复横跳,复制cookie、打开开发者工具、粘贴、刷新……
我看着都觉得累。
这还不是最要命的。最要命的是,这种重复劳动每天要占用一个人四五个小时,而且不管你多小心,一个月总会有那么几次串号。轻则限流,重则连坐封店。
市面上的群控软件他试过三款。第一款用了一周被风控,第二款就是个按键精灵套壳,第三款直接要求把账号密码托管到云端——他吓得连夜改密码。
我放下手机,想了一整夜。
第二天早晨我给他回了一条消息:“三个月后给你方案。”
那三个月里,我把自己关在书房,用Python写了一套带UI的店群自动化管理系统,取名Alien。
今天这篇文章,我把Alien从0到1的完整复盘写出来。不吹牛,不藏私,全是真实代码、架构设计和一线踩坑记录。
一、店群行业最痛的“体力活”:切号、切号、还是切号
先还原一个真实场景。
你手里有80个拼多多店铺。每天早晨,你要做三件事:
- 登录所有店铺后台,检查订单和留言(80个店 × 1.5分钟 = 120分钟)
- 导出前一天的订单,手工对账(约1小时)
3. 上架新品,一个一个店重复操作(约2小时)
加起来,一天5个小时纯手工。这还不算处理售后、改价、报活动的时间。
所以店群老板普遍的做法是:招人。一个运营月薪8000,最多管30个店。80个店至少要3个人。加上社保、电脑、代理IP,一个月硬成本3万以上。
更难受的是,人不是机器。切号的时候忘了清缓存,串了;复制cookie的时候多复制了一行,串了;两个窗口同时操作没注意,串了。
串号的后果是什么?平台检测到多个账号在同一设备/同一IP/同一浏览器指纹下登录,判定为“关联”,轻则限流,重则全部封店。
一夜回到解放前,不是开玩笑。
通用的脚本更不靠谱。几百块钱一套,本质就是个按键精灵。平台页面改个按钮位置,脚本就废了。风控系统稍微升级,账号就封一片。
所以我的判断是:必须从底层重新设计一套系统。 让软件代替人做环境隔离和任务调度,影刀RPA只负责执行具体操作。
二、Alien架构:三层分离,各司其职
Alien不是要替代影刀RPA。恰恰相反,影刀是我最信任的“执行单元”。它的流程录制、元素识别、异常重试,比自己手写selenium稳定十倍。
Alien做的是影刀不擅长的两件事:
- 环境隔离:为每个店铺创建独立的浏览器profile,注入随机指纹,绑定独立代理。
- 任务调度:控制同时运行的窗口数量(默认22个),管理任务队列,自动回收资源。
架构图(文字版):
┌──────────────────────────────────────┐
│ PyQt6 控制面板(UI层) │
│ 环境管理 │ 编排流设计 │ 实时日志 │
└─────────────────┬────────────────────┘
│
┌─────────────────▼────────────────────┐
│ Python 调度核心(逻辑层) │
│ 并发控制(Semaphore) │ 优先级队列 │
│ 环境生命周期 │ 失败重试 │
└─────────────────┬────────────────────┘
│
┌─────────────────▼────────────────────┐
│ 浏览器隔离矩阵(环境层) │
│ profile_001 │ profile_002 │ ... │
│ 独立指纹 │ 独立代理 │ 独立缓存 │
└─────────────────┬────────────────────┘
│
┌─────────────────▼────────────────────┐
│ 影刀RPA(执行层) │
│ 附着到调试端口,执行.flow文件 │
└──────────────────────────────────────┘
用户只看到第一个层。下面三层全自动运行。
三、核心模块A:浏览器环境隔离矩阵——每个店铺都是一台“虚拟电脑”
3.1 UI设计:让老板一眼看懂“隔离”在哪里
第一版Alien是命令行。我用print输出“环境1启动成功”。客户看了五分钟,皱着眉头问:“这玩意靠谱吗?”
我明白了:UI就是信任。
我用PyQt6重写了整个前端。主界面分三块:
左侧分组树:无限级嵌套。老板可以建“TK美区”、“TK东南亚”、“拼多多主站”。拖拽卡片就能移动店铺到不同分组。
中间卡片墙:每张卡片一个店铺。卡片上只放老板最关心的信息:
- 店铺名称(大号字体)
- 代理IP的国家(国旗emoji)
- Cookie有效期(绿色≥7天,黄色3-7天,红色≤2天)
- 状态灯:🟢空闲 🟡执行中 🔴异常
鼠标悬停时,卡片放大,显示更多信息:最后登录时间、今日任务次数。
右侧操作面板:底部三个核心按钮——批量导入、手动打开、批量校验。
- 批量导入:下载Excel模板,填上店铺名、平台、代理IP。一键导入,Alien自动为每个店铺创建独立profile,注入随机指纹,测试代理连通性。一百个店,三分钟。
- 手动打开:选中一个或多个店铺,点击后Alien会用各自的独立profile启动Chrome窗口。老板可以亲自进去查看后台,用完直接关掉。这个功能被老板们评为“最实用的功能”——因为信任需要亲眼看到。
- 批量校验:一键检查所有店铺的cookie和代理是否有效,失效的标红。
有个客户第一次用批量导入时,盯着进度条看了三十秒,然后说:“我以前招个人建环境,搞了两天还错了一堆。你这三分钟完事了?”
3.2 技术拆解:如何做到“店铺级”的独立环境
核心思想:每个店铺对应一个独立的Chrome用户数据目录(user-data-dir)。目录里不仅存cookie和缓存,还包含经过修改的Preferences文件,用来伪造浏览器指纹。
创建新环境的代码:
import shutil
import json
import random
from pathlib import Path
from typing import Optional
class EnvironmentFactory:
BLANK_PROFILE = Path("./assets/blank_profile") # 干净的Chrome模板
PROFILES_ROOT = Path("./alien_profiles")
@classmethod
def create(cls, env_id: str, platform: str, proxy: Optional[str] = None) -> Path:
"""创建一个隔离环境,返回profile路径"""
target = cls.PROFILES_ROOT / env_id
# 如果已存在,先备份再删除(防止残留损坏数据)
if target.exists():
backup = target.with_suffix(".bak")
shutil.move(str(target), str(backup))
shutil.rmtree(str(backup), ignore_errors=True)
# 从空白模板拷贝
shutil.copytree(cls.BLANK_PROFILE, target)
# 注入随机指纹
cls._inject_fingerprint(target, platform)
# 写入代理配置
if proxy:
(target / "proxy.txt").write_text(proxy, encoding="utf-8")
# 写入环境元数据
meta = {"env_id": env_id, "platform": platform, "created_at": str(Path.cwd())}
(target / "meta.json").write_text(json.dumps(meta, indent=2))
return target
@classmethod
def _inject_fingerprint(cls, profile_path: Path, platform: str):
"""修改Preferences,伪造浏览器指纹"""
prefs_file = profile_path / "Default" / "Preferences"
if not prefs_file.exists():
return
with open(prefs_file, 'r', encoding='utf-8') as f:
prefs = json.load(f)
# 1. 屏幕分辨率随机化(不同店铺看起来不是同一台电脑)
widths = [1366, 1440, 1536, 1600, 1920, 2560]
width = random.choice(widths)
height = int(width * 0.5625) + random.randint(-20, 20)
prefs.setdefault("web_prefs", {})["screen"] = {
"width": width,
"height": height,
}
# 2. 硬件并发数随机(模拟不同配置的电脑)
prefs["hardware_concurrency"] = random.choice([2, 4, 6, 8, 12])
# 3. WebGL渲染器随机(避免统一显示“SwiftShader”)
renderers = ["ANGLE (NVIDIA)", "ANGLE (Intel)", "Google Inc.", "SwiftShader"]
prefs.setdefault("webgl", {})["renderer"] = random.choice(renderers)
# 4. 时区随机(根据代理位置,这里简化随机)
timezones = ["America/New_York", "America/Los_Angeles", "Europe/London", "Asia/Tokyo"]
prefs["timezone"] = random.choice(timezones)
with open(prefs_file, 'w', encoding='utf-8') as f:
json.dump(prefs, f, indent=2)
启动一个环境的浏览器,并让影刀能够连接:
import subprocess
import socket
import time
import sys
def launch_chrome(env_id: str, debug_port: int) -> subprocess.Popen:
profile_dir = Path(f"./alien_profiles/{env_id}")
if not profile_dir.exists():
raise FileNotFoundError(f"环境 {env_id} 不存在")
# 读取代理
proxy_file = profile_dir / "proxy.txt"
proxy = None
if proxy_file.exists():
proxy = proxy_file.read_text(encoding="utf-8").strip()
# 找到Chrome可执行文件
chrome_exe = "C:/Program Files/Google/Chrome/Application/chrome.exe"
if not Path(chrome_exe).exists():
chrome_exe = "./assets/chrome_portable/chrome.exe"
cmd = [
chrome_exe,
f"--user-data-dir={profile_dir}",
f"--remote-debugging-port={debug_port}",
"--no-first-run",
"--no-default-browser-check",
"--disable-blink-features=AutomationControlled", # 隐藏自动化特征
"--disable-features=TranslateUI,PasswordImport",
"--disable-background-networking",
"--disable-sync",
]
if proxy:
cmd.append(f"--proxy-server={proxy}")
# Windows下不显示控制台窗口
creationflags = subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, creationflags=creationflags)
# 等待端口就绪,最多10秒
for _ in range(30):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
if sock.connect_ex(('127.0.0.1', debug_port)) == 0:
sock.close()
return proc
sock.close()
time.sleep(0.3)
raise TimeoutError(f"环境 {env_id} 启动超时,端口 {debug_port} 未响应")
3.3 踩坑实录:Chrome的“修复”机制把我的指纹覆盖了
Alien上线第三天,客户打电话说:“林工,我查了十几个店的指纹,发现分辨率全一样,都是1920x1080。”
我心里咯噔一下。
远程连上去,打开一个店铺的Preferences文件,发现里面我写的指纹配置全没了,被Chrome恢复成了默认值。
查了两天才搞明白:Chrome在启动时会“验证”Preferences文件的合法性。如果某些字段的格式不对,或者值超出了合理范围,Chrome就会认为文件损坏,然后自动用默认值覆盖。
解决方案:改用Chrome官方支持的字段,并通过CDP(Chrome DevTools Protocol)在启动后二次注入部分属性。
这个bug让我熬了一个通宵。但修好之后,指纹再也没有被重置过。
四、核心模块B:调度编排——让22个窗口不乱、不抢、不崩
环境隔离好了,下一个问题:怎么让这些环境有序地执行任务?
你不能同时开100个窗口——机器扛不住。你也不能串行跑——100个店×5分钟=8小时,一天都跑不完。
4.1 并发数22:实测出来的黄金分割点
我用一台32GB内存、8核i7的物理机做了压测。
| 并发窗口 | CPU占用 | 内存占用 | 任务完成时间(100任务) | 稳定性 |
|---|---|---|---|---|
| 10 | 30% | 6GB | 50分钟 | 100% |
| 15 | 45% | 9GB | 34分钟 | 100% |
| 20 | 62% | 13GB | 26分钟 | 98% |
| 22 | 73% | 15GB | 24分钟 | 96% |
| 25 | 88% | 18GB | 23分钟 | 72% |
22是性能与稳定性的最佳平衡点。再高一点,稳定性就断崖式下跌。
Alien默认并发上限就是22。用户可以手动调高,但会收到警告。
4.2 调度器核心:信号量 + 环境锁 + 优先级队列
import threading
import queue
from dataclasses import dataclass, field
from typing import Callable, Optional
@dataclass(order=True)
class Task:
priority: int = 5 # 1最高,10最低
task_id: str = field(compare=False)
env_id: str = field(compare=False)
flow_file: str = field(compare=False)
retries: int = field(default=3, compare=False)
callback: Optional[Callable] = field(default=None, compare=False)
class AlienScheduler:
def __init__(self, max_concurrent: int = 22):
self.max_concurrent = max_concurrent
self.semaphore = threading.Semaphore(max_concurrent)
self.task_queue = queue.PriorityQueue()
self.env_locks = {} # env_id -> threading.Lock
self.running_tasks = {} # env_id -> task_id
self.lock = threading.Lock()
self._running = True
# 启动工作线程池
for i in range(max_concurrent):
t = threading.Thread(target=self._worker, name=f"Worker-{i}", daemon=True)
t.start()
def register_env(self, env_id: str):
with self.lock:
if env_id not in self.env_locks:
self.env_locks[env_id] = threading.Lock()
def submit(self, task: Task) -> bool:
with self.lock:
if task.env_id in self.running_tasks:
print(f"环境 {task.env_id} 已有任务 {self.running_tasks[task.env_id]} 正在执行,拒绝新任务")
return False
self.task_queue.put((task.priority, task))
return True
def _worker(self):
while self._running:
try:
priority, task = self.task_queue.get(timeout=1)
except queue.Empty:
continue
# 全局并发控制
self.semaphore.acquire()
# 环境独占锁
env_lock = self.env_locks.get(task.env_id)
if env_lock:
env_lock.acquire()
with self.lock:
self.running_tasks[task.env_id] = task.task_id
t = threading.Thread(target=self._execute, args=(task, env_lock))
t.start()
def _execute(self, task: Task, env_lock: Optional[threading.Lock]):
try:
success = self._run_flow_on_env(task.env_id, task.flow_file)
if not success and task.retries > 0:
task.retries -= 1
# 重试时降级优先级
self.task_queue.put((task.priority + 1, task))
else:
if task.callback:
task.callback(task, success)
finally:
with self.lock:
if self.running_tasks.get(task.env_id) == task.task_id:
del self.running_tasks[task.env_id]
if env_lock:
env_lock.release()
self.semaphore.release()
self.task_queue.task_done()
def _run_flow_on_env(self, env_id: str, flow_file: str) -> bool:
import requests
try:
resp = requests.post("http://127.0.0.1:18888/run_flow", json={
"flow_path": flow_file,
"env_id": env_id
}, timeout=300)
return resp.status_code == 200 and resp.json().get("success", False)
except Exception:
return False
双重控制的精髓:
Semaphore保证全局不超过22个任务同时执行。env_locks保证同一个店铺不会同时跑两个任务(防止状态冲突)。PriorityQueue让紧急任务可以插队。- 失败自动重试,重试时优先级降低,避免死循环。
4.3 编排流:把复杂业务变成“一键启动”
老板不想每天手动选流程。Alien的“编排流”功能允许预设一套完整的业务流水线。
界面描述(模拟截图):
- 左侧组件库:启动环境、执行影刀流程、等待、条件判断、循环、发送钉钉通知、结束。
- 中间画布:拖拽组件到画布,用鼠标连线。支持分支和循环。
- 右侧属性面板:每个组件的详细参数。
一个真实的“TikTok美区自动上货+报活动”编排流:
- 循环:遍历环境组“美区主店”
2. 执行流程:
export_products.flow(导出商品数据)- 本地脚本:运行
compress_images.py(压缩图片) - 执行流程:
batch_publish.flow(批量发布) - 条件判断:如果发布成功数 > 10,执行第6步,否则跳过
- 执行流程:
apply_activity.flow(报名活动)
- 本地脚本:运行
- 循环结束
- 发送钉钉:汇总所有店铺的执行结果
保存后,老板每天早上点一下“运行编排流”,然后去喝茶。以前需要两个人盯半天的工作,现在20分钟跑完。
五、底层工程封装:为什么客户觉得“这软件像大厂出品”
5.1 PyQt6:从黑框框到专业桌面软件
Alien的第一版是命令行。客户看了直摇头。
我用PyQt6重写了整个前端。花了两周时间死磕样式表,最终实现了:
- 深灰主题 + 蓝绿色强调色,长时间使用不累眼。
- 所有字体14px以上(老板们年纪都不小)。
- 每个操作都有进度条或加载动画。
- 实时日志用不同颜色区分INFO/WARNING/ERROR。
- 系统托盘支持,可以最小化到后台继续跑。
客户第二次看到界面时说:“这个看着正规多了。”
5.2 Nuitka打包:双击即用的黑盒交付
独立软件最大的交付难题:客户电脑上没有Python,没有依赖,甚至没有Chrome。
Nuitka是解决方案。它把Python代码编译成C++,再静态链接成单个exe。
打包命令:
nuitka --standalone --onefile ^
--windows-console-mode=disable ^
--enable-plugin=pyqt6 ^
--include-data-dir=./assets=assets ^
--include-data-dir=./blank_profile=blank_profile ^
--windows-icon-from-ico=alien.ico ^
--output-file=Alien.exe ^
main.py
生成的Alien.exe约120MB。客户双击,两秒钟出现登录窗口。没有控制台黑框,没有“缺少DLL”,不需要装任何运行库。
我还内置了便携版Chrome。如果系统检测不到Chrome,Alien会自动解压并使用内置版本。客户零配置。
5.3 安全验证:保护劳动成果
独立开发者也要吃饭。Alien做了三层简单的防护:
- 机器码绑定:读取CPU序列号+主板序列号+MAC地址,生成设备指纹。客户把指纹发给我,我用RSA私钥生成license文件(含有效期)。
- 核心模块加密:调度器和环境管理模块用Cython编译成
.pyd,反编译难度极大。 - 联网心跳(可选):每24小时验证一次license是否有效。客户也可以关闭,改用纯本地验证。
这套方案挡不住顶级黑客,但拦住了99%“拷贝到另一台电脑就能用”的想法。
六、那些年我踩过的“生产环境炸弹”
最后分享三个让我记忆深刻的故障。
炸弹1:内存泄漏,跑两天就崩
现象:Alien连续运行48小时后,内存占用从500MB涨到7GB,然后崩溃。
排查:任务结束后,虽然chrome进程关了,但Python里还保留着对subprocess.Popen对象的引用。引用计数不为零,垃圾回收不执行。
修复:在_execute的finally块里,显式del proc并调用gc.collect()。
炸弹2:代理切换时新环境用了旧IP
现象:客户反馈A店的后台IP显示在香港,但代理配的是美国。
排查:Alien启动新环境时,前一个环境的chrome进程还没完全退出。新进程启动时,发现已经有chrome进程(即将退出),于是复用了该进程,导致--proxy-server参数被忽略。
修复:启动新环境前,强制杀死所有使用同一user-data-dir的chrome进程,等待500ms,再启动。
炸弹3:Windows更新后所有环境打不开
现象:客户电脑自动更新重启后,Alien报错“找不到Chrome”。
排查:更新后Chrome的安装路径变了(从Program Files变成了Program Files (x86))。我的代码里写死了路径。
修复:动态检测Chrome。先读注册表,找不到再搜索常见目录,再找不到就用内置便携版。
每一个炸弹都让我通宵过。但也正是这些经历,让Alien从“能跑”变成了“稳跑”。
七、结语:技术是杠杆,用对了省下几十万
Alien不是什么黑科技。它就是Python、Chrome命令行参数、影刀RPA的合理组合。
但它解决了一个真实存在的、高频发生的、让人痛苦的业务问题:多店铺防关联和自动化调度。
现在这套系统跑在十几个工作室的电脑上,日均处理店铺数超过2000。累计为这些工作室节省的人力成本,保守估计超过五十万。
我做这件事的初衷很简单:不想再看到那些老板们凌晨两点还在手动切号。
如果你也在被同样的问题折磨,或者对这套架构感兴趣,欢迎交流。
我是林焱,一个还在写代码、接电话、半夜查日志的独立开发者。
(全文完)
