yangy
90aec37461
- Created .env.example for environment variable configuration. - Added docker-compose.yml for service orchestration. - Implemented frequency response extraction in extract_frequency_response.py and convert_to_frequency_db.py. - Generated output files: frequency_response_detailed.json, frequency_response_points.json, frequency_response.csv, and frequency_response_curve.png. - Included sample measurement data for FiiO FA19 in CSV format.
133 lines
5.4 KiB
Python
133 lines
5.4 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, Query
|
|
from sqlalchemy.orm import Session
|
|
from typing import List, Optional
|
|
import logging
|
|
from database import get_db
|
|
from models.brand import Brand
|
|
from schemas import BrandCreate, BrandUpdate, BrandResponse
|
|
from response import ApiResponse, PageData
|
|
|
|
# Configure logging
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/brands", tags=["brands"])
|
|
|
|
|
|
@router.get("/", response_model=ApiResponse)
|
|
def get_brands(
|
|
skip: int = Query(0, ge=0, description="跳过记录数"),
|
|
limit: int = Query(100, ge=1, le=1000, description="返回记录数"),
|
|
name: Optional[str] = Query(None, description="品牌名称（支持模糊查询）"),
|
|
db: Session = Depends(get_db)
|
|
):
|
|
"""获取所有品牌列表（支持按名称模糊查询）"""
|
|
try:
|
|
logger.info(f"Getting brands: skip={skip}, limit={limit}, name={name}")
|
|
query = db.query(Brand)
|
|
if name:
|
|
query = query.filter(Brand.name.like(f"%{name}%"))
|
|
|
|
total = query.count()
|
|
brands = query.offset(skip).limit(limit).all()
|
|
|
|
logger.info(f"Found {len(brands)} brands, total={total}")
|
|
|
|
if not brands:
|
|
logger.warning("No brands found")
|
|
return ApiResponse(code=2, msg="empty", data=None)
|
|
|
|
# 转换为字典列表
|
|
brands_data = [brand.to_dict() for brand in brands]
|
|
|
|
return ApiResponse(code=1, msg="success", data={"items": brands_data, "total": total, "skip": skip, "limit": limit})
|
|
except Exception as e:
|
|
logger.error(f"Error getting brands: {str(e)}", exc_info=True)
|
|
return ApiResponse(code=0, msg="error", data=None)
|
|
|
|
|
|
@router.get("/{brand_id}", response_model=ApiResponse)
|
|
def get_brand(brand_id: int, db: Session = Depends(get_db)):
|
|
"""获取单个品牌"""
|
|
try:
|
|
logger.info(f"Getting brand: id={brand_id}")
|
|
brand = db.query(Brand).filter(Brand.id == brand_id).first()
|
|
if not brand:
|
|
logger.warning(f"Brand not found: id={brand_id}")
|
|
return ApiResponse(code=2, msg="empty", data=None)
|
|
logger.info(f"Brand found: {brand.to_dict()}")
|
|
return ApiResponse(code=1, msg="success", data=brand.to_dict())
|
|
except Exception as e:
|
|
logger.error(f"Error getting brand {brand_id}: {str(e)}", exc_info=True)
|
|
return ApiResponse(code=0, msg="error", data=None)
|
|
|
|
|
|
@router.post("/", response_model=ApiResponse)
|
|
def create_brand(brand: BrandCreate, db: Session = Depends(get_db)):
|
|
"""创建新品牌"""
|
|
try:
|
|
logger.info(f"Creating brand: name={brand.name}")
|
|
# 检查是否已存在
|
|
existing = db.query(Brand).filter(Brand.name == brand.name).first()
|
|
if existing:
|
|
logger.warning(f"Brand already exists: name={brand.name}")
|
|
return ApiResponse.error(msg="品牌名称已存在", code=0)
|
|
|
|
db_brand = Brand(name=brand.name)
|
|
db.add(db_brand)
|
|
db.commit()
|
|
db.refresh(db_brand)
|
|
logger.info(f"Brand created successfully: id={db_brand.id}, name={db_brand.name}")
|
|
return ApiResponse.success(data=db_brand.to_dict(), msg="品牌创建成功")
|
|
except Exception as e:
|
|
logger.error(f"Error creating brand: {str(e)}", exc_info=True)
|
|
db.rollback()
|
|
return ApiResponse.error(msg=f"创建失败：{str(e)}", code=0)
|
|
|
|
|
|
@router.put("/{brand_id}", response_model=ApiResponse)
|
|
def update_brand(brand_id: int, brand: BrandUpdate, db: Session = Depends(get_db)):
|
|
"""更新品牌"""
|
|
try:
|
|
logger.info(f"Updating brand: id={brand_id}, data={brand.dict()}")
|
|
db_brand = db.query(Brand).filter(Brand.id == brand_id).first()
|
|
if not db_brand:
|
|
logger.warning(f"Brand not found: id={brand_id}")
|
|
return ApiResponse.no_data(msg="品牌不存在")
|
|
|
|
# 如果更新名称，检查是否冲突
|
|
if brand.name and brand.name != db_brand.name:
|
|
existing = db.query(Brand).filter(Brand.name == brand.name).first()
|
|
if existing:
|
|
logger.warning(f"Brand name already exists: name={brand.name}")
|
|
return ApiResponse.error(msg="品牌名称已存在", code=0)
|
|
db_brand.name = brand.name
|
|
|
|
db.commit()
|
|
db.refresh(db_brand)
|
|
logger.info(f"Brand updated successfully: id={db_brand.id}")
|
|
return ApiResponse.success(data=db_brand.to_dict(), msg="品牌更新成功")
|
|
except Exception as e:
|
|
logger.error(f"Error updating brand {brand_id}: {str(e)}", exc_info=True)
|
|
db.rollback()
|
|
return ApiResponse.error(msg=f"更新失败：{str(e)}", code=0)
|
|
|
|
|
|
@router.delete("/{brand_id}", response_model=ApiResponse)
|
|
def delete_brand(brand_id: int, db: Session = Depends(get_db)):
|
|
"""删除品牌"""
|
|
try:
|
|
logger.info(f"Deleting brand: id={brand_id}")
|
|
db_brand = db.query(Brand).filter(Brand.id == brand_id).first()
|
|
if not db_brand:
|
|
logger.warning(f"Brand not found: id={brand_id}")
|
|
return ApiResponse.no_data(msg="品牌不存在")
|
|
|
|
db.delete(db_brand)
|
|
db.commit()
|
|
logger.info(f"Brand deleted successfully: id={brand_id}")
|
|
return ApiResponse.success(msg="删除成功")
|
|
except Exception as e:
|
|
logger.error(f"Error deleting brand {brand_id}: {str(e)}", exc_info=True)
|
|
db.rollback()
|
|
return ApiResponse.error(msg=f"删除失败：{str(e)}", code=0)
|