一个人写了一套店群矩阵自动化软件：我是如何干掉繁琐切号流程的

三个人的工作室，每月工资两万四。现在一台电脑加我的软件，全干完了。

去年夏天，一个做TikTok美区店群的老板请我喝酒。

喝到第三瓶，他突然不说话，盯着手机屏幕发呆。我问怎么了。

“又有一个店铺关联违规。”他把手机转过来给我看，“这已经是这个月第三个了。招了三个人，每天切号、上架、对账，到头来还是串。”

他咽了口酒：“林工，你懂技术，有没有办法搞一套系统，让这些店各跑各的，别互相干扰？”

我当时没说行，也没说不行。

回去之后我想了三天，然后开始写Alien。

九个月后的今天，那家工作室已经从三个人缩减到一个人。那个人每天的工作就是：打开Alien，点一下“启动编排流”，然后去喝茶。

这篇文章，我把Alien从0到1的全过程复盘一遍。

不讲虚的，只讲硬核的实现、真实的坑、以及老板们看了会心动的功能。

一、店群老板的“体力活”困境：招人比开店还累

picture.image 很多人以为做店群的核心是选品、流量、转化。

错。

核心是别被封号。

为了“防关联”，老板们每天要做的重复劳动能写满一张A4纸：

picture.image

每天早晨登录几十个店铺，检查cookie是否过期
每个店铺必须用不同的浏览器、不同的IP、不同的缓存目录
切换店铺时，要手动清理浏览器缓存、换代理、重启浏览器
上架商品要一个一个店操作，报活动要一个一个店点进去
晚上还要对账，从几十个后台导出订单，手工合并

picture.image 这些活，招人能解决吗？能，但要花多少钱？

一个熟练的店群运营，月薪八千起。管三十个店就到头了。一百个店最少要三个人。加上社保、电脑、代理费，一个月硬成本三万。

更难受的是人还会出错。谁也不能保证连续八小时盯着几十个窗口不出岔子。

而那些几百块钱一套的“群控脚本”？我见过太多。本质就是个按键精灵，平台页面一改版就废，风控系统一升级就封一片。

picture.image 老板们不是不想自动化，是市面上的东西太烂。

picture.image 所以我决定：自己写一套。

二、破局：Python底层 + 影刀执行，各取所长

我不打算从头造一个浏览器自动化引擎。那是重复造轮子。

影刀RPA已经做得很好。它的流程录制、元素识别、异常重试，比我手写selenium稳定太多。

影刀的短板在哪？环境管理和任务调度。

影刀打开浏览器，默认用的是同一个user-data-dir。你开十个窗口，十个店的cookie全混在一起。想隔离？可以，每条流程里手动指定不同路径。但一百个店你就要写一百个条件分支，维护起来想死。

另外，影刀没有“并发控制”的概念。你同时启动十个流程，它就给你开十个浏览器。内存爆了也不管，卡死了也不重启。

所以Alien的定位很清晰：

Alien做“大脑”和“骨架”：管理所有店铺的隔离环境，控制并发数量，调度任务优先级。
影刀做“手脚”：在Alien已经开好的浏览器窗口里，执行具体的点击、输入、读取操作。

这样，影刀不需要知道有多少个店、每个店的代理是什么、指纹长什么样。它只关心一件事：这个页面上，我要点哪里，输入什么。

底层复杂的东西，Alien全包了。

三、核心模块A：浏览器环境隔离矩阵——告别串号

3.1 UI：老板不用看文档就知道怎么用

Alien的环境管理界面，设计目标只有一个：让没碰过电脑的大姐也能三分钟上手。

界面分为三个区：

左侧——分组树

支持无限级嵌套。老板可以建“TK美区”、“TK东南亚”、“拼多多主站”三大组，每个大组下再分子组，比如“美区-服装”、“美区-家居”。

右键分组可以批量操作：全组换代理、全组校验登录态、全组导出报表。

中间——环境卡片墙

每个店铺是一张独立卡片。卡片上不堆砌信息，只显示老板最关心的：

店铺名称（大号字体）
平台小图标（一眼识别）
代理IP的国家国旗
状态灯：🟢空闲 🟡执行中 🔴异常
Cookie剩余天数（绿色≥7天，黄色3-6天，红色≤2天）

鼠标悬停时，卡片轻微放大，显示更多信息：最后登录时间、今日任务次数。

右侧——操作面板

点击卡片后，右侧出现详细配置：

代理IP地址和端口（可修改）
用户代理字符串（可编辑）
时区设置
备注信息

底部的两个大按钮是所有老板的救星：

“批量导入”：下载Excel模板，填上店铺名、平台、代理IP、账号密码（可选）。点一下导入，Alien自动为每一行创建隔离环境，生成随机指纹，测试代理连通性。一百个店，三分钟。

“手动打开选中环境”：勾选一个或多个店铺，点这个按钮，Alien会用各自独立的profile启动Chrome窗口，自动应用代理和指纹。老板可以像操作普通电脑一样进去看后台。用完直接关掉，不影响下次自动化。

有个客户跟我说过一句话，我记到现在：

“我以前进一个店铺后台，要翻笔记找账号密码、手动切代理、清缓存，三分钟起步。现在勾一下，点个按钮，十秒就进去了。这省下来的时间，够我再看十个品。”

3.2 技术实现：每个店铺都是一个“独立小电脑”

核心思想：每个店铺对应一个独立的Chrome用户数据目录（user-data-dir）。目录里不仅存cookie和缓存，还包含经过修改的Preferences文件，用来伪造浏览器指纹。

