Skip to content

About

token共享池

Topics

Resources

Stars

1 star

Watchers

0 watching

Forks

Repository files navigation

Token池共享系统

大模型 API Key 共享平台,将分散的个人 API Key 集中池化,按需分配、自动回收,避免 Key 闲置浪费与手动协调冲突。

核心机制

共享(Share)

用户同意共享协议后,系统自动在 中转站 平台创建名为 token-pool 的 API Key,加密存储至本地数据库。创建后该 Key 在 中转站 上自动禁用,防止无人占用时被意外调用。

抢占(Seize)

已共享 Key 的用户可在池广场抢占他人空闲的 Key。抢占成功后:

  • 本地状态标记为"已占用",每人限占一个
  • 中转站 上该 Key 自动启用(用 Key 拥有者的 access_token 操作),抢占者可正常使用
  • 若抢占时发现 中转站 上该 Key 已不存在(被删除),系统自动重建并更新本地记录

释放(Release)

占用者主动释放 Key 后:

  • 本地状态恢复为"可用",占用者清空
  • 中转站 上该 Key 自动禁用(用 Key 拥有者的 access_token 操作),回归池中等待下次抢占
  • 同样支持自动重建:若 中转站 上 Key 已丢失,释放时自动重建后立即禁用

退出共享(Withdraw)

Key 拥有者退出共享后:

  • 中转站 上该 token-pool 条目自动注销(DELETE)
  • 本地 Key 软删除,若有占用者则自动释放
  • 用户退出后需重新登录,再次同意协议才能重新共享

状态流转图

用户同意协议 → 创建 token-pool Key(中转站 禁用)
    ↓
池广场列表可见(本地可用 / 中转站 禁用)
    ↓
他人抢占 → 本地已占用 / 中转站 启用 → 使用额度
    ↓
占用者释放 → 本地可用 / 中转站 禁用 → 回到池中
    ↓
拥有者退出 → 中转站 注销 / 本地软删除

所有 中转站 操作(启用/禁用/注销/重建)失败时不阻塞主流程,仅记录日志,本地状态正常更新。

技术栈

  • 后端:Django 3.2 + DRF + PostgreSQL
  • 前端:Vue 3 + Element Plus(Django TemplateView 渲染)
  • 部署:Docker + Nginx(HTTPS)+ Gunicorn
  • 认证:LDAP 登录 + JWT
  • 外部依赖:中转站 平台(Key 创建/额度查询/启用禁用)

快速开始

1. 克隆脚手架

git clone <脚手架仓库地址> your-project
cd your-project

2. 初始化项目

使用 Claude Code 的 /project-init 命令,交互式完成项目初始化:

  1. 输入系统英文名(将作为 URL 路径前缀、服务名等)
  2. 输入系统中文名和作用描述

skill 会自动修改 urls.py、CLAUDE.md、README.md、service.json 四个文件。

3. 手动完成初始化

skill 完成后,手动执行:

# 重命名根目录(按系统英文名)
mv demo-backend your-project-name

# 进入项目目录
cd your-project-name

# 删除脚手架的 git 历史,建立新项目仓库
rm -rf .git
git init
git add -A
git commit -m "init: 初始化项目"

4. 安装依赖并启动

cd src
pip install -r requirements.txt
python manage.py runserver

目录结构

