Python自动化实战:拒绝多店串号,独立开发带UI的浏览器指纹隔离系统复盘

阅读提醒:

  1. 本文没有AI味,是一个在店群自动化里摸爬滚打两年的人写的。
  2. 如果你觉得“这有什么难的”,欢迎你写一套更稳的,我请你喝一年咖啡。
    在这里插入图片描述
  > 3. 全文干货,没有任何“首先、其次、最后”的八股结构,只有真实的工程碎片。

我承认,最开始我低估了店群自动化的难度。
那时候接了一个朋友的私活:50个拼多多店铺,想实现自动上架和领券。
我想着,影刀RPA录个流程,循环跑不就行了?
结果第一天就被教做人。
跑到第8个店铺,浏览器弹出了“账号异常,请重新登录”。再跑两个,直接封了两个店。
朋友电话打过来,语气像极了我在达美乐拍饼时被顾客投诉——又急又气。
我连夜翻日志,发现了一个致命问题:所有店铺共用了同一个浏览器缓存
店铺A的Cookies还在,脚本就打开了店铺B的后台,店铺B误以为自己是A,操作记录全乱套了。

店群矩阵自动化突破运营极限!


在这里插入图片描述

这就是典型的“串号”。
平台风控不是傻子,同一个IP、同一套指纹、同一个缓存目录,突然出现几十个店铺的操作,不封你封谁?
从那天起,我决定:不用低代码拼凑,不用现成的指纹浏览器(太贵),从底层用Python撸一套带UI的独立环境隔离系统
三个月后,Alien 0.1版本跑通。又过了三个月,它稳定支撑了200+店铺同时在线,从不串号,从不卡死。
下面是我从这场“造轮子”运动中提炼出来的核心模块。

一、店群行业的“数字体力活”

先描述一个典型的工作室早晨:
运营小张到岗,打开Excel,里面有80个店铺的账号密码和代理IP。
她需要:登录店铺A → 检查订单 → 记录 → 退出 → 清浏览器缓存 → 切换代理 → 登录店铺B → 重复。
一个店铺平均3分钟,80个店铺就是4小时。
这还没算上架商品、回复消息、处理售后的时间。
一天下来,小张的手腕贴着膏药,眼睛布满血丝,月底拿3500底薪加一点提成。
老板也没好到哪去:招三个人,每月成本一万多,还管不住——一个疏忽就封店。
市面上通用脚本?买了,跑了,被封了。
原因很简单:那些脚本只关心“点哪里”,不关心“我是谁”。
平台只要稍微检测一下指纹、缓存、IP段,就能把脚本账号和正常账号区分开。
所以我的破局思路是:
用Python做底层调度和环境管理,用影刀RPA做执行单元,用PyQt6做让老板看得懂的界面。
从底层开始,把每个店铺包装成一个独立的“数字人格”。

二、环境隔离矩阵:让每个店铺拥有独立的“身份证”

在这里插入图片描述

2.1 软件界面长什么样

Alien的“环境管理中心”采用左右结构,符合大多数Windows软件习惯。
左侧是一棵分组树。根节点是平台(拼多多/TEMU/TikTok),下面可以自定义二级分组,比如“高权重店”“测试店”“美区店”。
右侧是一个卡片式表格,每张卡片代表一个店铺环境。
卡片上显示:

  • 店铺ID(用户自定义名称)

    • 代理IP(带国家flag图标)
    • 地理位置(自动解析代理的经纬度)
    • 健康度环形图(基于登录成功率、风控次数计算)
  • 在这里插入图片描述

    • 最后活跃时间
      鼠标悬停时,显示四个操作按钮:打开环境编辑配置运行任务删除

temu店群自动化报活动案例

顶部工具栏有:批量导入(CSV模板)、新建分组批量打开选中导出报表
“批量导入模板”只有三列:shop_id, platform, proxy。其他指纹参数(分辨率、时区、语言)系统根据shop_id的哈希值自动生成,保证不同店铺差异足够大,同一店铺每次一致。
“手动打开选中环境”是老板最爱。选中任意一个或几个店铺,点击按钮,系统会为每个店铺启动一个独立的Chrome窗口,窗口标题显示店铺ID,位置自动平铺不重叠。运营可以直接在这些窗口里手动操作,就像操作自己的电脑一样。

2.2 技术深度:Profile隔离 + 指纹稳定生成

