部署Epic自动推送领取免费游戏

部署Epic自动推送领取免费游戏

最近看到很多人都在 Epic 免费领《霍格沃茨之遗》,网络搜索时看到了一个项目可以自动检测 Epic 的免费游戏并将其加入到购物车后推送付款链接,我们只需要点击链接结算就行了。研究了一下,发现自带的推送不太好用所以用 py 实现监测项目日志并通过pushplus进行推送

项目地址:https://hub.docker.com/r/charlocharlie/epicgames-freegames

项目文档:https://claabs.github.io/epicgames-freegames-node

部署pushplus-log-watcher

如果使用项目自带的 webhook 推送方式,他推送日志不直观,不进链接都不知道领取的是啥免费游戏,所以用 python 写了个监测容器日志并推送的项目

imageimage

先创建一个目录存放所有的数据

mkdir epicgames-freegames && cd epicgames-freegames

创建一个 watcher 目录存放 py 代码

mkdir watcher && cd watcher

创建app.py和依赖文件requirements.txt

#app.py
import os
import re
import time
import requests
import docker

TARGET_CONTAINER = os.getenv("TARGET_CONTAINER", "epicgames-freegames")
PUSHPLUS_TOKEN = os.getenv("PUSHPLUS_TOKEN", "").strip()
TITLE_PREFIX = os.getenv("TITLE_PREFIX", "Epic 自动领取").strip()
WINDOW_SECONDS = int(os.getenv("WINDOW_SECONDS", "900"))
SINCE_SECONDS = int(os.getenv("SINCE_SECONDS", "600"))  # 启动时回看最近 N 秒日志

if not PUSHPLUS_TOKEN:
    raise SystemExit("Missing env PUSHPLUS_TOKEN")

RE_USER = re.compile(r'user:\s*"([^"]+)"')
RE_ACCOUNT = re.compile(r'accountEmail:\s*"([^"]+)"')
RE_URL = re.compile(r'url:\s*"([^"]+)"')
RE_QUOTED = re.compile(r'"([^"]+)"')

def now():
    return int(time.time())

def norm_account(s: str) -> str:
    return (s or "").strip()

def pushplus_send(title: str, content_md: str):
    payload = {
        "token": PUSHPLUS_TOKEN,
        "title": title,
        "content": content_md,
        "template": "markdown",
    }
    r = requests.post("https://www.pushplus.plus/send", json=payload, timeout=20)
    r.raise_for_status()

last_purchase = {}  # account -> {ts,url}
last_games = {}     # account -> {ts,games}

def try_merge_and_send(account: str):
    account = norm_account(account)
    p = last_purchase.get(account)
    g = last_games.get(account)
    if not p or not g:
        return
    if abs(p["ts"] - g["ts"]) > WINDOW_SECONDS:
        return

    games = g.get("games") or []
    url = p.get("url") or ""

    if games:
        head = games[0]
        more = f" 等{len(games)}款" if len(games) > 1 else ""
        title = f"{TITLE_PREFIX} - 可领取:{head}{more}"
    else:
        title = f"{TITLE_PREFIX} - PURCHASE"

    lines = []
    lines.append(f"**账号:** `{account}`")
    lines.append("**原因:** `PURCHASE`")
    if games:
        lines.append("")
        lines.append("**可领取内容:**")
        for n in games:
            lines.append(f"- {n}")
    if url:
        lines.append("")
        lines.append(f"[领取链接]({url})")

    pushplus_send(title, "\n".join(lines))

    # 清掉避免重复
    last_purchase.pop(account, None)
    last_games.pop(account, None)

def attach_and_watch():
    client = docker.from_env()

    # 找容器
    try:
        c = client.containers.get(TARGET_CONTAINER)
    except Exception:
        # 模糊匹配
        match = None
        for cc in client.containers.list(all=True):
            if TARGET_CONTAINER in (cc.name or ""):
                match = cc
                break
        if not match:
            raise SystemExit(f"Cannot find container: {TARGET_CONTAINER}")
        c = match

    print(f"[watcher] attached to container: {c.name} ({c.id[:12]})", flush=True)

    in_checkout = False
    checkout_ctx = {"account": None, "url": None, "ts": 0}

    in_games = False
    games_ctx = {"account": None, "games": [], "collecting": False, "ts": 0}

    stream = c.logs(stream=True, follow=True, since=now() - SINCE_SECONDS)

    for raw in stream:
        line = raw.decode("utf-8", errors="ignore").rstrip()

        # 1) checkout 通知块(固定就是 PURCHASE)
        if "Dispatching checkout notification" in line:
            in_checkout = True
            checkout_ctx = {"account": None, "url": None, "ts": now()}
            m = RE_USER.search(line) or RE_ACCOUNT.search(line)
            if m:
                checkout_ctx["account"] = norm_account(m.group(1))
            continue

        if in_checkout:
            m = RE_USER.search(line) or RE_ACCOUNT.search(line)
            if m:
                checkout_ctx["account"] = norm_account(m.group(1))

            m = RE_URL.search(line)
            if m:
                checkout_ctx["url"] = m.group(1)

            if checkout_ctx["account"] and checkout_ctx["url"]:
                last_purchase[checkout_ctx["account"]] = {
                    "ts": checkout_ctx["ts"],
                    "url": checkout_ctx["url"],
                }
                try_merge_and_send(checkout_ctx["account"])
                in_checkout = False
            continue

        # 2) 未购买游戏块
        if "Unpurchased free games" in line:
            in_games = True
            games_ctx = {"account": None, "games": [], "collecting": False, "ts": now()}
            m = RE_USER.search(line) or RE_ACCOUNT.search(line)
            if m:
                games_ctx["account"] = norm_account(m.group(1))
            continue

        if in_games:
            m = RE_USER.search(line) or RE_ACCOUNT.search(line)
            if m:
                games_ctx["account"] = norm_account(m.group(1))

            if "purchasableGames" in line:
                games_ctx["collecting"] = True
                games_ctx["games"].extend(RE_QUOTED.findall(line))
                continue

            if games_ctx["collecting"]:
                games_ctx["games"].extend(RE_QUOTED.findall(line))
                if "]" in line:
                    in_games = False
                    # 去重保序
                    seen = set()
                    uniq = []
                    for g in games_ctx["games"]:
                        g = g.strip()
                        if g and g not in seen:
                            uniq.append(g)
                            seen.add(g)
                    games_ctx["games"] = uniq

                    if games_ctx["account"]:
                        last_games[games_ctx["account"]] = {
                            "ts": games_ctx["ts"],
                            "games": games_ctx["games"],
                        }
                        try_merge_and_send(games_ctx["account"])
            continue