token-pool/
├── Dockerfile                    # Docker镜像构建文件
├── docker-compose.yml            # 本地容器编排
├── README.md                     # 项目说明文档
│
└── src/                         # 业务代码根目录
    ├── manage.py                 # Django管理命令入口
    ├── requirements.txt          # Python依赖清单
    ├── right_config.json         # 平台菜单注册配置
    ├── start.sh                  # 生产环境启动脚本(gunicorn)
    │
    ├── config/                     # Django项目配置包
    │   ├── __init__.py
    │   ├── urls.py               # 根URL路由(注册所有ViewSet)
    │   ├── wsgi.py               # WSGI入口(根据env_type选择配置)
    │   └── asgi.py               # ASGI入口
    │
    ├── settings/                 # 多环境配置
    │   ├── conf.ini              # INI格式数据库配置(按环境分段)
    │   ├── settings.py           # 基础Django配置(DB、DRF、中间件、缓存等)
    │   ├── dev.py                # 开发环境(DEBUG=True)
    │   ├── local.py              # 本地环境(DEBUG=True)
    │   ├── prod.py               # 生产环境(DEBUG=False)
    │   ├── test.py               # 测试环境(DEBUG=True)
    │   └── gunicorn.py           # Gunicorn服务配置
    │
    ├── apps/                     # 业务模块目录(新增模块在此创建子目录)
    │   ├── base/                 # 基类模块
    │   │   ├── baseviews.py      # ViewSet基类 + 统一响应格式
    │   │   └── models.py         # 抽象基Model(含时间戳、软删除字段)
    │   └── test/                 # 示例模块(开发参考)
    │       ├── models.py         # 数据模型
    │       ├── serializers.py    # DRF序列化器
    │       ├── views.py          # ViewSet视图
    │       └── tests.py          # 单元测试
    │
    ├── utils/                    # 工具类目录
    │   ├── authentication.py     # JWT认证(JwtAuthentication、AuthedUser)
    │   ├── middleware.py         # 中转站客户端(MiddlewareClient,获取组件信息)
    │   ├── common.py             # 通用工具(密码加密、DRF错误格式化、DB连接清理装饰器)
    │   ├── configure.py          # INI配置解析器(CONF_ATTR)
    │   ├── custom_enum.py        # 枚举类定义(IntegerChoices)
    │   ├── exception.py          # DRF自定义异常处理器
    │   ├── logger.py             # 日志工具(get_logger,按天轮转、统一配置)
    │   ├── middleware.py         # 自定义中间件
    │   ├── pagination.py         # 分页工具(StandardPagination + paginate辅助函数)
    │
    ├── hooks/                    # 平台生命周期钩子
    │   ├── install.py            # 安装钩子(注册菜单、初始化SQL)
    │   ├── uninstall.py          # 卸载钩子(注销菜单)
    │
    ├── jobs/                     # 定时任务目录(预留)
    │   └── __init__.py
    ├── scripts/                  # 运维脚本
    │   ├── init_sql.py           # SQL初始化脚本
    │
    ├── sql/                      # 数据库SQL
    │   ├── init_user_db.sh       # 创建数据库角色与授权
    │   ├── pg_struct.sql         # Schema结构
    │   ├── pg_data.sql           # 初始数据(预留)
    │   ├── patch.sql             # 补丁SQL(预留)
    │
    └── static/                   # 前端dist静态文件目录

核心文件说明

文件 作用 开发时是否需修改
config/urls.py URL路由注册,新增模块必须在此注册ViewSet 必须
apps/base/baseviews.py ViewSet基类与统一响应封装,新视图继承对应基类 仅扩展基类时
apps/base/models.py 抽象Model基类,新Model必须继承 仅扩展字段时
utils/authentication.py JWT认证实现,请求头读取Token字段 一般不修改
utils/pagination.py 分页工具,列表视图必须使用paginate()函数 一般不修改
utils/middleware.py 中转站客户端,用于获取中转站组件信息 一般不修改
utils/logger.py 统一日志工具,按天轮转保留30天,提供get_logger() 一般不修改
utils/exception.py 全局异常处理,返回统一错误响应 一般不修改
utils/configure.py INI配置解析,自动根据env_type选择配置段 一般不修改
settings/settings.py Django核心配置,新App需加入INSTALLED_APPS 添加App时
settings/conf.ini 数据库连接配置,按环境分段 新环境时
right_config.json 平台菜单注册结构 新增菜单时
service.json Chart包服务定义 版本发布时

开发规范

1. 新增业务模块

每个业务模块在 apps/ 下创建独立子目录,必须包含以下四个文件:

