100 lines
2.6 KiB
Python
100 lines
2.6 KiB
Python
import db
|
|
|
|
|
|
def list_upgrade_logs(limit: int = 200) -> list[dict]:
|
|
conn = db.get_conn()
|
|
rows = conn.execute(
|
|
"""
|
|
SELECT id,device_id,from_ver,to_ver,result,create_time
|
|
FROM upgrade_logs ORDER BY create_time DESC LIMIT ?
|
|
""",
|
|
(limit,),
|
|
).fetchall()
|
|
conn.close()
|
|
return [
|
|
{
|
|
"id": row["id"],
|
|
"device_id": row["device_id"],
|
|
"app_id": "",
|
|
"from_version": row["from_ver"],
|
|
"to_version": row["to_ver"],
|
|
"result": row["result"],
|
|
"create_time": row["create_time"],
|
|
}
|
|
for row in rows
|
|
]
|
|
|
|
|
|
def delete_upgrade_log(log_id: int) -> int:
|
|
conn = db.get_conn()
|
|
cur = conn.execute("DELETE FROM upgrade_logs WHERE id=?", (log_id,))
|
|
conn.commit()
|
|
deleted = cur.rowcount
|
|
conn.close()
|
|
return deleted
|
|
|
|
|
|
def clear_upgrade_logs() -> int:
|
|
conn = db.get_conn()
|
|
cur = conn.execute("DELETE FROM upgrade_logs")
|
|
conn.commit()
|
|
deleted = cur.rowcount
|
|
conn.close()
|
|
return deleted
|
|
|
|
|
|
def list_download_logs(app_id: str = "", limit: int = 300) -> list[dict]:
|
|
conn = db.get_conn()
|
|
if app_id:
|
|
rows = conn.execute("SELECT * FROM download_logs WHERE app_id=? ORDER BY id DESC LIMIT ?", (app_id, limit)).fetchall()
|
|
else:
|
|
rows = conn.execute("SELECT * FROM download_logs ORDER BY id DESC LIMIT ?", (limit,)).fetchall()
|
|
conn.close()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def delete_download_log(log_id: int) -> int:
|
|
conn = db.get_conn()
|
|
cur = conn.execute("DELETE FROM download_logs WHERE id=?", (log_id,))
|
|
conn.commit()
|
|
deleted = cur.rowcount
|
|
conn.close()
|
|
return deleted
|
|
|
|
|
|
def clear_download_logs(app_id: str = "") -> int:
|
|
conn = db.get_conn()
|
|
if app_id:
|
|
cur = conn.execute("DELETE FROM download_logs WHERE app_id=?", (app_id,))
|
|
else:
|
|
cur = conn.execute("DELETE FROM download_logs")
|
|
conn.commit()
|
|
deleted = cur.rowcount
|
|
conn.close()
|
|
return deleted
|
|
|
|
|
|
def list_audit_logs(limit: int = 300) -> list[dict]:
|
|
conn = db.get_conn()
|
|
rows = conn.execute("SELECT * FROM admin_audit_logs ORDER BY id DESC LIMIT ?", (limit,)).fetchall()
|
|
conn.close()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def delete_audit_log(log_id: int) -> int:
|
|
conn = db.get_conn()
|
|
cur = conn.execute("DELETE FROM admin_audit_logs WHERE id=?", (log_id,))
|
|
conn.commit()
|
|
deleted = cur.rowcount
|
|
conn.close()
|
|
return deleted
|
|
|
|
|
|
def clear_audit_logs() -> int:
|
|
conn = db.get_conn()
|
|
cur = conn.execute("DELETE FROM admin_audit_logs")
|
|
conn.commit()
|
|
deleted = cur.rowcount
|
|
conn.close()
|
|
return deleted
|