非你网 - 第2页共78页 - 挥泪别痴心，只怨情长天未怜。盼君忆我千百世，真爱永不悔。

8月 04

发表于 2026年8月4日 • 207 views • 没有评论!

客户端向服务端推流，前端 h5 播放流。

SRS(Simple Realtime Server)是一个简单高效的实时视频服务器，支持RTMP、WebRTC、HLS、HTTP-FLV、SRT等多种实时流媒体协议。Oryx是一个一体化、开箱即用、开源的视频解决方案，可部署在云上或自建机房，以直播和WebRTC等能力赋能你的业务。

阅读全文 >>

8月 03

Error Class “Psy\Readline\Interactive\Helper\TokenHelper” not found.

发表于 2026年8月3日 • 178 views • 没有评论!

php

php artisan tinker; 使用时，报错 Error Class "Psy\Readline\Interactive\Helper\TokenHelper" not found. 处理。

https://packagist.org/packages/psy/psysh

psy 版本：Psy Shell v0.12.24

root@a2129439b046:/www/tcrj.laravel# php artisan tinker;
Psy Shell v0.12.24 (PHP 8.5.5 — cli) by Justin Hileman
New PHP manual is available (latest: 3.1.1). Update with `doc --update-manual`

tinker 版本：

root@a2129439b046:/www/tcrj.laravel# composer show laravel/tinker
name : laravel/tinker
descrip. : Powerful REPL for the Laravel framework.
keywords : REPL, Tinker, laravel, psysh
versions : * v3.0.2
released : 2026-03-17, 4 months ago
type : library
license : MIT License (MIT) (OSI approved) https://spdx.org/licenses/MIT.html#licenseText
homepage :
source : [git] https://github.com/laravel/tinker.git 4faba77764bd33411735936acdf30446d058c78b
dist : [zip] https://api.github.com/repos/laravel/tinker/zipball/4faba77764bd33411735936acdf30446d058c78b 4faba77764bd33411735936acdf30446d058c78b
path : /www/tcrj.laravel/vendor/laravel/tinker
names : laravel/tinker

进入 vendor 中，确实没看到 TokenHelper 这个类。手动去官网复制一份下来。
https://github.com/bobthecow/psysh/blob/main/src/Readline/Interactive/Helper/TokenHelper.php

进入到 vendor 目录，创建 TokenHelper.php 文件，复制代码进去。

<?php

/*
 * This file is part of Psy Shell.
 *
 * (c) 2012-2026 Justin Hileman
 *
 * For the full copyright and license information, please view the LICENSE
 * file that was distributed with this source code.
 */

namespace Psy\Readline\Interactive\Helper;

/**
 * Shared token analysis utilities.
 */
class TokenHelper
{
    /** @var int[] Token types that indicate a statement continues on the next line */
    private const TRAILING_TOKEN_OPS = [
        \T_OBJECT_OPERATOR, \T_PAAMAYIM_NEKUDOTAYIM,
        \T_BOOLEAN_AND, \T_BOOLEAN_OR, \T_LOGICAL_AND, \T_LOGICAL_OR,
        \T_DOUBLE_ARROW, \T_COALESCE, \T_SPACESHIP,
    ];

    /** @var string[] Single-character operators that indicate continuation */
    private const TRAILING_CHAR_OPS = ['+', '-', '*', '/', '%', '.', '=', '&', '|', '^', '<', '>', ','];

    /**
     * Check whether the last non-whitespace token is a trailing operator.
     *
     * Used to determine whether a statement continues onto the next line.
     *
     * @param array $tokens token_get_all tokens
     */
    public static function hasTrailingOperator(array $tokens): bool
    {
        if (empty($tokens)) {
            return false;
        }

        $lastToken = null;
        for ($i = \count($tokens) - 1; $i >= 0; $i--) {
            $token = $tokens[$i];
            if (\is_array($token) && $token[0] === \T_WHITESPACE) {
                continue;
            }

            $lastToken = $token;
            break;
        }

        if ($lastToken === null) {
            return false;
        }

        if (\is_array($lastToken)) {
            return \in_array($lastToken[0], self::TRAILING_TOKEN_OPS, true);
        }

        return \in_array($lastToken, self::TRAILING_CHAR_OPS, true);
    }
}

再重启 composer。

composer dump-autoload -o

再使用 tinker 就可以了。

阅读全文 >>

7月 28