apps/your_module/
├── __init__.py          # 包标识(空文件)
├── models.py            # 数据模型,必须继承 AbstractTimeFiledModel
├── serializers.py       # DRF序列化器
├── views.py             # ViewSet,必须继承 baseviews 中的基类

Model 规范

所有 Model 必须继承 AbstractTimeFiledModel,自动获得 create_time、update_time、creator、is_deleted 字段。使用软删除(is_deleted=False),禁止物理删除记录。

from apps.base.models import AbstractTimeFiledModel
from utils import custom_enum

class YourModel(AbstractTimeFiledModel):
    name = models.CharField(max_length=64, verbose_name="名称")
    status = models.SmallIntegerField(choices=custom_enum.YourEnum.choices, verbose_name="状态")

    class Meta:
        db_table = "your_table"

枚举类在 utils/custom_enum.py 中使用 IntegerChoices 定义:

class YourEnum(models.IntegerChoices):
    TYPE_A = 1, "类型A"
    TYPE_B = 2, "类型B"

ViewSet 规范

根据权限需求选择继承基类:

基类 权限 适用场景
AnyLogin 无认证、无权限 公开接口,平台时默认使用此基类,接口权限由平台控制
BaseView / OperatorView 需登录(IsAuthenticated) 常规业务接口
SuperUserView 需管理员(IsAdminUser) 系统管理接口
from apps.base import baseviews
from utils import pagination

class YourView(baseviews.BaseView):
    queryset = models.YourModel.objects.all()
    serializer_class = serializers.YourSerializer
    pagination_class = pagination.StandardPagination

    def list(self, request, **kwargs):
        queryset = self.get_queryset().filter(is_deleted=False).order_by('-id')
        result = pagination.paginate(self, queryset)
        return baseviews.ResponseOK(result)

自定义 Action 使用 DRF 的 @action 装饰器:

from rest_framework.decorators import action

@action(detail=False, methods=['POST'], url_path='custom-action')
def custom_action(self, request, **kwargs):
    # 业务逻辑
    return baseviews.ResponseOK(data)

Serializer 规范

  • 查询/列表场景使用 ModelSerializer
  • 创建/更新场景使用普通 Serializer,在 validate() 中做跨字段校验
  • 校验失败时使用 ToolUtil.format_drf_error() 格式化错误信息
from rest_framework import serializers

class YourSerializer(serializers.ModelSerializer):
    create_time = serializers.DateTimeField(format='%Y-%m-%d %H:%M:%S')

    class Meta:
        model = models.YourModel
        exclude = ['update_time']

class YourCreateSerializer(serializers.Serializer):
    name = serializers.CharField(max_length=64)

    def validate(self, attrs):
        # 跨字段校验逻辑
        return attrs

URL 注册

在 src/config/urls.py 中注册新 ViewSet:

from apps.your_module import views as your_views

router.register(rf'{SYS_NAME}/v1/your_module', your_views.YourView, basename='your_module')

URL 路径规则(SYS_NAME 在 src/config/urls.py 中定义):

  • 简单页面接口:{SYS_NAME}/api/xxx(函数视图)
  • RESTful API:{SYS_NAME}/v1/xxx(ViewSet路由)

2. 统一响应格式

所有 API 必须返回统一 JSON 响应结构:

{
    "code": 2000,
    "message": "success",
    "data": { ... }
}

响应码对照表:

类 code 含义
ResponseOK 2000 成功
ResponseError 5000 服务器错误
ResponseBadRequest 4000 参数错误
ResponseForbidden 4003 无权限
ResponseNotFound 4004 资源不存在
ResponseExpectationFailed 4017 业务条件未满足

使用示例:

return baseviews.ResponseOK({"key": "value"})              # 成功
return baseviews.ResponseOK(pagination.paginate(self, qs))  # 分页成功
return baseviews.ResponseBadRequest("参数不合法")             # 参数错误
return baseviews.ResponseError("操作失败")                   # 服务端错误

3. 分页规范

列表接口必须使用 pagination.paginate() 辅助函数,返回结构包含 total 字段:

