safesight/control/scripts/deploy/import-assets.py
tian 98e375dc36 chore: 合并 safesight-control 仓库为 control/ 子目录
项目合并为单仓库:设备端(根) + 管理端(control/)。
两端 git 历史完整保留(git subtree 合并)。

git-subtree-dir: control
git-subtree-mainline: 923ab10d5c
git-subtree-split: 70738edb31
2026-08-02 11:13:26 +08:00

76 lines
2.6 KiB
Python

#!/usr/bin/env python3
"""SafeSight 初始资产导入脚本
将 edge 项目 configs 目录下的模板/场景/覆盖层导入管理端数据库
用法: sudo python3 import-assets.py /tmp/configs
"""
import sys, os, json, sqlite3
from datetime import datetime, timezone
def main():
configs_dir = sys.argv[1] if len(sys.argv) > 1 else "/tmp/configs"
db_path = "/opt/safesightd/data/app.db"
if not os.path.isfile(db_path):
print(f"错误: 数据库文件不存在: {db_path}")
sys.exit(1)
conn = sqlite3.connect(db_path)
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
kinds = [
("templates", "templates", ["name", "description", "body_json", "created_at", "updated_at"]),
("profiles", "profiles", ["name", "primary_template_name", "business_name", "description", "body_json", "created_at", "updated_at"]),
("overlays", "overlays", ["name", "description", "body_json", "created_at", "updated_at"]),
]
print("========== SafeSight 资产导入 ==========")
print(f"数据库: {db_path}")
print(f"配置源: {configs_dir}\n")
for dir_name, table, cols in kinds:
src_dir = os.path.join(configs_dir, dir_name)
if not os.path.isdir(src_dir):
print(f" ⚠ 目录不存在: {src_dir}")
continue
count = 0
for fname in sorted(os.listdir(src_dir)):
if not fname.endswith(".json"):
continue
fpath = os.path.join(src_dir, fname)
name = fname[:-5] # remove .json
with open(fpath, "r", encoding="utf-8") as f:
body = f.read()
placeholders = ", ".join(["?" for _ in cols])
col_names = ", ".join(cols)
if table == "templates":
values = [name, "", body, now, now]
elif table == "overlays":
values = [name, "", body, now, now]
else: # profiles
values = [name, "", "", "", body, now, now]
try:
conn.execute(
f"INSERT OR IGNORE INTO {table} ({col_names}) VALUES ({placeholders})",
values,
)
count += 1
print(f"{dir_name.rstrip('s')}: {name}")
except Exception as e:
print(f"{name}: {e}")
conn.commit()
if count > 0:
print(f" → 导入 {count}\n")
conn.close()
print("========== 导入完成 ==========")
print("\n请执行: sudo systemctl restart safesightd")
if __name__ == "__main__":
main()