使用 node.js 自动部署

发表于 2026年7月28日 • 248 views • 没有评论!

linux

现在有这样一个需求。本地推送前端代码到 codeberg 后，服务端拉取代码，打包后，放到对应的位置。

服务器服务，全是由 docker 构建，文件复制移动全靠 volume 。

构建 api 接口服务

新建 webhook-server 目录，里边有下边几个文件。

.
├── Dockerfile
├── package.json
└── server.js

Dockerfile：

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(); }

运行 api 接口服务

这里是完整的目录和文件。

.
├── deploy.sh
├── docker-compose.yml
├── setup.sh
└── webhook-server
    ├── Dockerfile
    ├── package.json
    └── server.js

deploy.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
fi

bash.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

nginx 代理

    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 配置 Forgejo

密钥文件和服务器上的文件一致哈。

可以点击测试推送进行测试，也可以主动推送代码进行测试。

阅读全文 >>

7月 28

laravel 框架使用 spatie/laravel-backup 备份数据库和文件

发表于 2026年7月28日 • 180 views • 没有评论!

linux

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]);
}
}

然后可以去完善定时任务相关的逻辑。

阅读全文 >>

7月 23

安装 Woodpecker CI

发表于 2026年7月23日 • 208 views • 没有评论!

linux,工具

jenkins 太占 cpu，内存了，小服务器难搞。准备试试 Woodpecker CI。

官网： https://woodpecker-ci.org/docs/administration/installation/docker-compose

安装

创建 woodpecker 目录，创建 docker-compose.yaml 文件。

mkdir woodpecker

cd woodpecker

vim docker-compose.yaml

填入 docker-compose.yaml 内容。

services:
  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: true

配置 .env 文件

vim .env

# 配置
WOODPECKER_GITHUB_CLIENT=
WOODPECKER_GITHUB_SECRET=
WOODPECKER_AGENT_SECRET=

阅读全文 >>

7月 21

使用 docker 安装 jenkins，并配置 nginx 代理

发表于 2026年7月21日 • 215 views • 没有评论!

linux

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_home

创建 docker-compose.yaml 文件。

services:
  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

修改已存在的 nginx 配置，添加代理，就使用 cicd 作为子路径代理。新增代理配置。

    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/;
    }

重启 nginx。

docker restart xr-nginx

运行并设置用户名和密码

打开网站： https://xxx.xxx.com/cicd

查找密码，输入密码，继续。

# 查看密码
cat ./jenkins home/secrets/initialAdminPassword

安装插件，可以选择安装默认插件。安装过程到最后，如果有失败的可以继续安装，也可以跳过进入到下一步。
配置第一个管理员用户，这个比较重要，要记住账号和密码。
继续直到最后。
安装完成，进入 jenkins 首页。

开始使用

docker compose down

从安装到卸载，jenkins 这玩意，我这破服务器玩不好。

阅读全文 >>

7月 20

2026年7月的一些新闻

发表于 2026年7月20日 • 198 views • 没有评论!

前端

前端圈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

阅读全文 >>

6月 23

lightroom 预设下载

发表于 2026年6月23日 • 249 views • 没有评论!

工具

https://fixthephoto.com/free-lightroom-landscape-presets

阅读全文 >>

6月 15

部署和使用 dify 【7】新增注册接口，获取用户信息等接口

发表于 2026年6月15日 • 264 views • 没有评论!

诗和远方

运行起 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

阅读全文 >>

6月 15

部署和使用 dify 【6】api-token 对应用户改造

发表于 2026年6月15日 • 252 views • 没有评论!

linux,python

虽然 dify 支持创建多个 api-token,也修改了 api_tokens 表，加上了 account_id 进行关联。但是，通过 api-token 转换成 current_user 的时候，这个 user 总是 tenant_account_joins 表里 role 字段是 owner 的用户。既然想给每个用户都有单独的 token，就要 current_user 能对应到每个用户身上。

api-key 到 account 的转换

从 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,
        )

阅读全文 >>

Page 2 of 78
<
1
2
3
4
5
6
7
8
9
10
...
78
>

2026年 10月
一	二	三	四	五	六	日
			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

Write the Code. Change the World.

构建 api 接口服务

运行 api 接口服务

nginx 代理

codeberg 配置 Forgejo

安装

通过定时任务来进行备份

安装

安装

运行并设置用户名和密码

开始使用

api-key 到 account 的转换