def list(self, request, **kwargs):
    queryset = self.get_queryset().filter(is_deleted=False).order_by('-id')
    result = pagination.paginate(self, queryset)
    return baseviews.ResponseOK(result)

分页参数:page(页码)、page_size(每页条数,默认10)。

4. 认证规范

  • JWT Token 通过请求头 Token 字段传递(非标准 Authorization: Bearer)
  • JWT Secret 在 settings/settings.py 的 JWT_SECRET 中配置,生产环境必须设置
  • 解码后用户信息通过 request.user 访问(AuthedUser 对象,含 user_id、username、role)

5. 环境配置规范

通过环境变量 ENV_TYPE 切换运行环境:

ENV_TYPE 配置文件 说明
local(默认) settings/local.py 本地开发
dev settings/dev.py 开发环境
test settings/test.py 测试环境
prod settings/prod.py 生产环境

数据库连接在 settings/conf.ini 中按 [db_{ENV_TYPE}] 分段配置,由 utils/configure.py 自动解析为 CONF_ATTR 对象。

6. 后台任务规范

  • 定时任务放在 jobs/ 目录
  • 异步任务使用 django-q,任务函数需加 @clean_db_connections_decorator 装饰器防止数据库连接丢失
  • 定时调度使用 apscheduler

7. 日志规范

项目中需要日志时,必须使用 utils/logger.py 的 get_logger() 函数,方便统一管理日志配置:

from utils.logger import get_logger

LOGGER = get_logger('your_module.log')

LOGGER.info('操作成功')
LOGGER.warning('注意事项')
LOGGER.error('操作失败')

日志特性:

  • 日志文件存放在 src/logs/ 目录,按天轮转(midnight),保留30天
  • 同时输出到文件和终端(双Handler)
  • 自动过滤 Django 弃用警告(RemovedInDjango*Warning)
  • 模块级预置:logger_app = get_logger('app.log') 可直接导入使用

8. 中转站客户端规范

通过 utils/middleware.py 的 middleware_client 获取中转站上的组件信息:

from utils.middleware import middleware_client

# 获取组件信息(传入组件英文名称)
info = middleware_client.get_service_component_info('b-vuln-backend')

认证方式:自动从 CONF_ATTR 读取 common_sessionid 和 common_csrftoken,构造请求头代理到中转站。

9. SQL规范

  • Schema定义:sql/pg_struct.sql
  • 初始数据:sql/pg_data.sql
  • 补丁脚本:sql/patch.sql(增量变更)
  • 数据库角色创建:sql/init_user_db.sh

AI 开发约束加载流程

使用 Claude Code 进行 AI 辅助开发时,每次对话会自动加载以下约束文件,确保生成的代码符合脚手架规范:

加载顺序

CLAUDE.md(项目根目录)
  ├── 项目概述、常用命令、架构概要
  ├── 脚手架检查清单摘要(表格形式速查)
  └── .ai-harness/rules/ 详细规则引用链接
        ├── scaffold.md    → 脚手架规范(模型继承、ViewSet基类、响应格式、分页、软删除、枚举、路由注册)
        ├── security.md    → 安全约束(凭据管理、输入验证、认证要求、日志安全、密码存储)
        ├── naming.md      → 命名规范(类名、表名、ViewSet、Serializer、URL、枚举)
        ├── architecture.md → 架构约束(模块位置、通用能力、跨模块调用、配置读取、依赖引入)
        └── code-style.md  → 代码风格(格式化、注释、YAGNI、异常处理)

具体流程

  1. CLAUDE.md 自动注入 — Claude Code 启动时自动读取项目根目录的 CLAUDE.md,作为系统级指令注入到对话上下文中
  2. 规则引用生效 — CLAUDE.md 中通过 Markdown 链接引用 .ai-harness/rules/ 下的各规范文件,AI 在需要时按链接读取详细规则
  3. 脚手架检查清单 — CLAUDE.md 以表格形式摘要关键检查项,AI 在每次代码修改前都会对照检查
  4. 违规即阻断 — CLAUDE.md 明确声明:违反 .ai-harness/rules/ 中任何规则即不符合脚手架规范,必须在继续之前修复

