Compare commits

..

1 Commits

3 changed files with 82 additions and 8 deletions
+1
View File
@@ -39,3 +39,4 @@ Thumbs.db
# ===== 工具状态 / 临时文件 =====
.reasonix/
cmd_out.txt
uav.db.zip
+1 -1
View File
@@ -14,7 +14,7 @@ BS 架构的无人机资产管理库:管理无人机基本信息(机型、
- **飞行透视表**:行=日期、列=编号,单元格=当日飞行次数,支持机型与日期范围筛选
- **按机型透视表**:行 = 日期、列 = 飞机编号、单元格 = 当日状态;无变更日期自动沿用前一状态;支持自定义日期范围(默认近 30 天)
- **状态颜色区分**:待组装=灰、待调试=橙、待飞=绿、损坏=红
- **手动导出 Excel**:点击「导出 Excel」按钮,下载 `无人机台账.xlsx`(含"无人机台账"与"状态历史"两个工作表),数据以点击时刻为准
- **手动导出 Excel**:点击「导出 Excel」按钮,下载 `无人机台账.xlsx`;不含日期时导出「无人机台账(含每架飞行次数)+ 状态历史 + 飞行历史」,选择日期时导出「每日状态矩阵 + 状态变更明细 + 无人机台账 + 飞行历史」;飞行历史 sheet 行=飞机编号、首列=飞行次数总计、后列=每日飞行记录;数据以点击时刻为准
## 技术栈
+80 -7
View File
@@ -251,6 +251,39 @@ def _current_status(conn: sqlite3.Connection, drone_id: int) -> Optional[dict]:
# ---------------------------------------------------------------- Excel 台账
def _flights_day_map(flight_rows) -> dict:
"""将飞行记录行按 (drone_id -> {date: [摘要]}) 分组,同日多条保留。"""
day_map: dict = {}
for f in flight_rows:
parts = []
if f["duration_min"]:
parts.append(f"{f['duration_min']}分钟")
if f["takeoff"] or f["landing"]:
parts.append(f"{f['takeoff']}→{f['landing']}")
if f["remark"]:
parts.append(f["remark"])
summary = " ".join(parts) if parts else "飞行"
day_map.setdefault(f["drone_id"], {}).setdefault(f["flight_date"], []).append(summary)
return day_map
def _write_flight_history_sheet(ws, rows, dates, day_map) -> None:
"""写入「飞行历史」sheet:行=飞机编号,首列=次数总计,后列=每日飞行记录。"""
ws.append(["飞机编号", "飞行次数总计"] + list(dates))
for r in rows:
dm = day_map.get(r["id"], {})
total = sum(len(v) for v in dm.values())
row_vals = [r["serial_no"], total]
for ds in dates:
row_vals.append("\n".join(dm.get(ds, [])))
ws.append(row_vals)
# 每日单元格换行显示
for row in ws.iter_rows(min_row=2, min_col=3):
for cell in row:
if cell.value:
cell.alignment = Alignment(vertical="center", wrap_text=True)
def export_to_excel(
models: Optional[List[str]] = None,
start: Optional[str] = None,
@@ -361,16 +394,31 @@ def export_to_excel(
# ---- Sheet3: 无人机台账 ----
ws3 = wb.create_sheet("无人机台账")
ws3.append(["机型", "编号", "链路ID", "飞控ID", "飞控版本",
"导航版本", "当前状态", "备注", "创建时间"])
"导航版本", "当前状态", "飞行次数", "备注", "创建时间"])
for r in rows:
cnt = conn.execute(
"SELECT COUNT(*) FROM flight_records "
"WHERE drone_id = ? AND flight_date BETWEEN ? AND ?",
(r["id"], dates[0], dates[-1]),
).fetchone()[0]
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"],
r["cur_status"] or "", cnt, r["remark"], r["created_at"],
])
sheets = [ws, ws2, ws3]
# ---- Sheet4: 飞行历史(范围内) ----
flight_rows = conn.execute(
"SELECT f.drone_id, f.flight_date, f.duration_min, f.takeoff, f.landing, f.remark "
"FROM flight_records f JOIN drones d ON d.id = f.drone_id "
f"{where} AND f.flight_date BETWEEN ? AND ? "
"ORDER BY f.drone_id, f.flight_date, f.id",
params + (dates[0], dates[-1]),
).fetchall()
ws4 = wb.create_sheet("飞行历史")
_write_flight_history_sheet(ws4, rows, dates, _flights_day_map(flight_rows))
sheets = [ws, ws2, ws3, ws4]
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 "
@@ -378,22 +426,47 @@ def export_to_excel(
"ORDER BY s.changed_at, s.id",
params,
).fetchall()
# 全部飞行记录(用于次数统计与飞行历史 sheet)
flight_rows = conn.execute(
"SELECT f.drone_id, f.flight_date, f.duration_min, f.takeoff, f.landing, f.remark "
"FROM flight_records f JOIN drones d ON d.id = f.drone_id "
f"{where} ORDER BY f.drone_id, f.flight_date, f.id",
params,
).fetchall()
if flight_rows:
d0 = date.fromisoformat(min(f["flight_date"] for f in flight_rows))
d1 = date.fromisoformat(max(f["flight_date"] for f in flight_rows))
dates = []
d = d0
while d <= d1:
dates.append(d.isoformat())
d += timedelta(days=1)
else:
dates = []
# 每机每日飞行摘要
day_map = _flights_day_map(flight_rows)
ws = wb.active
ws.title = "无人机台账"
ws.append(["机型", "编号", "链路ID", "飞控ID", "飞控版本",
"导航版本", "当前状态", "备注", "创建时间"])
"导航版本", "当前状态", "飞行次数", "备注", "创建时间"])
for r in rows:
cnt = conn.execute(
"SELECT COUNT(*) FROM flight_records WHERE drone_id = ?",
(r["id"],),
).fetchone()[0]
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"],
r["cur_status"] or "", cnt, 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]
ws4 = wb.create_sheet("飞行历史")
_write_flight_history_sheet(ws4, rows, dates, day_map)
sheets = [ws, ws2, ws4]
# ---- 样式 ----
for sheet in sheets: