""" main.py FastAPI 主应用：设备管理平台后端服务。 启动方式： uvicorn main:app --reload --host 0.0.0.0 --port 8000 访问文档： http://localhost:8000/docs (Swagger UI) http://localhost:8000/redoc (ReDoc) """ from fastapi import FastAPI, HTTPException, Depends from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import RedirectResponse from contextlib import asynccontextmanager from database import init_db, seed_demo_data, get_db from models import ( DeviceCheckRequest, LicenseGenerateRequest, ActivateReportRequest, CheckResponse, LicenseResponse, ActivateResponse, DeviceInfo, ) from services import build_license_data, encrypt_license, decrypt_license # ═══════════════════════════════════════════════════════════════ # 生命周期：启动时初始化数据库 # ═══════════════════════════════════════════════════════════════ @asynccontextmanager async def lifespan(app: FastAPI): """应用启动时自动创建表并插入演示数据""" init_db() seed_demo_data() yield # 关闭时可做清理（此处无需） app = FastAPI( title="设备管理平台 API", description="基于 FastAPI + SQLite 的轻量级设备管理后端，支持设备校验、授权文件生成、激活上报。", version="1.0.0", lifespan=lifespan, ) # 允许跨域（方便前端/APP 调试） app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"], ) # ═══════════════════════════════════════════════════════════════ # 根路由：自动跳转到 Swagger 文档 # ═══════════════════════════════════════════════════════════════ @app.get("/", include_in_schema=False) def root(): return RedirectResponse(url="/docs") # ═══════════════════════════════════════════════════════════════ # 1. 设备校验接口 # ═══════════════════════════════════════════════════════════════ @app.post("/api/devices/check", response_model=CheckResponse, summary="设备校验") def device_check(payload: DeviceCheckRequest): """ APP 启动时调用，校验设备 SN 是否合法、是否已激活。 - 若 SN 不存在 → 非法设备，拒绝激活 - 若 SN 存在但未激活 → 允许进入激活流程 - 若 SN 已激活 → 正常使用 """ with get_db() as db: row = db.execute( "SELECT status FROM devices WHERE sn = ?", (payload.sn,) ).fetchone() if not row: return CheckResponse( valid=False, activated=False, message="设备 SN 不存在，请联系管理员注册" ) status = row["status"] if status == "已激活": return CheckResponse( valid=True, activated=True, message="设备已激活，正常使用" ) elif status == "已禁用": return CheckResponse( valid=True, activated=False, message="设备已被禁用，请联系客服" ) else: return CheckResponse( valid=True, activated=False, message="设备待激活，请获取授权文件并激活" ) # ═══════════════════════════════════════════════════════════════ # 2. 生成加密授权文件接口 # ═══════════════════════════════════════════════════════════════ @app.post("/api/licenses/generate", response_model=LicenseResponse, summary="生成加密授权文件") def generate_license(payload: LicenseGenerateRequest): """ 管理后台调用，为指定设备生成绑定 SN 的加密授权文件。 流程： 1. 校验设备 SN 是否存在 2. 构造授权数据（模块列表 + 有效期） 3. XOR 加密（密钥由 SN 派生，一机一密） 4. 返回 Base64 加密字符串 APP 收到后必须用相同 SN 才能解密。 """ with get_db() as db: row = db.execute( "SELECT sn FROM devices WHERE sn = ?", (payload.sn,) ).fetchone() if not row: raise HTTPException(status_code=404, detail="设备 SN 不存在") # 构造明文授权数据 license_data = build_license_data( device_sn=payload.sn, modules=payload.modules, valid_days=payload.valid_days, ) # 加密（绑定 SN） encrypted = encrypt_license(license_data) return LicenseResponse( success=True, encrypted_license=encrypted, raw_license=license_data.model_dump(), # 调试用，生产环境可去掉 message="授权文件生成成功，已绑定设备 SN" ) # ═══════════════════════════════════════════════════════════════ # 3. 上报激活状态接口 # ═══════════════════════════════════════════════════════════════ @app.post("/api/devices/activate", response_model=ActivateResponse, summary="上报激活状态") def report_activation(payload: ActivateReportRequest): """ 设备主机首次联网后调用，上报激活结果。 - 若上报 "已激活" → 数据库更新状态与激活时间 - 若上报 "激活失败" → 状态保持原样，记录失败 """ with get_db() as db: row = db.execute( "SELECT status FROM devices WHERE sn = ?", (payload.sn,) ).fetchone() if not row: raise HTTPException(status_code=404, detail="设备 SN 不存在") current_status = row["status"] # 只有"待激活"或"已禁用"的设备允许重新激活 if payload.status == "已激活": db.execute( """ UPDATE devices SET status = '已激活', activated_at = datetime('now', 'localtime') WHERE sn = ? """, (payload.sn,), ) db.commit() # 查询更新后的时间 updated = db.execute( "SELECT activated_at FROM devices WHERE sn = ?", (payload.sn,) ).fetchone() return ActivateResponse( success=True, sn=payload.sn, new_status="已激活", activated_at=updated["activated_at"], message="设备激活成功，已记录激活时间" ) else: return ActivateResponse( success=False, sn=payload.sn, new_status=current_status, message="激活失败，未更新状态" ) # ═══════════════════════════════════════════════════════════════ # 辅助接口（方便调试和管理） # ═══════════════════════════════════════════════════════════════ @app.get("/api/devices", summary="获取所有设备列表") def list_devices(): """列出当前所有设备及其状态""" with get_db() as db: rows = db.execute( "SELECT sn, status, activated_at, created_at FROM devices ORDER BY created_at DESC" ).fetchall() return [dict(r) for r in rows] @app.post("/api/devices/register", summary="注册新设备") def register_device(sn: str): """向平台注册一个新的设备 SN""" with get_db() as db: try: db.execute( "INSERT INTO devices (sn, status) VALUES (?, '待激活')", (sn,), ) db.commit() return {"success": True, "sn": sn, "message": "设备注册成功"} except Exception: raise HTTPException(status_code=400, detail="设备 SN 已存在") @app.post("/api/licenses/verify", summary="验证授权文件") def verify_license(sn: str, encrypted_license: str): """ 调试验证：传入加密授权文件和设备 SN，验证能否正确解密。 用于测试一机一密绑定是否生效。 """ decrypted = decrypt_license(encrypted_license, sn) if decrypted: return { "valid": True, "decrypted": decrypted, "message": "授权文件解密成功，SN 匹配" } return { "valid": False, "decrypted": None, "message": "授权文件无效或 SN 不匹配" } # ═══════════════════════════════════════════════════════════════ # 运行入口（直接 python main.py 启动） # ═══════════════════════════════════════════════════════════════ if __name__ == "__main__": import uvicorn uvicorn.run("main:app", host="localhost", port=8000, reload=True)