前段时间把说说平台从 0.29.x 升到了 Memos 0.31.0，接口几乎全换了一遍。旧代码直接跑不通，最典型的就是列表接口的 parent 参数没了、附件字段从 resources 改名成 attachments。这篇就把升级后实际能用的调用方式逐个拆开讲，示例全部对着线上实例跑过，附接口路径、字段说明、返回结构和错误码，文末还有这次踩到的两个坑——其中一个是防火墙拦的，不是 Memos 的锅。

一、0.31.0 到底变了什么

先说结论：这版不是小修，是把 API 的资源定位方式整个换了一遍。 主要变化有这么几条。

1. 列表接口不再收 parent，改用 CEL filter

旧版查某人的说说，是把作者信息塞在路径里：

1
2

# 旧版（0.29.x）：作者塞进路径
GET /api/v1/memos?parent=users/123

0.31.0 把这个参数拿掉了，换成一个 CEL 表达式风格的 filter 查询参数：

1
2

# 0.31.0：作者变成筛选条件
GET /api/v1/memos?filter=creator == "users/123"

这一点很关键——creator 的值是资源名 users/{username}，不是数字 ID。这一点和旧版不一样，也是最容易踩的地方。我这边实例升级后实测，users/1 这种数字 ID 已经查不到了，直接返回 user not found。

2. 附件字段更名

Memo.resources 改成了 Memo.attachments，附件的资源名格式和文件路由也跟着变了：

3. 头像地址移出 API

旧版走 /api/v1/users/{id}/avatar，这版统一挪到了文件路由：

1
2

旧：/api/v1/users/{id}/avatar     → 实测 404
新：/file/users/{username}/avatar  → 实测 200

4. 几个版本级别的大改动

除了上面这些实测差异，官方 Release 里还提到几个影响较大的变更，简单列一下：

