File size: 4,058 Bytes
cef20af | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | #!/usr/bin/env python3
"""
Patch database.py: adds an ota_update table + set_latest_ota_update /
get_latest_ota_update functions, alongside the existing app_version pattern.
Run from the same directory as database.py:
python3 patch_ota_database.py
"""
path = "database.py"
with open(path, "r") as f:
src = f.read()
# --- Patch 1: add ota_update table next to app_version table ---------------
old_1 = ''' c.execute("""
CREATE TABLE IF NOT EXISTS app_version (
id INTEGER PRIMARY KEY,
version TEXT NOT NULL,
apk_url TEXT NOT NULL,
notes TEXT,
updated_at INTEGER NOT NULL
)
""")
conn.commit()
conn.close()'''
new_1 = ''' c.execute("""
CREATE TABLE IF NOT EXISTS app_version (
id INTEGER PRIMARY KEY,
version TEXT NOT NULL,
apk_url TEXT NOT NULL,
notes TEXT,
updated_at INTEGER NOT NULL
)
""")
c.execute("""
CREATE TABLE IF NOT EXISTS ota_update (
id INTEGER PRIMARY KEY,
update_id TEXT NOT NULL,
runtime_version TEXT NOT NULL,
launch_asset_json TEXT NOT NULL,
assets_json TEXT NOT NULL,
notes TEXT,
created_at INTEGER NOT NULL
)
""")
conn.commit()
conn.close()'''
assert old_1 in src, "Patch 1 anchor not found -- database.py may have changed since this script was written."
assert src.count(old_1) == 1, "Patch 1 anchor is not unique -- aborting to avoid a wrong replace."
src = src.replace(old_1, new_1)
# --- Patch 2: add set/get_latest_ota_update functions -----------------------
old_2 = '''def get_latest_app_version():
conn = get_conn()
try:
row = conn.execute("SELECT version, apk_url, notes, updated_at FROM app_version WHERE id = 1").fetchone()
return dict(row) if row else None
finally:
conn.close()'''
new_2 = '''def get_latest_app_version():
conn = get_conn()
try:
row = conn.execute("SELECT version, apk_url, notes, updated_at FROM app_version WHERE id = 1").fetchone()
return dict(row) if row else None
finally:
conn.close()
def set_latest_ota_update(update_id, runtime_version, launch_asset_json, assets_json, notes=None):
"""Always overwrites the single row (id=1) -- there's only ever one
'latest' OTA update, same pattern as set_latest_app_version. Callers
pass launch_asset_json/assets_json as already-serialized JSON strings
(the exact objects the Expo Updates manifest needs)."""
conn = get_conn()
try:
conn.execute("""
INSERT INTO ota_update (id, update_id, runtime_version, launch_asset_json, assets_json, notes, created_at)
VALUES (1, ?, ?, ?, ?, ?, ?)
ON CONFLICT(id) DO UPDATE SET update_id=excluded.update_id, runtime_version=excluded.runtime_version,
launch_asset_json=excluded.launch_asset_json, assets_json=excluded.assets_json,
notes=excluded.notes, created_at=excluded.created_at
""", (update_id, runtime_version, launch_asset_json, assets_json, notes, int(time.time())))
conn.commit()
return True
finally:
conn.close()
def get_latest_ota_update():
conn = get_conn()
try:
row = conn.execute("""
SELECT update_id, runtime_version, launch_asset_json, assets_json, notes, created_at
FROM ota_update WHERE id = 1
""").fetchone()
return dict(row) if row else None
finally:
conn.close()'''
assert old_2 in src, "Patch 2 anchor not found -- database.py may have changed since this script was written."
assert src.count(old_2) == 1, "Patch 2 anchor is not unique -- aborting to avoid a wrong replace."
src = src.replace(old_2, new_2)
with open(path, "w") as f:
f.write(src)
print("database.py patched successfully.")
print(" - added ota_update table")
print(" - added set_latest_ota_update()")
print(" - added get_latest_ota_update()")
|