该后台使用 vite + ts + pnpm + vue3 + element-plus + tailwindcss 等技术栈构成。没有添加任意可视化图标等插件。以最小功能，最基础功能展现。用户可以额外添加可使用的插件逻辑。
该后台后端使用 php8.2 + laravel 10 + mysql
该后台后端 go 语言版本开发中。将使用 gframe2.5.2
源码： https://github.com/vini123/simpleAdmin
在线体验： https://www.zeipan.com/admin
权限以及密码一键复位: https://v3test.yuepaibao.com/admin/api/reset
测试账号以及密码： zhoulin@xiangrong.pro、 111111 （如果发现登录不了，可一键复位谢谢）
房间传送门2： https://www.mlxiu.com/room/70188
现在有这样一个需求。本地推送前端代码到 codeberg 后，服务端拉取代码，打包后，放到对应的位置。
服务器服务，全是由 docker 构建，文件复制移动全靠 volume 。
新建 webhook-server 目录，里边有下边几个文件。
.
├── Dockerfile
├── package.json
└── server.jsDockerfile：
FROM node:latest
WORKDIR /app
# 新增：安装docker客户端
RUN apt-get update && apt-get install -y ca-certificates curl \
&& curl -fsSL https://mirrors.aliyun.com/docker-ce/linux/static/stable/x86_64/docker-27.3.1.tgz \
| tar -xz -C /usr/local/bin --strip-components=1 docker/docker \
&& apt-get clean && rm -rf /var/lib/apt/lists/*
COPY package.json ./
RUN npm install --omit=dev
COPY server.js ./
EXPOSE 9000
CMD ["node", "server.js"]package.json
{
"name": "webhook-server",
"version": "1.0.0",
"private": true,
"dependencies": {
"express": "^5.0"
}
}server.js
/**
* Webhook 接收服务
*
* 接收 Codeberg (Gitea) 的 push 通知，验证签名后触发 deploy.sh
*/
const express = require('express');
const crypto = require('crypto');
const { spawn } = require('child_process');
const PORT = parseInt(process.env.WEBHOOK_PORT || '9000', 10);
const SECRET = process.env.WEBHOOK_SECRET || '';
const DEPLOY_SCRIPT = process.env.DEPLOY_SCRIPT || '/opt/frontend-cicd/deploy.sh';
const TARGET_BRANCH = `refs/heads/${process.env.BRANCH || 'main'}`;
if (!SECRET || SECRET === 'CHANGE_ME') {
console.error('[ERROR] 请先设置 WEBHOOK_SECRET');
process.exit(1);
}
const app = express();
app.use('/webhook', express.raw({ type: 'application/json' }));
app.post('/webhook/deploy', (req, res) => {
const body = req.body.toString('utf8');
const sig = req.headers['x-gitea-signature'] || '';
const event = req.headers['x-gitea-event'] || '';
// 验证 HMAC-SHA256 签名
const expected = crypto.createHmac('sha256', SECRET).update(body).digest('hex');
if (sig !== expected) {
console.warn(`[${ts()}] 签名验证失败`);
return res.status(403).json({ error: 'Invalid signature' });
}
if (event !== 'push') return res.json({ msg: `忽略事件: ${event}` });
let payload;
try { payload = JSON.parse(body); } catch { return res.status(400).end(); }
const ref = payload.ref || '';
if (ref !== TARGET_BRANCH) return res.json({ msg: `忽略分支: ${ref}` });
const pusher = payload.pusher?.login || '?';
const commits = (payload.commits || []).length;
console.log(`[${ts()}] ${pusher} 推送了 ${commits} 个提交到 ${ref}，开始部署...`);
// 先返回 200，避免 Codeberg 超时
res.json({ msg: 'deploy triggered' });
console.log(`[${ts()}] 开始执行部署脚本: ${DEPLOY_SCRIPT}`);
const child = spawn('bash', [DEPLOY_SCRIPT], { timeout: 600000 });
child.stdout.on('data', (data) => {
console.log(`[${ts()}] [部署stdout] ${data.toString()}`);
});
child.stderr.on('data', (data) => {
console.error(`[${ts()}] [部署stderr] ${data.toString()}`);
});
child.on('close', (code) => {
if (code === 0) {
console.log(`[${ts()}] 部署完成，退出码:${code}`);
} else {
console.error(`[${ts()}] 部署失败，脚本退出码:${code}`);
}
});
child.on('error', (err) => {
console.error(`[${ts()}] 启动脚本异常:`, err);
});
});
app.get('/health', (_, res) => {
res.json({ ok: true, uptime: process.uptime(), time: new Date().toISOString() });
});
app.listen(PORT, () => {
console.log(`Webhook 服务已启动，端口 ${PORT}`);
console.log(` POST /webhook/deploy (分支: ${TARGET_BRANCH})`);
console.log(` GET /health`);
});
function ts() { return new Date().toISOString(); }这里是完整的目录和文件。
.
├── deploy.sh
├── docker-compose.yml
├── setup.sh
└── webhook-server
├── Dockerfile
├── package.json
└── server.jsdeploy.sh
#!/bin/bash
# ============================================================
# 自动部署脚本 —— 兼容 webhook docker 容器运行
# ============================================================
set -euo pipefail
# ======================== 配置（修改路径！） ========================
PROJECT_DIR="/alidata/code/editor.manhuariji.com"
DEPLOY_DIR="/alidata/www/editor.manhuariji.com/public"
DOCKER_IMAGE="xr-node:20"
CONTAINER_NAME="editor-mhrj"
BUILD_CMD="npm run build:h5"
BUILD_OUTPUT="dist"
DETAIL_OUTPUT="build/h5"
BRANCH="main"
# 【修改点1】日志放到容器内/app目录，拥有写入权限！
LOG_FILE="/app/editor-mhrj-deploy.log"
LOCK_FILE="/tmp/editor-mhrj-deploy.lock"
# =====================================================
# ---- 防并发 ----
if [ -f "$LOCK_FILE" ]; then
echo "$(date '+%Y-%m-%d %H:%M:%S') [WARN] 有部署正在运行，跳过" >> "$LOG_FILE"
exit 0
fi
touch "$LOCK_FILE"
trap 'rm -f "$LOCK_FILE"' EXIT
# log函数：同时输出到日志文件 AND 控制台（调试关键！）
log() {
local msg="$(date '+%Y-%m-%d %H:%M:%S') $1"
echo "$msg" >> "$LOG_FILE"
echo "$msg" # 打印到终端，Node服务可以捕获
}
log "========================================"
log "开始部署"
log "========================================"
# ---- 1. Git 拉取代码 ----
log "[1/3] cd $PROJECT_DIR && git pull ..."
cd "$PROJECT_DIR"
PREV=$(git rev-parse --short HEAD 2>/dev/null || echo "none")
# 强制同步远端代码，丢弃本地所有修改（适合自动部署服务器）
log "[INFO] 重置本地代码，对齐远端origin/${BRANCH}"
git fetch origin "${BRANCH}" >> "$LOG_FILE" 2>&1
git reset --hard "origin/${BRANCH}" >> "$LOG_FILE" 2>&1
NEW=$(git rev-parse --short HEAD)
#if [ "$PREV" = "$NEW" ]; then
# log "没有新提交 ($NEW)，跳过构建"
# exit 0
# fi
log "代码更新: $PREV -> $NEW"
# ---- 2. Docker 容器内构建 ----
# 新增：清理宿主机缓存，腾出内存
log "[INFO] 清理系统缓存释放内存"
sync; echo 3 > /proc/sys/vm/drop_caches
log "[2/3] Docker 容器构建 ($DOCKER_IMAGE) ..."
log " npm install + $BUILD_CMD"
docker run --rm \
--memory=2G \
--memory-swap=3G \
--name "$CONTAINER_NAME" \
-v "${PROJECT_DIR}":/www \
-w /www \
"$DOCKER_IMAGE" \
bash -c "
npm config set registry https://registry.npmmirror.com
npm install
NODE_OPTIONS='--max-old-space-size=1400' npm run build:h5
" >> "$LOG_FILE" 2>&1
if [ $? -ne 0 ]; then
log "[ERROR] 构建失败！"
exit 1
fi
if [ ! -d "${PROJECT_DIR}/${BUILD_OUTPUT}" ]; then
log "[ERROR] 构建产物目录 '${BUILD_OUTPUT}' 不存在"
exit 1
fi
log "构建完成，产物: ${PROJECT_DIR}/${BUILD_OUTPUT}"
# ---- 3. 部署 ----
log "[3/3] 部署到 $DEPLOY_DIR ..."
# 备份逻辑保留（可选）
#if [ -d "$DEPLOY_DIR" ] && [ "$(ls -A "$DEPLOY_DIR" 2>/dev/null)" ]; then
# BACKUP="${DEPLOY_DIR}.backup"
# rm -rf "$BACKUP"
# mv "$DEPLOY_DIR" "$BACKUP"
# log "旧版本备份到 $BACKUP"
#fi
mkdir -p "$DEPLOY_DIR"
SRC="${PROJECT_DIR}/${BUILD_OUTPUT}/${DETAIL_OUTPUT}"
# 使用 cp -r 源头目录/. 方案，完美规避所有通配符问题
if [ -d "$SRC" ]; then
cp -r "${SRC}/." "${DEPLOY_DIR}/"
else
log "[ERROR] 构建目录不存在: $SRC"
exit 1
fi
log "========================================"
log "部署完成！($PREV -> $NEW)"
log "========================================"
# ---- 日志截断 ----
MAX_LINES=3000
if [ -f "$LOG_FILE" ]; then
LINES=$(wc -l < "$LOG_FILE")
if [ "$LINES" -gt "$MAX_LINES" ]; then
tail -n "$MAX_LINES" "$LOG_FILE" > "${LOG_FILE}.tmp"
mv "${LOG_FILE}.tmp" "$LOG_FILE"
fi
fibash.sh
#!/bin/bash
# ============================================================
# 一键初始化脚本
#
# 用法：
# 1. scp -r frontend-cicd root@服务器IP:/opt/
# 2. 编辑 /opt/frontend-cicd/.env（填仓库地址和 Secret）
# 3. bash /opt/frontend-cicd/setup.sh
# ============================================================
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
source "$SCRIPT_DIR/.env"
echo "=========================================="
echo " editor.manhuariji.com 自动部署初始化"
echo "=========================================="
# ---- 1. 检查环境 ----
echo ""
echo "[1/5] 检查环境 ..."
for cmd in git docker; do
if ! command -v $cmd &>/dev/null; then
echo "[ERROR] $cmd 未安装"
exit 1
fi
done
echo " git: $(git --version)"
echo " docker: $(docker --version)"
if [ "$WEBHOOK_SECRET" = "CHANGE_ME" ]; then
echo ""
echo "[ERROR] 请先编辑 .env，把 WEBHOOK_SECRET 改成随机字符串"
echo " 生成方法: openssl rand -hex 24"
exit 1
fi
# ---- 2. 克隆项目 ----
echo ""
echo "[2/5] 项目代码 ..."
if [ -d "$PROJECT_DIR/.git" ]; then
echo " 已存在，git pull ..."
cd "$PROJECT_DIR" && git pull
else
echo " 克隆 $REPO_URL ..."
mkdir -p "$(dirname "$PROJECT_DIR")"
git clone "$REPO_URL" "$PROJECT_DIR"
fi
# ---- 3. 首次构建验证 ----
echo ""
echo "[3/5] 首次构建验证 ..."
cd "$PROJECT_DIR"
docker run --rm \
--name "$CONTAINER_NAME" \
-v "${PROJECT_DIR}":/www \
-w /www \
"$DOCKER_IMAGE" \
bash -c "npm install && $BUILD_CMD"
if [ -d "$BUILD_OUTPUT" ]; then
mkdir -p "$(dirname "$DEPLOY_DIR")"
rm -rf "$DEPLOY_DIR"
mv "$BUILD_OUTPUT" "$DEPLOY_DIR"
echo " 首次部署完成 -> $DEPLOY_DIR"
else
echo "[WARN] 构建产物不存在，跳过部署"
fi
# ---- 4. 准备 Webhook 镜像依赖 ----
echo ""
echo "[4/5] 准备 Webhook 镜像 ..."
# 把宿主机的 docker CLI 复制到 webhook-server 目录，Dockerfile 会 COPY 进去
# 这样构建镜像时不需要拉任何外部镜像
cp "$(which docker)" "$SCRIPT_DIR/webhook-server/docker"
echo " 已复制 docker CLI -> webhook-server/docker"
# ---- 5. 启动 Webhook ----
echo ""
echo "[5/5] 启动 Webhook 服务 ..."
cd "$SCRIPT_DIR"
docker compose up -d --build
echo ""
echo "=========================================="
echo " 初始化完成！"
echo "=========================================="
echo ""
echo " Webhook: http://服务器IP:$WEBHOOK_PORT/webhook/deploy"
echo " Secret: $WEBHOOK_SECRET"
echo ""
echo " 验证："
echo " curl http://localhost:$WEBHOOK_PORT/health"
echo " docker logs editor-mhrj-webhook"
echo ""
echo " 下一步："
echo " 1. 阿里云安全组放行 $WEBHOOK_PORT 端口"
echo " 2. Codeberg → Settings → Webhooks → Add Webhook (Gitea)"
echo " URL: http://你的服务器IP:$WEBHOOK_PORT/webhook/deploy"
echo " Secret: $WEBHOOK_SECRET"
echo " Event: Push Events"
echo " 3. 本地 git push 验证"
echo ""docker-compose.yaml
"docker-compose.yml" 25L, 851C 7,5 全部
services:
webhook-server:
build: ./webhook-server
container_name: editor-mhrj-webhook
restart: always
privileged: true
volumes:
- ./webhook-server/server.js:/app/server.js
# 挂载 docker.sock，让容器能调用宿主机的 docker 执行 deploy.sh
- /var/run/docker.sock:/var/run/docker.sock
- /root/.ssh:/root/.ssh:ro
# 挂载部署脚本（只读）
- ./deploy.sh:/app/deploy.sh:ro
- /alidata/code/editor.manhuariji.com:/alidata/code/editor.manhuariji.com
- /alidata/www/editor.manhuariji.com/public:/alidata/www/editor.manhuariji.com/public
environment:
- WEBHOOK_SECRET=${WEBHOOK_SECRET}
- WEBHOOK_PORT=${WEBHOOK_PORT:-9000}
- BRANCH=${BRANCH:-main}
- DEPLOY_SCRIPT=/app/deploy.sh
networks:
- xrnet
networks:
xrnet:
external: true.env
# ======================== 按需修改 ========================
# Codeberg 仓库 SSH 地址（仅 setup.sh 首次 clone 用，deploy.sh 走已有的 remote）
REPO_URL=git@codeberg.org:manhuariji/manhuariji.h5.git
# 项目在宿主机上的路径
PROJECT_DIR=/alidata/code/editor.manhuariji.com
# 部署目标（Nginx public 目录）
DEPLOY_DIR=/alidata/www/editor.manhuariji.com/public
# 构建用的 Docker 镜像（你自己的镜像）
DOCKER_IMAGE=xr-node:20
# 容器名
CONTAINER_NAME=editor-mhrj
# 构建命令
BUILD_CMD=npm run build:h5
# 构建产物目录名
BUILD_OUTPUT=dist
# 打包输出详情目录
DETAIL_OUTPUT=build/h5/*
# 部署分支
BRANCH=main
# ---- WEBHOOK SECRET（和 git 无关！）----
# 这是 Codeberg 和你的 webhook 服务之间的"暗号"，用来验证 HTTP 请求确实来自 Codeberg。
# 防止别人往你的 webhook 地址随便发 POST 触发部署。
# 生成方法：openssl rand -hex 24
# 这个值要和 Codeberg Webhook 设置页填的 Secret 一致。
# WEBHOOK_SECRET=CHANGE_ME
WEBHOOK_SECRET=b85cb03fd42507f5f921ac4534e76155c0045aed49d0236a
# Webhook 监听端口
WEBHOOK_PORT=9000运行起来: docker compose up -d
location /cicd/ {
proxy_pass http://webhook-server:9000/;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header X-Forwarded-Host $host:$server_port;
proxy_set_header X-Forwarded-Port $server_port;
proxy_set_header X-Forwarded-Context /cicd;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_connect_timeout 60s;
proxy_read_timeout 300s;
proxy_send_timeout 300s;
}
location = /cicd {
return 301 /cicd/;
}然后通过域名就可以开始部署了： https://xxx.com/cicd/webhook/deploy
登录 codeberg，进入项目，点击设置，切换到 web钩子页签。再点击添加web钩子， 选择 Forgejo。开始进行配置。
密钥文件和服务器上的文件一致哈。
可以点击 测试推送进行测试，也可以主动推送代码进行测试。
spatie/laravel-backup 框架不仅可以备份数据库，还能方便的备份文件，以及通知。
https://github.com/spatie/laravel-backup
https://spatie.be/docs/laravel-backup/v10/introduction
composer require spatie/laravel-backup
php artisan vendor:publish --provider="Spatie\Backup\BackupServiceProvider" --tag=backup-config备份相关的配置文件在 config/backup.php 里。可以根据自己的情况调整处理。
如果是使用 php 镜像运行的项目，可能会出现下边这样的错误。
Starting backup...\nDumping database test_holovision...\nBackup failed because: The dump process failed with a none successful exitcode.\nExitcode\n========\n127: Command not found\n\nOutput\n======\n\n\nError Output\n============\nsh: 1: mysqldump: not found\n.\n#0 /www/www.holovision.com/vendor/spatie/db-dumper/src/DbDumper.php(200):
就是 mysqldump 不存在。这个需要在镜像里单独安装。推荐装 mariadb 的，不装 mysql 的。
可以进入容器，在容器中安装完成测试。
apt-get update
apt-get install -y --no-install-recommends mariadb-client
rm -rf /var/lib/apt/lists/确定 ok 的话，可以在构建 php 镜像时，加入 mysqldump 的安装的部分。
# 安装 mariadb-client（内含 mysqldump）
RUN apt-get update && apt-get install -y --no-install-recommends \
mariadb-client \
&& rm -rf /var/lib/apt/lists/*新建 BackupCommand .
php artisan make:command BackupCommand填入代码
option('force');
$backupSetting = DatabaseBackupSetting::first();
if (!$backupSetting) {
Log::channel('backup')->error('备份设置未配置');
return;
}
if (!$backupSetting->is_enable && !$force) {
Log::channel('backup')->error('备份设置已关闭:' . $force);
return;
}
$last_backup_time = $backupSetting->last_backup_time;
if ($last_backup_time) {
$need_backup = false;
if ($backupSetting->backup_interval == 1) {
// 按天进行备份
if (Carbon::parse($last_backup_time)->addDays(1)->addMinutes(-2)->lt(Carbon::now())) {
$need_backup = true;
}
} elseif ($backupSetting->backup_interval == 2) {
// 按周进行备份
if (Carbon::parse($last_backup_time)->addweeks(1)->addMinutes(-2)->lt(Carbon::now())) {
$need_backup = true;
}
} elseif ($backupSetting->backup_interval == 3) {
// 按月进行备份
if (Carbon::parse($last_backup_time)->addMonths(1)->addMinutes(-2)->lt(Carbon::now())) {
$need_backup = true;
}
}
if (!$need_backup) {
Log::channel('backup')->info('还未到备份时间', ['now' => Carbon::now()->toDateTimeString(), 'last_backup_time' => $last_backup_time, 'backup_interval' => $last_backup_time]);
return;
}
}
Log::channel('backup')->info('数据备份开始:' . $force);
$databaseBackupLog = DatabaseBackupLog::create([
'type' => $force ? 'manual' : 'auto',
'status' => 'running',
'started_at' => Carbon::now()->toDateTimeString()
]);
$exitCode = Artisan::call('backup:run', [
'--only-db' => true,
]);
$output = Artisan::output();
if ($exitCode != 0) {
$databaseBackupLog->update([
'status' => 'failure'
]);
Log::channel('backup')->error('数据备份失败:', [$output]);
return;
}
$databaseBackupLog->update([
'status' => 'success',
'finished_at' => Carbon::now()->toDateTimeString()
]);
$backupSetting->update([
'last_backup_time' => $databaseBackupLog->started_at
]);
Log::channel('backup')->info('数据备份完成', [$output]);
}
}然后可以去完善定时任务相关的逻辑。
jenkins 太占 cpu，内存了，小服务器难搞。准备试试 Woodpecker CI。
官网： https://woodpecker-ci.org/docs/administration/installation/docker-compose
mkdir woodpecker
cd woodpecker
vim docker-compose.yamlservices:
woodpecker-server:
image: woodpeckerci/woodpecker-server:v3
volumes:
- ./data/woodpecker-server-data:/var/lib/woodpecker/
environment:
- WOODPECKER_OPEN=true
- WOODPECKER_HOST=${WOODPECKER_HOST}
- WOODPECKER_GITHUB=true
- WOODPECKER_GITHUB_CLIENT=${WOODPECKER_GITHUB_CLIENT}
- WOODPECKER_GITHUB_SECRET=${WOODPECKER_GITHUB_SECRET}
- WOODPECKER_AGENT_SECRET=${WOODPECKER_AGENT_SECRET}
networks:
- xrnet
woodpecker-agent:
image: woodpeckerci/woodpecker-agent:v3
command: agent
restart: always
depends_on:
- woodpecker-server
volumes:
- ./data/woodpecker-agent-config:/etc/woodpecker
- /var/run/docker.sock:/var/run/docker.sock
environment:
- WOODPECKER_SERVER=woodpecker-server:9000
- WOODPECKER_AGENT_SECRET=${WOODPECKER_AGENT_SECRET}
networks:
- xrnet
networks:
xrnet:
external: truevim .env
# 配置
WOODPECKER_GITHUB_CLIENT=
WOODPECKER_GITHUB_SECRET=
WOODPECKER_AGENT_SECRET=Jenkins 是一个开源的自动化服务器，它支持软件开发的持续集成和持续部署（CI/CD）。它允许开发团队自动化构建、测试和部署应用程序，从而提高软件开发的效率和质量。
下边来使用 docker 安装 jenkins，并配置 nginx 代理，进行子路径部署。就是 jenkins 部署到非根目录下，jenkins 只是主域名访问的一部分。
jenkins 官网下载地址：
https://www.jenkins.io/download/
mkdir jenkins
cd jenkins
mkdir jenkins_home
chown 1000:1000 jenkins_homeservices:
jenkins:
image: jenkins/jenkins:2.568.1
container_name: jenkins
restart: always
volumes:
- ./jenkins_home:/var/jenkins_home
# 挂载宿主机 docker.sock，让 Jenkins 能调用宿主机的 docker
- /var/run/docker.sock:/var/run/docker.sock
- /usr/bin/docker:/usr/bin/docker
# 挂载宿主机的 SSH 密钥（git clone Codeberg 用）
- /root/.ssh:/var/jenkins_home/.ssh:ro
environment:
- TZ:Asia/Shanghai
command: >
--prefix=/cicd
networks:
- xrnet
networks:
xrnet:
external: true这里的网络必须和宿主机中 nginx 的网络相同。因为要部署到子路径下，所以要配置 prefix。
docker compose up -d location /cicd/ {
proxy_pass http://jenkins:8080;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;
proxy_set_header X-Forwarded-Host $host:$server_port;
proxy_set_header X-Forwarded-Port $server_port;
proxy_set_header X-Forwarded-Context /cicd;
proxy_http_version 1.1;
proxy_set_header Upgrade $http_upgrade;
proxy_set_header Connection "upgrade";
proxy_connect_timeout 60s;
proxy_read_timeout 300s;
proxy_send_timeout 300s;
}
location = /cicd {
return 301 /cicd/;
}docker restart xr-nginx# 查看密码
cat ./jenkins home/secrets/initialAdminPassword
docker compose down从安装到卸载，jenkins 这玩意，我这破服务器玩不好。
前端圈7月大地震：TS 7发布、Bun 11天从Zig迁Rust、Kimi K3登顶、大厂裁员50%：
https://juejin.cn/post/7663095491467411462
https://juejin.cn/post/7663889052081389604
Vue 3.6 发布、Vue 4.0 启动、Vite 8 已落地:
https://juejin.cn/post/7662256328783200319
阿里要求全员卸载 Claude Code：事件始末与深层逻辑:
https://juejin.cn/post/7663822459978137646
运行起 dify 后，在控制台里可以看到登录调用的接口文件定义在这里：api\controllers\console\auth\login.py 现在想扩展新的接口，可以在当前目录下新建一个 register.py 文件，加入以下代码。
import os
import hashlib
import functools
import time
import logging
import secrets
import base64
from flask import make_response, request
from flask_restx import Resource
from pydantic import BaseModel, Field, field_validator, ValidationError
from sqlalchemy import select, update
from typing import Literal, Optional
from controllers.common.fields import SimpleResultOptionalDataResponse
from controllers.common.schema import (
register_response_schema_models,
register_schema_models,
)
from controllers.console import console_ns
from libs.helper import EmailStr
from libs.helper import timezone as validate_timezone_string
from libs.password import hash_password
from constants.languages import get_valid_language
from models import Account, Tenant, TenantAccountJoin
from extensions.ext_database import db
from services.errors.account import AccountRegisterError
from controllers.console.wraps import (
decrypt_password_field,
setup_required,
)
from sqlalchemy import func
from models import ApiToken
logger = logging.getLogger(__name__)
# 签名校验装饰器
def private_api_auth_required(f):
@functools.wraps(f)
def decorated(*args, **kwargs):
private_secret = os.environ.get("CONSOLE_PRIVATE_API_SECRET")
if not private_secret or len(private_secret) != 12:
return {"result": "fail", "message": "Private API service unavailable"}, 503
req_time = request.headers.get("X-API-TIME", "")
req_sign = request.headers.get("X-API-SIGN", "")
if not req_time or not req_sign:
return {"result": "fail", "message": "Missing required headers"}, 400
try:
timestamp = int(req_time)
if timestamp > 9999999999:
timestamp = timestamp // 1000
except (ValueError, TypeError):
return {"result": "fail", "message": "Invalid timestamp"}, 400
if abs(int(time.time()) - timestamp) > 300:
return {"result": "fail", "message": "Request expired"}, 401
calc_sign = (
hashlib.md5(f"{private_secret}{req_time}".encode()).hexdigest().lower()
)
if calc_sign != req_sign.lower():
return {"result": "fail", "message": "Invalid signature"}, 401
return f(*args, **kwargs)
return decorated
# 请求模型：字段改为 name，timezone 默认 Asia/Shanghai
class RegisterPayload(BaseModel):
email: EmailStr = Field(..., description="邮箱")
name: str = Field(..., min_length=1, max_length=50, description="用户名")
password: str = Field(..., description="明文密码")
language: str | None = Field(default=None, description="界面语言")
timezone: str = Field(
default="Asia/Shanghai", description="时区，默认 Asia/Shanghai"
)
@field_validator("timezone")
@classmethod
def validate_timezone(cls, value):
return validate_timezone_string(value) if value else "Asia/Shanghai"
# 获取用户详情
class UserInfoPayload(BaseModel):
account_id: str = Field(..., description="用户ID")
class GenerateApiTokenPayload(BaseModel):
account_id: str = Field(..., description="用户ID")
tenant_id: str = Field(..., description="租户ID")
# 限定只能是 dataset / app
type: Literal["dataset", "app"] = Field(
..., description="令牌类型，仅支持 dataset、app"
)
# 非必填，默认 None
app_id: Optional[str] = Field(None, description="应用ID，type 为 app 时必填")
@field_validator("app_id")
def check_app_id_required(cls, v, values):
# values.data 获取模型已解析的其他字段
token_type = values.data.get("type")
if token_type == "app" and not v:
raise ValueError("当令牌类型为 app 时，app_id 不能为空")
if token_type == "dataset" and v is not None:
raise ValueError("当令牌类型为 dataset 时，无需传入 app_id")
return v
register_schema_models(
console_ns,
RegisterPayload,
UserInfoPayload,
GenerateApiTokenPayload,
)
register_response_schema_models(console_ns, SimpleResultOptionalDataResponse)
def get_first_tenant():
stmt = select(Tenant).order_by(Tenant.created_at)
return db.session.scalar(stmt)
# 对齐 Dify 原生密码哈希逻辑
def create_password_hash(raw_pwd: str) -> tuple[str, str]:
salt = secrets.token_bytes(16)
hashed = hash_password(raw_pwd, salt)
return base64.b64encode(hashed).decode("utf-8"), base64.b64encode(salt).decode(
"utf-8"
)
@console_ns.route("/register")
class RegisterApi(Resource):
@setup_required
@private_api_auth_required
@console_ns.expect(console_ns.models[RegisterPayload.__name__])
@console_ns.response(
200, "Success", console_ns.models[SimpleResultOptionalDataResponse.__name__]
)
@decrypt_password_field
def post(self):
# 单独捕获参数校验异常
try:
args = RegisterPayload.model_validate(console_ns.payload)
except ValidationError as ve:
err_info = ve.errors()[0]
field = ".".join(map(str, err_info["loc"]))
msg = f"Parameter error: {field} - {err_info['msg']}"
return {"result": "fail", "message": msg}, 403
try:
email = args.email.strip().lower()
name = args.name.strip()
raw_pwd = args.password.strip()
lang = get_valid_language(args.language)
tz = args.timezone.strip()
# 邮箱去重
exist_stmt = select(Account).where(Account.email == email)
if db.session.scalar(exist_stmt):
raise AccountRegisterError("Email already exists")
# 生成密码
pwd_hash, pwd_salt = create_password_hash(raw_pwd)
# 新建账号（字段完全对齐数据库）
new_account = Account(
email=email,
name=name,
password=pwd_hash,
password_salt=pwd_salt,
interface_language=lang,
timezone=tz,
)
db.session.add(new_account)
db.session.flush()
# 租户处理
tenant = get_first_tenant()
if not tenant:
tenant = Tenant(name=f"{name}'s Workspace")
db.session.add(tenant)
db.session.flush()
is_first_tenant = get_first_tenant() is None
join_rel = TenantAccountJoin(
account_id=new_account.id,
tenant_id=tenant.id,
role="owner" if is_first_tenant else "admin",
current=True,
)
db.session.add(join_rel)
# 更新当前租户标记
update_all = (
update(TenantAccountJoin)
.where(TenantAccountJoin.account_id == new_account.id)
.values(current=False)
)
db.session.execute(update_all)
update_curr = (
update(TenantAccountJoin)
.where(
TenantAccountJoin.account_id == new_account.id,
TenantAccountJoin.tenant_id == tenant.id,
)
.values(current=True)
)
db.session.execute(update_curr)
# 统一提交，事务原子化
db.session.commit()
return make_response(
{
"result": "success",
"data": {
"account_id": str(new_account.id),
"email": new_account.email,
"name": new_account.name,
"tenant_id": str(tenant.id),
"tenant_name": tenant.name,
},
}
)
except Exception as e:
db.session.rollback()
logger.exception("Register failed, rollback transaction")
if isinstance(e, AccountRegisterError):
return {"result": "fail", "message": str(e)}, 400
return {"result": "fail", "message": "Register failed"}, 500
# ===================== 获取用户基本信息 =====================
@console_ns.route("/user/info")
class UserInfoApi(Resource):
@setup_required
@private_api_auth_required
def get(self):
try:
# 从 URL 查询参数取值，而非 payload
account_id = request.args.get("account_id", "").strip()
if not account_id:
return {"result": "fail", "message": "account_id is required"}, 403
args = UserInfoPayload(account_id=account_id)
except ValidationError as ve:
err_info = ve.errors()[0]
field = ".".join(map(str, err_info["loc"]))
msg = f"Parameter error: {field} - {err_info['msg']}"
return {"result": "fail", "message": msg}, 403
try:
account_id = args.account_id.strip()
# 1. 查询单条用户基础信息
user_stmt = select(
Account.id,
Account.email,
Account.name,
Account.status,
Account.timezone,
Account.interface_language,
Account.created_at,
).where(Account.id == account_id)
user_row = db.session.execute(user_stmt).first()
if not user_row:
return {"result": "success", "data": None, "message": "用户不存在"}
# 2. 查询该用户关联的租户信息
tenant_stmt = (
select(
TenantAccountJoin.account_id,
TenantAccountJoin.tenant_id,
TenantAccountJoin.role,
TenantAccountJoin.current,
Tenant.name.label("tenant_name"),
Tenant.created_at.label("tenant_created_at"),
)
.join(Tenant, TenantAccountJoin.tenant_id == Tenant.id)
.where(TenantAccountJoin.account_id == account_id)
)
tenant_row = db.session.execute(tenant_stmt).first()
# 查询用户的 api_tokens
api_token_stmt = select(
ApiToken.id,
ApiToken.tenant_id,
ApiToken.account_id,
ApiToken.type,
ApiToken.token,
ApiToken.app_id,
ApiToken.last_used_at,
).where(ApiToken.account_id == account_id)
api_token_rows = db.session.execute(api_token_stmt).all()
api_token_list = []
for row in api_token_rows:
api_token_list.append(
{
"id": str(row.id),
"tenant_id": str(row.tenant_id),
"account_id": str(row.account_id),
"type": row.type,
"token": row.token,
"app_id": row.app_id,
"last_used_at": (
row.last_used_at.isoformat() if row.last_used_at else None
),
}
)
# 3. 组装最终返回数据
user_data = {
"user_id": str(user_row.id),
"email": user_row.email,
"name": user_row.name,
"timezone": user_row.timezone,
"interface_language": user_row.interface_language,
"created_at": (
user_row.created_at.isoformat() if user_row.created_at else None
),
"tenant": tenant_row,
"api_token_list": api_token_list,
}
return {"result": "success", "data": user_data}
except Exception as e:
logger.exception("Get user list failed")
return {"result": "fail", "message": "Query failed"}, 500
# ===================== 创建 api key =====================
@console_ns.route("/generate/api_key")
class GenerateApiTokenApi(Resource):
@setup_required
@private_api_auth_required
@console_ns.expect(console_ns.models[GenerateApiTokenPayload.__name__])
def post(self):
try:
args = GenerateApiTokenPayload.model_validate(console_ns.payload)
except ValidationError as ve:
err_info = ve.errors()[0]
field = ".".join(map(str, err_info["loc"]))
msg = f"Parameter error: {field} - {err_info['msg']}"
return {"result": "fail", "message": msg}, 403
try:
account_id = args.account_id.strip()
tenant_id = args.tenant_id.strip()
type = args.type
app_id = args.app_id
# 一个租户最多可以创建 10 个 api tokens(todo 要破掉)
current_key_count = db.session.scalar(
select(func.count(ApiToken.id)).where(ApiToken.tenant_id == tenant_id)
)
if current_key_count >= 10:
logger.warning(
f"User {account_id} has reached maximum API key limit (10)"
)
return {"result": "fail", "message": "API key limit reached"}, 403
# 2. 生成API Key（和Dify原生生成规则完全一致）
token_prefix = "sk-" # Dify原生用户API Key前缀
key = ApiToken.generate_api_key(token_prefix, 24)
# 3. 创建并保存API Token
api_token = ApiToken()
api_token.tenant_id = tenant_id
api_token.account_id = account_id
api_token.type = type
api_token.token = key
if app_id:
api_token.app_id = app_id
db.session.add(api_token)
db.session.commit()
return {"result": "success", "data": {"token": key}}
except Exception as e:
logger.exception("Get user list failed")
return {"result": "fail", "message": "Query failed"}, 500
# ===================== 用户列表接口（含租户 + API Token） =====================
@console_ns.route("/user/list")
class UserListApi(Resource):
@setup_required
@private_api_auth_required
def get(self):
try:
# 解析分页参数
page = request.args.get("page", 1, type=int)
page_size = request.args.get("page_size", 20, type=int)
# 限制最大条数，防护大数据查询
page_size = min(page_size, 100)
offset = (page - 1) * page_size
# 1. 查询用户总数
total_stmt = select(func.count(Account.id))
total = db.session.scalar(total_stmt) or 0
# 2. 分页查询用户基础信息
user_stmt = (
select(
Account.id,
Account.email,
Account.name,
Account.timezone,
Account.interface_language,
Account.created_at,
)
.order_by(Account.created_at.desc())
.offset(offset)
.limit(page_size)
)
user_rows = db.session.execute(user_stmt).all()
user_id_list = [row.id for row in user_rows]
# 无用户直接返回空列表
if not user_id_list:
return {
"result": "success",
"data": {
"total": total,
"page": page,
"page_size": page_size,
"list": [],
},
}
# 3. 批量查询 用户-租户 关联信息（修复变量名笔误）
tenant_stmt = (
select(
TenantAccountJoin.account_id,
TenantAccountJoin.tenant_id,
TenantAccountJoin.role,
TenantAccountJoin.current,
Tenant.name.label("tenant_name"),
Tenant.created_at.label("tenant_created_at"),
)
.join(Tenant, TenantAccountJoin.tenant_id == Tenant.id)
.where(TenantAccountJoin.account_id.in_(user_id_list))
)
# 此处原代码写错：tenant_rows → tenant_stmt
tenant_rows = db.session.execute(tenant_stmt).all()
# 按用户ID分组租户数据
tenant_map = {}
for row in tenant_rows:
aid = row.account_id
if aid not in tenant_map:
tenant_map[aid] = []
tenant_map[aid].append(
{
"tenant_id": str(row.tenant_id),
"tenant_name": row.tenant_name,
"role": row.role,
"is_current": row.current,
"tenant_created_at": (
row.tenant_created_at.isoformat()
if row.tenant_created_at
else None
),
}
)
# 5. 组装最终返回数据
user_list = []
for user in user_rows:
user_id = user.id
user_list.append(
{
"user_id": str(user_id),
"email": user.email,
"name": user.name,
"timezone": user.timezone,
"interface_language": user.interface_language,
"created_at": (
user.created_at.isoformat() if user.created_at else None
),
"tenant_list": tenant_map.get(user_id, []),
# "api_token_list": token_map.get(user_id, []),
}
)
return {
"result": "success",
"data": {
"total": total,
"page": page,
"page_size": page_size,
"list": user_list,
},
}
except Exception as e:
logger.exception("Get user list failed")
return {"result": "fail", "message": "Query failed"}, 500
并且将该文件注册进去。修改 api/controllers/console/init.py, 添加 "register" 就好。
然后看看最近这几个功能修改涉及到的文件。
最后，如果有需要可以去打一个镜像。
docker compose build api虽然 dify 支持创建多个 api-token,也修改了 api_tokens 表，加上了 account_id 进行关联。但是，通过 api-token 转换成 current_user 的时候，这个 user 总是 tenant_account_joins 表里 role 字段是 owner 的用户。既然想给每个用户都有单独的 token，就要 current_user 能对应到每个用户身上。
从 post("/v1/datasets") 接口里，往回找，就可以找到用户绑定的地方：api\controllers\service_api\dataset\dataset.py
这里就有当前用户的获取： assert isinstance(current_user, Account)
assert isinstance(current_user, Account)在 api\controllers\service_api\wraps.py 里，有用户查询相关的逻辑。
tenant_account_join = db.session.execute(
select(Tenant, TenantAccountJoin)
.where(Tenant.id == api_token.tenant_id)
.where(TenantAccountJoin.tenant_id == Tenant.id)
.where(TenantAccountJoin.role.in_(["owner"]))
.where(Tenant.status == TenantStatus.NORMAL)
).one_or_none() # TODO: only owner information is required, so only one is returned.
if tenant_account_join:
tenant, ta = tenant_account_join
account = db.session.get(Account, ta.account_id)
# Login admin
if account:
account.current_tenant = tenant
current_app.login_manager._update_request_context_with_user(account) # type: ignore
user_logged_in.send(current_app._get_current_object(), user=current_user) # type: ignore
else:
raise Unauthorized("Tenant owner account does not exist.")既然之前在 api_tokens 表中加了 account_id 字段，那么优先就应该从 api_tokens 表查询。如果这个表列没查出对应的 account_id, 再走原来的逻辑。
def validate_dataset_token[R](view: Callable[..., R]) -> Callable[..., R]:
positional_parameters = [
parameter
for parameter in inspect.signature(view).parameters.values()
if parameter.kind
in (inspect.Parameter.POSITIONAL_ONLY, inspect.Parameter.POSITIONAL_OR_KEYWORD)
]
expects_bound_instance = bool(
positional_parameters and positional_parameters[0].name in {"self", "cls"}
)
@wraps(view)
def decorated(*args: object, **kwargs: object) -> R:
api_token = validate_and_get_api_token("dataset")
if api_token.account_id:
account = db.session.get(Account, api_token.account_id)
if not account:
raise Unauthorized(
"By account_id. Token associated account does not exist."
)
tenant = db.session.get(Tenant, api_token.tenant_id)
if not tenant or tenant.status == TenantStatus.ARCHIVE:
raise Forbidden("Workspace is invalid or archived")
account.current_tenant = tenant
current_app.login_manager._update_request_context_with_user(account) # type: ignore
user_logged_in.send(current_app._get_current_object(), user=account)
else:
# Flask may pass URL path parameters positionally, so inspect both kwargs and args.
dataset_id = kwargs.get("dataset_id")
if not dataset_id and args:
potential_id = args[0]
try:
str_id = str(potential_id)
if len(str_id) == 36 and str_id.count("-") == 4:
dataset_id = str_id
except Exception:
logger.exception("Failed to parse dataset_id from positional args")
if dataset_id:
dataset_id = str(dataset_id)
dataset = db.session.scalar(
select(Dataset)
.where(
Dataset.id == dataset_id,
Dataset.tenant_id == api_token.tenant_id,
)
.limit(1)
)
if not dataset:
raise NotFound("Dataset not found.")
if not dataset.enable_api:
raise Forbidden("Dataset api access is not enabled.")
tenant_account_join = db.session.execute(
select(Tenant, TenantAccountJoin)
.where(Tenant.id == api_token.tenant_id)
.where(TenantAccountJoin.tenant_id == Tenant.id)
.where(TenantAccountJoin.role.in_(["owner"]))
.where(Tenant.status == TenantStatus.NORMAL)
).one_or_none() # TODO: only owner information is required, so only one is returned.
if tenant_account_join:
tenant, ta = tenant_account_join
account = db.session.get(Account, ta.account_id)
# Login admin
if account:
account.current_tenant = tenant
current_app.login_manager._update_request_context_with_user(account) # type: ignore
user_logged_in.send(current_app._get_current_object(), user=current_user) # type: ignore
else:
raise Unauthorized("Tenant owner account does not exist.")
else:
raise Unauthorized("Tenant does not exist.")
if expects_bound_instance:
if not args:
raise TypeError(
"validate_dataset_token expected a bound resource instance."
)
return view(args[0], api_token.tenant_id, *args[1:], **kwargs)
return view(api_token.tenant_id, *args, **kwargs)
return decorated只加了：
if api_token.account_id:
account = db.session.get(Account, api_token.account_id)
if not account:
raise Unauthorized(
"By account_id. Token associated account does not exist."
)
tenant = db.session.get(Tenant, api_token.tenant_id)
if not tenant or tenant.status == TenantStatus.ARCHIVE:
raise Forbidden("Workspace is invalid or archived")
account.current_tenant = tenant
current_app.login_manager._update_request_context_with_user(account) # type: ignore
user_logged_in.send(current_app._get_current_object(), user=account)到这里看似是 ok 了。但是还是会报错。在 api_token = validate_and_get_api_token("dataset") 里，api_token 可能来自 ApiTokenCache。如果来自 ApiTokenCache 这里就会报错。在 ApiTokenCache 里根本就没有定要 account_id。如果不走缓存，之前在修改数据库表的时候，在模型中已经加入了 account_id 字段，自然不会报错。现在只需要在ApiTokenCache 中加入 account_id 即可。
# api\services\api_token_service.py
id: str
app_id: str | None
tenant_id: str | None
account_id: str | None
type: str
token: str
last_used_at: datetime | None
created_at: datetime | None
cached = CachedApiToken(
id=str(api_token.id),
app_id=str(api_token.app_id) if api_token.app_id else None,
tenant_id=str(api_token.tenant_id) if api_token.tenant_id else None,
account_id=str(api_token.account_id) if api_token.account_id else None,
type=api_token.type,
token=api_token.token,
last_used_at=api_token.last_used_at,
created_at=api_token.created_at,
)dify 默认的数据库表，不一定是自己想要的。这个时候就需要取修改数据库表。修改数据库表最爽的方式就是用迁移文件。在 dify 原本的迁移目录下新增迁移文件会出现迁移版本错位的问题。比如你当前建立了一个迁移文件，过段时间拉 dify 官网代码合并后，迁移有改动，这个时候就会出现迁移版本对不上。为了解决这个问题，自定义个人迁移目录，新建迁移文件就很有必要了。
现在想给 api_tokens 表新增一个 account_id 字段。因为之前的 api_tokens 表只和 tenant 表进行关联，满足不了需求。
修改 docker-compose.yaml 的 api 的配置。主要修改 volumes，因为默认是没有挂载目录，就会出现在容器中生成的文件在宿主机上没有。
api:
volumes:
- ../api:/app/api
# 匿名卷：让 .venv 目录不被宿主机覆盖，保留镜像内的内容
- /app/api/.venv然后启动服务，开始干活。（从尝试到放弃）
# 启动 docker 服务
docker compose up -d
# 进入 api 容器
docker exec -it docker-api-1 bash
# 创建自定义迁移目录
flask db init --directory migrations_custom
# 查看官方迁移当前版本
flask db current
# 给自定义迁移库打基线（7bad07dc267d 是上一步查询出来的）
flask db stamp --directory migrations_custom 7bad07dc267d --purge想法是很美好，可是执行起来，各种问题就来了。所以，还是放弃自建迁移文件的想法了。既然迁移文件不好使，那就调用 sql 来达到实现修改字段的目的了。
在 api/extensions 目录下，新建 ext_database_custom.py 文件。代码入下。
# ext_database_custom.py
from sqlalchemy import text as sql_text
from extensions.ext_database import db
from dify_app import DifyApp
def init_custom_db_fields():
"""初始化自定义数据库字段/索引，启动自动执行"""
try:
with db.engine.connect() as conn:
# 新增 account_id 字段
conn.execute(
sql_text(
"ALTER TABLE api_tokens ADD COLUMN IF NOT EXISTS account_id UUID;"
)
)
# 新增联合索引
conn.execute(
sql_text(
"CREATE INDEX IF NOT EXISTS api_token_account_idx ON api_tokens(account_id, type);"
)
)
conn.commit()
print("[CustomDB] 自定义字段&索引初始化完成")
except Exception as e:
print(f"[CustomDB] 初始化异常: {str(e)}")
# 关键：实现 Dify 扩展标准接口 init_app
def init_app(app: DifyApp):
# 这里要在 app 上下文里执行，否则 db 可能没绑定
with app.app_context():
init_custom_db_fields()
然后在 api/app_factory.py 的 initialize_extensions 函数中，引入注册插件。放在 ext_database 后边即可。
最后执行 docker compose restart 就可以了。
运行起来，查看数据库，会看到 在 api_tokens 表中已经有 account_id 字段了。
虽然数据库表里字段有了。但是模型这里也要加。
# api\models\model.py
class ApiToken(Base):
account_id = mapped_column(StringUUID, nullable=True)
拉取 dify 官方源代码构建服务。虽然能跑起来，正常运行。可是很多功能不是自己想要的，或者自己想要更多的功能。这个时候就需要修改源码。 dify 源码都提供了，修改后，重新构建镜像使用不就可以实现愿望了吗。
修改 docker/docker-compose.yaml 文件，删掉官方的镜像，使用本地的镜像构建。前端 web 页面和 api 接口的配置修改。
api:
<<: *shared-api-worker-config
image: user-dify-api:1.14.2
user: root
build:
context: ..
dockerfile: api/Dockerfile
web:
image: user-dify-web:1.14.2
build:
context: ..
dockerfile: web/Dockerfile
这样就好了。如果需要增加环境变量，在 .env 和 .env.example 还有 docker-compose.yaml 中加入就好。 volumes 也看着办。