认证: 企业微信 OAuth2.0
1.2 系统模块
2. 数据库设计
-- 用户表
CREATE TABLE users (
id SERIAL PRIMARY KEY,
wecom_userid VARCHAR(64) UNIQUE NOT NULL,
name VARCHAR(100) NOT NULL,
avatar_url TEXT,
department JSON,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
last_login_at TIMESTAMP WITH TIME ZONE
);
-- 会话表
CREATE TABLE chat_sessions (
id SERIAL PRIMARY KEY,
user_id INTEGER REFERENCES users(id),
title VARCHAR(200),
model VARCHAR(50),
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
-- 消息表
CREATE TABLE messages (
id SERIAL PRIMARY KEY,
session_id INTEGER REFERENCES chat_sessions(id),
role VARCHAR(20) CHECK (role IN ('user', 'assistant', 'system')),
content TEXT,
created_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP
);
3. 后端实现
3.1 配置文件
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
# 数据库配置
DATABASE_URL: str = "postgresql://user:password@localhost:5432/openwebui"
# Redis配置
REDIS_URL: str = "redis://localhost:6379"
# 企业微信配置
WECOM_CORPID: str
WECOM_CORPSECRET: str
WECOM_AGENT_ID: int
WECOM_REDIRECT_URI: str
# JWT配置
JWT_SECRET_KEY: str
JWT_ALGORITHM: str = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES: int = 60 * 24
class Config:
env_file = ".env"
3.2 认证服务
from fastapi import HTTPException
import httpx
from typing import Dict
class WecomAuthService:
def __init__(self, settings):
self.settings = settings
self.access_token = None
async def get_access_token(self) -> str:
async with httpx.AsyncClient() as client:
response = await client.get(
"https://qyapi.weixin.qq.com/cgi-bin/gettoken",
params={
"corpid": self.settings.WECOM_CORPID,
"corpsecret": self.settings.WECOM_CORPSECRET
}
)
data = response.json()
if data.get("errcode") != 0:
raise HTTPException(status_code=401, detail="获取access_token失败")
return data["access_token"]
async def get_user_info(self, code: str) -> Dict:
token = await self.get_access_token()
async with httpx.AsyncClient() as client:
response = await client.get(
"https://qyapi.weixin.qq.com/cgi-bin/user/getuserinfo",
params={
"access_token": token,
"code": code
}
)
return response.json()
3.3 数据模型
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from .database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
wecom_userid = Column(String, unique=True, index=True)
name = Column(String)
avatar_url = Column(String)
created_at = Column(DateTime(timezone=True), server_default=func.now())
last_login_at = Column(DateTime(timezone=True), onupdate=func.now())
4. 前端实现
4.1 登录组件
企业微信登录
4.2 用户状态管理
import { defineStore } from 'pinia'
export const useAuthStore = defineStore('auth', {
state: () => ({
user: null,
token: null,
}),
actions: {
async login(code: string) {
try {
const response = await fetch('/api/auth/wecom/login', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify({ code }),
})
const data = await response.json()
this.token = data.token
this.user = data.user
localStorage.setItem('token', data.token)
} catch (error) {
console.error('登录失败:', error)
throw error
}
},
},
})
5. 部署说明
5.1 环境要求
- Python 3.8+
- Node.js 16+
- PostgreSQL 13+
- Redis 6+
5.2 安装步骤
- 克隆项目并安装依赖
git clone https://github.com/open-webui/open-webui.git
cd open-webui
# 后端依赖
cd backend
pip install -r requirements.txt
# 前端依赖
cd ../frontend
npm install
- 配置环境变量
DATABASE_URL=postgresql://user:password@localhost:5432/openwebui
REDIS_URL=redis://localhost:6379
WECOM_CORPID=your_corpid
WECOM_CORPSECRET=your_secret
WECOM_AGENT_ID=your_agent_id
JWT_SECRET_KEY=your_secret_key
- 初始化数据库
cd backend
alembic upgrade head
- 启动服务
# 后端服务
uvicorn main:app --host 0.0.0.0 --port 8000
# 前端服务
cd ../frontend
npm run dev
6. 测试计划
6.1 单元测试
import pytest
from services.auth import WecomAuthService
async def test_wecom_auth():
service = WecomAuthService(settings)
token = await service.get_access_token()
assert token is not None
6.2 集成测试检查清单
