feat(wacli-export): export daily chat messages to csv
This commit is contained in:
Executable
+99
@@ -0,0 +1,99 @@
|
||||
#!/usr/bin/env -S uv run --script
|
||||
|
||||
# /// script
|
||||
# dependencies = []
|
||||
# ///
|
||||
|
||||
import argparse
|
||||
import csv
|
||||
import logging
|
||||
import sqlite3
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def parse_args() -> argparse.Namespace:
|
||||
parser = argparse.ArgumentParser(
|
||||
description="Export daily WhatsApp chat messages to CSV files.",
|
||||
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
|
||||
)
|
||||
parser.add_argument(
|
||||
"--db",
|
||||
"-d",
|
||||
type=Path,
|
||||
default=Path.home() / ".wacli" / "wacli.db",
|
||||
help="Path to wacli SQLite database",
|
||||
)
|
||||
parser.add_argument(
|
||||
"--output",
|
||||
"-o",
|
||||
type=Path,
|
||||
default=Path.cwd(),
|
||||
help="Output directory for CSV files",
|
||||
)
|
||||
parser.add_argument(
|
||||
"chat_jid",
|
||||
type=str,
|
||||
help="Chat JID to export (e.g. 905341231231@s.whatsapp.net)",
|
||||
)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main() -> None:
|
||||
args = parse_args()
|
||||
|
||||
if not args.db.exists():
|
||||
log.error("Database not found: %s", args.db)
|
||||
raise SystemExit(1)
|
||||
|
||||
args.output.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
filename_prefix = f"whatsapp_{args.chat_jid.split('@')[0]}"
|
||||
|
||||
conn = sqlite3.connect(args.db)
|
||||
conn.row_factory = sqlite3.Row
|
||||
|
||||
rows = conn.execute(
|
||||
"""
|
||||
SELECT sender_name, ts, display_text
|
||||
FROM messages
|
||||
WHERE chat_jid = ?
|
||||
ORDER BY ts ASC
|
||||
""",
|
||||
(args.chat_jid,),
|
||||
).fetchall()
|
||||
|
||||
conn.close()
|
||||
|
||||
if not rows:
|
||||
log.warning("No messages found for JID: %s", args.chat_jid)
|
||||
return
|
||||
|
||||
daily: dict[str, list[tuple[str, str, str]]] = {}
|
||||
for row in rows:
|
||||
day = datetime.fromtimestamp(row["ts"], tz=timezone.utc).strftime("%Y-%m-%d")
|
||||
ts_iso = datetime.fromtimestamp(row["ts"], tz=timezone.utc).strftime(
|
||||
"%Y-%m-%d %H:%M:%S"
|
||||
)
|
||||
daily.setdefault(day, []).append(
|
||||
(row["sender_name"] or "", ts_iso, row["display_text"] or "")
|
||||
)
|
||||
|
||||
written = 0
|
||||
for day in sorted(daily):
|
||||
day_path = args.output / f"{filename_prefix}_{day}.csv"
|
||||
with day_path.open("w", newline="") as f:
|
||||
w = csv.writer(f)
|
||||
w.writerow(["sender_name", "timestamp", "display_text"])
|
||||
w.writerows(daily[day])
|
||||
written += 1
|
||||
log.info("Wrote %s (%d messages)", day_path, len(daily[day]))
|
||||
|
||||
log.info("Done — %d files written to %s", written, args.output)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user