Files

73 lines
3.0 KiB
Python

import os
import subprocess
import sqlite3
import logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')
def run_adb_command(command: str):
full_cmd = f"adb -s 192.168.1.193:5555 shell {command}"
try:
result = subprocess.run(full_cmd, shell=True, check=True, capture_output=True, text=True)
return result.stdout.strip()
except subprocess.CalledProcessError as e:
return None
def analyze_specific_db(db_path):
filename = os.path.basename(db_path)
logging.info(f"Đang kéo file {filename} về server...")
# Copy DB ra thư mục temp trên android
run_adb_command(f"su -c 'cp {db_path} /sdcard/temp_{filename}'")
run_adb_command(f"su -c 'chmod 666 /sdcard/temp_{filename}'")
# Pull về server armbian
subprocess.run(f"adb -s 192.168.1.193:5555 pull /sdcard/temp_{filename} .", shell=True, capture_output=True)
run_adb_command(f"su -c 'rm /sdcard/temp_{filename}'")
if os.path.exists(filename):
logging.info(f"Đã lấy được {filename}. Đang đọc cấu trúc SQLite...")
try:
conn = sqlite3.connect(filename)
cursor = conn.cursor()
# Liệt kê các bảng
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
tables = [t[0] for t in cursor.fetchall()]
logging.info(f"Các bảng trong DB: {tables}")
# Tìm bảng chat
chat_table = None
for t in ["chat", "messages", "msg", "chat_msg"]:
if t in tables:
chat_table = t
break
if not chat_table and tables:
chat_table = tables[0] # lấy đại bảng đầu tiên nếu không thấy tên quen thuộc
if chat_table:
logging.info(f"Phát hiện bảng tin nhắn: {chat_table}")
cursor.execute(f"PRAGMA table_info({chat_table});")
cols = cursor.fetchall()
logging.info(f"Cấu trúc bảng {chat_table}:")
for col in cols:
logging.info(f" - {col[1]} ({col[2]})")
# In thử 3 tin nhắn mới nhất
logging.info(f"--- 3 tin nhắn mới nhất trong bảng {chat_table} ---")
try:
cursor.execute(f"SELECT * FROM {chat_table} ORDER BY _rowid_ DESC LIMIT 3;")
rows = cursor.fetchall()
for r in rows:
logging.info(str(r)[:200] + "...")
except Exception as e:
logging.error(f"Lỗi khi đọc dữ liệu: {e}")
conn.close()
logging.info(f"Xong file {filename}!\n")
except Exception as e:
logging.error(f"Lỗi SQLite: {e}")
if __name__ == "__main__":
analyze_specific_db("/data/data/com.zing.zalo/databases/154819133/group_677752884.db")