def main():
    # 容器可能会“跑完就退出”,logs 流会断;所以循环重连
    while True:
        try:
            attach_and_watch()
        except Exception as e:
            print(f"[watcher] reconnecting after error: {e}", flush=True)
            time.sleep(2)

if __name__ == "__main__":
    main()
docker==7.1.0
requests==2.32.3

部署epicgames-freegames

回到epicgames-freegames目录下并创建config目录存放项目所需的配置文件,我们需要创建config.json5文件,根据自己情况填写对应配置

webPortalConfig 配置是因为第一次登录或者出验证时我们需要访问这个这个反代域名 (方案 A) 或者本地 ip 的 43000 端口 (方案 B) 进行验证。方案 A 主要是方便不与部署项目的服务在同一网络时使用,根据自己情况选择方案 A 或者方案 B

{
  // 启动容器就跑一次(便于立刻触发登录/通知流程)
  runOnStartup: true,

  // 建议跑得频繁点:文档提示设备码刷新 token 过期约 8 小时
  // 所以 cron 最好 < 8 小时一次(例如每 6 小时)
  cronSchedule: "0 */6 * * *",

  logLevel: "info",

  // 你在 Asia/Taipei,可写这里或用容器环境变量 TZ
  timezone: "Asia/Taipei",

  // 搜索策略:weekly / all(默认 all)
  searchStrategy: "weekly",

  // Web Portal:用于“设备码登录/验证码处理”的跳转页面
  webPortalConfig: {
    // 方案 A:你有域名/反代/端口映射(推荐稳定)
    // baseUrl: "https://dome.xyz",

    // 方案 B:不方便公网访问,直接用 localtunnel(更省事)
    localtunnel: false,
  },

  // 多账号:只填 email(不需要 Epic 密码;v5 起移除了账号密码配置)
  accounts: [
    { email: "Epic账号邮箱" },
    // { email: "your_epic_email_2@example.com" },
  ],

  notifiers: [],
  // 因为我们使用上面python脚本的推送方式所以下面注释并且notifiers设为空数组了,如果想要使用原有的推送方式可以参考下面注释
  // 通知器:用 webhook 对接 PushPlus
  //notifiers: [
  //  {
  //    type: "webhook",
  //    url: "http://pushplus-relay:8080/epic",
  //    headers: { "Content-Type": "application/json" },
  //  }
    // 你也可以换成 email/discord/gotify/bark/ntfy/webhook 等(可配多个)
  //],
}

回到epicgames-freegames目录下创建docker-compose.yml文件,并根据自己的情况配置一下。

version: "2.4"

services:
  epicgames-freegames:
    image: charlocharlie/epicgames-freegames:debian
    container_name: epicgames-freegames
    restart: unless-stopped
    ports:
      - "43000:3000"   # Web portal/captcha 页面端口:contentReference[oaicite:8]{index=8}
    environment:
      - TZ=Asia/Taipei
    volumes:
      - ./config:/usr/app/config:rw  # 配置与 cookies 持久化:contentReference[oaicite:9]{index=9}
    # Chromium 更稳一点(可选但推荐)
    shm_size: "1gb"
    # 文档推荐 2g 内存上限:contentReference[oaicite:10]{index=10}
    mem_limit: 2g

  pushplus-log-watcher:
    image: python:3.12-slim
    container_name: pushplus-log-watcher
    restart: unless-stopped
    working_dir: /app
    volumes:
      - ./watcher:/app:ro
      - /var/run/docker.sock:/var/run/docker.sock   # ⚠️ 必须:读容器日志
    environment:
      - TARGET_CONTAINER=epicgames-freegames
      - PUSHPLUS_TOKEN=pushplus的推送秘钥				#⚠️  更改为自己的
      - TITLE_PREFIX=Epic 自动领取
      - WINDOW_SECONDS=900
    command: sh -c "pip install -r requirements.txt && python app.py"
    depends_on:
      - epicgames-freegames

创建完后直接启动就可以了

docker compose up -d

启动完成后,第一次需要访问一下 ip:43000 页面过一下验证,然后就会开始每 6 小时检测一次是否有免费游戏可以领取或者是否需要重新验证账号,如果有则会进行推送,如果项目有更新可以使用 docker compose pull 命令更新镜像

部署Apache APISIX®为API添加鉴权和IP白名单 2025-12-16

评论区