from fastapi import APIRouter, Depends, HTTPException, Query, Body, UploadFile, File, Form from sqlalchemy.orm import Session from typing import List, Optional import logging from dotenv import load_dotenv from database import get_db from models.ota import Ota from schemas import OtaCreate, OtaUpdate, OtaResponse from response import ApiResponse, PageData from ota_storage import ( OTA_MODEL_X8, OTA_MODEL_X9, OTA_UPLOAD_MODELS, read_upload_content_and_md5, save_x9_package_local, upload_x8_package_to_s3, ) load_dotenv() # Configure logging logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/ota", tags=["ota"]) @router.post("/upload-package", response_model=ApiResponse) async def upload_ota_package( model: str = Form(..., description="设备型号"), package_file: UploadFile = File(..., description="OTA 升级包"), ): """上传 OTA 升级包:计算 MD5;X8 上传 S3 为 LUXSIN_X8.PKG,X9 本地保存为 LUXSIN.PKG""" try: model = (model or "").strip() if model not in OTA_UPLOAD_MODELS: return ApiResponse( code=0, msg=f"当前仅支持为 {OTA_MODEL_X8}、{OTA_MODEL_X9} 上传升级包", data=None, ) if not package_file or not package_file.filename: return ApiResponse(code=0, msg="请选择升级包文件", data=None) logger.info( "Uploading OTA package: model=%s, filename=%s", model, package_file.filename, ) content, md5_hex = await read_upload_content_and_md5(package_file) if model == OTA_MODEL_X9: saved_name, download_url = save_x9_package_local(content, md5_hex) return ApiResponse( code=1, msg="success", data={ "md5": md5_hex, "filename": saved_name, "url": download_url, }, ) if model == OTA_MODEL_X8: saved_name, download_url, s3_key = upload_x8_package_to_s3( content, md5_hex ) return ApiResponse( code=1, msg="success", data={ "md5": md5_hex, "filename": saved_name, "url": download_url, "s3_key": s3_key, }, ) return ApiResponse(code=0, msg="不支持的设备型号", data=None) except ValueError as e: return ApiResponse(code=0, msg=str(e), data=None) except Exception as e: logger.error("Error uploading OTA package: %s", e, exc_info=True) return ApiResponse(code=0, msg=f"上传失败:{e}", data=None) @router.get("/", response_model=ApiResponse) def get_ota_list( skip: int = Query(0, ge=0, description="跳过记录数"), limit: int = Query(100, ge=1, le=1000, description="返回记录数"), verCode: Optional[int] = Query(None, description="按版本号查询"), verName: Optional[str] = Query(None, description="按版本名称模糊查询"), model: Optional[str] = Query(None, description="按设备型号模糊查询"), status: Optional[int] = Query(None, ge=0, le=1, description="按状态查询"), db: Session = Depends(get_db) ): """获取 OTA 版本列表(支持多种筛选条件)""" try: logger.info(f"Getting OTA list: skip={skip}, limit={limit}, verCode={verCode}, verName={verName}, model={model}, status={status}") query = db.query(Ota) # 应用筛选条件 if verCode is not None: query = query.filter(Ota.verCode == verCode) if verName: query = query.filter(Ota.verName.like(f"%{verName}%")) if model: query = query.filter(Ota.model.like(f"%{model}%")) if status is not None: query = query.filter(Ota.status == status) total = query.count() query = query.order_by(Ota.verCode.desc()) ota_list = query.offset(skip).limit(limit).all() logger.info(f"Found {len(ota_list)} OTA records, total={total}") if not ota_list: logger.warning("No OTA records found") return ApiResponse(code=2, msg="empty", data=None) # 转换为字典列表 ota_data = [ota.to_dict() for ota in ota_list] return ApiResponse(code=1, msg="success", data={"items": ota_data, "total": total, "skip": skip, "limit": limit}) except Exception as e: logger.error(f"Error getting OTA list: {str(e)}", exc_info=True) return ApiResponse(code=0, msg="error", data=None) @router.get("/{ota_id}", response_model=ApiResponse) def get_ota(ota_id: int, db: Session = Depends(get_db)): """获取单个 OTA 版本""" try: logger.info(f"Getting OTA: id={ota_id}") ota = db.query(Ota).filter(Ota.id == ota_id).first() if not ota: logger.warning(f"OTA not found: id={ota_id}") return ApiResponse(code=2, msg="empty", data=None) logger.info(f"OTA found: {ota.to_dict()}") return ApiResponse(code=1, msg="success", data=ota.to_dict()) except Exception as e: logger.error(f"Error getting OTA {ota_id}: {str(e)}", exc_info=True) return ApiResponse(code=0, msg="error", data=None) @router.post("/", response_model=ApiResponse) def create_ota(ota: OtaCreate, db: Session = Depends(get_db)): """创建新 OTA 版本""" try: logger.info(f"Creating OTA: verCode={ota.verCode}, verName={ota.verName}, model={ota.model}") # 检查版本号是否已存在 existing = db.query(Ota).filter( Ota.verCode == ota.verCode, Ota.model == ota.model ).first() if existing: logger.warning(f"OTA version already exists: verCode={ota.verCode}, model={ota.model}") return ApiResponse(code=0, msg="该版本已存在", data=None) db_ota = Ota(**ota.model_dump()) db.add(db_ota) db.commit() db.refresh(db_ota) logger.info(f"OTA created successfully: id={db_ota.id}, verCode={db_ota.verCode}") return ApiResponse(code=1, msg="success", data=db_ota.to_dict()) except Exception as e: logger.error(f"Error creating OTA: {str(e)}", exc_info=True) db.rollback() return ApiResponse(code=0, msg="error", data=None) @router.put("/{ota_id}", response_model=ApiResponse) def update_ota(ota_id: int, ota: OtaUpdate, db: Session = Depends(get_db)): """更新 OTA 版本""" try: logger.info(f"Updating OTA: id={ota_id}, data={ota.model_dump()}") db_ota = db.query(Ota).filter(Ota.id == ota_id).first() if not db_ota: logger.warning(f"OTA not found: id={ota_id}") return ApiResponse(code=2, msg="empty", data=None) # 如果更新版本号,检查是否冲突 if ota.verCode or ota.model: new_verCode = ota.verCode if ota.verCode is not None else db_ota.verCode new_model = ota.model if ota.model is not None else db_ota.model if new_verCode != db_ota.verCode or new_model != db_ota.model: existing = db.query(Ota).filter( Ota.verCode == new_verCode, Ota.model == new_model ).first() if existing: logger.warning(f"OTA version already exists: verCode={new_verCode}, model={new_model}") return ApiResponse(code=0, msg="该版本已存在", data=None) # 更新字段 update_data = ota.model_dump(exclude_unset=True) for field, value in update_data.items(): setattr(db_ota, field, value) db.commit() db.refresh(db_ota) logger.info(f"OTA updated successfully: id={db_ota.id}") return ApiResponse(code=1, msg="success", data=db_ota.to_dict()) except Exception as e: logger.error(f"Error updating OTA {ota_id}: {str(e)}", exc_info=True) db.rollback() return ApiResponse(code=0, msg="error", data=None) @router.delete("/{ota_id}", response_model=ApiResponse) def delete_ota(ota_id: int, db: Session = Depends(get_db)): """删除 OTA 版本""" try: logger.info(f"Deleting OTA: id={ota_id}") db_ota = db.query(Ota).filter(Ota.id == ota_id).first() if not db_ota: logger.warning(f"OTA not found: id={ota_id}") return ApiResponse(code=2, msg="empty", data=None) db.delete(db_ota) db.commit() logger.info(f"OTA deleted successfully: id={ota_id}") return ApiResponse(code=1, msg="success", data=None) except Exception as e: logger.error(f"Error deleting OTA {ota_id}: {str(e)}", exc_info=True) db.rollback() return ApiResponse(code=0, msg="error", data=None) @router.get("/latest/check", response_model=ApiResponse) def check_latest_ta( currentVerCode: int = Query(..., description="当前版本号"), model: str = Query(..., description="设备型号"), hw: Optional[int] = Query(None, description="硬件版本号"), db: Session = Depends(get_db) ): """检查是否有可用的 OTA 升级""" try: logger.info(f"Checking latest OTA: currentVerCode={currentVerCode}, model={model}, hw={hw}") # 构建查询条件 query = db.query(Ota).filter( Ota.status == 1, Ota.verCode > currentVerCode, Ota.model == model ) # 如果提供了硬件版本号,添加筛选条件 if hw is not None: query = query.filter(Ota.hw == hw) # 按版本号降序排列,获取最新版本 latest_ota = query.order_by(Ota.verCode.desc()).first() if not latest_ota: logger.info(f"No available OTA found for model={model}, currentVerCode={currentVerCode}") return ApiResponse(code=2, msg="empty", data=None) logger.info(f"Latest OTA found: verCode={latest_ota.verCode}, verName={latest_ota.verName}") return ApiResponse(code=1, msg="success", data=latest_ota.to_dict()) except Exception as e: logger.error(f"Error checking latest OTA: {str(e)}", exc_info=True) return ApiResponse(code=0, msg="error", data=None)