**Approach (brief)** - Treat files as evidence: do not modify originals. Work on copies. - Ensure WAL is applied so you read latest committed transactions by copying both mmssms.db and mmssms.db-wal to a temp working directory and opening the copy (SQLite will read WAL automatically when files are colocated). Optionally checkpoint to merge WAL into DB. - Query messages using UTC-range filtering; convert stored timestamps to/from UTC carefully. Group by thread_id (or address) to build threads.**Pseudocode / Python outline**python
import shutil, tempfile, sqlite3
from datetime import datetime, timezone, timedelta
# 1. copy evidence safely
orig_db = "mmssms.db"
orig_wal = "mmssms.db-wal"
workdir = tempfile.mkdtemp()
shutil.copy(orig_db, workdir)
if os.path.exists(orig_wal):
shutil.copy(orig_wal, workdir)
dbpath = os.path.join(workdir, "mmssms.db")
# 2. open DB (read-write on the copy so WAL can be checkpointed)
conn = sqlite3.connect(dbpath)
cur = conn.cursor()
# 3. checkpoint to merge WAL ensuring latest committed transactions are visible
cur.execute("PRAGMA wal_checkpoint(FULL);")
# 4. prepare UTC range (input start/end as ISO strings)
def to_epoch_ms(dt_utc: datetime):
return int(dt_utc.replace(tzinfo=timezone.utc).timestamp() * 1000)
start_utc = datetime.fromisoformat("2021-01-01T00:00:00").replace(tzinfo=timezone.utc)
end_utc = datetime.fromisoformat("2021-01-31T23:59:59").replace(tzinfo=timezone.utc)
start_ms, end_ms = to_epoch_ms(start_utc), to_epoch_ms(end_utc)
# 5. query messages table (column names vary by vendor; common: date, date_sent, body, thread_id, address)
sql = """
SELECT _id, thread_id, address, date, date_sent, type, body
FROM sms
WHERE date BETWEEN ? AND ?
ORDER BY thread_id, date ASC
"""
cur.execute(sql, (start_ms, end_ms))
rows = cur.fetchall()
# 6. build threads
threads = {}
for r in rows:
msg = { 'id': r[0], 'thread_id': r[1], 'address': r[2], 'date_ms': r[3], 'body': r[6] }
tid = msg['thread_id'] or msg['address']
threads.setdefault(tid, []).append(msg)
# 7. convert timestamps for display / analysis (keep original ms for evidence)
def ms_to_utc_iso(ms):
return datetime.fromtimestamp(ms/1000, tz=timezone.utc).isoformat()
for tid, msgs in threads.items():
for m in msgs:
m['date_utc'] = ms_to_utc_iso(m['date_ms'])
conn.close()