创建新环境的代码：

import os
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", "Australia/Sydney"]
        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)
    
    # 等待端口就绪
    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.2)
    
    raise TimeoutError(f"环境 {env_id} 启动超时")

3.3 踩坑：导入100个店，一半代理失效

第一次给客户部署时，批量导入了87个店铺。Alien显示全部创建成功。

第二天客户打电话来：“43个店报网络错误，上不了后台。”

排查发现，导入时我只保存了代理字符串，没有做连通性测试。客户Excel里有一半代理IP已经过期或者不可用。Alien傻乎乎地都创建了环境，等到执行任务时才发现连不上。

后来我在导入流程里增加了三步：

对每个代理发送HTTP请求测试（超时5秒）
测试失败的记录到错误列表，跳过创建
导入完成后生成一份报告：成功XX个，失败XX个，失败原因（超时/拒绝连接/认证失败）

客户再导入时，盯着报告看：“哦，这几个代理是坏的，我换一下。” 再也没有全挂的情况。

四、核心模块B：自动化调度编排——让22个窗口听话地干活

4.1 并发数22：实测出来的甜点值

我专门租了三台不同配置的物理机做压测。

最优配置（32GB内存，8核i7）：

并发10个窗口：内存6GB，CPU 30%，稳定率100%
并发15个：内存9.5GB，CPU 45%，稳定率100%
并发20个：内存13GB，CPU 62%，稳定率98%
并发22个：内存15GB，CPU 73%，稳定率95%
并发25个：内存18.5GB，CPU 88%，稳定率72%（频繁超时白屏）
并发28个：内存22GB，CPU 99%，稳定率40%（随时崩溃）

22是黄金分割点。再高一点，稳定性断崖式下跌。

Alien的默认并发上限就是22。用户可以手动调高，但我会弹窗警告：“超过22可能不稳定，建议只在64GB以上内存的机器尝试。”

4.2 调度器源码：信号量控制并发 + 优先级队列 + 环境锁

import threading
import queue
import time
from dataclasses import dataclass, field
from typing import Callable, Optional
import logging

logger = logging.getLogger("Scheduler")

@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=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:
                logger.warning(f"环境 {task.env_id} 已有任务 {self.running_tasks[task.env_id]} 正在执行，拒绝新任务")
                return False
        self.task_queue.put((task.priority, task))
        logger.info(f"任务 {task.task_id} 入队，环境 {task.env_id}，优先级 {task.priority}")
        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:
            logger.info(f"执行 {task.task_id} on {task.env_id} -> {task.flow_file}")
            success = self._run_flow(task.env_id, task.flow_file)
            
            if not success and task.retries > 0:
                task.retries -= 1
                logger.warning(f"{task.task_id} 失败，剩余重试 {task.retries}")
                # 重试时降低优先级
                self.task_queue.put((task.priority + 1, task))
            else:
                if task.callback:
                    task.callback(task, success)
                logger.info(f"{task.task_id} 完成，结果 {success}")
        except Exception as e:
            logger.error(f"{task.task_id} 异常: {e}")
        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(self, env_id: str, flow_file: str) -> bool:
        """调用影刀本地API执行流程"""
        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的“编排流”功能允许预设一套完整的业务流水线。

界面描述（截图级别还原）：

左侧组件库：启动环境、执行影刀流程、等待、条件判断、循环、发送钉钉通知、结束
中间画布：拖拽组件到画布，用鼠标连线。支持分支和循环。
右侧属性面板：每个组件的详细参数。比如“执行影刀流程”要选择.flow文件路径，“条件判断”要写表达式。

一个拼多多“自动上货+改价+报活动”的真实编排流：

循环：遍历环境组“所有主店” 2. 执行流程：export_products.flow（导出商品数据）
1. 本地脚本：运行compress_images.py（压缩图片）
2. 执行流程：batch_publish.flow（批量发布）
3. 条件判断：如果发布成功数 > 10，执行步骤6，否则跳过
4. 执行流程：apply_activity.flow（报名秒杀活动）
发送钉钉：汇总所有店铺的执行结果

保存后，老板每天只需要做一件事：打开Alien，点一下“运行编排流”，然后去喝咖啡。

有个客户把每天的工作从3小时压缩到了20分钟。他跟我说：“我原来觉得自动化很玄乎，现在觉得是我不够懒。”

五、底层工程封装：为什么客户觉得“这软件像大厂出品”

5.1 PyQt6：从黑框框到专业桌面软件

Alien的第一版是命令行。我用print输出“环境1启动成功”、“环境2执行中”。

给第一个客户演示时，他看了五分钟，皱着眉头问：“你这东西怎么连个界面都没有？靠谱吗？”

我明白了：UI就是信任。

我用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地址，生成设备指纹。客户把指纹发给我，我用私钥签名生成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、PyQt、影刀RPA的合理组合。

但它解决了一个真实存在的、高频发生的、让人痛苦的业务问题：多店铺防关联和高并发自动化。

现在这套系统跑在十几家工作室的电脑上，日均处理店铺数超过2000。累计为这些工作室节省的人力成本，保守估计超过五十万。

我做这件事的初衷很简单：不想再看到那些老板们凌晨两点还在手动切号。

如果你也在被同样的问题折磨，或者对这套架构感兴趣，欢迎交流。

我是林焱，一个还在写代码、接电话、半夜查日志的独立开发者。

（全文完）