Skill 自动加载

除了规则文件,AI 还会通过 Skill 机制加载 .ai-harness/skills/ 下的操作指引:

Skill 触发场景 作用
scaffold 新增模块、修改模型/视图/序列化器 强制执行脚手架规范
module-creator 创建新业务模块 生成符合规范的 models/serializers/views/URL
review 代码变更审查 检查安全、架构、脚手架合规性
project-init 新团队克隆项目后初始化 交互式修改系统名称、描述

如何修改约束

  • 调整全局指令:修改 CLAUDE.md(如新增检查项、调整摘要表格)
  • 调整详细规则:修改 .ai-harness/rules/ 下对应的规范文件
  • 调整 Skill 行为:修改 .ai-harness/skills/ 下对应 Skill 的 SKILL.md

注意:所有修改需提交到 Git,确保团队成员使用 AI 开发时加载一致的约束。

Demo样例说明

apps/test/ 是完整的示例模块,展示脚手架各层协作方式:

Model — MavenEnforcerRules

# apps/test/models.py
class MavenEnforcerRules(models.Model):
    title = models.CharField(max_length=128, verbose_name="规则标题")
    rule_type = models.SmallIntegerField(choices=TaskTypeEnum.choices, verbose_name="规则类型")
    rules = models.JSONField(default=dict, verbose_name="规则详情")
    # ... 含 create_time、is_deleted 等字段

Serializer — 两种类型

  • MavenEnforcerRuleSerializer:ModelSerializer,用于列表查询,格式化时间、枚举label展示
  • CreateBannedDependenciesSerializer:普通Serializer,用于创建参数校验,含 validate() 跨字段校验

ViewSet — TestView

# apps/test/views.py
class TestView(baseviews.AnyLogin):       # 无需认证的公开接口
    queryset = models.MavenEnforcerRules.objects.all()
    serializer_class = serializers.MavenEnforcerRuleSerializer
    pagination_class = pagination.StandardPagination

    def list(self, request, **kwargs):    # 标准列表(分页+软删除过滤)
        queryset = self.get_queryset().filter(is_deleted=False).order_by('-id')
        result = pagination.paginate(self, queryset)
        return baseviews.ResponseOK(result)

    @action(detail=False, methods=['POST'], url_path='banned')  # 自定义Action
    def create_banned_dependency_rule(self, request, **kwargs):
        # Serializer校验 + 统一响应

URL注册

# config/urls.py
router.register(rf'{SYS_NAME}/v1/test', test_views.TestView, basename='test')

生成的API端点(当前 SYS_NAME = 'bbb-vuln'):

  • GET /bbb-vuln/v1/test — 列表
  • POST /bbb-vuln/v1/test — 创建
  • GET /bbb-vuln/v1/test/{id} — 详情
  • POST /bbb-vuln/v1/test/banned — 自定义Action

本地开发

# 1. 安装依赖
cd src
pip install -r requirements.txt

# 2. 默认使用local环境,配置ENV_TYPE切换环境
export ENV_TYPE=local   # 可选:local/dev/test/prod

# 3. 启动开发服务器
python manage.py runserver

依赖清单(核心)

包 版本 用途
Django 3.2.25 Web框架
djangorestframework 3.15.1 REST API框架
django-filter 21.1 查询过滤
django-redis-cache 2.1.0 Redis缓存
django-cors-headers 3.4.0 CORS跨域
requests 2.24.0 HTTP客户端
redis 3.5.3 Redis客户端
PyJWT 1.7.1 JWT认证
psycopg2-binary 2.9.9 PostgreSQL驱动
gunicorn 20.0.4 WSGI生产服务器
simple-background-task 0.1.0 后台任务
pyyaml 6.0.1 YAML解析

About

token共享池

Topics

Resources

Stars

1 star

Watchers

0 watching

Forks

Releases

Packages

Contributors

Languages