init combine

This commit is contained in:
古月 2025-03-03 22:28:34 +08:00
parent 14236e1e2a
commit 6f2a39de3b
31 changed files with 3120 additions and 0 deletions

0
app/__init__.py Normal file
View File

13
app/api/api.py Normal file
View File

@ -0,0 +1,13 @@
from fastapi import APIRouter
from app.api.endpoints import types, configs, users, auth, pages
api_router = APIRouter()
# 添加页面路由,不带前缀
api_router.include_router(pages.router, tags=["pages"])
# API路由不带前缀因为前缀在main.py中添加
api_router.include_router(auth.router, prefix="/auth", tags=["auth"])
api_router.include_router(types.router, prefix="/types", tags=["types"])
api_router.include_router(configs.router, prefix="/configs", tags=["configs"])
api_router.include_router(users.router, prefix="/users", tags=["users"])

129
app/api/deps.py Normal file
View File

@ -0,0 +1,129 @@
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from jose import JWTError, jwt
from typing import Optional
from datetime import datetime, timedelta
from app.core.config import settings
from app.models.database import get_db
from app.models.user import User
from app.models.config import Config
# 设置 auto_error=False 使其在没有令牌时不抛出异常
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token", auto_error=False)
# 修改 is_privilege_mode 函数
from app.core.config import settings
async def is_privilege_mode(db: AsyncSession) -> bool:
"""检查是否启用了特权模式"""
# 首先检查环境变量中的特权模式设置
if settings.PRIVILEGE_MODE:
return True
# 如果环境变量中没有启用特权模式,则检查数据库中的配置
try:
result = await db.execute(
select(Config).where(
Config.key == "privilege_mode"
)
)
config = result.scalar_one_or_none()
return config is not None and config.value.lower() == "true"
except Exception as e:
print(f"检查特权模式出错: {e}")
return False
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
"""创建访问令牌"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")
return encoded_jwt
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
"""获取当前用户(必须已登录)"""
# 检查特权模式
if await is_privilege_mode(db):
# 特权模式下,返回管理员用户
result = await db.execute(select(User).where(User.role == "admin").limit(1))
admin_user = result.scalar_one_or_none()
if admin_user:
return admin_user
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无法验证凭据",
headers={"WWW-Authenticate": "Bearer"},
)
if not token:
raise credentials_exception
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
result = await db.execute(select(User).where(User.username == username))
user = result.scalar_one_or_none()
if user is None:
raise credentials_exception
return user
async def get_current_user_optional(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> Optional[User]:
"""获取当前用户可选未登录返回None"""
# 检查特权模式
if await is_privilege_mode(db):
# 特权模式下,返回管理员用户
result = await db.execute(select(User).where(User.role == "admin").limit(1))
admin_user = result.scalar_one_or_none()
if admin_user:
return admin_user
if not token:
return None
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
username: str = payload.get("sub")
if username is None:
return None
except JWTError:
return None
result = await db.execute(select(User).where(User.username == username))
user = result.scalar_one_or_none()
return user
async def get_current_active_user(
current_user: User = Depends(get_current_user),
) -> User:
"""获取当前活跃用户"""
return current_user
async def get_current_admin_user(
current_user: User = Depends(get_current_user),
) -> User:
"""获取当前管理员用户"""
if current_user.role != "admin":
raise HTTPException(status_code=403, detail="权限不足")
return current_user

19
app/api/deps/__init__.py Normal file
View File

@ -0,0 +1,19 @@
# 从模块中导出所需的函数和类
from app.api.deps.auth import (
create_access_token,
get_current_user,
get_current_user_optional,
get_current_active_user,
get_current_admin_user,
is_privilege_mode
)
# 确保这些函数可以直接从 app.api.deps 导入
__all__ = [
"create_access_token",
"get_current_user",
"get_current_user_optional",
"get_current_active_user",
"get_current_admin_user",
"is_privilege_mode"
]

142
app/api/deps/auth.py Normal file
View File

@ -0,0 +1,142 @@
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from jose import JWTError, jwt
from typing import Optional
from datetime import datetime, timedelta
from app.core.config import settings
from app.models.database import get_db
from app.models.user import User
from app.models.config import Config
# 设置 auto_error=False 使其在没有令牌时不抛出异常
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/token", auto_error=False)
# 修改 is_privilege_mode 函数
from app.core.config import settings
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from app.models.config import Config
import logging
logger = logging.getLogger(__name__)
async def is_privilege_mode(db: AsyncSession = None) -> bool:
"""检查是否启用了特权模式"""
# 首先检查环境变量中的特权模式设置
logger.info(f"检查特权模式,环境变量设置为: {settings.PRIVILEGE_MODE}")
if settings.PRIVILEGE_MODE:
logger.info("特权模式已启用(通过环境变量)")
return True
# 如果环境变量中没有启用特权模式,则检查数据库中的配置
if db:
try:
result = await db.execute(
select(Config).where(
Config.key == "privilege_mode"
)
)
config = result.scalar_one_or_none()
is_enabled = config is not None and config.value.lower() == "true"
logger.info(f"特权模式数据库配置: {is_enabled}")
return is_enabled
except Exception as e:
logger.error(f"检查特权模式出错: {e}")
return False
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
"""创建访问令牌"""
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm="HS256")
return encoded_jwt
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> User:
"""获取当前用户(必须已登录)"""
# 检查特权模式
if await is_privilege_mode(db):
# 特权模式下,返回管理员用户
result = await db.execute(select(User).where(User.role == "admin").limit(1))
admin_user = result.scalar_one_or_none()
if admin_user:
return admin_user
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="无法验证凭据",
headers={"WWW-Authenticate": "Bearer"},
)
if not token:
raise credentials_exception
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
except JWTError:
raise credentials_exception
result = await db.execute(select(User).where(User.username == username))
user = result.scalar_one_or_none()
if user is None:
raise credentials_exception
return user
async def get_current_user_optional(
token: str = Depends(oauth2_scheme),
db: AsyncSession = Depends(get_db)
) -> Optional[User]:
"""获取当前用户可选未登录返回None"""
# 检查特权模式
if await is_privilege_mode(db):
# 特权模式下,返回管理员用户
result = await db.execute(select(User).where(User.role == "admin").limit(1))
admin_user = result.scalar_one_or_none()
if admin_user:
return admin_user
if not token:
return None
try:
payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
username: str = payload.get("sub")
if username is None:
return None
except JWTError:
return None
result = await db.execute(select(User).where(User.username == username))
user = result.scalar_one_or_none()
return user
async def get_current_active_user(
current_user: User = Depends(get_current_user),
) -> User:
"""获取当前活跃用户"""
return current_user
async def get_current_admin_user(
current_user: User = Depends(get_current_user),
) -> User:
"""获取当前管理员用户"""
if current_user.role != "admin":
raise HTTPException(status_code=403, detail="权限不足")
return current_user

190
app/api/endpoints/auth.py Normal file
View File

@ -0,0 +1,190 @@
from fastapi import APIRouter, Depends, HTTPException, status, Form, Request
from fastapi.security import OAuth2PasswordRequestForm
from fastapi.responses import HTMLResponse, RedirectResponse
from fastapi.templating import Jinja2Templates
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
import bcrypt
from datetime import timedelta
from pathlib import Path
from app.models.database import get_db
from app.models.user import User
from app.api.deps import create_access_token, is_privilege_mode
router = APIRouter()
# 设置模板目录
templates = Jinja2Templates(directory=str(Path(__file__).parent.parent.parent / "templates"))
@router.post("/token")
async def login_for_access_token(
form_data: OAuth2PasswordRequestForm = Depends(),
db: AsyncSession = Depends(get_db)
):
"""
获取访问令牌
"""
# 打印接收到的用户名,用于调试
print(f"接收到的用户名: {form_data.username}")
# 检查特权模式
if await is_privilege_mode(db):
# 特权模式下,返回管理员用户的令牌
result = await db.execute(select(User).where(User.role == "admin").limit(1))
admin_user = result.scalar_one_or_none()
if admin_user:
# 创建访问令牌
access_token_expires = timedelta(minutes=60 * 24) # 24小时
access_token = create_access_token(
data={"sub": admin_user.username}, expires_delta=access_token_expires
)
return {
"access_token": access_token,
"token_type": "bearer",
"username": admin_user.username,
"role": admin_user.role
}
# 查询用户
result = await db.execute(select(User).where(User.username == form_data.username))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或密码错误",
headers={"WWW-Authenticate": "Bearer"},
)
# 验证密码
is_password_correct = bcrypt.checkpw(
form_data.password.encode(),
user.hashed_password.encode()
)
if not is_password_correct:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或密码错误",
headers={"WWW-Authenticate": "Bearer"},
)
# 创建访问令牌
access_token_expires = timedelta(minutes=60 * 24) # 24小时
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {
"access_token": access_token,
"token_type": "bearer",
"username": user.username,
"role": user.role
}
# 添加一个直接接收表单数据的端点以防OAuth2PasswordRequestForm不工作
@router.post("/login")
async def login_direct(
username: str = Form(...),
password: str = Form(...),
db: AsyncSession = Depends(get_db)
):
"""
直接接收表单数据的登录端点
"""
# 查询用户
result = await db.execute(select(User).where(User.username == username))
user = result.scalar_one_or_none()
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或密码错误"
)
# 验证密码
is_password_correct = bcrypt.checkpw(
password.encode(),
user.hashed_password.encode()
)
if not is_password_correct:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="用户名或密码错误"
)
# 创建访问令牌
access_token_expires = timedelta(minutes=60 * 24) # 24小时
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {
"access_token": access_token,
"token_type": "bearer",
"username": user.username,
"role": user.role
}
# 添加一个简单的登录处理函数
@router.post("/form-login", response_class=HTMLResponse)
async def login_form(
request: Request,
username: str = Form(...),
password: str = Form(...),
db: AsyncSession = Depends(get_db)
):
"""处理表单登录"""
try:
# 查询用户
result = await db.execute(select(User).where(User.username == username))
user = result.scalar_one_or_none()
if not user:
return templates.TemplateResponse(
"login.html",
{"request": request, "error": "用户名或密码错误"}
)
# 验证密码
is_password_correct = bcrypt.checkpw(
password.encode(),
user.hashed_password.encode()
)
if not is_password_correct:
return templates.TemplateResponse(
"login.html",
{"request": request, "error": "用户名或密码错误"}
)
# 创建访问令牌
access_token_expires = timedelta(minutes=60 * 24) # 24小时
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
# 设置cookie
response = RedirectResponse(url="/", status_code=303)
response.set_cookie(
key="access_token",
value=f"Bearer {access_token}",
httponly=True,
max_age=60 * 60 * 24, # 24小时
samesite="lax"
)
return response
except Exception as e:
return templates.TemplateResponse(
"login.html",
{"request": request, "error": f"登录失败: {str(e)}"}
)
# 添加检查特权模式的端点
@router.get("/privilege-mode")
async def check_privilege_mode(db: AsyncSession = Depends(get_db)):
"""检查是否启用了特权模式"""
privilege_mode = await is_privilege_mode(db)
return {"privilege_mode": privilege_mode}

View File

@ -0,0 +1,419 @@
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func, and_, or_
from typing import List, Optional
from app.models.database import get_db
from app.models.config import Config
from app.models.type import Type
from app.schemas.config import ConfigCreate, ConfigUpdate, Config as ConfigSchema, ConfigList, ConfigSearch
from app.api.deps import get_current_active_user, get_current_admin_user, is_privilege_mode
from app.models.user import User
# 添加缺少的导入
from app import schemas
router = APIRouter()
@router.post("", response_model=ConfigSchema)
async def create_config(
config_data: ConfigCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_active_user)
):
"""
创建新的配置项
"""
# 查找或创建类型
result = await db.execute(select(Type).where(Type.type_name == config_data.type_name))
db_type = result.scalars().first()
if not db_type:
# 如果类型不存在,创建新类型
db_type = Type(type_name=config_data.type_name, description=f"自动创建的类型: {config_data.type_name}")
db.add(db_type)
await db.flush() # 获取新创建类型的ID
# 检查同一类型下是否已存在相同key的配置
result = await db.execute(
select(Config).where(
and_(
Config.type_id == db_type.type_id,
Config.key == config_data.key
)
)
)
if result.scalars().first():
raise HTTPException(status_code=400, detail=f"类型 '{config_data.type_name}' 下已存在键 '{config_data.key}'")
# 创建新配置
db_config = Config(
type_id=db_type.type_id,
key=config_data.key,
value=config_data.value,
key_description=config_data.key_description
)
db.add(db_config)
await db.commit()
await db.refresh(db_config)
# 构建返回结果
result = ConfigSchema(
config_id=db_config.config_id,
type_id=db_config.type_id,
key=db_config.key,
value=db_config.value,
key_description=db_config.key_description,
created_at=db_config.created_at,
updated_at=db_config.updated_at,
type_name=db_type.type_name
)
return result
@router.get("", response_model=ConfigList)
async def get_configs(
skip: int = 0,
limit: int = 100,
type_name: Optional[str] = None,
key: Optional[str] = None,
value: Optional[str] = None,
exact_match: bool = False,
db: AsyncSession = Depends(get_db)
):
"""
获取配置列表支持按类型值筛选
"""
# 构建查询条件
conditions = []
if type_name:
# 按类型名称筛选
result = await db.execute(select(Type).where(Type.type_name == type_name))
db_type = result.scalars().first()
if not db_type:
# 如果类型不存在,返回空列表
return {"configs": [], "total": 0}
conditions.append(Config.type_id == db_type.type_id)
if key:
# 按键筛选
if exact_match:
conditions.append(Config.key == key)
else:
conditions.append(Config.key.like(f"%{key}%"))
if value:
# 按值筛选
if exact_match:
conditions.append(Config.value == value)
else:
conditions.append(Config.value.like(f"%{value}%"))
# 构建查询
query = select(Config, Type.type_name).join(Type)
if conditions:
query = query.where(and_(*conditions))
# 查询总数
count_query = select(func.count()).select_from(Config).join(Type)
if conditions:
count_query = count_query.where(and_(*conditions))
result = await db.execute(count_query)
total = result.scalar()
# 查询配置列表
query = query.offset(skip).limit(limit)
result = await db.execute(query)
rows = result.all()
# 构建返回结果
configs = []
for row in rows:
config, type_name = row
configs.append(
ConfigSchema(
config_id=config.config_id,
type_id=config.type_id,
key=config.key,
value=config.value,
key_description=config.key_description,
created_at=config.created_at,
updated_at=config.updated_at,
type_name=type_name
)
)
return {"configs": configs, "total": total}
@router.get("/{config_id}", response_model=ConfigSchema)
async def get_config(
config_id: int,
db: AsyncSession = Depends(get_db)
):
"""
获取指定配置项的详细信息
"""
result = await db.execute(
select(Config, Type.type_name)
.join(Type)
.where(Config.config_id == config_id)
)
row = result.first()
if not row:
raise HTTPException(status_code=404, detail=f"配置ID {config_id} 不存在")
config, type_name = row
return ConfigSchema(
config_id=config.config_id,
type_id=config.type_id,
key=config.key,
value=config.value,
key_description=config.key_description,
created_at=config.created_at,
updated_at=config.updated_at,
type_name=type_name
)
@router.put("/{config_id}", response_model=ConfigSchema)
async def update_config(
config_id: int,
config_data: ConfigUpdate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_active_user)
):
"""
更新配置项
"""
result = await db.execute(
select(Config, Type.type_name)
.join(Type)
.where(Config.config_id == config_id)
)
row = result.first()
if not row:
raise HTTPException(status_code=404, detail=f"配置ID {config_id} 不存在")
config, type_name = row
# 更新配置
if config_data.value is not None:
config.value = config_data.value
if config_data.key_description is not None:
config.key_description = config_data.key_description
await db.commit()
await db.refresh(config)
return ConfigSchema(
config_id=config.config_id,
type_id=config.type_id,
key=config.key,
value=config.value,
key_description=config.key_description,
created_at=config.created_at,
updated_at=config.updated_at,
type_name=type_name
)
@router.delete("/{config_id}", response_model=dict)
async def delete_config(
config_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
"""
删除配置项仅管理员
"""
result = await db.execute(select(Config).where(Config.config_id == config_id))
config = result.scalars().first()
if not config:
raise HTTPException(status_code=404, detail=f"配置ID {config_id} 不存在")
await db.delete(config)
await db.commit()
return {"message": f"配置ID {config_id} 已成功删除"}
@router.post("/search", response_model=ConfigList)
async def search_configs(
search_data: ConfigSearch,
skip: int = 0,
limit: int = 100,
db: AsyncSession = Depends(get_db)
):
"""
高级搜索配置项
"""
# 构建查询条件
conditions = []
if search_data.type_name:
# 按类型名称筛选
result = await db.execute(select(Type).where(Type.type_name == search_data.type_name))
db_type = result.scalars().first()
if not db_type:
# 如果类型不存在,返回空列表
return {"configs": [], "total": 0}
conditions.append(Config.type_id == db_type.type_id)
if search_data.key:
# 按键筛选
if search_data.exact_match:
conditions.append(Config.key == search_data.key)
else:
conditions.append(Config.key.like(f"%{search_data.key}%"))
if search_data.value:
# 按值筛选
if search_data.exact_match:
conditions.append(Config.value == search_data.value)
else:
conditions.append(Config.value.like(f"%{search_data.value}%"))
# 构建查询
query = select(Config, Type.type_name).join(Type)
if conditions:
query = query.where(and_(*conditions))
# 查询总数
count_query = select(func.count()).select_from(Config).join(Type)
if conditions:
count_query = count_query.where(and_(*conditions))
result = await db.execute(count_query)
total = result.scalar()
# 查询配置列表
query = query.offset(skip).limit(limit)
result = await db.execute(query)
rows = result.all()
# 构建返回结果
configs = []
for row in rows:
config, type_name = row
configs.append(
ConfigSchema(
config_id=config.config_id,
type_id=config.type_id,
key=config.key,
value=config.value,
key_description=config.key_description,
created_at=config.created_at,
updated_at=config.updated_at,
type_name=type_name
)
)
return {"configs": configs, "total": total}
@router.get("/type/{type_name}/key/{key}", response_model=ConfigSchema)
async def get_config_by_type_and_key(
type_name: str,
key: str,
db: AsyncSession = Depends(get_db)
):
"""
通过类型名称和键获取配置
"""
# 查找类型
result = await db.execute(select(Type).where(Type.type_name == type_name))
db_type = result.scalars().first()
if not db_type:
raise HTTPException(status_code=404, detail=f"类型 '{type_name}' 不存在")
# 查找配置
result = await db.execute(
select(Config)
.where(
and_(
Config.type_id == db_type.type_id,
Config.key == key
)
)
)
config = result.scalars().first()
if not config:
raise HTTPException(status_code=404, detail=f"类型 '{type_name}' 下不存在键 '{key}'")
return ConfigSchema(
config_id=config.config_id,
type_id=config.type_id,
key=config.key,
value=config.value,
key_description=config.key_description,
created_at=config.created_at,
updated_at=config.updated_at,
type_name=type_name
)
@router.delete("/type/{type_name}/key/{key}", response_model=dict)
async def delete_config_by_type_and_key(
type_name: str,
key: str,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
"""
通过类型名称和键删除配置仅管理员
"""
# 查找类型
result = await db.execute(select(Type).where(Type.type_name == type_name))
db_type = result.scalars().first()
if not db_type:
raise HTTPException(status_code=404, detail=f"类型 '{type_name}' 不存在")
# 查找配置
result = await db.execute(
select(Config)
.where(
and_(
Config.type_id == db_type.type_id,
Config.key == key
)
)
)
config = result.scalars().first()
if not config:
raise HTTPException(status_code=404, detail=f"类型 '{type_name}' 下不存在键 '{key}'")
# 修改删除配置项的端点s
@router.delete("/{type_name}/{key}", response_model=dict)
async def delete_config(
type_name: str,
key: str,
db: AsyncSession = Depends(get_db),
# current_user: User = Depends(get_current_user_optional)
current_user: User = Depends(get_current_admin_user)
):
"""
删除配置项
"""
# 检查特权模式
if not current_user and not await is_privilege_mode(db):
raise HTTPException(status_code=401, detail="需要认证")
# 查询配置项
result = await db.execute(
select(Config).join(Type).where(
Type.type_name == type_name,
Config.key == key
)
)
config = result.scalar_one_or_none()
if not config:
raise HTTPException(status_code=404, detail=f"找不到配置项: {type_name}.{key}")
# 删除配置项
await db.delete(config)
await db.commit()
# return config1212
return {"message": "配置项已删除"}

128
app/api/endpoints/pages.py Normal file
View File

@ -0,0 +1,128 @@
from fastapi import APIRouter, Request, Depends
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
from pathlib import Path
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from app.models.database import get_db
from app.api.deps import get_current_user_optional # 修改这里,导入正确的函数
from app.models.user import User
from app.models.type import Type
from app.models.config import Config
from typing import Optional
router = APIRouter()
templates = Jinja2Templates(directory=str(Path(__file__).parents[2] / "templates"))
# 修改根路径处理函数
@router.get("/", response_class=HTMLResponse)
async def index(request: Request, db: AsyncSession = Depends(get_db)):
"""首页"""
# 查询类型数量
result = await db.execute(select(func.count()).select_from(Type))
type_count = result.scalar()
# 查询配置项数量
result = await db.execute(select(func.count()).select_from(Config))
config_count = result.scalar()
return templates.TemplateResponse(
"index.html",
{
"request": request,
"type_count": type_count,
"config_count": config_count
}
)
@router.get("/types", response_class=HTMLResponse)
async def types_page(
request: Request,
search: Optional[str] = None,
db: AsyncSession = Depends(get_db),
current_user: Optional[User] = Depends(get_current_user_optional)
):
"""配置类型页面"""
# 构建查询
query = select(Type)
if search:
query = query.where(Type.type_name.contains(search) | Type.description.contains(search))
# 执行查询
result = await db.execute(query)
types = result.scalars().all()
return templates.TemplateResponse(
"types.html",
{
"request": request,
"current_user": current_user,
"types": types,
"search": search
}
)
@router.get("/configs", response_class=HTMLResponse)
async def configs_page(
request: Request,
type_name: Optional[str] = None,
search: Optional[str] = None,
db: AsyncSession = Depends(get_db),
current_user: Optional[User] = Depends(get_current_user_optional)
):
"""配置项页面"""
# 获取所有类型
types_result = await db.execute(select(Type))
types = types_result.scalars().all()
# 构建查询
query = select(Config)
if type_name:
# 获取类型ID
type_result = await db.execute(select(Type).where(Type.type_name == type_name))
selected_type = type_result.scalar_one_or_none()
if selected_type:
query = query.where(Config.type_id == selected_type.type_id)
if search:
query = query.where(Config.key.contains(search) | Config.value.contains(search) | Config.key_description.contains(search))
# 执行查询
result = await db.execute(query)
configs = result.scalars().all()
# 获取每个配置的类型信息
for config in configs:
if not hasattr(config, 'type') or config.type is None:
type_result = await db.execute(select(Type).where(Type.type_id == config.type_id))
config.type = type_result.scalar_one_or_none()
return templates.TemplateResponse(
"configs.html",
{
"request": request,
"current_user": current_user,
"configs": configs,
"types": types,
"selected_type": selected_type if type_name else None,
"search": search
}
)
@router.get("/login", response_class=HTMLResponse)
async def login_page(
request: Request,
current_user: Optional[User] = Depends(get_current_user_optional)
):
"""登录页面"""
if current_user:
# 如果用户已登录,重定向到首页
from fastapi.responses import RedirectResponse
return RedirectResponse(url="/")
return templates.TemplateResponse(
"login.html",
{
"request": request
}
)

146
app/api/endpoints/types.py Normal file
View File

@ -0,0 +1,146 @@
from fastapi import APIRouter, Depends, HTTPException, Query
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from typing import List, Optional
from app.models.database import get_db
from app.models.type import Type
from app.models.config import Config
from app.schemas.type import TypeCreate, TypeUpdate, Type as TypeSchema, TypeList
from app.api.deps import get_current_active_user, get_current_admin_user, get_current_user_optional, is_privilege_mode
from app.models.user import User
router = APIRouter()
# 删除重复的路由,只保留一个获取所有类型的路由
@router.get("", response_model=TypeList)
async def get_types(
skip: int = 0,
limit: int = 100,
search: Optional[str] = None,
db: AsyncSession = Depends(get_db)
):
"""
获取所有配置类型
"""
# 构建查询条件
query = select(Type)
if search:
query = query.where(Type.type_name.contains(search) | Type.description.contains(search))
# 查询总数
count_query = select(func.count()).select_from(query.subquery())
result = await db.execute(count_query)
total = result.scalar()
# 查询类型列表
result = await db.execute(query.offset(skip).limit(limit))
types = result.scalars().all()
return {"types": types, "total": total}
@router.post("", response_model=TypeSchema)
async def create_type(
type_data: TypeCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user_optional)
):
"""
创建新的配置类型
"""
# 检查特权模式
if not current_user and not await is_privilege_mode(db):
raise HTTPException(status_code=401, detail="需要认证")
# 检查类型名是否已存在
result = await db.execute(select(Type).where(Type.type_name == type_data.type_name))
if result.scalars().first():
raise HTTPException(status_code=400, detail="类型名已存在")
# 创建新类型
db_type = Type(**type_data.dict())
db.add(db_type)
await db.commit()
await db.refresh(db_type)
return db_type
@router.get("/{type_name}", response_model=TypeSchema)
async def get_type(
type_name: str,
db: AsyncSession = Depends(get_db)
):
"""
获取指定类型的详细信息
"""
result = await db.execute(select(Type).where(Type.type_name == type_name))
db_type = result.scalars().first()
if not db_type:
raise HTTPException(status_code=404, detail=f"类型 '{type_name}' 不存在")
return db_type
@router.put("/{type_name}", response_model=TypeSchema)
async def update_type(
type_name: str,
type_data: TypeUpdate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user_optional)
):
"""
更新配置类型
"""
# 检查特权模式
if not current_user and not await is_privilege_mode(db):
raise HTTPException(status_code=401, detail="需要认证")
# 查找类型
result = await db.execute(select(Type).where(Type.type_name == type_name))
db_type = result.scalars().first()
if not db_type:
raise HTTPException(status_code=404, detail=f"类型 '{type_name}' 不存在")
# 更新类型
if type_data.description is not None:
db_type.description = type_data.description
await db.commit()
await db.refresh(db_type)
return db_type
@router.delete("/{type_name}", response_model=TypeSchema)
async def delete_type(
type_name: str,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user_optional)
):
"""
删除配置类型
"""
# 检查特权模式
if not current_user and not await is_privilege_mode(db):
raise HTTPException(status_code=401, detail="需要认证")
# 查询类型
result = await db.execute(
select(Type).where(Type.type_name == type_name)
)
type_obj = result.scalar_one_or_none()
if not type_obj:
raise HTTPException(status_code=404, detail=f"找不到类型: {type_name}")
# 检查是否有关联的配置项
configs_result = await db.execute(
select(Config).where(Config.type_id == type_obj.type_id)
)
configs = configs_result.scalars().all()
if configs:
raise HTTPException(status_code=400, detail=f"类型 {type_name} 下有 {len(configs)} 个配置项,请先删除这些配置项")
# 删除类型
await db.delete(type_obj)
await db.commit()
return type_obj

131
app/api/endpoints/users.py Normal file
View File

@ -0,0 +1,131 @@
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from typing import List, Optional
import bcrypt
from app.models.database import get_db
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate, User as UserSchema, UserList
from app.api.deps import get_current_active_user, get_current_admin_user
router = APIRouter()
@router.get("", response_model=UserList)
async def get_users(
skip: int = 0,
limit: int = 100,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
"""
获取所有用户仅管理员
"""
# 查询总数
result = await db.execute(select(func.count()).select_from(User))
total = result.scalar()
# 查询用户列表
result = await db.execute(select(User).offset(skip).limit(limit))
users = result.scalars().all()
return {"users": users, "total": total}
@router.post("", response_model=UserSchema)
async def create_user(
user_data: UserCreate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
"""
创建新用户仅管理员
"""
# 检查用户名是否已存在
result = await db.execute(select(User).where(User.username == user_data.username))
if result.scalars().first():
raise HTTPException(status_code=400, detail="用户名已存在")
# 哈希密码
hashed_password = bcrypt.hashpw(user_data.password.encode(), bcrypt.gensalt()).decode()
# 创建新用户
db_user = User(
username=user_data.username,
password=hashed_password,
role=user_data.role
)
db.add(db_user)
await db.commit()
await db.refresh(db_user)
return db_user
@router.get("/me", response_model=UserSchema)
async def read_users_me(
current_user: User = Depends(get_current_active_user)
):
"""
获取当前登录用户信息
"""
return current_user
@router.get("/{user_id}", response_model=UserSchema)
async def get_user(
user_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
"""
获取指定用户信息仅管理员
"""
result = await db.execute(select(User).where(User.user_id == user_id))
user = result.scalars().first()
if not user:
raise HTTPException(status_code=404, detail=f"用户ID {user_id} 不存在")
return user
@router.put("/{user_id}", response_model=UserSchema)
async def update_user(
user_id: int,
user_data: UserUpdate,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
"""
更新用户信息仅管理员
"""
result = await db.execute(select(User).where(User.user_id == user_id))
user = result.scalars().first()
if not user:
raise HTTPException(status_code=404, detail=f"用户ID {user_id} 不存在")
# 更新用户信息
if user_data.password:
user.password = bcrypt.hashpw(user_data.password.encode(), bcrypt.gensalt()).decode()
if user_data.role:
user.role = user_data.role
await db.commit()
await db.refresh(user)
return user
@router.delete("/{user_id}", response_model=dict)
async def delete_user(
user_id: int,
db: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_admin_user)
):
"""
删除用户仅管理员
"""
# 不允许删除自己
if user_id == current_user.user_id:
raise HTTPException(status_code=400, detail="不能删除当前登录的用户")
result = await db.execute(select(User).where(User.user_id == user_id))
user = result.scalars().first()
if not user:
raise HTTPException(status_code=404, detail=f"用户ID {user_id} 不存在")
await db.delete(user)
await db.commit()
return {"message": f"用户ID {user_id} 已成功删除"}

99
app/api/pages.py Normal file
View File

@ -0,0 +1,99 @@
from fastapi import APIRouter, Request, Depends
from fastapi.templating import Jinja2Templates
from pathlib import Path
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, func
from app.models.database import get_db
from app.models.type import Type
from app.models.config import Config
from app.api.deps import get_current_user_optional
# 将 router 改名为 page_router
page_router = APIRouter()
# 设置模板目录
templates = Jinja2Templates(directory=Path(__file__).parent.parent / "templates")
@page_router.get("/")
async def index_page(request: Request, current_user = Depends(get_current_user_optional), db: AsyncSession = Depends(get_db)):
"""首页"""
# 查询类型数量
result = await db.execute(select(func.count()).select_from(Type))
types_count = result.scalar()
# 查询配置项数量
result = await db.execute(select(func.count()).select_from(Config))
configs_count = result.scalar()
return templates.TemplateResponse(
"index.html",
{"request": request, "current_user": current_user,
"types_count": types_count,
"configs_count": configs_count
}
)
@page_router.get("/login")
async def login_page(request: Request):
"""登录页面"""
return templates.TemplateResponse("login.html", {"request": request})
@page_router.get("/types")
async def types_page(
request: Request,
search: str = None,
db: AsyncSession = Depends(get_db),
current_user = Depends(get_current_user_optional)
):
"""配置类型页面"""
query = select(Type)
if search:
query = query.where(Type.type_name.contains(search) | Type.description.contains(search))
result = await db.execute(query)
types = result.scalars().all()
return templates.TemplateResponse(
"types.html",
{
"request": request,
"types": types,
"current_user": current_user
}
)
@page_router.get("/configs")
async def configs_page(
request: Request,
type_name: str = None,
search: str = None,
db: AsyncSession = Depends(get_db),
current_user = Depends(get_current_user_optional)
):
"""配置项页面"""
# 获取所有类型
types_result = await db.execute(select(Type))
types = types_result.scalars().all()
# 构建查询
query = select(Config).join(Type)
if type_name:
query = query.where(Type.type_name == type_name)
if search:
query = query.where(Config.key.contains(search) | Config.value.contains(search) | Config.key_description.contains(search))
result = await db.execute(query)
configs = result.scalars().all()
return templates.TemplateResponse(
"configs.html",
{
"request": request,
"configs": configs,
"types": types,
"current_type": type_name,
"current_user": current_user
}
)

27
app/core/config.py Normal file
View File

@ -0,0 +1,27 @@
import os
from pydantic_settings import BaseSettings
from dotenv import load_dotenv
# 加载.env文件
load_dotenv()
class Settings(BaseSettings):
"""应用配置"""
# 应用名称
APP_NAME: str = "配置中心"
# 数据库URL
DATABASE_URL: str = os.getenv("DATABASE_URL", "sqlite:///./config_center.db")
# 密钥
SECRET_KEY: str = os.getenv("SECRET_KEY", "your-secret-key-here")
# 特权模式
PRIVILEGE_MODE: bool = os.getenv("PRIVILEGE_MODE", "True").lower() == "true"
class Config:
env_file = ".env"
# 允许额外字段
extra = "ignore"
settings = Settings()

78
app/main.py Normal file
View File

@ -0,0 +1,78 @@
from fastapi import FastAPI, Request # 添加 Request 导入
from fastapi.middleware.cors import CORSMiddleware
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from fastapi.responses import RedirectResponse
import uvicorn
import logging
from pathlib import Path
from app.api.api import api_router
from app.api.pages import page_router
from app.core.config import settings
from app.models.database import init_db
# 配置日志
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
handlers=[
logging.StreamHandler(),
logging.FileHandler(filename="app.log", encoding="utf-8")
]
)
logger = logging.getLogger(__name__)
# 设置模板目录
templates = Jinja2Templates(directory=str(Path(__file__).parent / "templates"))
print("*"*100)
print("templates:", templates)
print(str(Path(__file__).parent / "templates"))
app = FastAPI(
title=settings.APP_NAME,
description="配置中心API",
version="1.0.0",
docs_url="/api/docs",
redoc_url="/api/redoc",
openapi_url="/api/openapi.json"
)
# 设置CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # 在生产环境中应该设置为特定域名
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# 挂载静态文件
app.mount("/static", StaticFiles(directory=str(Path(__file__).parent / "static")), name="static")
# 设置模板目录
templates = Jinja2Templates(directory=Path(__file__).parent / "templates")
# 注册API路由
app.include_router(api_router, prefix="/api")
# 注册页面路由
app.include_router(page_router, prefix="/page")
# 根路径重定向到首页
@app.get("/")
async def root():
# 将根路径重定向到页面路由的首页
return RedirectResponse(url="/page/")
# 初始化数据库
@app.on_event("startup")
async def startup_event():
await init_db()
logger.info("数据库初始化完成")
@app.on_event("shutdown")
async def shutdown_event():
logger.info("应用关闭中...")
if __name__ == "__main__":
uvicorn.run("app.main:app", host="0.0.0.0", port=8000, reload=True)

3
app/models/base.py Normal file
View File

@ -0,0 +1,3 @@
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()

25
app/models/config.py Normal file
View File

@ -0,0 +1,25 @@
from sqlalchemy import Column, Integer, String, ForeignKey, DateTime, func, UniqueConstraint
from sqlalchemy.orm import relationship
from app.models.database import Base
class Config(Base):
__tablename__ = "configs"
config_id = Column(Integer, primary_key=True, index=True, autoincrement=True)
type_id = Column(Integer, ForeignKey("types.type_id"), nullable=False)
key = Column(String, index=True, nullable=False)
value = Column(String, nullable=False)
key_description = Column(String, nullable=True)
created_at = Column(DateTime, default=func.now())
updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
# 关联类型
type = relationship("Type", back_populates="configs")
# 确保同一类型下的key唯一
__table_args__ = (
UniqueConstraint('type_id', 'key', name='uix_type_key'),
)
def __repr__(self):
return f"<Config(config_id={self.config_id}, key='{self.key}', value='{self.value}')>"

79
app/models/database.py Normal file
View File

@ -0,0 +1,79 @@
import logging
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from sqlalchemy import text, select, inspect
from app.core.config import settings
import bcrypt
from app.models.base import Base
from app.models.user import User
from app.models.type import Type
# SQLite异步URL需要使用aiosqlite
DATABASE_URL = settings.DATABASE_URL.replace("sqlite:///", "sqlite+aiosqlite:///")
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
# 在创建默认配置类型后添加特权模式配置
async def init_db():
"""初始化数据库,创建所有表"""
try:
# 检查表是否存在
async with engine.connect() as conn:
# 检查users表是否存在
result = await conn.execute(text("SELECT name FROM sqlite_master WHERE type='table' AND name='users'"))
table_exists = result.scalar() is not None
if not table_exists:
# 如果表不存在,创建所有表
async with engine.begin() as conn2:
await conn2.run_sync(Base.metadata.create_all)
# 创建默认数据
async with AsyncSessionLocal() as session:
# 创建默认管理员用户
password = "admin123" # 默认密码
salt = bcrypt.gensalt()
hashed_password = bcrypt.hashpw(password.encode(), salt)
admin_user = User(
username="admin",
hashed_password=hashed_password.decode(),
role="admin"
)
session.add(admin_user)
# 创建默认配置类型
default_type = Type(
type_name="default",
description="默认配置类型"
)
session.add(default_type)
# 添加特权模式配置(默认关闭)
from app.models.config import Config
privilege_config = Config(
type_id=1, # 默认类型ID为1
key="privilege_mode",
value="true",
key_description="特权模式设置为true时所有用户无需登录即可使用所有功能"
)
session.add(privilege_config)
await session.commit()
logging.info("数据库初始化成功,创建了默认管理员用户(admin/admin123)和默认配置类型")
else:
logging.info("数据库表已存在,跳过初始化")
except Exception as e:
logging.error(f"数据库初始化失败: {e}")
raise
async def get_db():
"""获取数据库会话"""
db = AsyncSessionLocal()
try:
yield db
finally:
await db.close()

17
app/models/type.py Normal file
View File

@ -0,0 +1,17 @@
from sqlalchemy import Column, Integer, String, DateTime, func
from sqlalchemy.orm import relationship
from app.models.database import Base
class Type(Base):
__tablename__ = "types"
type_id = Column(Integer, primary_key=True, index=True, autoincrement=True)
type_name = Column(String, unique=True, index=True, nullable=False)
description = Column(String, nullable=True)
created_at = Column(DateTime, default=func.now())
# 关联配置项
configs = relationship("Config", back_populates="type", cascade="all, delete-orphan")
def __repr__(self):
return f"<Type(type_id={self.type_id}, type_name='{self.type_name}')>"

16
app/models/user.py Normal file
View File

@ -0,0 +1,16 @@
from sqlalchemy import Column, Integer, String, DateTime
from datetime import datetime
from app.models.base import Base
class User(Base):
__tablename__ = "users"
user_id = Column(Integer, primary_key=True, index=True)
username = Column(String, unique=True, index=True)
hashed_password = Column(String)
role = Column(String, default="user") # admin 或 user
created_at = Column(DateTime, default=datetime.utcnow)
updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
def __repr__(self):
return f"<User(user_id={self.user_id}, username='{self.username}', role='{self.role}')>"

37
app/schemas/config.py Normal file
View File

@ -0,0 +1,37 @@
from pydantic import BaseModel, Field
from typing import Optional, List, Any
from datetime import datetime
class ConfigBase(BaseModel):
key: str = Field(..., description="配置键")
value: str = Field(..., description="配置值")
key_description: Optional[str] = Field(None, description="配置键描述")
class ConfigCreate(ConfigBase):
type_name: Optional[str] = Field("default", description="配置类型名称")
class ConfigUpdate(BaseModel):
value: Optional[str] = Field(None, description="配置值")
key_description: Optional[str] = Field(None, description="配置键描述")
class ConfigInDB(ConfigBase):
config_id: int
type_id: int
created_at: datetime
updated_at: datetime
class Config:
orm_mode = True
class Config(ConfigInDB):
type_name: str
class ConfigList(BaseModel):
configs: List[Config]
total: int
class ConfigSearch(BaseModel):
type_name: Optional[str] = Field(None, description="配置类型名称")
key: Optional[str] = Field(None, description="配置键")
value: Optional[str] = Field(None, description="配置值")
exact_match: bool = Field(False, description="是否精确匹配")

37
app/schemas/type.py Normal file
View File

@ -0,0 +1,37 @@
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
class TypeBase(BaseModel):
"""配置类型基础模型"""
type_name: str = Field(..., description="类型名称")
description: Optional[str] = Field(None, description="类型描述")
class TypeCreate(TypeBase):
"""创建配置类型的请求模型"""
pass
class TypeUpdate(BaseModel):
"""更新配置类型的请求模型"""
type_name: Optional[str] = Field(None, description="类型名称")
description: Optional[str] = Field(None, description="类型描述")
class TypeInDB(TypeBase):
type_id: int
created_at: datetime
class Config:
orm_mode = True
class Type(TypeInDB):
"""配置类型响应模型"""
type_id: int
created_at: Optional[datetime] = None
updated_at: Optional[datetime] = None
class Config:
from_attributes = True # 替代旧版的 orm_mode = True
class TypeList(BaseModel):
types: List[Type]
total: int

37
app/schemas/user.py Normal file
View File

@ -0,0 +1,37 @@
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
class UserBase(BaseModel):
username: str = Field(..., description="用户名")
class UserCreate(UserBase):
password: str = Field(..., description="密码")
role: Optional[str] = Field("user", description="用户角色")
class UserUpdate(BaseModel):
password: Optional[str] = Field(None, description="密码")
role: Optional[str] = Field(None, description="用户角色")
class UserInDB(UserBase):
user_id: int
role: str
created_at: datetime
class Config:
orm_mode = True
class User(UserInDB):
pass
class UserList(BaseModel):
users: List[User]
total: int
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: Optional[str] = None
role: Optional[str] = None

432
app/static/css/style.css Normal file
View File

@ -0,0 +1,432 @@
:root {
--primary-color: #1976d2;
--primary-light: #63a4ff;
--primary-dark: #004ba0;
--secondary-color: #f5f5f5;
--text-color: #333;
--text-light: #666;
--border-color: #e0e0e0;
--success-color: #4caf50;
--warning-color: #ff9800;
--danger-color: #f44336;
}
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
line-height: 1.6;
color: var(--text-color);
background-color: #f9f9f9;
}
.container {
max-width: 1200px;
margin: 0 auto;
padding: 0 15px;
}
/* 导航栏样式 */
header {
background-color: var(--primary-color);
color: white;
box-shadow: 0 2px 4px rgba(0, 0, 0, 0.1);
}
nav {
display: flex;
justify-content: space-between;
align-items: center;
padding: 1rem;
max-width: 1200px;
margin: 0 auto;
}
.nav-brand {
font-size: 1.5rem;
font-weight: bold;
}
.nav-links {
display: flex;
list-style: none;
}
.nav-links li {
margin-left: 1.5rem;
}
.nav-links a {
color: white;
text-decoration: none;
font-weight: 500;
transition: color 0.3s;
}
.nav-links a:hover {
color: rgba(255, 255, 255, 0.8);
}
/* 主内容区域 */
main {
min-height: calc(100vh - 120px);
padding: 2rem 0;
}
/* 卡片样式 */
.card {
background-color: white;
border-radius: 4px;
box-shadow: 0 1px 3px rgba(0, 0, 0, 0.1);
padding: 1.5rem;
margin-bottom: 1.5rem;
}
.card-header {
display: flex;
justify-content: space-between;
align-items: center;
margin-bottom: 1rem;
padding-bottom: 0.5rem;
border-bottom: 1px solid var(--border-color);
}
.card-title {
font-size: 1.25rem;
font-weight: 500;
}
/* 表格样式 */
.table-container {
overflow-x: auto;
}
table {
width: 100%;
border-collapse: collapse;
margin-bottom: 1rem;
}
th, td {
padding: 0.75rem;
text-align: left;
border-bottom: 1px solid var(--border-color);
}
th {
background-color: var(--secondary-color);
font-weight: 500;
}
tr:hover {
background-color: rgba(0, 0, 0, 0.02);
}
/* 按钮样式 */
.btn {
display: inline-block;
padding: 0.5rem 1rem;
border-radius: 4px;
font-weight: 500;
text-align: center;
cursor: pointer;
transition: all 0.3s;
border: none;
}
.btn-primary {
background-color: var(--primary-color);
color: white;
}
.btn-primary:hover {
background-color: var(--primary-dark);
}
.btn-secondary {
background-color: var(--secondary-color);
color: var(--text-color);
}
.btn-secondary:hover {
background-color: #e0e0e0;
}
.btn-success {
background-color: var(--success-color);
color: white;
}
.btn-success:hover {
background-color: #3d8b40;
}
.btn-danger {
background-color: var(--danger-color);
color: white;
}
.btn-danger:hover {
background-color: #d32f2f;
}
.btn-sm {
padding: 0.25rem 0.5rem;
font-size: 0.875rem;
}
/* 表单样式 */
.form-group {
margin-bottom: 1rem;
}
.form-label {
display: block;
margin-bottom: 0.5rem;
font-weight: 500;
}
.form-control {
display: block;
width: 100%;
padding: 0.5rem;
font-size: 1rem;
border: 1px solid var(--border-color);
border-radius: 4px;
transition: border-color 0.3s;
}
.form-control:focus {
border-color: var(--primary-color);
outline: none;
}
/* 搜索框 */
.search-container {
margin-bottom: 1.5rem;
}
.search-box {
display: flex;
gap: 1rem;
flex-wrap: wrap;
align-items: center;
}
.search-box select {
min-width: 150px;
max-width: 200px;
}
.search-input {
flex: 1;
min-width: 200px;
}
/* 标签样式 */
.tag {
display: inline-block;
padding: 0.25rem 0.5rem;
border-radius: 4px;
font-size: 0.875rem;
margin-right: 0.5rem;
margin-bottom: 0.5rem;
background-color: var(--primary-light);
color: white;
}
/* 页脚样式 */
footer {
background-color: var(--secondary-color);
padding: 1rem 0;
text-align: center;
color: var(--text-light);
}
/* 响应式设计 */
@media (max-width: 768px) {
nav {
flex-direction: column;
align-items: flex-start;
}
.nav-links {
margin-top: 1rem;
flex-direction: column;
}
.nav-links li {
margin-left: 0;
margin-bottom: 0.5rem;
}
.auth-links {
margin-top: 10px;
}
.card-header {
flex-direction: column;
align-items: flex-start;
}
.card-header button {
margin-top: 10px;
}
.search-box {
flex-direction: column;
}
}
/* 在现有样式的基础上添加以下内容 */
/* 导航栏右侧认证链接样式 */
.auth-links {
display: flex;
align-items: center;
}
.auth-links a, .auth-links button {
color: white;
text-decoration: none;
margin-left: 1rem;
font-weight: 500;
}
.auth-links button {
background: transparent;
border: 1px solid white;
padding: 0.25rem 0.5rem;
border-radius: 4px;
cursor: pointer;
transition: all 0.3s;
}
.auth-links button:hover {
background-color: rgba(255, 255, 255, 0.1);
}
/* 特权模式标识样式 */
.badge {
display: inline-block;
padding: 0.25rem 0.5rem;
font-size: 0.75rem;
font-weight: bold;
border-radius: 10px;
margin-right: 0.5rem;
}
.badge-warning {
background-color: var(--warning-color);
color: white;
}
/* 消息提示样式 */
.message {
position: fixed;
top: 20px;
right: 20px;
padding: 15px;
border-radius: 4px;
color: white;
z-index: 1001;
box-shadow: 0 2px 10px rgba(0,0,0,0.2);
}
.message-info {
background-color: var(--primary-color);
}
.message-success {
background-color: var(--success-color);
}
.message-warning {
background-color: var(--warning-color);
color: white;
}
.message-danger {
background-color: var(--danger-color);
}
/* 警告框样式 */
.alert {
padding: 12px;
margin-bottom: 15px;
border-radius: 4px;
}
.alert-danger {
background-color: #f8d7da;
color: #721c24;
border: 1px solid #f5c6cb;
}
.alert-success {
background-color: #d4edda;
color: #155724;
border: 1px solid #c3e6cb;
}
/* 模态框样式 */
.modal {
display: none;
position: fixed;
z-index: 1000;
left: 0;
top: 0;
width: 100%;
height: 100%;
background-color: rgba(0,0,0,0.5);
}
.modal-content {
background-color: white;
margin: 10% auto;
padding: 20px;
border-radius: 4px;
width: 80%;
max-width: 500px;
box-shadow: 0 5px 15px rgba(0,0,0,0.3);
}
.modal-header {
display: flex;
justify-content: space-between;
align-items: center;
border-bottom: 1px solid var(--border-color);
padding-bottom: 10px;
margin-bottom: 15px;
}
.modal-header h2 {
margin: 0;
font-size: 1.25rem;
}
.close {
color: var(--text-light);
font-size: 28px;
font-weight: bold;
cursor: pointer;
}
.close:hover {
color: var(--text-color);
}
/* 响应式设计补充 */
@media (max-width: 768px) {
.auth-links {
margin-top: 1rem;
align-self: flex-start;
}
.auth-links a, .auth-links button {
margin-left: 0;
margin-right: 1rem;
}
}

85
app/static/js/main.js Normal file
View File

@ -0,0 +1,85 @@
// 通用函数
function showMessage(message, type = 'info') {
const messageContainer = document.getElementById('message-container');
if (!messageContainer) return;
const messageElement = document.createElement('div');
messageElement.className = `alert alert-${type}`;
messageElement.textContent = message;
messageContainer.appendChild(messageElement);
// 3秒后自动消失
setTimeout(() => {
messageElement.remove();
}, 3000);
}
// 表单提交处理
async function handleFormSubmit(event, url, method = 'POST', successCallback) {
event.preventDefault();
const form = event.target;
const formData = new FormData(form);
const data = {};
formData.forEach((value, key) => {
data[key] = value;
});
try {
const response = await fetch(url, {
method: method,
headers: {
'Content-Type': 'application/json',
},
body: JSON.stringify(data),
});
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '操作失败');
}
const result = await response.json();
showMessage('操作成功', 'success');
if (successCallback) {
successCallback(result);
}
} catch (error) {
showMessage(error.message, 'danger');
}
}
// 删除确认
function confirmDelete(message = '确定要删除吗?') {
return confirm(message);
}
// 页面加载完成后执行
document.addEventListener('DOMContentLoaded', function() {
// 初始化消息容器
if (!document.getElementById('message-container')) {
const container = document.createElement('div');
container.id = 'message-container';
container.style.position = 'fixed';
container.style.top = '20px';
container.style.right = '20px';
container.style.zIndex = '1000';
document.body.appendChild(container);
}
// 初始化表单提交事件
document.querySelectorAll('form[data-submit-ajax]').forEach(form => {
const url = form.getAttribute('action');
const method = form.getAttribute('method') || 'POST';
form.addEventListener('submit', event => {
handleFormSubmit(event, url, method, () => {
// 默认成功回调:刷新页面
window.location.reload();
});
});
});
});

193
app/templates/base.html Normal file
View File

@ -0,0 +1,193 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{% block title %}配置中心{% endblock %}</title>
<link rel="stylesheet" href="/static/css/style.css">
{% block extra_css %}{% endblock %}
</head>
<body>
<!-- 导航栏 - 删除重复的导航栏 -->
<header>
<nav>
<ul class="nav-links">
<li><a href="/page/">首页</a></li>
<li><a href="/page/types">配置类型</a></li>
<li><a href="/page/configs">配置项</a></li>
</ul>
<div class="auth-links">
<span id="privilegeMode" class="badge badge-warning" style="display: none;">特权模式</span>
<span id="userInfo" style="display: none;">
欢迎,<span id="username"></span>
<button onclick="logout()" class="btn btn-sm">退出</button>
</span>
<a href="/page/login" id="loginLink">登录</a>
</div>
</nav>
</header>
<main>
<div class="container">
{% block content %}{% endblock %}
</div>
</main>
<footer>
<div class="container">
<p>&copy; 2025 Guyue. 保留所有权利.</p>
</div>
</footer>
<!-- 消息提示 -->
<div id="message" class="message" style="display: none;"></div>
<!-- 在 script 标签中添加检查特权模式的代码 -->
<script>
// 检查登录状态
function checkLoginStatus() {
const token = localStorage.getItem('access_token');
const username = localStorage.getItem('username');
if (token && username) {
// 已登录
document.getElementById('userInfo').style.display = 'inline-block';
document.getElementById('username').textContent = username;
document.getElementById('loginLink').style.display = 'none';
// 添加令牌到所有API请求
addTokenToFetch();
} else {
// 未登录
document.getElementById('userInfo').style.display = 'none';
document.getElementById('loginLink').style.display = 'inline-block';
// 检查特权模式
checkPrivilegeMode();
}
}
// 检查特权模式
async function checkPrivilegeMode() {
try {
const response = await fetch('/api/auth/privilege-mode');
const data = await response.json();
if (data.privilege_mode) {
document.getElementById('privilegeMode').style.display = 'inline-block';
console.log('特权模式已启用');
} else {
document.getElementById('privilegeMode').style.display = 'none';
}
} catch (error) {
console.error('检查特权模式失败:', error);
}
}
// 添加令牌到所有fetch请求
function addTokenToFetch() {
const originalFetch = window.fetch;
window.fetch = function(url, options = {}) {
const token = localStorage.getItem('access_token');
if (token && url.startsWith('/api/')) {
options.headers = options.headers || {};
options.headers['Authorization'] = `Bearer ${token}`;
}
return originalFetch(url, options);
};
}
// 退出登录
function logout() {
localStorage.removeItem('access_token');
localStorage.removeItem('username');
localStorage.removeItem('role');
window.location.reload();
}
// 显示消息
function showMessage(text, type = 'info') {
const messageElement = document.getElementById('message');
messageElement.textContent = text;
messageElement.className = `message message-${type}`;
messageElement.style.display = 'block';
setTimeout(() => {
messageElement.style.display = 'none';
}, 3000);
}
// 确认删除
function confirmDelete(message) {
return confirm(message);
}
// 页面加载时检查登录状态
document.addEventListener('DOMContentLoaded', function() {
checkLoginStatus();
// 处理表单AJAX提交
document.querySelectorAll('form[data-submit-ajax]').forEach(form => {
form.addEventListener('submit', async function(event) {
event.preventDefault();
const formData = new FormData(form);
const method = form.method || 'POST';
const url = form.action;
try {
let options = {
method: method,
};
// 根据不同的方法处理数据
if (method === 'GET') {
// GET请求不需要body
} else if (formData.has('file')) {
// 如果有文件使用FormData
options.body = formData;
} else {
// 否则使用JSON
const jsonData = {};
formData.forEach((value, key) => {
jsonData[key] = value;
});
options.headers = {
'Content-Type': 'application/json',
};
options.body = JSON.stringify(jsonData);
}
const response = await fetch(url, options);
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '操作失败');
}
showMessage('操作成功', 'success');
// 关闭模态框
const modal = form.closest('.modal');
if (modal) {
modal.style.display = 'none';
}
// 重新加载页面
setTimeout(() => {
window.location.reload();
}, 1000);
} catch (error) {
showMessage(error.message, 'danger');
}
});
});
});
</script>
{% block extra_js %}{% endblock %}
</body>
</html>

255
app/templates/configs.html Normal file
View File

@ -0,0 +1,255 @@
{% extends "base.html" %}
{% block title %}配置项 - 配置中心{% endblock %}
{% block content %}
<div class="container">
<div class="card">
<div class="card-header">
<h1 class="card-title">配置项</h1>
<!-- 添加新增按钮 -->
<button class="btn btn-primary" onclick="showAddConfigModal()">添加配置</button>
</div>
<div class="card-body">
<div class="search-container">
<form class="search-box" action="/page/configs" method="get">
<select name="type_name" class="form-control" onchange="this.form.submit()">
<option value="">所有类型</option>
{% for type in types %}
<option value="{{ type.type_name }}" {% if current_type == type.type_name %}selected{% endif %}>{{ type.type_name }}</option>
{% endfor %}
</select>
<input type="text" name="search" class="form-control search-input" placeholder="搜索键名或值..." value="{{ request.query_params.get('search', '') }}">
<button type="submit" class="btn btn-primary">搜索</button>
{% if request.query_params.get('search') or request.query_params.get('type_name') %}
<a href="/page/configs" class="btn btn-secondary">清除</a>
{% endif %}
</form>
</div>
<div class="table-container">
<table>
<thead>
<tr>
<th>类型</th>
<th>键名</th>
<th></th>
<th>描述</th>
<th>操作</th>
</tr>
</thead>
<tbody>
{% for config in configs %}
<tr>
<td>{{ config.type.type_name }}</td>
<td>{{ config.key }}</td>
<td>{{ config.value }}</td>
<td>{{ config.key_description }}</td>
<td>
<!-- 添加编辑和删除按钮 -->
<button class="btn btn-sm btn-secondary" onclick="showEditConfigModal('{{ config.type.type_name }}', '{{ config.key }}', '{{ config.value }}', '{{ config.key_description }}')">编辑</button>
<button class="btn btn-sm btn-danger" onclick="confirmDeleteConfig('{{ config.type.type_name }}', '{{ config.key }}')">删除</button>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div>
</div>
<!-- 添加配置模态框 -->
<div id="addConfigModal" class="modal">
<div class="modal-content">
<div class="modal-header">
<h2>添加配置项</h2>
<span class="close" onclick="closeAddConfigModal()">&times;</span>
</div>
<form id="addConfigForm">
<div class="form-group">
<label for="configType" class="form-label">类型</label>
<select id="configType" name="type_name" class="form-control" required>
{% for type in types %}
<option value="{{ type.type_name }}">{{ type.type_name }}</option>
{% endfor %}
</select>
</div>
<div class="form-group">
<label for="configKey" class="form-label">键名</label>
<input type="text" id="configKey" name="key" class="form-control" required>
</div>
<div class="form-group">
<label for="configValue" class="form-label"></label>
<input type="text" id="configValue" name="value" class="form-control" required>
</div>
<div class="form-group">
<label for="configDescription" class="form-label">描述</label>
<textarea id="configDescription" name="key_description" class="form-control" rows="3"></textarea>
</div>
<button type="submit" class="btn btn-primary">保存</button>
</form>
</div>
</div>
<!-- 编辑配置模态框 -->
<div id="editConfigModal" class="modal">
<div class="modal-content">
<div class="modal-header">
<h2>编辑配置项</h2>
<span class="close" onclick="closeEditConfigModal()">&times;</span>
</div>
<form id="editConfigForm">
<input type="hidden" id="editConfigType" name="type_name">
<input type="hidden" id="editConfigKey" name="key">
<div class="form-group">
<label for="editConfigValue" class="form-label"></label>
<input type="text" id="editConfigValue" name="value" class="form-control" required>
</div>
<div class="form-group">
<label for="editConfigDescription" class="form-label">描述</label>
<textarea id="editConfigDescription" name="key_description" class="form-control" rows="3"></textarea>
</div>
<button type="submit" class="btn btn-primary">保存</button>
</form>
</div>
</div>
{% endblock %}
{% block extra_js %}
<script>
// 添加配置模态框
function showAddConfigModal() {
document.getElementById('addConfigModal').style.display = 'block';
}
function closeAddConfigModal() {
document.getElementById('addConfigModal').style.display = 'none';
}
// 编辑配置模态框
function showEditConfigModal(typeName, key, value, description) {
document.getElementById('editConfigType').value = typeName;
document.getElementById('editConfigKey').value = key;
document.getElementById('editConfigValue').value = value;
document.getElementById('editConfigDescription').value = description;
document.getElementById('editConfigModal').style.display = 'block';
}
function closeEditConfigModal() {
document.getElementById('editConfigModal').style.display = 'none';
}
// 确认删除配置
function confirmDeleteConfig(typeName, key) {
if (confirm(`确定要删除配置 "${typeName}.${key}" 吗?`)) {
deleteConfig(typeName, key);
}
}
// 添加配置表单提交
document.getElementById('addConfigForm').addEventListener('submit', async function(event) {
event.preventDefault();
const typeName = document.getElementById('configType').value;
const key = document.getElementById('configKey').value;
const value = document.getElementById('configValue').value;
const description = document.getElementById('configDescription').value;
try {
const response = await fetch('/api/configs/', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${localStorage.getItem('access_token')}`
},
body: JSON.stringify({
type_name: typeName,
key: key,
value: value,
key_description: description
})
});
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '添加配置失败');
}
// 关闭模态框并刷新页面
closeAddConfigModal();
showMessage('添加配置成功', 'success');
setTimeout(() => {
window.location.reload();
}, 1000);
} catch (error) {
showMessage(error.message, 'danger');
}
});
// 编辑配置表单提交
document.getElementById('editConfigForm').addEventListener('submit', async function(event) {
event.preventDefault();
const typeName = document.getElementById('editConfigType').value;
const key = document.getElementById('editConfigKey').value;
const value = document.getElementById('editConfigValue').value;
const description = document.getElementById('editConfigDescription').value;
try {
const response = await fetch(`/api/configs/${typeName}/${key}`, {
method: 'PUT',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${localStorage.getItem('access_token')}`
},
body: JSON.stringify({
value: value,
key_description: description
})
});
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '更新配置失败');
}
// 关闭模态框并刷新页面
closeEditConfigModal();
showMessage('更新配置成功', 'success');
setTimeout(() => {
window.location.reload();
}, 1000);
} catch (error) {
showMessage(error.message, 'danger');
}
});
// 删除配置
async function deleteConfig(typeName, key) {
try {
console.log(`正在删除配置: ${typeName}.${key}`); // 添加调试日志
const response = await fetch(`/api/configs/${encodeURIComponent(typeName)}/${encodeURIComponent(key)}`, {
method: 'DELETE',
headers: {
'Authorization': `Bearer ${localStorage.getItem('access_token')}`
}
});
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '删除配置失败');
}
showMessage('删除配置成功', 'success');
setTimeout(() => {
window.location.reload();
}, 1000);
} catch (error) {
console.error('删除配置出错:', error); // 添加错误日志
showMessage(error.message, 'danger');
}
}
</script>
{% endblock %}

44
app/templates/index.html Normal file
View File

@ -0,0 +1,44 @@
{% extends "base.html" %}
{% block title %}首页 - 配置中心{% endblock %}
{% block content %}
<div class="card">
<div class="card-header">
<h1 class="card-title">欢迎使用配置中心</h1>
</div>
<div class="card-body">
<p>配置中心是一个统一管理各类配置的平台,提供便捷的配置创建、修改、删除和查询功能。</p>
<div style="margin-top: 2rem; display: flex; gap: 2rem;">
<div class="card" style="flex: 1;">
<h3>配置类型</h3>
<p style="font-size: 2rem; margin: 1rem 0; color: var(--primary-color);">
{{ types_count|default(0) }}
</p>
<a href="/api/types" class="btn btn-primary">查看类型</a>
</div>
<div class="card" style="flex: 1;">
<h3>配置项</h3>
<p style="font-size: 2rem; margin: 1rem 0; color: var(--primary-color);">
{{ configs_count|default(0) }}
</p>
<a href="/api/configs" class="btn btn-primary">查看配置</a>
</div>
</div>
<div style="margin-top: 2rem;">
<h2>快速入门</h2>
<ul style="margin-left: 1.5rem; margin-top: 0.5rem;">
<li><a href="/api/types">类型管理</a>中创建新的配置类型</li>
<li><a href="/api/configs">配置管理</a>中添加配置项</li>
<li>使用搜索功能快速查找配置</li>
{% if current_user and current_user.role == 'admin' %}
<li><a href="/api/users">用户管理</a>中管理用户权限</li>
{% endif %}
</ul>
</div>
</div>
</div>
{% endblock %}

95
app/templates/login.html Normal file
View File

@ -0,0 +1,95 @@
{% extends "base.html" %}
{% block title %}登录 - 配置中心{% endblock %}
{% block content %}
<div class="card" style="max-width: 500px; margin: 0 auto; margin-top: 50px;">
<div class="card-header">
<h1 class="card-title">用户登录</h1>
</div>
<div class="card-body">
<div id="login-error" class="alert alert-danger" style="display: none;"></div>
<form id="loginForm">
<div class="form-group">
<label for="username" class="form-label">用户名</label>
<input type="text" id="username" name="username" class="form-control" required>
</div>
<div class="form-group">
<label for="password" class="form-label">密码</label>
<input type="password" id="password" name="password" class="form-control" required>
</div>
<button type="submit" class="btn btn-primary" style="width: 100%;">登录</button>
</form>
</div>
</div>
{% endblock %}
{% block extra_js %}
<script>
document.getElementById('loginForm').addEventListener('submit', async function(event) {
event.preventDefault();
// 直接从表单元素获取值
const usernameInput = document.getElementById('username');
const passwordInput = document.getElementById('password');
if (!usernameInput || !passwordInput) {
console.error('找不到用户名或密码输入框');
showMessage('登录表单错误', 'danger');
return;
}
const username = usernameInput.value.trim();
const password = passwordInput.value;
// 检查用户名和密码是否为空
if (!username || !password) {
const errorElement = document.getElementById('login-error');
errorElement.textContent = '用户名和密码不能为空';
errorElement.style.display = 'block';
return;
}
console.log('提交登录,用户名:', username); // 调试输出
try {
// 使用URLSearchParams直接构建表单数据
const formData = new URLSearchParams();
formData.append('username', username);
formData.append('password', password);
const response = await fetch('/api/auth/token', {
method: 'POST',
headers: {
'Content-Type': 'application/x-www-form-urlencoded',
},
body: formData.toString(),
});
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '登录失败');
}
const data = await response.json();
// 保存令牌到本地存储
localStorage.setItem('access_token', data.access_token);
localStorage.setItem('username', data.username);
localStorage.setItem('role', data.role);
// 显示成功消息
showMessage('登录成功,正在跳转...', 'success');
// 重定向到首页
setTimeout(() => {
window.location.href = '/page';
}, 1000);
} catch (error) {
const errorElement = document.getElementById('login-error');
errorElement.textContent = error.message;
errorElement.style.display = 'block';
}
});
</script>
{% endblock %}

222
app/templates/types.html Normal file
View File

@ -0,0 +1,222 @@
{% extends "base.html" %}
{% block title %}配置类型 - 配置中心{% endblock %}
{% block content %}
<div class="container">
<div class="card">
<div class="card-header">
<h1 class="card-title">配置类型</h1>
<!-- 添加新增按钮 -->
<button class="btn btn-primary" onclick="showAddTypeModal()">添加类型</button>
</div>
<div class="card-body">
<div class="search-container">
<form class="search-box" action="/page/types" method="get">
<input type="text" name="search" class="form-control search-input" placeholder="搜索类型名称或描述..." value="{{ request.query_params.get('search', '') }}">
<button type="submit" class="btn btn-primary">搜索</button>
{% if request.query_params.get('search') %}
<a href="/page/types" class="btn btn-secondary">清除</a>
{% endif %}
</form>
</div>
<div class="table-container">
<table>
<thead>
<tr>
<th>类型名称</th>
<th>描述</th>
<th>操作</th>
</tr>
</thead>
<tbody>
{% for type in types %}
<tr>
<td>
<!-- 添加链接,点击类型名称跳转到对应的配置项列表 -->
<a href="/page/configs?type_name={{ type.type_name }}">{{ type.type_name }}</a>
</td>
<td>{{ type.description }}</td>
<td>
<!-- 添加编辑和删除按钮 -->
<button class="btn btn-sm btn-secondary" onclick="showEditTypeModal('{{ type.type_name }}', '{{ type.description }}')">编辑</button>
<button class="btn btn-sm btn-danger" onclick="confirmDeleteType('{{ type.type_name }}')">删除</button>
</td>
</tr>
{% endfor %}
</tbody>
</table>
</div>
</div>
</div>
</div>
<!-- 添加类型模态框 -->
<div id="addTypeModal" class="modal">
<div class="modal-content">
<div class="modal-header">
<h2>添加配置类型</h2>
<span class="close" onclick="closeAddTypeModal()">&times;</span>
</div>
<form id="addTypeForm">
<div class="form-group">
<label for="typeName" class="form-label">类型名称</label>
<input type="text" id="typeName" name="type_name" class="form-control" required>
</div>
<div class="form-group">
<label for="typeDescription" class="form-label">描述</label>
<textarea id="typeDescription" name="description" class="form-control" rows="3"></textarea>
</div>
<button type="submit" class="btn btn-primary">保存</button>
</form>
</div>
</div>
<!-- 编辑类型模态框 -->
<div id="editTypeModal" class="modal">
<div class="modal-content">
<div class="modal-header">
<h2>编辑配置类型</h2>
<span class="close" onclick="closeEditTypeModal()">&times;</span>
</div>
<form id="editTypeForm">
<input type="hidden" id="editTypeName" name="type_name">
<div class="form-group">
<label for="editTypeDescription" class="form-label">描述</label>
<textarea id="editTypeDescription" name="description" class="form-control" rows="3"></textarea>
</div>
<button type="submit" class="btn btn-primary">保存</button>
</form>
</div>
</div>
{% endblock %}
{% block extra_js %}
<script>
// 添加类型模态框
function showAddTypeModal() {
document.getElementById('addTypeModal').style.display = 'block';
}
function closeAddTypeModal() {
document.getElementById('addTypeModal').style.display = 'none';
}
// 编辑类型模态框
function showEditTypeModal(typeName, description) {
document.getElementById('editTypeName').value = typeName;
document.getElementById('editTypeDescription').value = description;
document.getElementById('editTypeModal').style.display = 'block';
}
function closeEditTypeModal() {
document.getElementById('editTypeModal').style.display = 'none';
}
// 确认删除类型
function confirmDeleteType(typeName) {
if (confirm(`确定要删除类型 "${typeName}" 吗?`)) {
deleteType(typeName);
}
}
// 添加类型表单提交
document.getElementById('addTypeForm').addEventListener('submit', async function(event) {
event.preventDefault();
const typeName = document.getElementById('typeName').value;
const description = document.getElementById('typeDescription').value;
try {
const response = await fetch('/api/types/', {
method: 'POST',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${localStorage.getItem('access_token')}`
},
body: JSON.stringify({
type_name: typeName,
description: description
})
});
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '添加类型失败');
}
// 关闭模态框并刷新页面
closeAddTypeModal();
showMessage('添加类型成功', 'success');
setTimeout(() => {
window.location.reload();
}, 1000);
} catch (error) {
showMessage(error.message, 'danger');
}
});
// 编辑类型表单提交
document.getElementById('editTypeForm').addEventListener('submit', async function(event) {
event.preventDefault();
const typeName = document.getElementById('editTypeName').value;
const description = document.getElementById('editTypeDescription').value;
try {
const response = await fetch(`/api/types/${typeName}`, {
method: 'PUT',
headers: {
'Content-Type': 'application/json',
'Authorization': `Bearer ${localStorage.getItem('access_token')}`
},
body: JSON.stringify({
description: description
})
});
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '更新类型失败');
}
// 关闭模态框并刷新页面
closeEditTypeModal();
showMessage('更新类型成功', 'success');
setTimeout(() => {
window.location.reload();
}, 1000);
} catch (error) {
showMessage(error.message, 'danger');
}
});
// 删除类型
async function deleteType(typeName) {
try {
console.log(`正在删除类型: ${typeName}`); // 添加调试日志
const response = await fetch(`/api/types/${encodeURIComponent(typeName)}`, {
method: 'DELETE',
headers: {
'Authorization': `Bearer ${localStorage.getItem('access_token')}`
}
});
if (!response.ok) {
const errorData = await response.json();
throw new Error(errorData.detail || '删除类型失败');
}
showMessage('删除类型成功', 'success');
setTimeout(() => {
window.location.reload();
}, 1000);
} catch (error) {
console.error('删除类型出错:', error); // 添加错误日志
showMessage(error.message, 'danger');
}
}
</script>
{% endblock %}

BIN
config_center.db Normal file

Binary file not shown.

11
requirements.txt Normal file
View File

@ -0,0 +1,11 @@
fastapi==0.103.1
uvicorn==0.23.2
sqlalchemy==2.0.20
alembic==1.12.0
pydantic==2.3.0
python-jose==3.3.0
passlib==1.7.4
python-multipart==0.0.6
bcrypt==4.0.1
jinja2==3.1.2
aiosqlite==0.19.0

11
run.py Normal file
View File

@ -0,0 +1,11 @@
import uvicorn
from app.main import app
if __name__ == "__main__":
uvicorn.run(
"app.main:app",
host="0.0.0.0",
port=8000,
reload=True,
log_level="info"
)