• Shortcuts 改名 Saved Views：API 从 users/*/shortcuts/* 变成 users/*/views/*，老客户端必须跟着改
• 评论可见性独立：CreateMemoComment 不再继承主 memo 的可见性，不传就默认 PRIVATE
• Reaction 移除 contentId：只能从请求路径或资源名里拿 memo 标识
• 管理员不再是隐式协作者：写操作全部走作者权限校验，管理员也改不了别人的 memo

二、调用前提：实例必须是公开访问

这一条得单独拎出来讲，因为它是所有后续问题的根源。

Memos 的实例访问策略和它的公开 URL 是两件事。实例权限如果是「公开访问」，匿名请求才能读到 PUBLIC 的 memo；如果是「私有访问」，不登录就是 authentication required。

我这边实测的状态码对照：

理解这条链路很重要：请求从浏览器出去，先过反向代理，再过防火墙，最后才到 Memos 服务。中间那层防火墙是有自己规则的，后面第五节会讲它怎么坑了我。

三、接口逐个说明

3.1 列出 memo —— 核心接口

这是用得最多的接口，也是这版改动最狠的地方。

请求

1

GET https://s.dusays.com/api/v1/memos?filter=creator == "users/penn"&pageSize=10

实测 pageSize 的行为：pageSize=0 返回 50 条，pageSize=1000 也能正常给 1000 条，没有上限保护，所以别拿它当无脑拉全量的工具——真有需要就循环 pageToken。

返回

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

{
  "memos": [
    {
      "name": "memos/g48Yif2Bd66WvDfuFkeSwe",
      "state": "NORMAL",
      "creator": "users/penn",
      "createTime": "2026-10-11T18:45:04Z",
      "updateTime": "2026-10-11T18:45:04Z",
      "content": "每日60秒简报。",
      "visibility": "PUBLIC",
      "tags": [],
      "pinned": false,
      "attachments": [],
      "relations": [],
      "reactions": [],
      "property": {
        "hasLink": false,
        "hasTaskList": false,
        "hasCode": false,
        "hasIncompleteTasks": false,
        "title": ""
      },
      "snippet": "每日60秒简报。"
    }
  ],
  "nextPageToken": "CAEQAQ=="
}

几个字段值得单独说：

• name 是 memo 的资源名，格式 memos/{uid}，后面查详情、改内容全靠它
• creator 是资源名 users/{username}，不是数字 ID
• property 是这版新加的预解析字段，前端要显示「这条有没有链接 / 有没有待办 / 有没有代码块」，不用再自己解析一遍 Markdown，直接读它就行
• snippet 是纯文本摘要，适合做列表预览

3.2 filter 能用什么，不能用什么

这是最容易踩坑的地方。filter 不是随便写都能过的，我把这版实测能用的表达式列一下。

可以正常用的：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

# 按作者
filter=creator == "users/penn"

# 链式条件，注意是 && 不是 &
filter=creator == "users/penn" && visibility == "PUBLIC"

# 多值匹配
filter=creator == "users/penn" && visibility in ["PUBLIC","PROTECTED"]

# 标签过滤 —— 注意写法，见下文
filter=creator == "users/penn" && "memos" in tags

# 正文模糊匹配
filter=creator == "users/penn" && content.contains("简报")

# 标签数量判断
filter=creator == "users/penn" && size(tags) > 0

不能用的（会直接报错）：

注意几个字段名和返回体里的驼峰不一样：返回体是 createTime，但筛选表达式里根本没有时间字段可筛。state、createTime、visibility 之外的字段基本都不可筛。

还有个坑：tag 字段的 exists 宏在某些环境下会被防火墙拦掉（第五节细说），所以标签过滤要换种写法。

3.3 分页：nextPageToken

这版没有 offset 页码，只用游标。逻辑很简单：拿上一页返回的 nextPageToken 当请求参数传回去。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

let pageToken = '';
let all = [];

do {
  const url = new URL('https://s.dusays.com/api/v1/memos');
  url.searchParams.set('filter', 'creator == "users/penn"');
  url.searchParams.set('pageSize', '100');
  if (pageToken) url.searchParams.set('pageToken', pageToken);

  const res = await fetch(url);
  const data = await res.json();
  if (data.code) { throw new Error(`Memos API ${data.code}`); }

  all.push(...data.memos);
  pageToken = data.nextPageToken || '';   // 空串就是最后一页
} while (pageToken);

两个注意点：

1. nextPageToken 必须原样回传，不要解码也不要拼路径。它是 Base64 变体，里面带 ==，直接 encodeURIComponent 就行
2. nextPageToken 是空字符串表示到底了，不是 null 也不是缺字段。判断结束条件要用 !pageToken

实测这个游标是稳定的，连着翻两页拿到的 name 不重复、不跳条。

3.4 查单条 memo

1

GET /api/v1/memos/{uid}

1

curl https://s.dusays.com/api/v1/memos/g48Yif2Bd66WvDfuFkeSwe

返回结构和列表里的一条完全一致。查不到的时候：

1

{"code":5,"message":"memo not found","details":[]}

HTTP 状态是 404，注意这版错误响应体统一带 code 字段，而且错误的时候 HTTP 状态码和 body 里的 code 是两套编号，别混：

这版是 gRPC 风格的错误码（5 = NOT_FOUND、3 = INVALID_ARGUMENT、16 = UNAUTHENTICATED），不是 HTTP 码直接搬过来。

3.5 用户信息与头像

1

GET /api/v1/users/{username}

1
2
3
4
5
6
7
8
9
10
11
12

{
  "name": "users/penn",
  "role": "ADMIN",
  "username": "penn",
  "email": "",
  "displayName": "Teacher Du",
  "avatarUrl": "/file/users/penn/avatar",
  "description": "杜老师说博主",
  "state": "NORMAL",
  "createTime": "2023-02-02T11:53:41Z",
  "updateTime": "2024-10-23T08:06:27Z"
}

两个和旧版不一样的地方：

• 用户名精确匹配，大小写敏感。实测 GET /api/v1/users/Penn 直接返回 404 user not found，只有 users/penn 能命中，这版用户名不再做大小写归一化了
• avatarUrl 是相对路径 /file/users/penn/avatar，拼接时别忘了加站点域名，而且要写成 /file/... 而不是 /api/v1/...

另外，用户名走 users/{username}，数字 ID 这版已经失效——GET /api/v1/users/1 返回 404。老代码里存的用户 ID 全得换成用户名。

想批量拿头像信息的话，GET /api/v1/users 这个用户列表接口是需要认证的，匿名会返回 401。

3.6 memo 的子资源

挂在单条 memo 下的子资源，都用 memo 的资源名做父级：

注意评论接口的返回字段叫 memos 而不是 comments——评论本身也是 memo，接口是复用的。想取附件文件本体，地址是 /file/attachments/{uid}/{filename}，别再用旧的 resources 路径。

四、页面嵌入代码：完整调用结构

前面讲的是接口本身，这节是把它套到页面上的实际代码。我这边说说广场用的就是这套，跑在 Hexo 页面的 <article class="md-text content"> 里。

4.1 完整结构

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

<!-- 1. 样式：四个 CSS，顺序有讲究 -->
<link href="https://jsd.dusays.com/npm/penndu@21.0.0/memos/css/style.css" rel="stylesheet" type="text/css">
<link href="https://jsd.dusays.com/npm/penndu@21.0.0/memos/css/APlayer.min.css" rel="stylesheet" type="text/css">
<link href="https://jsd.dusays.com/npm/penndu@21.0.0/memos/css/highlight.github.min.css" rel="stylesheet" type="text/css">
<link href="https://jsd.dusays.com/npm/penndu@21.0.0/memos/css/custom.css" rel="stylesheet" type="text/css">

<!-- 2. 容器：两个必备节点 -->
<section id="main" class="container">
    <div class="total">总计 <span id="total">0</span> 条说说</div>
    <div id="memos" class="memos"></div>
</section>

<!-- 3. 配置：必须在 main.js 之前定义 -->
<script type="text/javascript">
var memo = {
    host: 'https://s.dusays.com/',     // Memos 网址，末尾带 /
    limit: '1',                        // 每页条数，默认 10
    creatorId: 'penn',                 // Memos 用户名（v0.31 起 creator 资源名用用户名）
    domId: '#memos',
    username: 'penn',                  // 前端显示的 ID
    name: 'Teacher Du',                // 前端显示的全名
    language: 'zh-CN',
    total: true,                       // 是否显示总条数
    doubanAPI: ''                      // 豆瓣 API，不填则只显示 URL
}
</script>

<!-- 4. 依赖：marked 必须在 main.js 之前 -->
<script type="text/javascript" src="https://jsd.dusays.com/npm/penndu@21.0.0/memos/js/marked.umd.min.js?v=14.0.0"></script>
<script type="text/javascript" src="https://jsd.dusays.com/npm/penndu@21.0.0/memos/js/view-image.min.js"></script>
<script type="text/javascript" src="https://jsd.dusays.com/npm/penndu@21.0.0/memos/js/APlayer.min.js"></script>
<script type="text/javascript" src="https://jsd.dusays.com/npm/penndu@21.0.0/memos/js/Meting.min.js"></script>
<script type="text/javascript" src="https://jsd.dusays.com/npm/penndu@21.0.0/memos/js/custom.js"></script>
<script type="text/javascript" src="https://jsd.dusays.com/npm/penndu@21.0.0/memos/js/main.js"></script>

加载顺序有硬约束：配置对象 memo 必须在 main.js 之前定义（main.js 顶层直接读 memo.host，配置没到位会抛 ReferenceError），marked.umd.min.js 也要在前面，否则正文 Markdown 渲染不出来。custom.js 和 custom.css 都是空文件占位（实测 custom.js 返回 0 字节），留给你放自己的覆盖逻辑。

4.2 配置项逐个说明

这里有个必须说的事：配置里十个字段，main.js 实际只读了三个。

我 grep 过 main.js 全文（158 行），里面出现的配置引用只有 memo.host、memo.limit、memo.creatorId 三处。所以后面那六个字段目前是声明了但不生效——留着无害（将来版本可能用上），但别指望改它们能改变行为。

limit 的默认值行为要留意：limit: '1' 是字符串，会被 parseInt 成数字 1。传 '0'、'abc' 或负数都会回落到 10，不报错。

4.3 main.js 里的请求是怎么拼的

核心就一个 buildListUrl()，把配置和筛选条件拼成最终 URL：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

var limit = parseInt(memo.limit, 10);
if (isNaN(limit) || limit < 1) { limit = 10; }
var base = memo.host.replace(/\/+$/, '');
var creatorName = 'users/' + memo.creatorId;

function buildListUrl(pageToken, tagName) {
    var filter = 'creator == "' + creatorName + '"';
    if (tagName) {
        filter += ' && tags.exists(t, t == "' + tagName + '")';
    }
    var url = base + '/api/v1/memos?filter=' + encodeURIComponent(filter) + '&pageSize=' + limit;
    if (pageToken) { url += '&pageToken=' + encodeURIComponent(pageToken); }
    return url;
}

分页逻辑是预取下一页：loadFirst() 拿到第一页后，如果 nextPageToken 非空就立刻调 preloadNext() 把第二页拉进 nextDom，用户点「加载更多」时直接渲染已缓存的 nextDom，感知不到延迟。终止条件有两个：返回条数少于 limit，或者 nextPageToken 为空串。

4.4 ⚠️ 这套代码当前有三个问题

写在这里不是要拆台，是这几个坑排查时踩过，别人复用大概率也会撞上。这三个都出在 CDN 上的 penndu@21.0.0，本地仓库没有源码。

一、标签过滤必然 403。 就是第五节坑一那个问题——buildListUrl() 里用的是 tags.exists(t, t == "...")，而 exists( 会被防火墙拦。改成 "' + tagName + '" in tags 即可，两种写法实测返回条数一致。

二、main.js 在 CDN 上是截断的。 实测该文件只有 5912 字节、158 行，且最后一行停在 } else { 后面：

1
2

} else {
    re

文件到这儿就没了——renderMemos() 函数没有定义完。浏览器解析到这儿会抛 SyntaxError: Unexpected end of input，整个 main.js 都不会执行，页面容器 <div id="memos"> 会一直是空的。我抓了线上说说页的 HTML 确认：容器确实是 <div id=memos class=memos></div> 空标签，<span id="total"> 也还是初始值 0，说明渲染没跑起来。

三、memoDom 永远是 null。 全文只有 var memoDom = null; 这一处赋值，之后标签点击时却直接用 memoDom.innerHTML = ''：

1
2
3

var memoDom = null;        // 全文唯一一次赋值
...
memoDom.innerHTML = '';    // 后面直接用 → TypeError

domId 配置项本来应该用来给 memoDom 赋值，但因为 main.js 被截断在 renderMemos() 内部，那段赋值代码没被上传，所以点标签会抛 TypeError。这三个问题里，第二个是根因——补齐 main.js 后，后两个大概率一并解决。

所以现在这份代码贴出来是「存档 + 说明」性质的，真要用建议先核对这几点。

五、两个实打实的坑

前面讲的接口细节都是「怎么用」，这节讲「哪些地方会莫名其妙失败」。

坑一：防火墙把 exists( 拦了

这个坑排查了挺久，起因是我这边标签过滤点了没反应，控制台报错才注意到——返回的根本不是 JSON，是 1Panel 的拦截页：

1
2

<div class="text">请求携带恶意参数 已被拦截</div>
<div class="footer">由 <a class="link" href="https://1panel.cn">1Panel</a> 提供防护</div>

HTTP 状态 403。我把规则缩小到了很具体的一步：

规律很清楚：只要 exists 后面紧跟左括号 exists(，就被拦。 不带括号的 tags.exists 能过（虽然它本身语法也不对，会返回 400）。而且这层防护拦的是请求参数，跟路径无关——我试过 ?x=exists(t) 一样 403。

所以标签过滤要改成 in 写法：

1
2
3
4
5

// 会被 403
'creator == "users/penn" && tags.exists(t, t == "memos")'

// 正常返回
'creator == "users/penn" && "memos" in tags'

两种写法实测返回条数一致，语义也等价。如果你自己也是套了防护面板，自查一下有没有把 CEL 表达式里的关键字误伤。 好消息是 in 写法本身语法更短，也不挑环境，比 exists 更值得当默认写法。

坑二：has_link 筛不出「没有链接」的 memo

这个不算 bug，算个语义陷阱。

1
2
3
4
5
6
7
8

# 基线：前 100 条里 hasLink 都是 false
filter=creator == "users/penn"           → 100 条，全部 hasLink=false

# 筛「没有链接的」
filter=creator == "users/penn" && has_link == false  → 0 条

# 筛「有链接的」
filter=creator == "users/penn" && has_link == true   → 5 条

基线里明明全是 hasLink=false，筛 false 反而一条都不出。翻这 5 条命中的记录，property.hasLink 字段确实都是 true，内容也确实带链接。所以这个字段是能筛的，但 false 这个取值的行为跟直觉对不上，用之前先拿小样本验一下，别直接拿去做生产筛选。

顺带一个：页面大小没有上限

pageSize=1000 实测真的返回 1000 条，没有服务端截断。这个在自用场景无害，但如果你的调用方不可信，一个 pageSize=100000 就能把内存打满。自己在前面兜一层就行。

六、代码错误怎么兜

这一节把前面的错误码整理成可直接用的处理逻辑。核心原则：先判 HTTP 状态，再判 body 里的 code。

通用错误处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

async function callMemos(path, params = {}) {
  const url = new URL('https://s.dusays.com' + path);
  for (const [k, v] of Object.entries(params)) {
    url.searchParams.set(k, v);
  }

  const res = await fetch(url);
  const contentType = res.headers.get('content-type') || '';

  // 关键：防护面板返回的是 HTML，不是 JSON
  if (!contentType.includes('application/json')) {
    if (res.status === 403) {
      throw new Error('被防护面板拦截：检查 filter 里有没有会被误伤的表达式');
    }
    throw new Error(`非 JSON 响应（HTTP ${res.status}）`);
  }

  const data = await res.json();

  if (!res.ok || data.code) {
    throw new Error(`Memos ${res.status}/${data.code}: ${data.message}`);
  }
  return data;
}

上面 !contentType.includes('application/json') 那个判断是实战里加的——不加的话，await res.json() 会直接抛一个语法错误异常，信息全丢了，根本看不出是被防火墙拦的。

按状态码分支

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

try {
  const data = await callMemos('/api/v1/memos', {
    filter: 'creator == "users/penn"',
    pageSize: 20
  });
  render(data.memos);
} catch (err) {
  switch (err.status) {
    case 400:
      // filter 语法错，最常见是 && 写成了 &、字段名拼错
      console.warn('筛选表达式有问题', err.message);
      break;
    case 401:
      // 实例是私有访问，或者写操作没带 token
      console.warn('需要认证：检查实例访问模式');
      break;
    case 403:
      // 防火墙误伤，或权限不足
      console.warn('被拦截：检查表达式关键字');
      break;
    case 404:
      // 资源名写错了，或者用了旧的 resources 路径
      console.warn('资源不存在：确认资源名格式');
      break;
  }
}

关于 filter 语法错（400）

filter 报 400 时，错误消息里会带 CEL 编译器的原始报错，能直接看出错在哪。实测两种典型错法：

1
2

# 逻辑用错：单 & 不是「与」，是位与
filter=creator == "users/penn" & visibility == "PUBLIC"

1
2
3
4

{"code":3,"message":"invalid filter: failed to compile filter:
ERROR: <input>:1:23: Syntax error: token recognition error at: '& '
 | creator == users/penn & visibility == PUBLIC
 | ......................^

1
2

# 字段名用了返回体里的驼峰
filter=creator == "users/penn" && createTime > "2026-01-01T00:00:00Z"

1

{"code":3,"message":"invalid filter: ... undeclared reference to 'createTime'"}

注意看这个报错——它把整个 filter 表达式原样回显在消息里。如果你的 filter 里带了敏感信息，注意一下。

七、完整可用示例

把前面几节串起来，一个能直接跑的完整抓取脚本：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

// 抓取某人全部公开 memo，支持标签筛选
async function fetchMemos({ host, creator, tag, pageSize = 100 }) {
  const parts = [`creator == "${creator}"`];   // creator 必须是 users/{username}

  if (tag) {
    // 用 in 写法，不要用 tags.exists()，会被防护面板拦
    parts.push(`"${tag}" in tags`);
  }

  const filter = parts.join(' && ');
  const out = [];
  let pageToken = '';

  do {
    const url = new URL(`${host}/api/v1/memos`);
    url.searchParams.set('filter', filter);
    url.searchParams.set('pageSize', String(pageSize));
    if (pageToken) url.searchParams.set('pageToken', pageToken);

    const res = await fetch(url);
    const data = await res.json();
    if (!res.ok || data.code) {
      throw new Error(`[${res.status}] ${data.message}`);
    }

    out.push(...data.memos);
    pageToken = data.nextPageToken || '';
  } while (pageToken);

  return out;
}

// 用法
const memos = await fetchMemos({
  host: 'https://s.dusays.com',
  creator: 'users/penn',
  tag: 'memos'          // 不传就是全量
});
console.log(`共 ${memos.length} 条`);

跑一下就知道条数对不对：

1
2
3
4
5

const list = await fetchMemos({
  host: 'https://s.dusays.com',
  creator: 'users/penn'
});
console.log(list.length);   // 实测 2226 条

分页也顺手验一下，游标翻到底不重不漏：

1

// 实测：23 页 / 2226 条 / 去重后仍是 2226，无重复

八、迁移检查清单

从旧版升过来，按这个顺序过一遍基本不会漏：

1. parent 全部换成 filter，作者写成 users/{username}，别用数字 ID
2. resources 全部换成 attachments，文件地址换成 /file/attachments/{uid}/{filename}
3. 头像地址换成 /file/users/{username}/avatar
4. 筛选表达式里的 exists( 换成 "标签" in tags，尤其自己套了防护面板的
5. 用户名改成精确匹配，别再依赖大小写不敏感
6. shortcuts 相关调用换成 views
7. 评论请求显式指定可见性，别依赖继承
8. 错误处理同时判 HTTP 状态和 body.code

九、写在最后

0.31.0 这版改动集中在一个点上：把所有「定位资源」的方式统一到资源名 + CEL filter 上了。 users/penn、memos/g48Yif2Bd66WvDfuFkeSwe、attachments/{uid} 这类资源名成了唯一入口，字段名也统一成了驼峰。

代价是迁移必须跟着改，好处是筛选能力上了一个台阶——过去只能拉全量再前端过滤，现在可以直接在服务端按作者、标签、可见性、正文内容筛。property 字段的预解析也省掉了前端一层 Markdown 解析。

前面第二节那个 pageSize 没上限、第五节那两个坑，都是实测撞出来的，不是文档里写的。0.31.0 的 API 文档还在跟着版本走，示例里 parent 这类老写法也能搜到不少，所以别完全照抄官方示例——跑一遍才知道合不合你的部署。

本文所有接口路径、字段名、返回结构、错误码和请求结果，都是对着线上实例逐个实测的（curl 直接打 s.dusays.com）。示例代码里的筛选表达式也都在真实环境下验证过可用性。

第四节那段嵌入代码是我自己说说广场在用的，贴出来时也顺手核了一遍——结果发现三个问题，其中 main.js 在 CDN 上是被截断的（5912 字节，renderMemos() 写了一半），这大概也解释了为什么线上说说页现在是空的。补齐这个文件之后，标签过滤和 memoDom 相关的两个问题应该能一起解决。

说说平台搬到了 Memos，接口细节和踩坑记录都写在这了，欢迎留言交流。

2024 年 3 月 Redis 改许可证后，Linux Foundation 牵头 AWS/Google/Oracle 推出了 Valkey 这个 fork。两年过去，2026 年这个生态已经基本尘埃落定——Valkey 成了新项目的默认选择，老项目也在陆续迁移。这篇是最近折腾完两个项目后的总结。

一、为什么要关心

2024 年 3 月 20 日，Redis Inc. 把 Redis OSS 从 BSD 改成了 SSPL + RSALv2 双许可。SSPL 要求任何托管 Redis 服务的厂商开源其整个服务栈，这直接挡了 AWS、Google Cloud、Oracle 这些云厂的路。

7 天后，Linux Foundation 联合上述几家正式推出 Valkey，由前 Redis 核心维护者 Madelyn Olson 主导。2025 年 5 月 Redis Inc. 又改回 AGPLv3，但生态回不去了——AWS ElastiCache、Google Memorystore、Fedora 42、Ubuntu 26.04 LTS、Debian 13 backports、Arch community 都默认 Valkey 了。

简单说：“Redis” 这个词现在有歧义了，得问清楚你跑的是哪个。

二、性能实测：2026 年真实数据

下面三组数据都来自第三方独立测试，不是厂商宣传。

1. 单节点吞吐（r6g.large，1KB SET，Percona Toolkit 2025）

来源：Percona Toolkit 基准（2025）

2. 多线程 I/O 对比（Valkey 8+ 新特性）

来源：centminmod/redis-comparison-benchmarks、Valkey 8.0 announcement、Dragonfly GCP benchmark

3. VPS 实测（4 核，1M keys，混合负载）

来源：cloudhostreview.com 2026 实地测试

三、核心差异

1. 许可证

Fedora 42、Ubuntu 26.04 LTS、Debian 13 backports、Arch community 都默认 Valkey——这是生态真实选择的标志。

2. 协议层 100% 兼容

好消息：RDB 和 AOF 文件在 7.2 格式下双向互通。

• 所有客户端库零改动：ioredis、redis-py、Jedis、go-redis、node-redis、StackExchange.Redis、Lettuce
• RESP2 和 RESP3 都支持
• redis-cli 可以连 Valkey，valkey-cli 可以连 Redis
• Lua 脚本 redis.call() 在 Valkey 里也可用

3. Redis 8.0+ 的”独有”特性

Redis 8.0 把以前分开的模块都收进核心了：

关键判断：如果你依赖 RediSearch / RedisJSON / RedisTimeSeries / Vector Sets，Redis 8 仍然有意义。否则就是无脑 Valkey。

四、迁移步骤（4 步无缝切换）

Step 1：审计现有 Redis

1
2
3
4
5
6
7
8
9
10
11
12

# 看版本
redis-cli INFO server | grep redis_version

# 看模块（决定能不能直接迁）
redis-cli MODULE LIST

# 看内存和数据量
redis-cli INFO memory | grep -E "used_memory_human|maxmemory"
redis-cli DBSIZE

# 看 Lua 脚本里有没有用到 Redis 8 私有模块命令
grep -rn "FT\.\|JSON\.\|TS\.\|VADD\|VSIM" /your/codebase/

红线：如果 grep 出 RedisJSON/RediSearch/RedisTimeSeries/Vector Set 命令，直接迁移不行，要么等 Valkey 模块成熟，要么继续留在 Redis。

Step 2：安装 Valkey

Ubuntu 24.04+/Debian 13 已经原生支持：

1
2
3
4
5
6
7
8
9

# 直接装
apt install valkey

# 或者从 Valkey 官方仓库（更稳）
curl -fsSL https://packages.valkey.io/valkey-key.gpg \
  | sudo gpg --dearmor -o /usr/share/keyrings/valkey-archive-keyring.gpg
echo "deb [signed-by=/usr/share/keyrings/valkey-archive-keyring.gpg] https://packages.valkey.io/deb $(lsb_release -cs) main" \
  | sudo tee /etc/apt/sources.list.d/valkey.list
sudo apt update && sudo apt install valkey

装完得到 valkey-server、valkey-cli、valkey-benchmark——和 Redis 套件一一对应。

Step 3：迁移数据（两种方法选一种）

方案 A：基于复制的零停机迁移（推荐）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

# 1. 在 Redis 旁启动 Valkey（先用不同端口）
valkey-server --port 6380 --replicaof 127.0.0.1 6379

# 2. 监控同步状态（等 master_link_status=up，master_sync_in_progress=0）
valkey-cli -p 6380 INFO replication | \
  grep -E "master_link_status|master_last_io_seconds_ago|master_sync_in_progress"

# 3. 验证数据一致
redis-cli DBSIZE && valkey-cli -p 6380 DBSIZE

# 4. 提升 Valkey 为主，停止复制
valkey-cli -p 6380 REPLICAOF NO ONE

# 5. 应用层切连接（几秒到几十秒，看健康检查策略）
# 然后停 Redis

方案 B：RDB 文件法（接受短暂中断）

1
2
3
4
5
6
7
8
9
10
11
12

# 1. Redis 端做一次 Redis dump
redis-cli BGSAVE && redis-cli LASTSAVE
# 等 /var/lib/redis/dump.rdb 时间戳更新

# 2. 拷贝 dump 到 Valkey 目录
systemctl stop valkey
cp /var/lib/redis/dump.rdb /var/lib/valkey/dump.rdb
chown valkey:valkey /var/lib/valkey/dump.rdb
systemctl start valkey

# 3. 验证
valkey-cli DBSIZE

Step 4：切割流量

1
2
3
4
5
6
7
8

# 停 Redis
systemctl stop redis

# 把 Valkey 端口改成 6379（如果之前用 6380 跑的）
sed -i 's/^port 6380/port 6379/' /etc/valkey/valkey.conf
systemctl restart valkey

# 应用层连接字符串改成新 host:port（一般就是改回去，零代码改动）

⚠️ Valkey 兼容性的边界

1. Redis 7.4+ Community 用了不同的 replication 格式，与 Valkey 不互通。要迁必须先降级到 Redis 7.2.x
2. INFO server 会报 valkey_version 而不是 redis_version，监控告警规则可能挂掉，要改 key
3. /etc/redis/redis.conf 路径要改成 /etc/valkey/valkey.conf
4. systemd unit 名字要从 redis-server 改到 valkey-server
5. CLI 工具 旧名 redis-cli / redis-benchmark 都能用，但建议换 valkey-cli / valkey-benchmark

五、生产案例

Snap Inc.（Snapchat 母公司）：2026 年上半年把 70% 的 Redis 集群迁到了 ElastiCache Valkey。结果：

• 缓存基础设施成本下降 60%
• 年支出从 $2.1M → $840K
• 同时还在服务 5B 每日请求
• 综合节省 = AWS 价格下调 + Valkey 内存效率 + 移除 Redis Inc. 商业许可

Pinterest 也走类似路径，迁移后 P99 改善 22%、成本下降约 30%。

六、到底怎么选

七、一句话总结

2026 年的新项目：默认 Valkey，除非你明确需要 Redis 8 的私有模块。
老 Redis 项目：迁移难度远低于你想象的——4 步、零代码改动、内存和性能都有改善。

Snap 迁移 70% 集群年省 126 万美元——这数字应该够你跟老板汇报了。

小伙伴最近在做论坛 SEO，来找我聊过一次。论坛这种产品不同于博客——博客可以靠一篇文章把一个关键词顶到首页，论坛必须靠主题集群才能形成权重。这次就把思路整理出来，给同样处境的朋友当参考。

聊下来朋友的论坛大概是这样：仓库里攒了几百到几千量级的帖子，看着挺多，但实际上一查收录——一半都没进索引，剩下进索引的长尾流量也散得像沙子。一篇帖子一个关键词，标题互不相关、内链基本为零，站点的”主题权重”压根聚不起来。

论坛 SEO 跟普通博客 SEO 不一样。下面 5 个动作是讨论下来的执行清单，按优先级排好。

一、内部链接织网（最低成本、最快见效）

这是最低成本、最快见效的一招。

论坛帖子之间最大的浪费就是 互不引用。每篇帖子都像一座孤岛，搜索引擎只能各自判定权重，然后告诉你”内容质量低、参考价值小”。

实操路径：

1. 按主题归档——把所有帖子按 10-15 个核心主题分桶（比如技术类、评测类、答疑类、灌水类）。具体怎么分取决于你论坛的实际内容
2. 每桶选 3-5 篇老帖作为”枢纽帖”——这些帖子通常是早期写、流量还行的
3. 每发新帖必带 3-5 条内链——指向同桶的枢纽帖和相关帖
4. 老帖回填——挑流量掉得最厉害的若干篇（10%、20% 都行，量力而为），每篇手工加 5-10 条指向新帖的链接

加内链这件事不难但费眼睛，工具可以用 Link Whisper（WordPress）或自己写个脚本扫同主题关键词自动推荐。Discuz 这种老牌论坛直接用模板钩子批量加也行。

二、用 Topic Cluster 把零散变成集群（内容架构）

存量不是多，是散。把它们绑成”一棵棵树”。

论坛主题 SEO 的标准模型就是 Topic Cluster——一个 Pillar（支柱页）+ 几十上百篇 Cluster（簇页）。比如论坛讨论 VPS，Pillar 页就是”VPS 选购完全指南”，所有 VPS 相关的帖子都链过来，权重就往 Pillar 集中。

落地步骤：

1. 挑 5-8 个核心词——每个词必须有商业价值或流量价值，别选太宽的（”技术”），也别选太窄的（”某品牌某型号 SSD 评测”）
2. 每个核心词写 1 篇 Pillar——3000-5000 字，覆盖该主题的所有方面
3. 把所有现有帖子按主题打标签——能归到 Pillar 下的就归，归不了的就保持原样或者合并/删除
4. 每篇 Cluster 帖子的开头和结尾都加 Pillar 链接——双向链接结构

我们讨论下来，朋友选的核心词是 VPS 推荐 / 域名注册 / WordPress 建站 / 论坛运营 / 服务器安全——五个 Pillar，具体能吸附多少帖子看你论坛的实际主题分布。

三、技术地基：让爬虫愿意多看几眼

内容再好，爬虫不来也白搭。

技术 SEO 不是越多越好，论坛这种成熟系统该做的就那几件：

1. 结构化数据——帖子上 DiscussionForumPosting 或 Article 的 JSON-LD，让 Google 能识别这是论坛帖不是普通页面
2. 分板块 Sitemap——按板块拆成多个 sitemap，每个 sitemap 单独提交，权重不互相稀释
3. 静态化——Discuz/Flarum 都支持静态化，TTFB 从 1 秒降到 100ms，对收录直接有利
4. 死链清理——用 screaming frog 或自己写脚本扫 404，做 301 重定向到相关帖
5. Canonical 标签——分页帖（?page=2）都要指向第一页，避免重复内容稀释权重

技术这块儿投入产出比最高的是 分板块 Sitemap + 结构化数据——前者让爬虫知道哪些板块是重点，后者让搜索结果展示更丰富的卡片。

四、让外链自己找上门（外链策略）

论坛最大的优势：内容天然就是外链磁铁。

论坛跟普通博客比有个先天优势——帖子本身经常被别人引用。一个技术解答、一个资源链接、一段代码示例，论坛用户和站长都喜欢贴来贴去。我们要做的就是把这种自发的外链引导到主域名上。

具体动作：

1. 资源类帖子单开”引用本帖”按钮——读者引用时自然带上主域名链接
2. 热门帖子主动投稿——挑流量前若干的，向同主题博客投稿，附带原文链接
3. 友链不要乱换——找 DA ≥ 论坛自身 50% 的同主题站点换，宁缺毋滥
4. 维基百科/Wikimedia 引用——技术类、数据类帖子可以尝试往英文 Wikipedia 提交引用源

外链这块不用急，论坛 SEO 最大的红利是时间复利——一篇优质帖子可能半年后才被某个博客引用，给你带来一个 DA 60+ 的反向链接。

五、用数据盯效果（监控与迭代）

没有数据支撑的 SEO 都是玄学。

SEO 是个慢变量，必须有一套监控体系才能判断动作是否有效：

1. Google Search Console——必须接，看”覆盖率/点击率/平均排名”三个核心指标
2. 百度站长平台——中文论坛百度流量还是要抓的
3. Ahrefs / SEMrush——监控外链增长和关键词排名
4. 自建日志分析——爬虫访问频率、深度、抓取预算

建议每周固定时间（周五下午）拉一次数据，4 周一个评估周期——一个动作做下去至少要看 4 周数据才有意义，不要每周看一次就调整方向。

写在最后

存量帖子不是负担，是财富。论坛上 SEO 真正的护城河是内容存量 + 时间复利——新站拼不过老站的核心原因就是别人攒了几年甚至十几年的内容。我们做的那点优化，本质上是把过去攒下来的财富重新激活而已。

具体动作优先级排个序：

最后补一句：论坛 SEO 不是一次性工程，是持续运营。存量攒到了不代表以后就躺平——每发一篇新帖都要回到”是否符合 Pillar 主题、是否带内链、是否配结构化数据”这三个问题上来。能把这个习惯坚持下去的论坛，权重早晚会上来。

这篇是和小伙伴讨论后的整理稿，实际执行结果等他那边跑满一个季度再来回填数据。如果有同样在做论坛 SEO 的朋友，欢迎评论区交流。

我家那只橘白猫叫「西瓜」。从 Looki 这两年的 DIARY 里我筛出 5 张他直接入镜的图。算是给西瓜的「日常 5 张」。

她抱着橘猫站在粉色窗帘前

Looki 那天的 DIARY 只留了一句：

她抱着橘猫站在粉色窗帘前。

「她」是爱人，「橘猫」是西瓜。

我家那只猫一旦被抱超过 30 秒就要挣扎——唯独爱人抱他能坚持一分钟以上。我没问过原因，大概是角度：左手托后腿，右手按肩膀，西瓜整个身体正好卡在怀里最舒服的那个凹陷处。

我后来跟 Looki 提过一句：西瓜的主人是爱人，我只是个铲屎 + 加猫粮 + 半夜被踩的工具人。

那天的 DIARY 里没写我。但图里那个抱着猫的人，手臂上是她今天刚换的银色镯子——她每次抱完西瓜，镯子上都会粘着一两根橘色的毛，她说这是「我家有猫的证据」。

游戏正打到一半，旁边的橘猫已经睡熟了

游戏正打到一半，旁边的橘猫已经睡熟了。

那天我开了一把 RTS。打到第 18 分钟，回头一看，西瓜已经在我键盘旁边睡着了，前爪搭在 W 键上。

我没敢动——怕一动他就醒了。

打完那一局大概用了 30 分钟。他那 30 分钟的呼噜声大概是 24 小时里最连续的一次，也算我的白噪音。

他醒过来的时候打了个哈欠，伸了个懒腰，跳下去找水喝。走之前回头看了我一眼——那一眼的意思是「你怎么还在打，我都睡醒了」。

MOBA 战斗正酣时橘猫在桌边打盹

MOBA 战斗正酣时橘猫在桌边打盹。

这条 DIARY 几乎是养猫的全部意义。我打 MOBA，西瓜在桌边打盹，互不干扰，各有各的战场。

唯一的问题是：西瓜打盹的位置刚好在我鼠标右边 20 厘米。每当我手肘一动，他就要抬头看一下，让我心虚得不行——好像我在偷偷做了什么对不起他的事。

后来我学会了一个窍门——每打一局就摸他两下。这样他就觉得你记得他，不会突然醒来把你送回泉水。

那段时间 Looki 又记下了好几条提到西瓜的 DIARY——但那些图里他都睡在镜头外。我翻了很久才明白：养猫这件事，最常发生的状态就是「他在画面里但没入镜」。

工作间隙撸了会儿猫

工作间隙撸了会儿猫，还玩了会儿小玩偶，感觉整个人都放松了下来，生活里的小美好总是藏在这些不起眼的瞬间里。

这一条几乎是我今年夏天工作的全部节奏。

撸猫不是为了撸猫，是为了把工作间隙里那一点点烦心事从脑子里挤出去。每次手指从西瓜的下巴划过头顶，划过背，划过尾巴根部——世界就安静了几秒钟。

撸完了，西瓜会抬头看我一眼，那一眼的意思是「够了」。

后来他学会了一件事：只要我开始揉他的肚子，就是工作做完了。他会很配合地翻身、伸懒腰、然后跳下桌去喝水。这是我们之间一种无声的契约。

外卖盒还没收，橘猫已经在凳子上打起了呼噜

外卖盒还没收，橘猫已经在凳子上打起了呼噜，我也跟着犯困，这大概就是吃饱喝足后的专属默契吧。

这一条最早，是 5 张图里时间最远的一张。

那天我刚下班回来，吃完外卖没来得及收拾，西瓜已经跳到餐凳上睡着了。

那是他第一次在我吃饭的地方打盹——之前他只在我书桌前打盹。我突然意识到一件事：

这只猫已经把整个家都当成自己的了。

一个月后他开始上桌。两个月后他开始跟人抢枕头。三个月后他开始在我胸口睡觉——整个家的边界都被他一点一点吃掉。

他不是占领，他是把家当成他应该待的地方。

写完这 5 张图，我又去翻了一遍 Looki 的 DIARY 目录——里面现在有 40+ 条提到西瓜。

他在我们家六年了。

一只猫陪人过日子，到底是他在陪我们，还是我们在陪他？

大概两者都是。

晚上回家，门口放着一个快递盒子——面单上没写是谁寄来的，只是个普通的纸箱。拆开才知道是礼盒，我猜是小小 API 站长寄来的中秋礼物。

门口那个纸箱

那天我回家，门口放着一个快递盒子——面单上没写是谁寄的，所以我拆开之前还以为就是个普通快递。外箱上能看到”小心开拆 勿用刀划”、”稻香村集团旗下品牌”这样的字样。

之后问了站长，才确认这是小小 API 那边寄来的中秋礼物。

礼盒长得挺用心

拆开外箱，是稻香村的礼盒。整体是青色主调，搭一条紫色的”低糖心意”色带，金色章印写着”蔗糖含量减 35%”，右下角标着”净含量 510 克”。

礼盒印刷比较细——骑车的人、跑步的人、月兔、亭子这些小元素排在一起，传统节气感和现代运动感搭得挺协调。单看外盒就觉得是有花心思的，不是那种群发批量采购的味道。

拆开里面是这样

打开礼盒，里面是若干独立包装的月饼。颜色和造型各有不同，从图里能看到的是 4×3 排布、12 块可见的样子，每一块都有金色独立包装。

一年前的那篇同题博文

一年前我写过一篇 《免费、快速、稳定的公益 API 平台推荐》，那次是从技术维度系统介绍过小小 API——接口多、响应快、站长维护用心这些我都记下了。今年没做新的对照评测，但借这件事再把它拿出来提一下。具体功能、性能、稳定性以官网最新公告为准。

小小 API 的近况（公开信息）

按 xxapi.cn 首页公开内容整理：

• 小小 API，定位专业的 API 服务平台
• 自述技术架构：分布式部署、HTTPS、签名校验、IP 白名单、SDK 工具
• 合作伙伴页面里能找到「杜老师说」

有一点变化：新闻动态显示 2026-06 汇率接口已启用收费套餐（日/月/季/年/次数）。这和早期”完全免费”的说法不一样了，调用前最好去看一眼官方公告。

顺手加了个 CDN

礼盒之外，最近还顺手做了一件小事——给小小 API 加了层 CDN。

小小 API 官方目前还没有配置 CDN 加速器，直接调用接口偶尔会卡，加载图片资源尤其慢。我前段时间自己用的时候碰到过几次图加载半天才出来，体验确实不算好。

为了缓解这个问题，我在自己的 CDN 上做了两个域名映射，让请求先走 CDN 边缘节点，再回源到小小 API：

• 访问域名 v2.dusays.com → 回源目标 v2.xxapi.cn
• 访问域名 images.dusays.com → 回源目标 images.xxapi.cn

改完之后我自己测了一轮：首屏加载从原来的 2~3 秒降到了 500ms 左右，图片资源命中也稳了不少。

站长那边什么时候上官方 CDN 我也不清楚，等官方有动静再切回去也不迟。这之前这两个域名可以先用着。

最后想说的是——

这份中秋礼物的心意我收到了，谢谢站长。中秋没回家乡，但有人记着这件事本身，已经比月饼本身更暖。

借这篇文章，也祝站长今年中秋快乐。

下次有人再寄东西过来，我争取第一时间更新文章。 😅

这两周杜老师做技术选型时接连被两件事扎到：MinIO 在 5 月悄悄砍掉了社区版的 Web Console，11 万行代码删除；Mattermost v11 给免费版加了”消息历史最多 1 万条”的限制——旧消息还躺在数据库里，但 UI 看不见了。摆在一起看，“屠龙的少年长出了龙鳞” 这事好像不再是个比喻。写篇文章，把这两年”开源变收费”的剧本翻出来聊聊。

两件小事

第一件：MinIO 砍 WebUI。 2025 年 5 月，MinIO 悄悄合并了一个 PR，把社区版的 Web Console 砍掉了。11 万行代码删除，只剩一个能看能下载的对象浏览器。你想改个 IAM 策略、看一眼监控指标、配个桶复制？对不起，请升级到 MinIO AIStor Enterprise Edition，起价 9.6 万美元一年。

第二件：Mattermost 限 1 万消息。 2025 年 12 月发布的 v11 给免费版加了一个神奇的限制——消息历史最多 1 万条。超出的旧消息依然躺在你自己的 PostgreSQL 数据库里，但 UI 看不见、搜不到。一所学校 2000 多用户、47 万条历史消息，瞬间被”折叠”。Mattermost 的官方说法是”防止过度部署”，社区里更准确的描述是 “bait-and-switch”（先诱后宰）。

两件事摆一起看，忽然就明白了一件事：“开源的尽头是收费”不是预言，是 2025 年起所有基础设施软件正在发生的现实。

MinIO 十年堕落史

MinIO 时间线：从 Apache 2.0 起步，到 2026-04 仓库归档

MinIO 的故事值得展开讲讲，因为它把”开源变收费”的剧本演得最完整：

更扎心的是它”留了条活路”的姿态——CEO Anand Babu Periasamy 在采访里说：

“开发者真正需要的是 API，不是 Web UI。”

听听这话：你都要花 9.6 万美元买的”管理界面”，他告诉你”你不需要”。但任何用过 MinIO 的人都懂——CLI 配 IAM、Replication、监控，是真的反人类。

社区的反应也很真实：

“Open source users aren’t confused. They’re insulted. Know the difference between community and capture?”（开源用户不是困惑，是被侮辱了。”社区”和”俘获”的区别，懂吗？）

MinIO 还没有改协议——它依然 AGPLv3，依然”开源”。但 GitHub 仓库归档那一刻起，”License 是开源”和”项目是开源”，已经是两回事了。

Mattermost 的”温柔一刀”

配图：聊天 UI 顶部红色告警进度条，旧消息逐渐灰化坠入虚空——“消息历史已达到上限”

Mattermost 是”软切”，MinIO 是”硬切”。但这两种切法背后，是同一套剧本。

Mattermost v11 的”免费版 Entry”（前身 E0）加了一堆”看起来很美”的限制：

• 📜 消息历史 10,000 条（旧消息还在 DB，但 UI 不可见）
• 📨 推送通知 10,000 条/月
• 📋 看板卡片 1,000 个
• 🎬 剧本运行 5 次/月
• 🎙️ 1:1 音频通话 40 分钟
• 🤖 AI 助手 250 次/月

加上 Team Edition（真·MIT 协议版）的限制：用户上限 250，GitLab SSO 没了。

懂了吗？这套组合拳的核心是——

先用”高级功能免费”把用户圈进来，再等用户用顺手、用出依赖，最后用”消息历史上限”逼你付费。

byteiota 那篇文章里有个精准的比喻：

“Messages remain in the database but become inaccessible through the UI. It’s digital hostage-taking.”（消息还在数据库，但 UI 看不到。这是数字人质劫持。）

而且 Mattermost 没有给过渡期：v11 一发布，所有现存实例都立刻生效。一家用 Mattermost 三年的学校管理员在 GitHub 上问：

“How can we trust Mattermost from now on?”（从今以后我们还能信 Mattermost 吗？）

这问题问得真好。

这不是孤例

把 MinIO、Mattermost 放进时间线看，会发现这早已是一股潮流：

10 年间，6 个明星项目、5 种”软硬兼施”的手法，结果都是同一个：免费版越来越难用，付费墙越来越高。

背后是同一套商业逻辑：

投资人要回报 → 公司不能永远靠”开源精神”养活 → 限制免费版、倒逼付费 → 社区骂声一片但已离不开 → 收费闭环跑通。

为什么受伤的总是我们

认真想一下，为什么”屠龙少年变恶龙”的剧本总是成立？

三个必要条件同时满足，结局就注定：

1. 单一公司控制项目（不是基金会、不是广泛社区）
2. 云厂商白嫖（AWS、Azure 拿开源赚钱但不贡献代码）
3. 投资人要求回报（烧了 VC 的钱总要吐出来）

三个条件里只要少一个，情况就不同：

• Apache、CNCF 旗下的项目（Kafka、Spark、Kubernetes）——背后是基金会，董事会多公司，没人能”绑架”项目
• Linux 内核、Python、PostgreSQL——无单一雇主控制，靠捐赠 + 多家公司雇佣贡献者
• 真正”屠龙者”和”恶龙”身份长期分离的项目——更不容易翻车

这就是为什么我们以后挑基础设施，要先看它背后的”治理结构”，再看技术。

我们该警惕什么？怎么做？

四个危险信号

看到这些，赶紧想退路：

1. License 突然变更（AGPLv3、SSPL、BSL 都是”危险牌”）
2. 免费版功能悄悄缩水（不加新功能是常态；删老功能是警报）
3. Docker 镜像停止分发（”你得自己编译”等于”我们不想伺候免费用户”）
4. 公司被收购 / 融资异常（新东家可能一上任就加速商业化）

选型的五条原则

1. 看治理，不看技术——优先 Apache / CNCF / Linux Foundation 旗下项目
2. 看母公司，不看项目——单一公司主导的明星项目都有”被收编”风险
3. 看 fork 活跃度——Elastic 有 OpenSearch、Terraform 有 OpenTofu、Redis 有 Valkey，这是项目”被社区接管”能力的体现
4. 数据不外置——任何”功能限制”如果影响你能看见自己服务器上的数据（比如 Mattermost 的消息限制），都是红旗
5. 关键路径双备——别把生产环境的命脉单押给一家商业公司控制的项目

MinIO / Mattermost 现在怎么选

MinIO 的替代：

• Ceph（重，老牌，复杂）
• Garage（轻量，新兴，Rust 写）
• SeaweedFS（速度快，专注海量小文件）
• RustFS（国产，对标 MinIO 刚起步）

Mattermost 的替代：

• Matrix（端到端加密，联邦架构，法国政府都在用）
• Rocket.Chat（Slack-like，定制能力强）
• Zulip（主题分流的邮件式讨论，开发组心头好）
• 或者直接退回 Mattermost Team Edition（虽然功能少，但 MIT 协议保底）

开源精神从未消失，只是换了地方

写到这里必须说一句：开源不是死了，只是回到了它该在的地方。

那些真正的”基础设施开源”——Linux、Kubernetes、PostgreSQL、Redis（社区版 AGPLv3）、Valkey、OpenSearch、OpenTofu——活得很好。它们有的被资本恶心过，但社区顶住了、有 fork、有基金会保护、有真正的自由。

反而是那些”曾经开源，现在想收费”的明星项目，正在被市场教训：

• Elastic 被 OpenSearch 抢市场份额
• HashiCorp 被 OpenTofu 抢基础设施客户
• Redis 不得不回归 AGPLv3
• MinIO 把名字搞臭了，仓库归档了

所以这句话送给所有还在认真做开源的公司：

你可以从开源出发，走向商业。但别走得太远——你走多远，社区的耐心就消失多远。等你回头，发现 Valkey 已经站在你曾经的位置了。

至于我们用户：保持清醒、多备一手、把数据握在自己手里，就够了。

你最近被哪个”开源变收费”的项目伤过？评论区聊聊。（我知道你会说又是”软广”——但这篇文章真的没收钱 😅）

每天有大量重复工作占用我们的时间:批量重命名文件、爬取数据、发送邮件、操作 Excel、自动部署……学会用 Python 把这些重复任务自动化,每天节省 2-3 小时不是梦。本文通过 12 个真实案例,带你入门 Python 自动化。

一、Python 自动化的优势

• 易学易用:语法简洁,几天就能上手
• 生态丰富:50 万+ 第三方包,几乎所有场景都有现成工具
• 跨平台:Windows / macOS / Linux 全支持
• 胶水语言:可以调用 shell 命令、调用其他程序、操控 Excel/Word

二、环境准备

安装 Python

1
2
3
4
5
6
7
8
9
10

# macOS
brew install python3

# Ubuntu / Debian
sudo apt install python3 python3-pip

# CentOS / RHEL
sudo yum install python3 python3-pip

# Windows:官网下载安装包

推荐工具

• IDE:VS Code / PyCharm
• 虚拟环境:venv / conda
• 包管理:pip / poetry

创建虚拟环境

1
2
3
4
5

python3 -m venv myenv
source myenv/bin/activate  # macOS/Linux
myenv\Scripts\activate     # Windows

pip install requests beautifulsoup4 pandas openpyxl

三、12 个实战案例

案例 1:批量重命名文件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import os
from pathlib import Path

def batch_rename(directory, prefix="file"):
    """批量重命名目录下的所有文件"""
    path = Path(directory)
    
    for i, file in enumerate(path.iterdir(), 1):
        if file.is_file():
            # 保留扩展名
            ext = file.suffix
            new_name = f"{prefix}_{i:03d}{ext}"
            new_path = file.parent / new_name
            file.rename(new_path)
            print(f"{file.name} → {new_name}")

# 使用
batch_rename("/path/to/photos", prefix="trip2026")

案例 2:批量压缩图片

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from PIL import Image
from pathlib import Path

def compress_images(directory, quality=70):
    """批量压缩图片到指定目录"""
    path = Path(directory)
    output_dir = path / "compressed"
    output_dir.mkdir(exist_ok=True)
    
    for img_file in path.glob("*.jpg"):
        img = Image.open(img_file)
        # 等比缩放
        max_size = (1920, 1080)
        img.thumbnail(max_size)
        
        output_path = output_dir / img_file.name
        img.save(output_path, "JPEG", quality=quality, optimize=True)
        
        # 对比大小
        old_size = img_file.stat().st_size / 1024
        new_size = output_path.stat().st_size / 1024
        print(f"{img_file.name}: {old_size:.1f}KB → {new_size:.1f}KB (-{100*(1-new_size/old_size):.1f}%)")

compress_images("/path/to/photos")

案例 3:读取与处理 Excel

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

import pandas as pd

# 读取
df = pd.read_excel("data.xlsx", sheet_name="Sheet1")

# 数据处理
df['年龄'] = pd.to_numeric(df['年龄'], errors='coerce')
df = df.dropna(subset=['年龄'])

# 分组统计
summary = df.groupby('部门')['工资'].agg(['mean', 'sum', 'count'])
print(summary)

# 输出到新的 Excel
with pd.ExcelWriter('output.xlsx') as writer:
    df.to_excel(writer, sheet_name='原始数据', index=False)
    summary.to_excel(writer, sheet_name='汇总统计')

案例 4:读写 Word 文档

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

from docx import Document
from docx.shared import Pt, RGBColor

# 创建 Word
doc = Document()

# 添加标题
doc.add_heading('Python 自动生成报告', 0)

# 添加段落
p = doc.add_paragraph('这是一个自动生成的 ')
run = p.add_run('粗体')
run.bold = True
run.font.size = Pt(14)

# 添加表格
table = doc.add_table(rows=3, cols=3)
table.style = 'Light Grid Accent 1'
for i in range(3):
    for j in range(3):
        cell = table.cell(i, j)
        cell.text = f'行 {i+1} 列 {j+1}'

doc.save('report.docx')

案例 5:网页内容抓取(requests + BeautifulSoup)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

import requests
from bs4 import BeautifulSoup
import csv

def scrape_jobs(keyword, pages=1):
    """抓取招聘信息"""
    headers = {
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
    }
    
    results = []
    
    for page in range(1, pages + 1):
        url = f"https://example.com/jobs?kw={keyword}&page={page}"
        resp = requests.get(url, headers=headers, timeout=10)
        resp.raise_for_status()
        
        soup = BeautifulSoup(resp.text, 'html.parser')
        
        for job_card in soup.select('.job-card'):
            title = job_card.select_one('.title').text.strip()
            company = job_card.select_one('.company').text.strip()
            salary = job_card.select_one('.salary').text.strip()
            
            results.append({
                'title': title,
                'company': company,
                'salary': salary
            })
    
    # 保存到 CSV
    with open(f'jobs_{keyword}.csv', 'w', newline='', encoding='utf-8') as f:
        writer = csv.DictWriter(f, fieldnames=['title', 'company', 'salary'])
        writer.writeheader()
        writer.writerows(results)
    
    print(f"已抓取 {len(results)} 条数据")

scrape_jobs("Python", pages=5)

案例 6:发送邮件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication

def send_email(to, subject, body, attachments=None):
    """发送邮件"""
    smtp_server = "smtp.gmail.com"
    smtp_port = 587
    username = "your_email@gmail.com"
    password = "app_password"  # 开启两步验证后用应用专用密码
    
    msg = MIMEMultipart()
    msg['From'] = username
    msg['To'] = to
    msg['Subject'] = subject
    
    msg.attach(MIMEText(body, 'html'))
    
    # 附件
    if attachments:
        for file_path in attachments:
            with open(file_path, 'rb') as f:
                attach = MIMEApplication(f.read())
                attach.add_header('Content-Disposition', 'attachment', 
                                 filename=file_path.split('/')[-1])
                msg.attach(attach)
    
    # 发送
    with smtplib.SMTP(smtp_server, smtp_port) as server:
        server.starttls()
        server.login(username, password)
        server.send_message(msg)
    
    print(f"邮件已发送到 {to}")

# 使用
send_email(
    "recipient@example.com",
    "每日数据报告",
    "<h1>今日数据</h1><p>详见附件</p>",
    attachments=["/path/to/report.pdf"]
)

案例 7:定时任务(schedule 库)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

import schedule
import time
from datetime import datetime

def daily_report():
    """每天早上 9 点生成报告"""
    print(f"[{datetime.now()}] 生成每日报告...")
    # 实际报告生成逻辑
    generate_report()

def backup_database():
    """每天凌晨 2 点备份数据库"""
    print(f"[{datetime.now()}] 备份数据库...")
    # 备份逻辑

# 配置任务
schedule.every().day.at("09:00").do(daily_report)
schedule.every().day.at("02:00").do(backup_database)
schedule.every(30).minutes.do(check_server_status)

print("定时任务已启动,等待执行...")

while True:
    schedule.run_pending()
    time.sleep(60)

案例 8:监控服务器状态

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

import psutil
import requests
import time

def monitor_server(threshold_cpu=80, threshold_mem=80, threshold_disk=90):
    """监控服务器 CPU/内存/磁盘"""
    
    cpu = psutil.cpu_percent(interval=1)
    mem = psutil.virtual_memory().percent
    disk = psutil.disk_usage('/').percent
    
    status = {
        'time': time.strftime('%Y-%m-%d %H:%M:%S'),
        'cpu': cpu,
        'memory': mem,
        'disk': disk
    }
    
    # 告警
    alerts = []
    if cpu > threshold_cpu:
        alerts.append(f"⚠️ CPU 使用率过高: {cpu}%")
    if mem > threshold_mem:
        alerts.append(f"⚠️ 内存使用率过高: {mem}%")
    if disk > threshold_disk:
        alerts.append(f"⚠️ 磁盘使用率过高: {disk}%")
    
    if alerts:
        # 发钉钉/飞书/邮件通知
        send_alert('\n'.join(alerts))
    
    return status

def send_alert(message):
    """发送告警到钉钉"""
    webhook = "https://oapi.dingtalk.com/robot/send?access_token=XXX"
    requests.post(webhook, json={
        "msgtype": "text",
        "text": {"content": message}
    })

# 每 5 分钟检测一次
while True:
    status = monitor_server()
    print(status)
    time.sleep(300)

案例 9:数据库批量操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

import pymysql
from contextlib import contextmanager

@contextmanager
def get_conn():
    """上下文管理器:自动管理连接"""
    conn = pymysql.connect(
        host='localhost',
        user='root',
        password='password',
        database='mydb',
        charset='utf8mb4'
    )
    try:
        yield conn
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()

# 批量插入
def batch_insert(data_list):
    with get_conn() as conn:
        with conn.cursor() as cursor:
            sql = "INSERT INTO users (name, email, age) VALUES (%s, %s, %s)"
            cursor.executemany(sql, data_list)
            print(f"插入了 {cursor.rowcount} 条记录")

# 使用
users = [
    ('张三', 'zhangsan@example.com', 25),
    ('李四', 'lisi@example.com', 30),
    ('王五', 'wangwu@example.com', 28),
]
batch_insert(users)

# 批量更新
def update_user_ages():
    with get_conn() as conn:
        with conn.cursor() as cursor:
            cursor.execute("UPDATE users SET age = age + 1 WHERE active = 1")
            print(f"更新了 {cursor.rowcount} 条记录")

案例 10:JSON 与 API 调用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44

import requests
import json

def call_api(endpoint, method='GET', data=None, headers=None):
    """通用 API 调用"""
    base_url = "https://api.example.com/v1"
    url = f"{base_url}/{endpoint}"
    
    default_headers = {
        'Authorization': 'Bearer YOUR_API_KEY',
        'Content-Type': 'application/json'
    }
    if headers:
        default_headers.update(headers)
    
    try:
        if method == 'GET':
            resp = requests.get(url, headers=default_headers, timeout=10)
        elif method == 'POST':
            resp = requests.post(url, headers=default_headers, 
                               data=json.dumps(data), timeout=10)
        elif method == 'PUT':
            resp = requests.put(url, headers=default_headers,
                              data=json.dumps(data), timeout=10)
        elif method == 'DELETE':
            resp = requests.delete(url, headers=default_headers, timeout=10)
        
        resp.raise_for_status()
        return resp.json()
    
    except requests.exceptions.RequestException as e:
        print(f"API 调用失败: {e}")
        return None

# 获取用户列表
users = call_api("users")
print(json.dumps(users, indent=2, ensure_ascii=False))

# 创建订单
new_order = call_api("orders", method='POST', data={
    "product_id": 123,
    "quantity": 2,
    "user_id": 456
})

案例 11:操作 PDF

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54

from reportlab.lib.pagesizes import A4
from reportlab.pdfgen import canvas
from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont
import PyPDF2

# 1. 生成 PDF
def create_pdf(filename, title, content_lines):
    """生成 PDF 报告"""
    c = canvas.Canvas(filename, pagesize=A4)
    width, height = A4
    
    # 中文字体(需要下载字体文件)
    pdfmetrics.registerFont(TTFont('SimSun', 'SimSun.ttf'))
    c.setFont('SimSun', 16)
    
    # 标题
    c.drawString(100, height - 100, title)
    
    # 内容
    c.setFont('SimSun', 12)
    y = height - 150
    for line in content_lines:
        c.drawString(100, y, line)
        y -= 20
        if y < 50:
            c.showPage()
            y = height - 50
    
    c.save()
    print(f"PDF 已生成: {filename}")

# 2. 合并 PDF
def merge_pdfs(output, inputs):
    """合并多个 PDF"""
    merger = PyPDF2.PdfMerger()
    for pdf in inputs:
        merger.append(pdf)
    merger.write(output)
    merger.close()
    print(f"PDF 已合并: {output}")

# 3. 提取 PDF 文本
def extract_text(pdf_path):
    """提取 PDF 文本"""
    reader = PyPDF2.PdfReader(pdf_path)
    text = ""
    for page in reader.pages:
        text += page.extract_text()
    return text

# 使用
create_pdf("report.pdf", "测试报告", ["第一行", "第二行", "第三行"])
merge_pdfs("merged.pdf", ["file1.pdf", "file2.pdf"])

案例 12:命令行工具开发(Click)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

import click

@click.group()
def cli():
    """文件处理工具集"""
    pass

@cli.command()
@click.argument('directory')
@click.option('--prefix', default='file', help='文件名前缀')
@click.option('--ext', default=None, help='只处理指定扩展名')
def rename(directory, prefix, ext):
    """批量重命名文件"""
    from pathlib import Path
    path = Path(directory)
    files = path.glob(f"*.{ext}") if ext else path.iterdir()
    
    for i, f in enumerate(files, 1):
        if f.is_file():
            new_name = f"{prefix}_{i:03d}{f.suffix}"
            f.rename(f.parent / new_name)
            click.echo(f"✓ {f.name} → {new_name}")

@cli.command()
@click.argument('source')
@click.argument('target')
@click.option('--quality', default=70, help='压缩质量 1-100')
def compress(source, target):
    """压缩图片"""
    from PIL import Image
    img = Image.open(source)
    img.save(target, "JPEG", quality=quality, optimize=True)
    click.echo(f"✓ 压缩完成: {target}")

@cli.command()
@click.argument('url')
@click.option('--output', '-o', default='output.html')
def fetch(url, output):
    """下载网页内容"""
    import requests
    resp = requests.get(url)
    with open(output, 'w', encoding='utf-8') as f:
        f.write(resp.text)
    click.echo(f"✓ 已保存到 {output}")

if __name__ == '__main__':
    cli()

使用方式:

1
2
3

python tools.py rename ./photos --prefix vacation --ext jpg
python tools.py compress photo.jpg small.jpg --quality 60
python tools.py fetch https://example.com -o page.html

四、自动化项目实战

综合项目:每日数据报告自动化

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174

"""
每日销售报告自动化:
1. 从数据库拉取数据
2. 用 pandas 分析
3. 生成 Excel + 图表
4. 生成 PDF 报告
5. 邮件发送给管理层
"""

import pandas as pd
import pymysql
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.application import MIMEApplication
from datetime import datetime, timedelta
import matplotlib.pyplot as plt

class DailyReport:
    def __init__(self):
        self.db_config = {
            'host': 'localhost',
            'user': 'report_user',
            'password': 'password',
            'database': 'sales'
        }
        self.email_config = {
            'smtp': 'smtp.gmail.com',
            'port': 587,
            'user': 'report@company.com',
            'password': 'app_password'
        }
    
    def fetch_data(self):
        """从数据库拉取昨日销售数据"""
        yesterday = (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%d')
        
        conn = pymysql.connect(**self.db_config)
        query = f"""
        SELECT order_id, product_name, category, amount, customer_id, created_at
        FROM orders
        WHERE DATE(created_at) = '{yesterday}'
        """
        df = pd.read_sql(query, conn)
        conn.close()
        return df, yesterday
    
    def analyze(self, df):
        """数据分析"""
        analysis = {
            '总订单数': len(df),
            '总销售额': df['amount'].sum(),
            '平均订单金额': df['amount'].mean(),
            '独立客户数': df['customer_id'].nunique(),
        }
        
        # 分类销售统计
        category_stats = df.groupby('category')['amount'].agg(['sum', 'count', 'mean'])
        
        # TOP 10 商品
        top_products = df.groupby('product_name')['amount'].sum().nlargest(10)
        
        return analysis, category_stats, top_products
    
    def generate_charts(self, df, date):
        """生成图表"""
        fig, axes = plt.subplots(2, 2, figsize=(14, 10))
        
        # 1. 各类别销售额
        category_sum = df.groupby('category')['amount'].sum()
        axes[0, 0].pie(category_sum.values, labels=category_sum.index, autopct='%1.1f%%')
        axes[0, 0].set_title('各类别销售额占比')
        
        # 2. 时段分布
        df['hour'] = pd.to_datetime(df['created_at']).dt.hour
        hourly = df.groupby('hour')['amount'].sum()
        axes[0, 1].bar(hourly.index, hourly.values)
        axes[0, 1].set_title('时段销售分布')
        axes[0, 1].set_xlabel('小时')
        
        # 3. 订单金额分布
        axes[1, 0].hist(df['amount'], bins=30, edgecolor='black')
        axes[1, 0].set_title('订单金额分布')
        axes[1, 0].set_xlabel('金额')
        
        # 4. TOP 10 商品
        top = df.groupby('product_name')['amount'].sum().nlargest(10)
        axes[1, 1].barh(top.index, top.values)
        axes[1, 1].set_title('TOP 10 商品')
        
        plt.tight_layout()
        chart_path = f'charts_{date}.png'
        plt.savefig(chart_path, dpi=100, bbox_inches='tight')
        plt.close()
        return chart_path
    
    def generate_excel(self, df, analysis, category_stats, top_products, date):
        """生成 Excel 报告"""
        excel_path = f'report_{date}.xlsx'
        
        with pd.ExcelWriter(excel_path, engine='openpyxl') as writer:
            # 概览
            pd.DataFrame([analysis]).T.to_excel(
                writer, sheet_name='概览', header=['数值'])
            
            # 明细
            df.to_excel(writer, sheet_name='订单明细', index=False)
            
            # 分类统计
            category_stats.to_excel(writer, sheet_name='分类统计')
            
            # TOP 商品
            top_products.to_excel(writer, sheet_name='TOP 商品')
        
        return excel_path
    
    def send_email(self, date, attachments, recipients):
        """发送邮件"""
        msg = MIMEMultipart()
        msg['Subject'] = f'每日销售报告 - {date}'
        
        html = f"""
        <h2>每日销售报告</h2>
        <p><strong>报告日期:</strong> {date}</p>
        <p>详细数据请查看附件。</p>
        <p><small>本报告由系统自动生成,请勿直接回复。</small></p>
        """
        msg.attach(MIMEText(html, 'html'))
        
        for file_path in attachments:
            with open(file_path, 'rb') as f:
                attach = MIMEApplication(f.read())
                attach.add_header('Content-Disposition', 'attachment',
                                 filename=file_path.split('/')[-1])
                msg.attach(attach)
        
        with smtplib.SMTP(self.email_config['smtp'], self.email_config['port']) as server:
            server.starttls()
            server.login(self.email_config['user'], self.email_config['password'])
            for recipient in recipients:
                msg['To'] = recipient
                server.send_message(msg)
                del msg['To']
        
        print(f"报告已发送给 {len(recipients)} 位收件人")
    
    def run(self):
        """执行完整流程"""
        print("开始生成每日报告...")
        
        # 1. 拉取数据
        df, date = self.fetch_data()
        print(f"  ✓ 拉取 {len(df)} 条订单数据")
        
        # 2. 分析
        analysis, category_stats, top_products = self.analyze(df)
        print(f"  ✓ 总销售额: ¥{analysis['总销售额']:.2f}")
        
        # 3. 图表
        chart_path = self.generate_charts(df, date)
        print(f"  ✓ 图表已生成: {chart_path}")
        
        # 4. Excel
        excel_path = self.generate_excel(df, analysis, category_stats, top_products, date)
        print(f"  ✓ Excel 已生成: {excel_path}")
        
        # 5. 邮件
        recipients = ['manager@company.com', 'ceo@company.com']
        self.send_email(date, [excel_path, chart_path], recipients)
        print("✓ 报告流程完成")

# 定时执行
if __name__ == '__main__':
    report = DailyReport()
    report.run()

配合 crontab 每天 8 点自动跑:

1

0 8 * * * cd /opt/reports && /usr/bin/python3 daily_report.py

五、Python 自动化最佳实践

1. 错误处理

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

import logging

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('app.log'),
        logging.StreamHandler()
    ]
)

def safe_operation(func):
    """装饰器:统一异常处理"""
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logging.error(f"{func.__name__} 执行失败: {e}", exc_info=True)
            # 发告警
            send_alert(f"任务失败: {func.__name__}")
    return wrapper

@safe_operation
def my_task():
    # 业务逻辑
    pass

2. 配置文件分离

1
2
3
4
5
6
7
8
9

# config.yaml
database:
  host: localhost
  port: 3306
  user: root

email:
  smtp: smtp.gmail.com
  port: 587

1
2
3
4

import yaml

with open('config.yaml') as f:
    config = yaml.safe_load(f)

3. 日志规范

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

import logging

logger = logging.getLogger(__name__)

# DEBUG:调试信息
logger.debug("变量 x = %s", x)

# INFO:正常流程
logger.info("开始处理 %s 条记录", count)

# WARNING:可恢复异常
logger.warning("网络超时,准备重试")

# ERROR:严重错误
logger.error("数据库连接失败")

# CRITICAL:致命错误
logger.critical("系统无法继续运行")

4. 类型注解

1
2
3
4
5
6
7
8

from typing import List, Dict, Optional

def process_data(items: List[Dict[str, str]], 
                filter_key: Optional[str] = None) -> List[Dict]:
    """处理数据(带类型注解)"""
    if filter_key:
        return [item for item in items if filter_key in item]
    return items

5. 测试

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

import unittest

def add(a, b):
    return a + b

class TestMath(unittest.TestCase):
    def test_add(self):
        self.assertEqual(add(1, 2), 3)
        self.assertEqual(add(-1, 1), 0)
    
    def test_add_negative(self):
        with self.assertRaises(TypeError):
            add("1", 2)

if __name__ == '__main__':
    unittest.main()

六、常用库速查

七、学习路径

入门(2 周)

1. Python 基础语法(变量、控制流、函数)
2. 文件读写、异常处理
3. requests + BeautifulSoup 爬虫
4. pandas 处理 Excel/CSV

进阶(4 周)

1. 数据库操作(SQLAlchemy)
2. 多线程/异步(asyncio)
3. 命令行工具开发(click)
4. 项目实战:自动化报告系统

高级(8 周)

1. Web 自动化(selenium/playwright)
2. 异步爬虫(scrapy/aiohttp)
3. 设计模式与架构
4. CI/CD、测试、Docker 化

总结

Python 自动化的核心思想:

1. 识别重复:哪些任务每周/每天都在做?
2. 拆解步骤:把任务拆成可编程的小步骤
3. 编写脚本:用 Python 串联步骤
4. 定时执行:让 cron 或任务调度器自动跑

哪些场景适合自动化:

• 重复 3 次以上的任务
• 数据格式固定的报表生成
• 大量文件的批量处理
• 监控告警、日志分析
• API 数据同步

哪些场景不要自动化:

• 一次性的简单任务
• 需要复杂判断的业务流程
• 经常变化的需求

最重要的:从今天开始,把你每天重复做的事,试着写个脚本替代它。一次写、永久用,这就是自动化的复利效应。

据 Akamai 报告,2026 年 Web 应用攻击同比增长 40%,其中注入攻击和身份认证失效占比最高。无论你是开发者还是运维,了解 Web 应用常见漏洞都是必修课。本文基于 OWASP Top 10(2021 版),通过真实案例 + 代码讲解,带你建立完整的 Web 安全防护意识。

一、什么是 OWASP

OWASP(Open Worldwide Application Security Project)是全球最权威的 Web 安全组织,定期发布 Top 10 漏洞榜单,代表最常见、最危险的 Web 安全风险。

OWASP Top 10 (2021):

1. A01 - 访问控制失效 (Broken Access Control)
2. A02 - 加密机制失效 (Cryptographic Failures)
3. A03 - 注入 (Injection)
4. A04 - 不安全设计 (Insecure Design)
5. A05 - 安全配置错误 (Security Misconfiguration)
6. A06 - 易受攻击和过时的组件 (Vulnerable Components)
7. A07 - 身份认证失效 (Identification & Authentication Failures)
8. A08 - 软件和数据完整性失效 (Software & Data Integrity Failures)
9. A09 - 安全日志和监控失效 (Security Logging & Monitoring Failures)
10. A10 - 服务端请求伪造 (SSRF)

二、A01 访问控制失效

漏洞描述

用户能访问未授权的资源或功能。比如普通用户访问管理后台、修改他人订单。

典型案例

1
2
3
4
5
6

# 漏洞代码:直接用客户端传入的 user_id 查询
@app.route('/api/orders')
def get_orders():
    user_id = request.args.get('user_id')  # 攻击者改成别人的 ID
    orders = Order.query.filter_by(user_id=user_id).all()
    return jsonify([o.to_dict() for o in orders])

攻击: