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

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

三个人的工作室,每月工资两万四,串号事故不断。现在:一台电脑,点一下启动,零串号,一个人看仪表盘就够了。

去年秋天,一个做TikTok美区的老客户半夜给我发来一段语音。

背景音里有人在骂骂咧咧,他声音压得很低:“林工,运营又把A店的货用B店的库存发出去了,亏了三千多。我实在受不了了,你能不能给我写套真的能隔离环境的系统?钱不是问题。”

我说:“你先别急,明天把你们现在的操作流程发给我看看。”

第二天他发来一个录屏。视频里,运营小哥同时开着二十多个Chrome窗口,每个窗口标签页写着不同的店铺名。切号的时候,鼠标在几个窗口间反复横跳,复制cookie、打开开发者工具、粘贴、刷新……

我看着都觉得累。

这还不是最要命的。最要命的是,这种重复劳动每天要占用一个人四五个小时,而且不管你多小心,一个月总会有那么几次串号。轻则限流,重则连坐封店。

市面上的群控软件他试过三款。第一款用了一周被风控,第二款就是个按键精灵套壳,第三款直接要求把账号密码托管到云端——他吓得连夜改密码。

我放下手机,想了一整夜。

第二天早晨我给他回了一条消息:“三个月后给你方案。”

那三个月里,我把自己关在书房,用Python写了一套带UI的店群自动化管理系统,取名Alien。

今天这篇文章,我把Alien从0到1的完整复盘写出来。不吹牛,不藏私,全是真实代码、架构设计和一线踩坑记录。

一、店群行业最痛的“体力活”:切号、切号、还是切号

先还原一个真实场景。

你手里有80个拼多多店铺。每天早晨,你要做三件事:

  1. 登录所有店铺后台,检查订单和留言(80个店 × 1.5分钟 = 120分钟)
  2. 导出前一天的订单,手工对账(约1小时)

picture.image 3. 上架新品,一个一个店重复操作(约2小时)

加起来,一天5个小时纯手工。这还不算处理售后、改价、报活动的时间。

所以店群老板普遍的做法是:招人。一个运营月薪8000,最多管30个店。80个店至少要3个人。加上社保、电脑、代理IP,一个月硬成本3万以上。

更难受的是,人不是机器。切号的时候忘了清缓存,串了;复制cookie的时候多复制了一行,串了;两个窗口同时操作没注意,串了。

picture.image

串号的后果是什么?平台检测到多个账号在同一设备/同一IP/同一浏览器指纹下登录,判定为“关联”,轻则限流,重则全部封店。

一夜回到解放前,不是开玩笑。

通用的脚本更不靠谱。几百块钱一套,本质就是个按键精灵。平台页面改个按钮位置,脚本就废了。风控系统稍微升级,账号就封一片。

所以我的判断是:必须从底层重新设计一套系统。 让软件代替人做环境隔离和任务调度,影刀RPA只负责执行具体操作。

picture.image

picture.image

二、Alien架构:三层分离,各司其职

picture.image Alien不是要替代影刀RPA。恰恰相反,影刀是我最信任的“执行单元”。它的流程录制、元素识别、异常重试,比自己手写selenium稳定十倍。

picture.image Alien做的是影刀不擅长的两件事:

  1. 环境隔离:为每个店铺创建独立的浏览器profile,注入随机指纹,绑定独立代理。
  2. 任务调度:控制同时运行的窗口数量(默认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任务)稳定性
1030%6GB50分钟100%
1545%9GB34分钟100%
2062%13GB26分钟98%
2273%15GB24分钟96%
2588%18GB23分钟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美区自动上货+报活动”编排流:

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

保存后,老板每天早上点一下“运行编排流”,然后去喝茶。以前需要两个人盯半天的工作,现在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做了三层简单的防护:

  1. 机器码绑定:读取CPU序列号+主板序列号+MAC地址,生成设备指纹。客户把指纹发给我,我用RSA私钥生成license文件(含有效期)。
  2. 核心模块加密:调度器和环境管理模块用Cython编译成.pyd,反编译难度极大。
  3. 联网心跳(可选):每24小时验证一次license是否有效。客户也可以关闭,改用纯本地验证。

这套方案挡不住顶级黑客,但拦住了99%“拷贝到另一台电脑就能用”的想法。

六、那些年我踩过的“生产环境炸弹”

最后分享三个让我记忆深刻的故障。

炸弹1:内存泄漏,跑两天就崩

现象:Alien连续运行48小时后,内存占用从500MB涨到7GB,然后崩溃。

排查:任务结束后,虽然chrome进程关了,但Python里还保留着对subprocess.Popen对象的引用。引用计数不为零,垃圾回收不执行。

修复:在_executefinally块里,显式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。累计为这些工作室节省的人力成本,保守估计超过五十万。

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

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

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

(全文完)

0
0
0
0
评论
未登录
暂无评论