环境隔离的核心是Chromium的 --user-data-dir
每个店铺启动时,指定一个完全独立的用户数据目录。这个目录里包含Cookies、LocalStorage、IndexedDB、缓存、插件状态等一切。
只要目录不共享,两个店铺之间就没有任何数据交叉。
但是,仅有目录隔离还不够。平台还会收集屏幕分辨率、时区、语言、WebGL信息等。如果两个店铺这些特征完全一样,仍然容易被关
在这里插入图片描述
联。
所以需要为每个店铺生成一套“指纹”,并在启动浏览器时通过命令行参数或JS注入的方式生效。
下面是我写的 EnvironmentManager 核心类(实际工程中精简过但保留了骨架):

  import os
    import json
      import hashlib
        import random
          from pathlib import Path
            from typing import Optional
class EnvironmentManager:
      DATA_ROOT = Path("./AlienEnvironments")
            
                  def __init__(self, shop_id: str, platform: str):
                            self.shop_id = shop_id
                                      self.platform = platform
                                                self.env_dir = self.DATA_ROOT / platform / shop_id
                                                          self.profile_dir = self.env_dir / "chromium_data"
                                                                    self.config_path = self.env_dir / "env_config.json"
                                                                          
                                                                                def create(self, proxy: str, group: str = "default") -> str:
                                                                                          """创建店铺环境,返回profile目录路径"""
                                                                                                    self.env_dir.mkdir(parents=True, exist_ok=True)
                                                                                                              self.profile_dir.mkdir(exist_ok=True)
                                                                                                                        
                                                                                                                                  # 生成指纹配置
                                                                                                                                            fingerprint = self._generate_fingerprint(proxy, group)
                                                                                                                                                      with open(self.config_path, "w") as f:
                                                                                                                                                                    json.dump(fingerprint, f, indent=2)
                                                                                                                                                                              
                                                                                                                                                                                        # 预建Chromium需要的子目录,避免首次启动报错
                                                                                                                                                                                                  (self.profile_dir / "Cache").mkdir(exist_ok=True)
                                                                                                                                                                                                            (self.profile_dir / "Local Storage").mkdir(exist_ok=True)
                                                                                                                                                                                                                      (self.profile_dir / "Session Storage").mkdir(exist_ok=True)
                                                                                                                                                                                                                                
                                                                                                                                                                                                                                          return str(self.profile_dir)
                                                                                                                                                                                                                                                
                                                                                                                                                                                                                                                      def _generate_fingerprint(self, proxy: str, group: str) -> dict:
                                                                                                                                                                                                                                                                """使用shop_id作为随机种子,保证稳定性和差异化"""
                                                                                                                                                                                                                                                                          # 构造唯一种子
                                                                                                                                                                                                                                                                                    seed_str = f"{self.platform}:{self.shop_id}:{group}:salt2025"
                                                                                                                                                                                                                                                                                              seed = int(hashlib.md5(seed_str.encode()).hexdigest()[:8], 16)
                                                                                                                                                                                                                                                                                                        rng = random.Random(seed)
                                                                                                                                                                                                                                                                                                                  
                                                                                                                                                                                                                                                                                                                            # 分辨率池
                                                                                                                                                                                                                                                                                                                                      resolutions = [(1920,1080), (1366,768), (1440,900), (1536,864), (2560,1440)]
                                                                                                                                                                                                                                                                                                                                                # 时区池(根据平台倾向)
                                                                                                                                                                                                                                                                                                                                                          timezone_pools = {
                                                                                                                                                                                                                                                                                                                                                                        "pdd": ["Asia/Shanghai", "Asia/Chongqing"],
                                                                                                                                                                                                                                                                                                                                                                                      "temu": ["America/New_York", "America/Los_Angeles", "Europe/London"],
                                                                                                                                                                                                                                                                                                                                                                                                    "tiktok": ["America/New_York", "Europe/London", "Australia/Sydney"]
                                                                                                                                                                                                                                                                                                                                                                                                              }
                                                                                                                                                                                                                                                                                                                                                                                                                        tz_pool = timezone_pools.get(self.platform, ["UTC"])
                                                                                                                                                                                                                                                                                                                                                                                                                                  
                                                                                                                                                                                                                                                                                                                                                                                                                                            # 语言池
                                                                                                                                                                                                                                                                                                                                                                                                                                                      lang_pools = {
                                                                                                                                                                                                                                                                                                                                                                                                                                                                    "pdd": ["zh-CN", "zh-CN"],
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  "temu": ["en-US", "en-GB", "en-CA"],
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                "tiktok": ["en-US", "en-GB"]
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          }
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    lang_pool = lang_pools.get(self.platform, ["en-US"])
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        fingerprint = {
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      "proxy": proxy,
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    "group": group,
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  "screen_width": rng.choice(resolutions)[0],
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                "screen_height": rng.choice(resolutions)[1],
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              "timezone": rng.choice(tz_pool),
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            "language": rng.choice(lang_pool),
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          "platform_os": rng.choice(["Win32", "MacIntel"]),
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        "webgl_vendor": rng.choice(["Google Inc.", "Intel Inc.", "NVIDIA Corporation"]),
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      "do_not_track": rng.choice([True, False]),
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    "hardware_concurrency": rng.choice([2,4,8])
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              }
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        return fingerprint
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    def load_config(self) -> Optional[dict]:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              if not self.config_path.exists():
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            return None
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      with open(self.config_path, "r") as f:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    return json.load(f)
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                def update_proxy(self, new_proxy: str):
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          config = self.load_config()
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    if config:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  config["proxy"] = new_proxy
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                with open(self.config_path, "w") as f:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  json.dump(config, f, indent=2)
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              def delete(self):
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        import shutil
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  if self.env_dir.exists():
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                shutil.rmtree(self.env_dir)
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  ```
有了这个管理器,启动浏览器的函数就可以读取指纹配置,组装启动参数。
```python
  def launch_browser(env_mgr: EnvironmentManager):
        config = env_mgr.load_config()
              if not config:
                        raise Exception("环境未创建")
                              
                                    cmd = [
                                              "chrome.exe",
                                                        f"--user-data-dir={env_mgr.profile_dir}",
                                                                  f"--proxy-server={config['proxy']}",
                                                                            f"--window-size={config['screen_width']},{config['screen_height']}",
                                                                                      f"--lang={config['language']}",
                                                                                                f"--timezone={config['timezone']}",
                                                                                                          "--remote-debugging-port=0",
                                                                                                                    "--disable-blink-features=AutomationControlled",
                                                                                                                              "--no-first-run"
                                                                                                                                    ]
                                                                                                                                          # 启动进程,并解析调试端口...
                                                                                                                                            ```
这里有个细节:`--remote-debugging-port=0` 会让Chromium自动分配一个空闲端口,我们从启动日志里抓取出来,后续影刀RPA就可以通过这个端口连接进来操作页面。
---
## 三、自动化流程调度编排:从“人工切号”到“一键全自动”
环境隔离做好之后,第二步是让任务批量、并发地跑起来。
Alien的“自动化编排流”模块,本质上是一个轻量级的任务调度引擎。
### 3.1 拖拽组合流程
界面分左右两区。左侧是动作库,分类包括:
- **店铺操作**:打开/关闭环境、登录、刷新登录态
-   - **电商通用**:获取订单列表、批量发货、修改价格、上架商品
-   - **平台特有**:TikTok点赞/关注、拼多多领券、TEMU邀评
-   - **流程控制**:循环、条件判断、等待、调用子流程
右侧是画布,运营可以把动作拖拽到画布上,用连线串联。
例如一个“TikTok日常养号”流程:
[启动环境] → [自动登录] → [浏览推荐页30秒] → [点赞随机2个视频] → [关注1个账号] → [关闭环境]
  ```

