1154 lines
45 KiB
Python
1154 lines
45 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""无人机资产管理系统 — FastAPI 后端
|
||
|
||
BS 架构：FastAPI + SQLite（Python 内置 sqlite3），前端为 static/ 下的原生页面。
|
||
状态按"每次修改一条记录（含时间戳）"保存，当前状态取最新一条。
|
||
"""
|
||
import sqlite3
|
||
from contextlib import asynccontextmanager
|
||
from datetime import date, datetime, timedelta
|
||
from pathlib import Path
|
||
from typing import List, Optional
|
||
|
||
from fastapi import FastAPI, HTTPException, Query
|
||
from fastapi.responses import FileResponse
|
||
from fastapi.staticfiles import StaticFiles
|
||
from pydantic import BaseModel
|
||
|
||
try:
|
||
import openpyxl
|
||
from openpyxl.styles import Alignment, Font, PatternFill
|
||
from openpyxl.utils import get_column_letter
|
||
except ImportError: # openpyxl 未安装时仅禁用 Excel 导出，不影响系统运行
|
||
openpyxl = None
|
||
|
||
BASE_DIR = Path(__file__).resolve().parent
|
||
DB_PATH = BASE_DIR / "uav.db"
|
||
STATIC_DIR = BASE_DIR / "static"
|
||
EXCEL_PATH = BASE_DIR / "无人机台账.xlsx"
|
||
|
||
STATUSES = ["待组装", "待调试", "待飞", "损坏"] # 内置默认状态
|
||
NO_STATUS = "__none__" # 筛选"无状态"专用标记
|
||
STATUS_COLORS = ["#9e9e9e", "#ff9800", "#4caf50", "#f44336"] # 对应内置颜色
|
||
DEFAULT_STATUS_COLORS = [ # 新增自定义状态时的自动配色候选
|
||
"#9e9e9e", "#ff9800", "#4caf50", "#f44336",
|
||
"#3b82f6", "#8b5cf6", "#ec4899", "#14b8a6",
|
||
"#f59e0b", "#64748b",
|
||
]
|
||
|
||
|
||
# ---------------------------------------------------------------- 数据库
|
||
def get_conn() -> sqlite3.Connection:
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA foreign_keys = ON")
|
||
return conn
|
||
|
||
|
||
def init_db() -> None:
|
||
with get_conn() as conn:
|
||
conn.executescript(
|
||
"""
|
||
CREATE TABLE IF NOT EXISTS drones (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
model TEXT NOT NULL, -- 机型
|
||
serial_no TEXT NOT NULL UNIQUE, -- 编号
|
||
link_id TEXT NOT NULL DEFAULT '', -- 链路ID
|
||
fc_id TEXT NOT NULL DEFAULT '', -- 飞控ID
|
||
fc_version TEXT NOT NULL DEFAULT '', -- 飞控版本号
|
||
nav_version TEXT NOT NULL DEFAULT '', -- 导航版本号
|
||
remark TEXT NOT NULL DEFAULT '', -- 备注
|
||
created_at TEXT NOT NULL -- 创建时间(ISO)
|
||
);
|
||
CREATE TABLE IF NOT EXISTS drone_status (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
drone_id INTEGER NOT NULL REFERENCES drones(id) ON DELETE CASCADE,
|
||
status TEXT NOT NULL, -- 主状态
|
||
remark TEXT NOT NULL DEFAULT '', -- 本次变更备注
|
||
changed_at TEXT NOT NULL -- 变更时间戳(ISO)
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_status_drone ON drone_status(drone_id, changed_at);
|
||
CREATE TABLE IF NOT EXISTS status_defs (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL UNIQUE, -- 状态名称
|
||
color TEXT NOT NULL DEFAULT '#9e9e9e', -- 展示颜色
|
||
sort_order INTEGER NOT NULL DEFAULT 0, -- 排序
|
||
is_builtin INTEGER NOT NULL DEFAULT 0 -- 是否内置（内置不可删）
|
||
);
|
||
CREATE TABLE IF NOT EXISTS flight_records (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
drone_id INTEGER NOT NULL REFERENCES drones(id) ON DELETE CASCADE,
|
||
flight_date TEXT NOT NULL, -- 飞行日期 YYYY-MM-DD
|
||
duration_min INTEGER NOT NULL DEFAULT 0, -- 飞行时长(分钟)
|
||
takeoff TEXT NOT NULL DEFAULT '', -- 起飞地点
|
||
landing TEXT NOT NULL DEFAULT '', -- 降落地点
|
||
remark TEXT NOT NULL DEFAULT '', -- 备注
|
||
created_at TEXT NOT NULL -- 录入时间
|
||
);
|
||
CREATE INDEX IF NOT EXISTS idx_flight_drone ON flight_records(drone_id, flight_date);
|
||
"""
|
||
)
|
||
# 存量库迁移：为已存在的 drones 表补充版本号列
|
||
cols = {r[1] for r in conn.execute("PRAGMA table_info(drones)")}
|
||
if "fc_version" not in cols:
|
||
conn.execute("ALTER TABLE drones ADD COLUMN fc_version TEXT NOT NULL DEFAULT ''")
|
||
if "nav_version" not in cols:
|
||
conn.execute("ALTER TABLE drones ADD COLUMN nav_version TEXT NOT NULL DEFAULT ''")
|
||
# 预置内置状态（幂等）
|
||
for i, (s, c) in enumerate(zip(STATUSES, STATUS_COLORS), start=1):
|
||
conn.execute(
|
||
"INSERT OR IGNORE INTO status_defs (name, color, sort_order, is_builtin) "
|
||
"VALUES (?, ?, ?, 1)",
|
||
(s, c, i),
|
||
)
|
||
|
||
|
||
@asynccontextmanager
|
||
async def lifespan(_app: FastAPI):
|
||
init_db()
|
||
yield
|
||
|
||
|
||
app = FastAPI(title="无人机资产管理系统", lifespan=lifespan)
|
||
|
||
|
||
# ---------------------------------------------------------------- 模型
|
||
class DroneCreate(BaseModel):
|
||
model: str
|
||
serial_no: str
|
||
link_id: str = ""
|
||
fc_id: str = ""
|
||
fc_version: str = ""
|
||
nav_version: str = ""
|
||
remark: str = ""
|
||
|
||
|
||
class DroneUpdate(BaseModel):
|
||
model: Optional[str] = None
|
||
serial_no: Optional[str] = None
|
||
link_id: Optional[str] = None
|
||
fc_id: Optional[str] = None
|
||
fc_version: Optional[str] = None
|
||
nav_version: Optional[str] = None
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class StatusChange(BaseModel):
|
||
status: str
|
||
remark: str = ""
|
||
changed_at: Optional[str] = None # 可选：指定变更时间（补充历史用）；缺省为当前时间
|
||
|
||
|
||
class StatusBatchChange(BaseModel):
|
||
ids: List[int]
|
||
status: str
|
||
remark: str = ""
|
||
changed_at: Optional[str] = None
|
||
|
||
|
||
class StatusRecordUpdate(BaseModel):
|
||
status: Optional[str] = None
|
||
remark: Optional[str] = None
|
||
changed_at: Optional[str] = None
|
||
|
||
|
||
class StatusDefCreate(BaseModel):
|
||
name: str
|
||
color: Optional[str] = None
|
||
|
||
|
||
class StatusDefUpdate(BaseModel):
|
||
name: Optional[str] = None
|
||
color: Optional[str] = None
|
||
|
||
|
||
class FlightCreate(BaseModel):
|
||
flight_date: str
|
||
duration_min: int = 0
|
||
takeoff: str = ""
|
||
landing: str = ""
|
||
remark: str = ""
|
||
|
||
|
||
class FlightUpdate(BaseModel):
|
||
flight_date: Optional[str] = None
|
||
duration_min: Optional[int] = None
|
||
takeoff: Optional[str] = None
|
||
landing: Optional[str] = None
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class FlightBatchCreate(BaseModel):
|
||
serial_nos: List[str]
|
||
flight_date: str
|
||
duration_min: int = 0
|
||
takeoff: str = ""
|
||
landing: str = ""
|
||
remark: str = ""
|
||
|
||
|
||
class FlightBatchUpdate(BaseModel):
|
||
ids: List[int]
|
||
flight_date: str
|
||
duration_min: Optional[int] = None
|
||
takeoff: Optional[str] = None
|
||
landing: Optional[str] = None
|
||
remark: Optional[str] = None
|
||
|
||
|
||
class FlightBatchDelete(BaseModel):
|
||
ids: List[int]
|
||
mode: str = "all" # all=全部 | date=按日期
|
||
flight_date: Optional[str] = None
|
||
|
||
|
||
class DroneBatchDelete(BaseModel):
|
||
ids: List[int]
|
||
|
||
|
||
def _check_date(s: str) -> str:
|
||
"""校验 YYYY-MM-DD 并返回规范化日期。"""
|
||
try:
|
||
return date.fromisoformat(s).isoformat()
|
||
except ValueError:
|
||
raise HTTPException(400, "日期格式应为 YYYY-MM-DD")
|
||
|
||
|
||
def _flight_row(row: sqlite3.Row) -> dict:
|
||
return {
|
||
"id": row["id"],
|
||
"drone_id": row["drone_id"],
|
||
"flight_date": row["flight_date"],
|
||
"duration_min": row["duration_min"],
|
||
"takeoff": row["takeoff"],
|
||
"landing": row["landing"],
|
||
"remark": row["remark"],
|
||
"created_at": row["created_at"],
|
||
}
|
||
|
||
|
||
def _drone_row(row: sqlite3.Row) -> dict:
|
||
return {
|
||
"id": row["id"],
|
||
"model": row["model"],
|
||
"serial_no": row["serial_no"],
|
||
"link_id": row["link_id"],
|
||
"fc_id": row["fc_id"],
|
||
"fc_version": row["fc_version"],
|
||
"nav_version": row["nav_version"],
|
||
"remark": row["remark"],
|
||
"created_at": row["created_at"],
|
||
}
|
||
|
||
|
||
def _current_status(conn: sqlite3.Connection, drone_id: int) -> Optional[dict]:
|
||
row = conn.execute(
|
||
"SELECT status, remark, changed_at FROM drone_status "
|
||
"WHERE drone_id = ? ORDER BY changed_at DESC, id DESC LIMIT 1",
|
||
(drone_id,),
|
||
).fetchone()
|
||
return dict(row) if row else None
|
||
|
||
|
||
# ---------------------------------------------------------------- Excel 台账
|
||
def export_to_excel(
|
||
models: Optional[List[str]] = None,
|
||
start: Optional[str] = None,
|
||
end: Optional[str] = None,
|
||
) -> Optional[str]:
|
||
"""导出 Excel 台账（手动触发）。
|
||
|
||
- 不指定日期：导出「无人机台账」+「状态历史」两个 sheet。
|
||
- 指定日期范围（start/end，YYYY-MM-DD）：导出「每日状态矩阵」
|
||
（行=日期、列=编号、状态按颜色着色）+「状态变更明细」(范围内) +
|
||
「无人机台账」(基础信息) 三个 sheet。
|
||
可按机型筛选（models 缺省导出全部）。失败返回 None。
|
||
"""
|
||
if openpyxl is None:
|
||
return None
|
||
try:
|
||
wb = openpyxl.Workbook()
|
||
|
||
where = ""
|
||
params: tuple = ()
|
||
if models:
|
||
where = f"WHERE d.model IN ({','.join('?' * len(models))})"
|
||
params = tuple(models)
|
||
|
||
with get_conn() as conn:
|
||
rows = conn.execute(
|
||
"SELECT d.*, s.status AS cur_status "
|
||
"FROM drones d "
|
||
"LEFT JOIN drone_status s ON s.id = ("
|
||
" SELECT id FROM drone_status WHERE drone_id = d.id "
|
||
" ORDER BY changed_at DESC, id DESC LIMIT 1) "
|
||
f"{where} "
|
||
"ORDER BY d.model, d.serial_no",
|
||
params,
|
||
).fetchall()
|
||
|
||
has_dates = bool(start and end)
|
||
if has_dates:
|
||
# ---- 日期范围解析 ----
|
||
d0, d1 = date.fromisoformat(start), date.fromisoformat(end)
|
||
if d0 > d1:
|
||
d0, d1 = d1, d0
|
||
dates = []
|
||
d = d0
|
||
while d <= d1:
|
||
dates.append(d.isoformat())
|
||
d += timedelta(days=1)
|
||
first_day = dates[0]
|
||
color_map = {
|
||
x["name"]: x["color"]
|
||
for x in conn.execute("SELECT name, color FROM status_defs").fetchall()
|
||
}
|
||
|
||
# 每架无人机逐日状态（无记录日期沿用前一状态）
|
||
drone_cells = []
|
||
for r in rows:
|
||
hist = conn.execute(
|
||
"SELECT status, changed_at FROM drone_status "
|
||
"WHERE drone_id = ? ORDER BY changed_at ASC, id ASC",
|
||
(r["id"],),
|
||
).fetchall()
|
||
last = ""
|
||
for h in hist:
|
||
if h["changed_at"][:10] < first_day:
|
||
last = h["status"]
|
||
else:
|
||
break
|
||
by_day = {}
|
||
for h in hist:
|
||
by_day[h["changed_at"][:10]] = h["status"]
|
||
cells = []
|
||
for ds in dates:
|
||
if ds in by_day:
|
||
last = by_day[ds]
|
||
cells.append(last)
|
||
drone_cells.append((r, cells))
|
||
|
||
# 范围内状态变更明细
|
||
statuses = conn.execute(
|
||
"SELECT d.serial_no, d.model, s.status, s.remark, s.changed_at "
|
||
"FROM drone_status s JOIN drones d ON d.id = s.drone_id "
|
||
f"{where} AND date(s.changed_at) BETWEEN ? AND ? "
|
||
"ORDER BY s.changed_at, s.id",
|
||
params + (dates[0], dates[-1]),
|
||
).fetchall()
|
||
|
||
# ---- Sheet1: 每日状态矩阵 ----
|
||
ws = wb.active
|
||
ws.title = "每日状态矩阵"
|
||
ws.append(["日期 \\ 编号"] + [dc[0]["serial_no"] for dc in drone_cells])
|
||
for i, ds in enumerate(dates):
|
||
ws.append([ds] + [dc[1][i] for dc in drone_cells])
|
||
for ri in range(2, len(dates) + 2):
|
||
for ci in range(2, len(drone_cells) + 2):
|
||
cell = ws.cell(ri, ci)
|
||
if cell.value:
|
||
col = color_map.get(cell.value, "#9e9e9e").lstrip("#")
|
||
cell.fill = PatternFill("solid", fgColor=col)
|
||
cell.font = Font(color="FFFFFF")
|
||
|
||
# ---- Sheet2: 状态变更明细 ----
|
||
ws2 = wb.create_sheet("状态变更明细")
|
||
ws2.append(["编号", "机型", "状态", "备注", "变更时间"])
|
||
for s in statuses:
|
||
ws2.append([s["serial_no"], s["model"], s["status"],
|
||
s["remark"], s["changed_at"]])
|
||
|
||
# ---- Sheet3: 无人机台账 ----
|
||
ws3 = wb.create_sheet("无人机台账")
|
||
ws3.append(["机型", "编号", "链路ID", "飞控ID", "飞控版本",
|
||
"导航版本", "当前状态", "备注", "创建时间"])
|
||
for r in rows:
|
||
ws3.append([
|
||
r["model"], r["serial_no"], r["link_id"], r["fc_id"],
|
||
r["fc_version"], r["nav_version"],
|
||
r["cur_status"] or "", r["remark"], r["created_at"],
|
||
])
|
||
sheets = [ws, ws2, ws3]
|
||
else:
|
||
# ---- 无日期：台账 + 状态历史 ----
|
||
statuses = conn.execute(
|
||
"SELECT d.serial_no, d.model, s.status, s.remark, s.changed_at "
|
||
"FROM drone_status s JOIN drones d ON d.id = s.drone_id "
|
||
f"{where} "
|
||
"ORDER BY s.changed_at, s.id",
|
||
params,
|
||
).fetchall()
|
||
ws = wb.active
|
||
ws.title = "无人机台账"
|
||
ws.append(["机型", "编号", "链路ID", "飞控ID", "飞控版本",
|
||
"导航版本", "当前状态", "备注", "创建时间"])
|
||
for r in rows:
|
||
ws.append([
|
||
r["model"], r["serial_no"], r["link_id"], r["fc_id"],
|
||
r["fc_version"], r["nav_version"],
|
||
r["cur_status"] or "", r["remark"], r["created_at"],
|
||
])
|
||
ws2 = wb.create_sheet("状态历史")
|
||
ws2.append(["编号", "机型", "状态", "备注", "变更时间"])
|
||
for s in statuses:
|
||
ws2.append([s["serial_no"], s["model"], s["status"],
|
||
s["remark"], s["changed_at"]])
|
||
sheets = [ws, ws2]
|
||
|
||
# ---- 样式 ----
|
||
for sheet in sheets:
|
||
for cell in sheet[1]:
|
||
cell.font = Font(bold=True, color="FFFFFF")
|
||
cell.fill = PatternFill("solid", fgColor="2563EB")
|
||
cell.alignment = Alignment(horizontal="center", vertical="center")
|
||
sheet.freeze_panes = "A2"
|
||
sheet.auto_filter.ref = sheet.dimensions
|
||
for col_idx, col_cells in enumerate(sheet.columns, start=1):
|
||
width = max(len(str(c.value)) if c.value is not None else 0
|
||
for c in col_cells)
|
||
sheet.column_dimensions[get_column_letter(col_idx)].width = min(max(width + 4, 10), 60)
|
||
|
||
wb.save(EXCEL_PATH)
|
||
return str(EXCEL_PATH)
|
||
except Exception as exc: # 导出失败不阻断业务操作
|
||
print(f"[Excel 导出失败] {exc}", flush=True)
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- 无人机 CRUD
|
||
@app.get("/api/drones")
|
||
def list_drones(
|
||
model: Optional[str] = Query(None),
|
||
serial_no: Optional[str] = Query(None),
|
||
fc_id: Optional[str] = Query(None),
|
||
status: Optional[str] = Query(None),
|
||
):
|
||
"""无人机列表，支持机型/编号/飞控ID/状态（当前状态）筛选。"""
|
||
with get_conn() as conn:
|
||
sql = "SELECT * FROM drones WHERE 1=1"
|
||
params: list = []
|
||
if model:
|
||
sql += " AND model = ?"
|
||
params.append(model)
|
||
if serial_no:
|
||
sql += " AND serial_no LIKE ?"
|
||
params.append(f"%{serial_no}%")
|
||
if fc_id:
|
||
sql += " AND fc_id LIKE ?"
|
||
params.append(f"%{fc_id}%")
|
||
sql += " ORDER BY id"
|
||
rows = conn.execute(sql, params).fetchall()
|
||
result = []
|
||
for r in rows:
|
||
cur = _current_status(conn, r["id"])
|
||
if status == NO_STATUS:
|
||
if cur:
|
||
continue
|
||
elif status:
|
||
if not cur or cur["status"] != status:
|
||
continue
|
||
item = _drone_row(r)
|
||
item["current_status"] = cur
|
||
result.append(item)
|
||
return result
|
||
|
||
|
||
@app.get("/api/drones/models")
|
||
def list_models():
|
||
"""所有机型（供筛选/透视表下拉框）。"""
|
||
with get_conn() as conn:
|
||
rows = conn.execute(
|
||
"SELECT DISTINCT model FROM drones ORDER BY model"
|
||
).fetchall()
|
||
return [r["model"] for r in rows]
|
||
|
||
|
||
@app.post("/api/drones", status_code=201)
|
||
def create_drone(body: DroneCreate):
|
||
model = body.model.strip()
|
||
serial_no = body.serial_no.strip()
|
||
if not model or not serial_no:
|
||
raise HTTPException(400, "机型与编号不能为空")
|
||
with get_conn() as conn:
|
||
exists = conn.execute(
|
||
"SELECT 1 FROM drones WHERE serial_no = ?", (serial_no,)
|
||
).fetchone()
|
||
if exists:
|
||
raise HTTPException(409, f"编号 {serial_no} 已存在")
|
||
cur = conn.execute(
|
||
"INSERT INTO drones (model, serial_no, link_id, fc_id, fc_version, nav_version, remark, created_at) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
|
||
(model, serial_no, body.link_id.strip(), body.fc_id.strip(),
|
||
body.fc_version.strip(), body.nav_version.strip(),
|
||
body.remark.strip(), date.today().isoformat()),
|
||
)
|
||
return _drone_row(
|
||
conn.execute("SELECT * FROM drones WHERE id = ?", (cur.lastrowid,)).fetchone()
|
||
)
|
||
|
||
|
||
@app.put("/api/drones/{drone_id}")
|
||
def update_drone(drone_id: int, body: DroneUpdate):
|
||
with get_conn() as conn:
|
||
row = conn.execute("SELECT * FROM drones WHERE id = ?", (drone_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "无人机不存在")
|
||
new_serial = body.serial_no.strip() if body.serial_no is not None else row["serial_no"]
|
||
new_model = body.model.strip() if body.model is not None else row["model"]
|
||
new_link = body.link_id.strip() if body.link_id is not None else row["link_id"]
|
||
new_fc = body.fc_id.strip() if body.fc_id is not None else row["fc_id"]
|
||
new_fcv = body.fc_version.strip() if body.fc_version is not None else row["fc_version"]
|
||
new_nav = body.nav_version.strip() if body.nav_version is not None else row["nav_version"]
|
||
new_remark = body.remark.strip() if body.remark is not None else row["remark"]
|
||
if not new_serial:
|
||
raise HTTPException(400, "编号不能为空")
|
||
if new_serial != row["serial_no"]:
|
||
exists = conn.execute(
|
||
"SELECT 1 FROM drones WHERE serial_no = ? AND id != ?",
|
||
(new_serial, drone_id),
|
||
).fetchone()
|
||
if exists:
|
||
raise HTTPException(409, f"编号 {new_serial} 已存在")
|
||
conn.execute(
|
||
"UPDATE drones SET model = ?, serial_no = ?, link_id = ?, fc_id = ?, "
|
||
"fc_version = ?, nav_version = ?, remark = ? WHERE id = ?",
|
||
(new_model, new_serial, new_link, new_fc, new_fcv, new_nav, new_remark, drone_id),
|
||
)
|
||
return _drone_row(
|
||
conn.execute("SELECT * FROM drones WHERE id = ?", (drone_id,)).fetchone()
|
||
)
|
||
|
||
|
||
@app.delete("/api/drones/{drone_id}", status_code=204)
|
||
def delete_drone(drone_id: int):
|
||
with get_conn() as conn:
|
||
row = conn.execute("SELECT 1 FROM drones WHERE id = ?", (drone_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "无人机不存在")
|
||
conn.execute("DELETE FROM drones WHERE id = ?", (drone_id,)) # 级联删除状态
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- 状态
|
||
@app.post("/api/drones/{drone_id}/status", status_code=201)
|
||
def change_status(drone_id: int, body: StatusChange):
|
||
"""修改主状态：写入一条含时间戳的记录。"""
|
||
status = body.status.strip()
|
||
with get_conn() as conn:
|
||
def_row = conn.execute(
|
||
"SELECT 1 FROM status_defs WHERE name = ?", (status,)
|
||
).fetchone()
|
||
if not def_row:
|
||
raise HTTPException(400, f"无效状态：{status} 不在状态列表中")
|
||
row = conn.execute("SELECT 1 FROM drones WHERE id = ?", (drone_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "无人机不存在")
|
||
now = datetime.now().isoformat(timespec="seconds")
|
||
ts = now
|
||
if body.changed_at:
|
||
raw = body.changed_at.strip().replace(" ", "T")
|
||
try:
|
||
dt = datetime.fromisoformat(raw)
|
||
ts = dt.strftime("%Y-%m-%dT%H:%M:%S")
|
||
except ValueError:
|
||
raise HTTPException(
|
||
400, "changed_at 格式无效，应为 YYYY-MM-DD 或 YYYY-MM-DD HH:MM[:SS]"
|
||
)
|
||
conn.execute(
|
||
"INSERT INTO drone_status (drone_id, status, remark, changed_at) VALUES (?, ?, ?, ?)",
|
||
(drone_id, status, body.remark.strip(), ts),
|
||
)
|
||
return {"drone_id": drone_id, "status": status, "remark": body.remark.strip(), "changed_at": ts}
|
||
|
||
|
||
@app.post("/api/drones/status/batch", status_code=201)
|
||
def batch_change_status(body: StatusBatchChange):
|
||
"""批量修改状态：一次给多架无人机写入同一条状态记录（支持补录时间）。"""
|
||
status = body.status.strip()
|
||
ids = list(dict.fromkeys(body.ids))
|
||
if not ids:
|
||
raise HTTPException(400, "请至少选择一架无人机")
|
||
with get_conn() as conn:
|
||
def_row = conn.execute(
|
||
"SELECT 1 FROM status_defs WHERE name = ?", (status,)
|
||
).fetchone()
|
||
if not def_row:
|
||
raise HTTPException(400, f"无效状态：{status} 不在状态列表中")
|
||
if body.changed_at:
|
||
raw = body.changed_at.strip().replace(" ", "T")
|
||
try:
|
||
ts = datetime.fromisoformat(raw).strftime("%Y-%m-%dT%H:%M:%S")
|
||
except ValueError:
|
||
raise HTTPException(
|
||
400, "changed_at 格式无效，应为 YYYY-MM-DD 或 YYYY-MM-DD HH:MM[:SS]"
|
||
)
|
||
else:
|
||
ts = datetime.now().isoformat(timespec="seconds")
|
||
remark = body.remark.strip()
|
||
updated, missing = [], []
|
||
for did in ids:
|
||
row = conn.execute("SELECT 1 FROM drones WHERE id = ?", (did,)).fetchone()
|
||
if not row:
|
||
missing.append(did)
|
||
continue
|
||
conn.execute(
|
||
"INSERT INTO drone_status (drone_id, status, remark, changed_at) VALUES (?, ?, ?, ?)",
|
||
(did, status, remark, ts),
|
||
)
|
||
updated.append(did)
|
||
return {"updated": updated, "missing": missing}
|
||
|
||
|
||
@app.get("/api/drones/{drone_id}/history")
|
||
def status_history(drone_id: int):
|
||
"""单机完整状态变更记录（时间正序）。"""
|
||
with get_conn() as conn:
|
||
row = conn.execute("SELECT * FROM drones WHERE id = ?", (drone_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "无人机不存在")
|
||
rows = conn.execute(
|
||
"SELECT id, status, remark, changed_at FROM drone_status "
|
||
"WHERE drone_id = ? ORDER BY changed_at ASC, id ASC",
|
||
(drone_id,),
|
||
).fetchall()
|
||
drone = _drone_row(row)
|
||
drone["current_status"] = _current_status(conn, drone_id)
|
||
return {"drone": drone, "history": [dict(r) for r in rows]}
|
||
|
||
|
||
@app.delete("/api/drones/{drone_id}/history/{record_id}", status_code=204)
|
||
def delete_status_record(drone_id: int, record_id: int):
|
||
"""删除单条状态历史记录（删除后当前状态取剩余最新一条）。"""
|
||
with get_conn() as conn:
|
||
row = conn.execute(
|
||
"SELECT 1 FROM drone_status WHERE id = ? AND drone_id = ?",
|
||
(record_id, drone_id),
|
||
).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "状态记录不存在")
|
||
conn.execute("DELETE FROM drone_status WHERE id = ?", (record_id,))
|
||
return None
|
||
|
||
|
||
@app.put("/api/drones/{drone_id}/history/{record_id}")
|
||
def update_status_record(drone_id: int, record_id: int, body: StatusRecordUpdate):
|
||
"""修改单条状态历史记录（状态/备注/变更时间；None 字段不修改）。"""
|
||
with get_conn() as conn:
|
||
rec = conn.execute(
|
||
"SELECT * FROM drone_status WHERE id = ? AND drone_id = ?",
|
||
(record_id, drone_id),
|
||
).fetchone()
|
||
if not rec:
|
||
raise HTTPException(404, "状态记录不存在")
|
||
new_status = body.status.strip() if body.status is not None else rec["status"]
|
||
def_row = conn.execute(
|
||
"SELECT 1 FROM status_defs WHERE name = ?", (new_status,)
|
||
).fetchone()
|
||
if not def_row:
|
||
raise HTTPException(400, f"无效状态：{new_status} 不在状态列表中")
|
||
new_remark = body.remark if body.remark is not None else rec["remark"]
|
||
new_ts = rec["changed_at"]
|
||
if body.changed_at:
|
||
raw = body.changed_at.strip().replace(" ", "T")
|
||
try:
|
||
new_ts = datetime.fromisoformat(raw).strftime("%Y-%m-%dT%H:%M:%S")
|
||
except ValueError:
|
||
raise HTTPException(
|
||
400, "changed_at 格式无效，应为 YYYY-MM-DD 或 YYYY-MM-DD HH:MM[:SS]"
|
||
)
|
||
conn.execute(
|
||
"UPDATE drone_status SET status = ?, remark = ?, changed_at = ? WHERE id = ?",
|
||
(new_status, new_remark, new_ts, record_id),
|
||
)
|
||
return {
|
||
"id": record_id,
|
||
"status": new_status,
|
||
"remark": new_remark,
|
||
"changed_at": new_ts,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------- 透视表
|
||
@app.get("/api/status-matrix")
|
||
def status_matrix(
|
||
model: str = Query(...),
|
||
serial_no: Optional[str] = Query(None),
|
||
fc_id: Optional[str] = Query(None),
|
||
status: Optional[str] = Query(None),
|
||
start: Optional[str] = Query(None),
|
||
end: Optional[str] = Query(None),
|
||
):
|
||
"""按机型返回日期×编号的状态矩阵（支持编号/飞控ID/当前状态筛选列）。
|
||
|
||
dates 为行（日期），drones[].statuses 与 dates 对齐；
|
||
单元格 = 截至该日 23:59 的最新状态，无记录日期沿用前一状态，
|
||
从未有过状态则为空字符串。
|
||
"""
|
||
try:
|
||
end_d = date.fromisoformat(end) if end else date.today()
|
||
start_d = date.fromisoformat(start) if start else end_d - timedelta(days=29)
|
||
except ValueError:
|
||
raise HTTPException(400, "日期格式应为 YYYY-MM-DD")
|
||
if start_d > end_d:
|
||
start_d, end_d = end_d, start_d
|
||
dates = []
|
||
d = start_d
|
||
while d <= end_d:
|
||
dates.append(d.isoformat())
|
||
d += timedelta(days=1)
|
||
first_day = dates[0]
|
||
|
||
with get_conn() as conn:
|
||
sql = "SELECT * FROM drones WHERE model = ?"
|
||
sql_params: list = [model]
|
||
if serial_no:
|
||
sql += " AND serial_no LIKE ?"
|
||
sql_params.append(f"%{serial_no}%")
|
||
if fc_id:
|
||
sql += " AND fc_id LIKE ?"
|
||
sql_params.append(f"%{fc_id}%")
|
||
sql += " ORDER BY serial_no"
|
||
rows = conn.execute(sql, sql_params).fetchall()
|
||
drones = []
|
||
for r in rows:
|
||
cur = _current_status(conn, r["id"])
|
||
if status == NO_STATUS:
|
||
if cur:
|
||
continue
|
||
elif status:
|
||
if not cur or cur["status"] != status:
|
||
continue
|
||
history = conn.execute(
|
||
"SELECT status, remark, changed_at FROM drone_status "
|
||
"WHERE drone_id = ? ORDER BY changed_at ASC, id ASC",
|
||
(r["id"],),
|
||
).fetchall()
|
||
# 初始值 = 范围开始日之前的最新状态（含备注）
|
||
last_status, last_remark = "", ""
|
||
for h in history:
|
||
if h["changed_at"][:10] < first_day:
|
||
last_status, last_remark = h["status"], h["remark"]
|
||
else:
|
||
break
|
||
by_day = {}
|
||
by_day_remark = {}
|
||
for h in history:
|
||
by_day[h["changed_at"][:10]] = h["status"] # 升序覆盖，保留当日最后一条
|
||
by_day_remark[h["changed_at"][:10]] = h["remark"]
|
||
cells = []
|
||
remark_cells = []
|
||
for ds in dates:
|
||
if ds in by_day:
|
||
last_status = by_day[ds]
|
||
last_remark = by_day_remark[ds]
|
||
cells.append(last_status)
|
||
remark_cells.append(last_remark)
|
||
drones.append(
|
||
{
|
||
"id": r["id"],
|
||
"serial_no": r["serial_no"],
|
||
"model": r["model"],
|
||
"statuses": cells,
|
||
"remarks": remark_cells,
|
||
}
|
||
)
|
||
return {
|
||
"model": model,
|
||
"start": dates[0],
|
||
"end": dates[-1],
|
||
"dates": dates,
|
||
"drones": drones,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------- 飞行历史
|
||
@app.get("/api/drones/{drone_id}/flights")
|
||
def list_flights(drone_id: int):
|
||
"""单机飞行历史列表（按飞行日期倒序）。"""
|
||
with get_conn() as conn:
|
||
row = conn.execute("SELECT * FROM drones WHERE id = ?", (drone_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "无人机不存在")
|
||
rows = conn.execute(
|
||
"SELECT * FROM flight_records WHERE drone_id = ? "
|
||
"ORDER BY flight_date DESC, id DESC",
|
||
(drone_id,),
|
||
).fetchall()
|
||
drone = _drone_row(row)
|
||
return {"drone": drone, "flights": [_flight_row(r) for r in rows]}
|
||
|
||
|
||
@app.post("/api/drones/{drone_id}/flights", status_code=201)
|
||
def create_flight(drone_id: int, body: FlightCreate):
|
||
"""新增一条飞行记录。"""
|
||
flight_date = _check_date(body.flight_date)
|
||
if body.duration_min < 0:
|
||
raise HTTPException(400, "飞行时长不能为负数")
|
||
with get_conn() as conn:
|
||
row = conn.execute("SELECT 1 FROM drones WHERE id = ?", (drone_id,)).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, "无人机不存在")
|
||
now = datetime.now().isoformat(timespec="seconds")
|
||
cur = conn.execute(
|
||
"INSERT INTO flight_records (drone_id, flight_date, duration_min, takeoff, landing, remark, created_at) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||
(drone_id, flight_date, body.duration_min, body.takeoff.strip(),
|
||
body.landing.strip(), body.remark.strip(), now),
|
||
)
|
||
rec = conn.execute(
|
||
"SELECT * FROM flight_records WHERE id = ?", (cur.lastrowid,)
|
||
).fetchone()
|
||
return _flight_row(rec)
|
||
|
||
|
||
@app.put("/api/drones/{drone_id}/flights/{flight_id}")
|
||
def update_flight(drone_id: int, flight_id: int, body: FlightUpdate):
|
||
"""修改飞行记录。"""
|
||
with get_conn() as conn:
|
||
rec = conn.execute(
|
||
"SELECT * FROM flight_records WHERE id = ? AND drone_id = ?",
|
||
(flight_id, drone_id),
|
||
).fetchone()
|
||
if not rec:
|
||
raise HTTPException(404, "飞行记录不存在")
|
||
new_date = _check_date(body.flight_date) if body.flight_date is not None else rec["flight_date"]
|
||
new_dur = body.duration_min if body.duration_min is not None else rec["duration_min"]
|
||
if new_dur < 0:
|
||
raise HTTPException(400, "飞行时长不能为负数")
|
||
new_takeoff = body.takeoff.strip() if body.takeoff is not None else rec["takeoff"]
|
||
new_landing = body.landing.strip() if body.landing is not None else rec["landing"]
|
||
new_remark = body.remark.strip() if body.remark is not None else rec["remark"]
|
||
conn.execute(
|
||
"UPDATE flight_records SET flight_date = ?, duration_min = ?, takeoff = ?, "
|
||
"landing = ?, remark = ? WHERE id = ?",
|
||
(new_date, new_dur, new_takeoff, new_landing, new_remark, flight_id),
|
||
)
|
||
rec = conn.execute(
|
||
"SELECT * FROM flight_records WHERE id = ?", (flight_id,)
|
||
).fetchone()
|
||
return _flight_row(rec)
|
||
|
||
|
||
@app.delete("/api/drones/{drone_id}/flights/{flight_id}", status_code=204)
|
||
def delete_flight(drone_id: int, flight_id: int):
|
||
"""删除飞行记录。"""
|
||
with get_conn() as conn:
|
||
rec = conn.execute(
|
||
"SELECT 1 FROM flight_records WHERE id = ? AND drone_id = ?",
|
||
(flight_id, drone_id),
|
||
).fetchone()
|
||
if not rec:
|
||
raise HTTPException(404, "飞行记录不存在")
|
||
conn.execute("DELETE FROM flight_records WHERE id = ?", (flight_id,))
|
||
return None
|
||
|
||
|
||
@app.post("/api/flights/batch", status_code=201)
|
||
def batch_create_flights(body: FlightBatchCreate):
|
||
"""统一添加飞行历史：一次给多个飞机编号添加同一条飞行记录。"""
|
||
flight_date = _check_date(body.flight_date)
|
||
if body.duration_min < 0:
|
||
raise HTTPException(400, "飞行时长不能为负数")
|
||
serials = [s.strip() for s in body.serial_nos if s.strip()]
|
||
if not serials:
|
||
raise HTTPException(400, "请至少选择一个飞机编号")
|
||
now = datetime.now().isoformat(timespec="seconds")
|
||
added, missing = [], []
|
||
with get_conn() as conn:
|
||
for s in serials:
|
||
row = conn.execute(
|
||
"SELECT id FROM drones WHERE serial_no = ?", (s,)
|
||
).fetchone()
|
||
if not row:
|
||
missing.append(s)
|
||
continue
|
||
conn.execute(
|
||
"INSERT INTO flight_records (drone_id, flight_date, duration_min, takeoff, landing, remark, created_at) "
|
||
"VALUES (?, ?, ?, ?, ?, ?, ?)",
|
||
(row["id"], flight_date, body.duration_min, body.takeoff.strip(),
|
||
body.landing.strip(), body.remark.strip(), now),
|
||
)
|
||
added.append(s)
|
||
return {"added": added, "missing": missing}
|
||
|
||
|
||
@app.get("/api/flight-matrix")
|
||
def flight_matrix(
|
||
model: str = Query(...),
|
||
serial_no: Optional[str] = Query(None),
|
||
fc_id: Optional[str] = Query(None),
|
||
start: Optional[str] = Query(None),
|
||
end: Optional[str] = Query(None),
|
||
):
|
||
"""飞行历史透视表：行=日期、列=编号、单元格=当日飞行次数（支持编号/飞控ID筛选列）。"""
|
||
try:
|
||
end_d = date.fromisoformat(end) if end else date.today()
|
||
start_d = date.fromisoformat(start) if start else end_d - timedelta(days=29)
|
||
except ValueError:
|
||
raise HTTPException(400, "日期格式应为 YYYY-MM-DD")
|
||
if start_d > end_d:
|
||
start_d, end_d = end_d, start_d
|
||
dates = []
|
||
d = start_d
|
||
while d <= end_d:
|
||
dates.append(d.isoformat())
|
||
d += timedelta(days=1)
|
||
with get_conn() as conn:
|
||
sql = "SELECT id, serial_no FROM drones WHERE model = ?"
|
||
sql_params: list = [model]
|
||
if serial_no:
|
||
sql += " AND serial_no LIKE ?"
|
||
sql_params.append(f"%{serial_no}%")
|
||
if fc_id:
|
||
sql += " AND fc_id LIKE ?"
|
||
sql_params.append(f"%{fc_id}%")
|
||
sql += " ORDER BY serial_no"
|
||
drones = conn.execute(sql, sql_params).fetchall()
|
||
result = []
|
||
for r in drones:
|
||
counts = {
|
||
x["flight_date"]: x["c"]
|
||
for x in conn.execute(
|
||
"SELECT flight_date, COUNT(*) AS c FROM flight_records "
|
||
"WHERE drone_id = ? AND flight_date BETWEEN ? AND ? "
|
||
"GROUP BY flight_date",
|
||
(r["id"], dates[0], dates[-1]),
|
||
)
|
||
}
|
||
# 每日期备注（当日多条用；拼接）
|
||
remarks: dict = {}
|
||
for x in conn.execute(
|
||
"SELECT flight_date, remark FROM flight_records "
|
||
"WHERE drone_id = ? AND flight_date BETWEEN ? AND ? "
|
||
"ORDER BY flight_date, id",
|
||
(r["id"], dates[0], dates[-1]),
|
||
):
|
||
rm = (x["remark"] or "").strip()
|
||
if rm:
|
||
remarks.setdefault(x["flight_date"], []).append(rm)
|
||
remark_cells = []
|
||
for ds in dates:
|
||
lst = remarks.get(ds)
|
||
remark_cells.append("；".join(lst) if lst else "")
|
||
result.append({
|
||
"id": r["id"],
|
||
"serial_no": r["serial_no"],
|
||
"counts": [counts.get(ds, 0) for ds in dates],
|
||
"remarks": remark_cells,
|
||
})
|
||
return {"model": model, "start": dates[0], "end": dates[-1], "dates": dates, "drones": result}
|
||
|
||
|
||
@app.post("/api/flights/batch-update")
|
||
def batch_update_flights(body: FlightBatchUpdate):
|
||
"""按日期批量修改：对所选飞机「该日期」的所有飞行记录统一更新（None 字段不修改）。"""
|
||
ids = list(dict.fromkeys(body.ids))
|
||
if not ids:
|
||
raise HTTPException(400, "请至少选择一架无人机")
|
||
flight_date = _check_date(body.flight_date)
|
||
if body.duration_min is not None and body.duration_min < 0:
|
||
raise HTTPException(400, "飞行时长不能为负数")
|
||
sets, params = [], []
|
||
if body.duration_min is not None:
|
||
sets.append("duration_min = ?")
|
||
params.append(body.duration_min)
|
||
if body.takeoff is not None:
|
||
sets.append("takeoff = ?")
|
||
params.append(body.takeoff.strip())
|
||
if body.landing is not None:
|
||
sets.append("landing = ?")
|
||
params.append(body.landing.strip())
|
||
if body.remark is not None:
|
||
sets.append("remark = ?")
|
||
params.append(body.remark.strip())
|
||
if not sets:
|
||
raise HTTPException(400, "请至少填写一个要修改的字段")
|
||
ph = ",".join("?" * len(ids))
|
||
params.extend(ids)
|
||
params.append(flight_date)
|
||
with get_conn() as conn:
|
||
cur = conn.execute(
|
||
f"UPDATE flight_records SET {', '.join(sets)} "
|
||
f"WHERE drone_id IN ({ph}) AND flight_date = ?",
|
||
params,
|
||
)
|
||
return {"updated_records": cur.rowcount}
|
||
|
||
|
||
@app.post("/api/flights/batch-delete")
|
||
def batch_delete_flights(body: FlightBatchDelete):
|
||
"""批量删除飞行记录：mode=all 删除全部，mode=date 按日期删除。"""
|
||
ids = list(dict.fromkeys(body.ids))
|
||
if not ids:
|
||
raise HTTPException(400, "请至少选择一架无人机")
|
||
sql = f"DELETE FROM flight_records WHERE drone_id IN ({','.join('?' * len(ids))})"
|
||
params: list = ids.copy()
|
||
if body.mode == "date":
|
||
if not body.flight_date:
|
||
raise HTTPException(400, "按日期删除需提供 flight_date")
|
||
sql += " AND flight_date = ?"
|
||
params.append(_check_date(body.flight_date))
|
||
elif body.mode != "all":
|
||
raise HTTPException(400, "mode 应为 all 或 date")
|
||
with get_conn() as conn:
|
||
cur = conn.execute(sql, params)
|
||
return {"deleted_records": cur.rowcount}
|
||
|
||
|
||
@app.post("/api/drones/batch-delete")
|
||
def batch_delete_drones(body: DroneBatchDelete):
|
||
"""批量删除整架无人机（级联删除其状态与飞行记录）。"""
|
||
ids = list(dict.fromkeys(body.ids))
|
||
if not ids:
|
||
raise HTTPException(400, "请至少选择一架无人机")
|
||
deleted, not_found = [], []
|
||
with get_conn() as conn:
|
||
for did in ids:
|
||
row = conn.execute("SELECT 1 FROM drones WHERE id = ?", (did,)).fetchone()
|
||
if not row:
|
||
not_found.append(did)
|
||
continue
|
||
conn.execute("DELETE FROM drones WHERE id = ?", (did,))
|
||
deleted.append(did)
|
||
return {"deleted": deleted, "not_found": not_found}
|
||
|
||
|
||
# ---------------------------------------------------------------- 状态管理（自定义状态）
|
||
@app.get("/api/status-types")
|
||
def list_status_types():
|
||
"""全部可用状态（含颜色与是否内置），按排序返回。"""
|
||
with get_conn() as conn:
|
||
rows = conn.execute(
|
||
"SELECT name, color, is_builtin FROM status_defs ORDER BY sort_order, id"
|
||
).fetchall()
|
||
return [
|
||
{"name": r["name"], "color": r["color"], "is_builtin": bool(r["is_builtin"])}
|
||
for r in rows
|
||
]
|
||
|
||
|
||
@app.post("/api/status-types", status_code=201)
|
||
def create_status_type(body: StatusDefCreate):
|
||
"""新增自定义状态；不填颜色时自动从调色板分配。"""
|
||
name = body.name.strip()
|
||
if not name:
|
||
raise HTTPException(400, "状态名称不能为空")
|
||
with get_conn() as conn:
|
||
exists = conn.execute(
|
||
"SELECT 1 FROM status_defs WHERE name = ?", (name,)
|
||
).fetchone()
|
||
if exists:
|
||
raise HTTPException(409, f"状态 {name} 已存在")
|
||
color = (body.color or "").strip()
|
||
if not color:
|
||
used = {r[0] for r in conn.execute("SELECT color FROM status_defs")}
|
||
color = next((c for c in DEFAULT_STATUS_COLORS if c not in used), "#64748b")
|
||
conn.execute(
|
||
"INSERT INTO status_defs (name, color, sort_order, is_builtin) VALUES (?, ?, "
|
||
"(SELECT COALESCE(MAX(sort_order), 0) + 1 FROM status_defs), 0)",
|
||
(name, color),
|
||
)
|
||
return {"name": name, "color": color, "is_builtin": False}
|
||
|
||
|
||
@app.put("/api/status-types/{name}")
|
||
def update_status_type(name: str, body: StatusDefUpdate):
|
||
"""改名 / 改颜色；改名会同步更新历史记录中的旧状态名。"""
|
||
with get_conn() as conn:
|
||
row = conn.execute(
|
||
"SELECT * FROM status_defs WHERE name = ?", (name,)
|
||
).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, f"状态 {name} 不存在")
|
||
new_name = body.name.strip() if body.name is not None else name
|
||
new_color = body.color.strip() if body.color is not None else row["color"]
|
||
if not new_name:
|
||
raise HTTPException(400, "状态名称不能为空")
|
||
if new_name != name:
|
||
dup = conn.execute(
|
||
"SELECT 1 FROM status_defs WHERE name = ? AND id != ?",
|
||
(new_name, row["id"]),
|
||
).fetchone()
|
||
if dup:
|
||
raise HTTPException(409, f"状态 {new_name} 已存在")
|
||
conn.execute(
|
||
"UPDATE status_defs SET name = ?, color = ? WHERE id = ?",
|
||
(new_name, new_color, row["id"]),
|
||
)
|
||
if new_name != name:
|
||
conn.execute(
|
||
"UPDATE drone_status SET status = ? WHERE status = ?",
|
||
(new_name, name),
|
||
)
|
||
return {"name": new_name, "color": new_color, "is_builtin": bool(row["is_builtin"])}
|
||
|
||
|
||
@app.delete("/api/status-types/{name}", status_code=204)
|
||
def delete_status_type(name: str):
|
||
"""删除自定义状态；内置状态与已被记录使用的状态不可删除。"""
|
||
with get_conn() as conn:
|
||
row = conn.execute(
|
||
"SELECT * FROM status_defs WHERE name = ?", (name,)
|
||
).fetchone()
|
||
if not row:
|
||
raise HTTPException(404, f"状态 {name} 不存在")
|
||
if row["is_builtin"]:
|
||
raise HTTPException(400, "内置状态不可删除")
|
||
used = conn.execute(
|
||
"SELECT COUNT(*) FROM drone_status WHERE status = ?", (name,)
|
||
).fetchone()[0]
|
||
if used:
|
||
raise HTTPException(400, f"状态 {name} 已被 {used} 条记录使用，无法删除")
|
||
conn.execute("DELETE FROM status_defs WHERE id = ?", (row["id"],))
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------- Excel 导出
|
||
@app.get("/api/export-excel")
|
||
def export_excel_api(
|
||
models: Optional[List[str]] = Query(None),
|
||
start: Optional[str] = Query(None),
|
||
end: Optional[str] = Query(None),
|
||
):
|
||
"""手动导出：生成并下载 Excel。
|
||
|
||
- models：机型筛选（可重复传，不传=全部）
|
||
- start/end：日期范围（YYYY-MM-DD）。不传日期导出「台账+历史」；
|
||
传入日期导出「每日状态矩阵+变更明细+台账」。
|
||
"""
|
||
try:
|
||
if (start and not end) or (end and not start):
|
||
raise HTTPException(400, "start 与 end 需同时提供")
|
||
except HTTPException:
|
||
raise
|
||
path = export_to_excel(models=models, start=start, end=end)
|
||
if not path or not Path(path).exists():
|
||
raise HTTPException(500, "Excel 导出失败，请检查 openpyxl 是否已安装")
|
||
parts = []
|
||
if models:
|
||
parts.append("_".join(models))
|
||
if start and end:
|
||
parts.append(f"{start}-{end}")
|
||
fname = "无人机台账.xlsx"
|
||
if parts:
|
||
safe = "".join(c if c not in r'\/:*?"<>|' else "_" for c in "_".join(parts))
|
||
fname = f"无人机台账-{safe}.xlsx"
|
||
return FileResponse(
|
||
path,
|
||
filename=fname,
|
||
media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------- 静态页面
|
||
class NoCacheStaticFiles(StaticFiles):
|
||
"""静态文件禁用缓存：每次请求都让浏览器重新校验（避免改版后看不到更新）。"""
|
||
|
||
def file_response(self, *args, **kwargs):
|
||
resp = super().file_response(*args, **kwargs)
|
||
resp.headers["Cache-Control"] = "no-cache"
|
||
return resp
|
||
|
||
|
||
app.mount("/", NoCacheStaticFiles(directory=STATIC_DIR, html=True), name="static")
|