100 lines
2.6 KiB
Python
Executable File
100 lines
2.6 KiB
Python
Executable File
#!/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()
|