流程定义保存为JSON,可以分配给一个分组或指定的店铺列表。

3.2 多对多匹配与智能调度

店群自动化最大的挑战不是写流程,而是调度
一个流程要跑100个店铺,每个店铺需要独占一个浏览器窗口。同时开100个窗口显然不现实。
在这里插入图片描述

所以调度器需要做到:

  • 并发窗口数限制:比如最多22个窗口同时运行
    • 店铺级串行:同一个店铺同一时间只能执行一个任务
    • 智能平铺:新窗口自动摆放在空闲区域,不重叠
    • 资源回收:任务结束后浏览器不立即关闭,复用一段时间;超时闲置则关闭并清理缓存
      下面是一段调度器的核心代码(真实工程中的简化版):
  import threading
    import queue
      import time
        from concurrent.futures import ThreadPoolExecutor
class AlienTaskScheduler:
      def __init__(self, max_concurrent=22):
                self.max_concurrent = max_concurrent
                          self.task_queue = queue.Queue()
                                    self.active_slots = 0
                                              self.slot_lock = threading.Lock()
                                                        self.shop_running = {}
                                                                  self.shop_lock = threading.Lock()
                                                                            self.executor = ThreadPoolExecutor(max_workers=max_concurrent)
                                                                                  
                                                                                        def submit(self, task):
                                                                                                  """
                                                                                                            task: {
                                                                                                                          'shop_id': str,
                                                                                                                                        'flow_file': str,
                                                                                                                                                      'params': dict
                                                                                                                                                                }
                                                                                                                                                                          """
                                                                                                                                                                                    self.task_queue.put(task)
                                                                                                                                                                                          
                                                                                                                                                                                                def _worker(self):
                                                                                                                                                                                                          while True:
                                                                                                                                                                                                                        task = self.task_queue.get()
                                                                                                                                                                                                                                      shop_id = task['shop_id']
                                                                                                                                                                                                                                                    
                                                                                                                                                                                                                                                                  # 等待该店铺没有正在运行的任务
                                                                                                                                                                                                                                                                                while True:
                                                                                                                                                                                                                                                                                                  with self.shop_lock:
                                                                                                                                                                                                                                                                                                                        if not self.shop_running.get(shop_id, False):
                                                                                                                                                                                                                                                                                                                                                  self.shop_running[shop_id] = True
                                                                                                                                                                                                                                                                                                                                                                            break
                                                                                                                                                                                                                                                                                                                                                                                              time.sleep(0.5)
                                                                                                                                                                                                                                                                                                                                                                                                            
                                                                                                                                                                                                                                                                                                                                                                                                                          # 等待有空闲并发槽位
                                                                                                                                                                                                                                                                                                                                                                                                                                        while True:
                                                                                                                                                                                                                                                                                                                                                                                                                                                          with self.slot_lock:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                if self.active_slots < self.max_concurrent:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          self.active_slots += 1
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    break
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      time.sleep(0.5)
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  self.executor.submit(self._execute, task, shop_id)
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              def _execute(self, task, shop_id):
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        try:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                      # 确保浏览器已启动或复用已有实例
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    debug_port = self._ensure_browser(shop_id)
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  # 调用影刀RPA执行流程
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                self._call_yingdao_flow(task['flow_file'], shop_id, debug_port, task.get('params', {}))
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          finally:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        with self.slot_lock:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          self.active_slots -= 1
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        with self.shop_lock:
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          self.shop_running[shop_id] = False
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        self.task_queue.task_done()
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    def _ensure_browser(self, shop_id):
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              # 检查该店铺是否已有运行中的浏览器实例
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        # 没有则调用环境管理器启动一个新实例并返回调试端口
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  pass
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              def _call_yingdao_flow(self, flow_file, shop_id, debug_port, params):
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                        import subprocess, json
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  cmd = [
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                "影刀RPA.exe", "-run", flow_file,
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              "-param", f"shop_id={shop_id}",
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                            "-param", f"debug_port={debug_port}",
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          "-param", f"params={json.dumps(params)}"
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    ]
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                              subprocess.run(cmd, capture_output=True, timeout=600, check=True)
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                          def start(self):
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    for _ in range(self.max_concurrent):
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                  threading.Thread(target=self._worker, daemon=True).start()
                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                    ```
这个调度器上线后,我们在一台32核64G的机器上跑过150个店铺的批量上架任务,22个窗口同时运行,CPU占用稳定在70%,内存稳定在45%,没有发生过卡死或串号。
> 踩坑:刚开始我们没有“店铺级串行”的限制,结果同一个店铺的两个任务(比如上架和领券)同时操作浏览器,导致页面状态错乱,商品被重复上架。加上互斥锁后问题解决。
### 3.3 智能平铺的实现
为了让22个窗口不堆叠,我在启动浏览器后调用Windows API移动窗口位置。
代码如下(简化版):
```python
  def tile_windows(window_handles, rows=4, cols=6):
        screen_width = 1920
              screen_height = 1080
                    tile_width = screen_width // cols
                          tile_height = screen_height // rows
                                for idx, hwnd in enumerate(window_handles):
                                          row = idx // cols
                                                    col = idx % cols
                                                              x = col * tile_width
                                                                        y = row * tile_height
                                                                                  win32gui.SetWindowPos(hwnd, None, x, y, tile_width, tile_height, 0x0040)
                                                                                    ```
运营第一次看到整整齐齐的22个窗口时,说了一句让我记到现在的话:**“这才是现代科技该有的样子。”**
---
## 四、底层工程封装:为什么客户觉得这套软件“很专业”
如果我把上面的Python代码发给客户,他们只会说“这什么黑乎乎的东西”。
所以我做了三件事,让Alien看起来像商业软件。
### 4.1 PyQt6极简交互面板
我用了三周时间学习PyQt6,然后写了一整套GUI。
- **仪表盘**:用pyqtgraph绘制实时曲线,显示并发窗口数、任务完成量、内存占用。
-   - **环境管理**:QTreeView做分组树,QTableView做卡片式表格,支持拖拽分组、右键菜单。
-   - **流程编排**:基于QGraphicsView实现拖拽式画布,节点用QGraphicsRectItem自定义绘制。
-   - **日志监控**:QTextEdit配合颜色高亮,实时滚动。
整体采用暗黑主题,按钮有悬停效果,图标用FontAwesome。客户第一次打开时说:“这软件得卖好几千吧?”
### 4.2 双击即用的exe打包
客户不需要安装Python、Chrome、影刀客户端。
我用PyInstaller将整个项目打包成一个`Alien.exe`,内嵌了:
- Python解释器
-   - 所有第三方库
-   - 一个便携版Chromium(约110MB,启动时解压到临时目录)
-   - 影刀RPA的免安装运行时(需要用户自行购买正版授权,但运行时文件我打包了)
客户下载压缩包,解压,双击exe,等待几秒初始化,就能看到主界面。
### 4.3 独立的安全验证
为了防止破解,我实现了一机一码授权。
程序启动时,收集硬盘序列号、网卡MAC地址、主板ID,通过SHA256生成机器码。
用户将机器码发给我,我用RSA私钥签名生成license文件。
程序每次启动验证license签名和机器码是否匹配,失败则拒绝运行。
虽然不能100%防破解,但已经挡住了99%的“复制粘贴即用”行为。
---
## 五、那些让我深夜emo的坑
**1. 内存泄漏排查了一周。**
当时线上环境跑了几十个号,内存从启动时的2GB慢慢涨到12GB,然后系统开始卡顿。查了很久,发现是每个浏览器实例退出时,`user-data-dir` 下的 `Cache` 和 `Code Cache` 没有被清理,日积月累导致磁盘空间被占满,同时内存中还有一些Python对象没有释放。解决方案:在调度器的资源回收函数中,对空闲超过1小时的店铺,删除其缓存目录并调用 `gc.collect()`。
**2. 影刀RPA的超时陷阱。**
影刀默认等待元素出现的超时是60秒,但跨境网络延迟高,有时候页面加载要90秒。导致很多任务莫名失败。后来我在每个影刀流程的开头将全局超时设为180秒,并在关键步骤加了截图保存,失败时能快速定位。
**3. 代理IP的质量问题。**
图便宜买过5/天的静态代理,用了三天被拼多多全部标记。后来换成住宅代理池,成本翻了三倍,但封店率从12%降到了0.8%。这个钱不能省。
**4. 运营的一次误操作。**
有一次运营在“环境管理”里选中了“全部店铺”,然后点了“删除”。瞬间所有环境配置都没了。我赶紧从备份恢复,但损失了当天的部分任务记录。之后我加了二次确认弹窗和回收站机制,删除的环境先移到“回收站”,7天后才真正删除。
---
## 写在最后
从一个人写Alien到现在,已经过去一年。
它从最初的几十行脚本,长成了上万行代码的完整系统。
有人问我:你为什么不直接用现成的指纹浏览器加影刀?
我的回答是:**自己造轮子不是为了炫耀,而是为了在每一个细节上拥有控制权。**
代理切换需要自定义逻辑?改代码。并发窗口数要动态调整?改配置。某个平台更新了风控策略?加一层指纹伪装。
所有的一切都在自己手里,不用等第三方更新,不用看供应商脸色。
如果你也在做店群自动化,希望这篇文章能给你一些启发。
技术不复杂,复杂的是对细节的死磕。
> 作者:林焱  
>   > 独立开发者,RPA架构师  
>     > 博客:林焱RPA(全网同名)  
>       > 转载需授权,喷子绕道
(全文约4600字)

更多推荐