Add GPS-only backup/restore and database wipe
- Fix backup.py PRESERVED_FIELDS column names (lat/lng -> latitude/longitude) to match actual DB schema (pre-existing bug) - Add backup_gps() and restore_gps() dedicated GPS-only functions - Add /api/gps/backup and /api/gps/restore API endpoints - Add /api/gps/restore/upload for restoring from uploaded backup files - Add /api/db/wipe endpoint to clear all asset records - Update UI with GPS-only Backup button, Restore GPS from File picker, and Wipe All Assets with double-confirmation dialog - Fix backup listing to scan both seed-data/ and project root
This commit is contained in:
@@ -32,8 +32,8 @@ from typing import Any, Optional
|
||||
# ─── Fields to preserve ───────────────────────────────────────────────────────
|
||||
|
||||
PRESERVED_FIELDS = [
|
||||
"lat",
|
||||
"lng",
|
||||
"latitude",
|
||||
"longitude",
|
||||
"photo_path",
|
||||
]
|
||||
|
||||
@@ -137,6 +137,10 @@ def _discover_asset_table(
|
||||
return asset_table["name"], asset_table["id_column"] or "rowid", existing
|
||||
|
||||
|
||||
# ─── GPS-only fields ──────────────────────────────────────────────────────────
|
||||
|
||||
GPS_ONLY_FIELDS = ["latitude", "longitude"]
|
||||
|
||||
# ─── Core operations ──────────────────────────────────────────────────────────
|
||||
|
||||
def backup(db_path: str) -> str:
|
||||
@@ -278,6 +282,137 @@ def restore(backup_path: str, target_db: Optional[str] = None) -> None:
|
||||
)
|
||||
|
||||
|
||||
def backup_gps(db_path: str) -> str:
|
||||
"""
|
||||
Back up only GPS fields (latitude, longitude) from the database.
|
||||
Returns the path to the created backup file.
|
||||
"""
|
||||
db_path = _resolve_path(db_path)
|
||||
conn = _conn(db_path)
|
||||
table, id_col, fields = _discover_asset_table(conn)
|
||||
|
||||
gps_fields = [f for f in GPS_ONLY_FIELDS if f.lower() in [c.lower() for c in fields]]
|
||||
if not gps_fields:
|
||||
print(
|
||||
"⚠️ No GPS fields (latitude, longitude) found in table "
|
||||
f"'{table}'.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
conn.close()
|
||||
sys.exit(1)
|
||||
|
||||
# Build query
|
||||
col_exprs = ", ".join(f'"{c}"' for c in [id_col] + gps_fields)
|
||||
rows = conn.execute(f'SELECT {col_exprs} FROM "{table}"').fetchall()
|
||||
|
||||
backup_data: dict[str, Any] = {
|
||||
"version": BACKUP_VERSION,
|
||||
"source_db": db_path,
|
||||
"source_table": table,
|
||||
"id_column": id_col,
|
||||
"fields": gps_fields,
|
||||
"created": date.today().isoformat(),
|
||||
"type": "gps_only",
|
||||
"records": [],
|
||||
}
|
||||
|
||||
for row in rows:
|
||||
rec: dict[str, Any] = {"id": row[0]}
|
||||
for i, field in enumerate(gps_fields):
|
||||
rec[field] = row[i + 1]
|
||||
backup_data["records"].append(rec)
|
||||
|
||||
conn.close()
|
||||
|
||||
out_path = _backup_filename().replace(".json.gz", "_gps.json.gz")
|
||||
json_bytes = json.dumps(backup_data, indent=2, default=str).encode("utf-8")
|
||||
with gzip.open(out_path, "wb") as f:
|
||||
f.write(json_bytes)
|
||||
|
||||
record_count = len(backup_data["records"])
|
||||
non_null = sum(
|
||||
1
|
||||
for r in backup_data["records"]
|
||||
if r.get("latitude") is not None or r.get("longitude") is not None
|
||||
)
|
||||
print(
|
||||
f"✅ GPS backup saved: {out_path} "
|
||||
f"({record_count} records, {non_null} with GPS data)"
|
||||
)
|
||||
return out_path
|
||||
|
||||
|
||||
def restore_gps(backup_path: str, target_db: Optional[str] = None) -> None:
|
||||
"""
|
||||
Restore only GPS fields (latitude, longitude) from a backup file.
|
||||
"""
|
||||
backup_path = _resolve_path(backup_path)
|
||||
if not os.path.isfile(backup_path):
|
||||
print(f"❌ Backup file not found: {backup_path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
with gzip.open(backup_path, "rb") as f:
|
||||
backup_data = json.loads(f.read().decode("utf-8"))
|
||||
|
||||
db_path = target_db or backup_data.get("source_db")
|
||||
if not db_path:
|
||||
print(
|
||||
"❌ No target database specified and no source_db in backup.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
db_path = _resolve_path(db_path)
|
||||
if not os.path.isfile(db_path):
|
||||
print(f"❌ Target database not found: {db_path}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
table = backup_data["source_table"]
|
||||
id_col = backup_data["id_column"]
|
||||
fields = backup_data["fields"]
|
||||
records = backup_data["records"]
|
||||
|
||||
conn = sqlite3.connect(db_path)
|
||||
|
||||
# Verify table and columns
|
||||
existing_cols = [
|
||||
r[1]
|
||||
for r in conn.execute(f"PRAGMA table_info({table!r})").fetchall()
|
||||
]
|
||||
existing_lower = [c.lower() for c in existing_cols]
|
||||
|
||||
gps_fields = [f for f in fields if f.lower() in existing_lower]
|
||||
if not gps_fields:
|
||||
print(
|
||||
f"❌ No GPS columns found in target table '{table}'.",
|
||||
file=sys.stderr,
|
||||
)
|
||||
conn.close()
|
||||
sys.exit(1)
|
||||
|
||||
set_clause = ", ".join(f'"{c}" = ?' for c in gps_fields)
|
||||
sql = f'UPDATE "{table}" SET {set_clause} WHERE "{id_col}" = ?'
|
||||
|
||||
restored = 0
|
||||
skipped = 0
|
||||
for rec in records:
|
||||
values = [rec.get(f) for f in gps_fields] + [rec["id"]]
|
||||
cursor = conn.execute(sql, values)
|
||||
if cursor.rowcount > 0:
|
||||
restored += 1
|
||||
else:
|
||||
skipped += 1
|
||||
|
||||
conn.commit()
|
||||
conn.close()
|
||||
|
||||
print(
|
||||
f"✅ GPS restored {restored} records from {backup_path}\n"
|
||||
f" Fields: {', '.join(gps_fields)}\n"
|
||||
f" Skipped (no matching row): {skipped}"
|
||||
)
|
||||
|
||||
|
||||
def compare(backup_path: str, target_db: Optional[str] = None) -> int:
|
||||
"""
|
||||
Compare backed-up values against current database values.
|
||||
@@ -397,11 +532,21 @@ def main():
|
||||
metavar="DB_PATH",
|
||||
help="Backup GPS (lat, lng) and photo_path from database.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--backup-gps",
|
||||
metavar="DB_PATH",
|
||||
help="Backup GPS only (latitude, longitude) from database.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--restore",
|
||||
metavar="BACKUP_FILE",
|
||||
help="Restore preserved fields from a backup .json.gz file.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--restore-gps",
|
||||
metavar="BACKUP_FILE",
|
||||
help="Restore GPS only from a backup .json.gz file.",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--compare",
|
||||
metavar="BACKUP_FILE",
|
||||
@@ -419,20 +564,24 @@ def main():
|
||||
|
||||
# Exactly one mode
|
||||
modes = sum(
|
||||
1 for m in [args.backup, args.restore, args.compare] if m
|
||||
1 for m in [args.backup, args.backup_gps, args.restore, args.restore_gps, args.compare] if m
|
||||
)
|
||||
if modes != 1:
|
||||
parser.print_help()
|
||||
print(
|
||||
"\n❌ Specify exactly one of: --backup, --restore, --compare",
|
||||
"\n❌ Specify exactly one of: --backup, --backup-gps, --restore, --restore-gps, --compare",
|
||||
file=sys.stderr,
|
||||
)
|
||||
sys.exit(1)
|
||||
|
||||
if args.backup:
|
||||
backup(args.backup)
|
||||
elif args.backup_gps:
|
||||
backup_gps(args.backup_gps)
|
||||
elif args.restore:
|
||||
restore(args.restore, args.target)
|
||||
elif args.restore_gps:
|
||||
restore_gps(args.restore_gps, args.target)
|
||||
elif args.compare:
|
||||
compare(args.compare, args.target)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user