"""
Generate a CSV of call and message history from a Google Voice Takeout.
CSV columns:
timestamp The date & time of the log in UTC
date The local date of the log
time The local time of the log
type The type of the log (Received, Placed, Missed, Voicemail, Text)
contact_id A unique, anonymized value representing the contact's number or name
contact_name The name of the contact
call_duration The duration of calls in HH:MM:SS
message_days The duration of a text message thread in days
message_count The number of messages in a text message thread
More details and instructions:
https://github.com/Mutual-Aid-Medford-and-Somerville/google-voice-history
"""
import argparse
import csv
import functools
import hashlib
import operator
import os
import re
import sys
import zipfile
from contextlib import contextmanager
from datetime import datetime
from typing import IO, Any, Dict, Iterable, Iterator, List, Optional
from xml.etree import ElementTree
CALL_PATTERN = (
r"Takeout/Voice/(?P<directory>Calls|Spam)/"
r"(?P<contact>.*?) - (?P<type>.+?) - (?P<timestamp>.+?)\.html"
)
CALL_FIELDS = [
"timestamp",
"date",
"time",
"type",
"contact_id",
"contact_name",
"call_duration",
"message_days",
"message_count",
]
# HACK: This is actually `Dict[str, Optional[Union[str, int]]]`
# But that would require lots of things like `cast(str, ...)` to pass mypy
# Might be better to use a dataclass
CallDict = Dict[str, Any]
@contextmanager
def pipeable() -> Iterator[None]:
"""
Silence noisy errors from `python my_script.py | head`.
https://docs.python.org/3/library/signal.html#note-on-sigpipe
"""
try:
yield
except BrokenPipeError:
os.dup2(os.open(os.devnull, os.O_WRONLY), sys.stdout.fileno())
sys.exit(1)
@pipeable()
def main() -> None:
"""Process command line arguments, parse the Takeout, and write a CSV."""
description, epilog = __doc__.strip().split("\n\n", 1)
parser = argparse.ArgumentParser(
description=description,
epilog=epilog,
formatter_class=argparse.RawDescriptionHelpFormatter,
)
parser.add_argument(
"takeout_path",
help="File path of Google Voice Takeout",
metavar="PATH",
)
# TODO: Add option for writing to a file?
# TODO: Add option for excluding columns, defaulting to ["contact"]
args = parser.parse_args()
calls = parse_takeout(args.takeout_path)
write_csv(calls, CALL_FIELDS, sys.stdout)
CONTACT_STATS = {
"total": 0,
"missing": 0,
"numbers": 0,
"names": 0,
}
def parse_takeout(path: str) -> List[CallDict]:
"""Parse all call metadata from a Google Takeout ZIP file."""
with zipfile.ZipFile(path) as takeout:
calls = match_calls(takeout.namelist())
calls = parse_calls(calls, takeout)
calls = sorted(calls, key=operator.itemgetter("timestamp"))
contact_total = sum(CONTACT_STATS[k] for k in ["missing", "numbers", "names"])
if contact_total != CONTACT_STATS["total"]:
raise ValueError(
f"Total contacts ({CONTACT_STATS['total']}) != "
f"sum of each type ({contact_total})"
)
return calls
def match_calls(filenames: List[str]) -> Iterable[CallDict]:
"""Parse some call metadata from call history filenames."""
for filename in filenames:
match = re.match(CALL_PATTERN, filename)
if not match:
continue
yield {
"filename": filename,
**match.groupdict(),
}
def parse_calls(
calls: Iterable[CallDict],
takeout: zipfile.ZipFile,
) -> Iterable[CallDict]:
"""Parse more call metadata from call history HTML files."""
for call in calls:
yield {
**call,
"timestamp": format_timestamp(call["timestamp"]),
**format_contact(call["contact"]),
**parse_file(call["filename"], takeout),
}
def format_timestamp(timestamp: str) -> str:
"""
Convert a non-standard timestamp to ISO8601.
2020-08-21T18_57_10Z => 2020-08-21T18:57:10-04:00
Matching datetime.isoformat(). For details, see:
https://docs.python.org/3/library/datetime.html#datetime.datetime.isoformat
"""
return timestamp.replace("_", ":").replace("Z", "+00:00")
def format_contact(contact: str) -> CallDict:
"""Convert a contact's name or number to a dictionary with a unique ID."""
CONTACT_STATS["total"] += 1
is_number = re.search(r"\d{10}", contact) is not None
if is_number:
CONTACT_STATS["numbers"] += 1
elif contact:
CONTACT_STATS["names"] += 1
else:
CONTACT_STATS["missing"] += 1
return {
"contact_id": anonymize(contact) if contact else None,
"contact_name": contact if not is_number else None,
}
ANONYMIZED_VALUES: Dict[str, str] = {}
@functools.lru_cache()
def anonymize(value: str) -> str:
"""Generate a unique ID for a value."""
# Using a small digest_size for readability
# In initial testing, didn't get collisions until digest_size == 2
digest = hashlib.blake2b(value.encode("utf-8"), digest_size=5).hexdigest()
current_value = ANONYMIZED_VALUES.setdefault(digest, value)
if current_value != value:
raise ValueError(f"Duplicate anonymization for {current_value} and {value}")
return digest
def parse_file(filename: str, takeout: zipfile.ZipFile) -> CallDict:
"""Parse a call's metadata from an HTML file."""
content = takeout.read(filename).decode("utf-8")
try:
xml = ElementTree.fromstring(
# HACK: Remove HTML tags and entities that cause XML parse errors
# This is fine because we don't care about the content
# Alternatively, use a proper HTML parser like lxml.html or pyquery
content.replace("<br>", "").replace("&", ""),
)
except ElementTree.ParseError as exc:
raise ValueError(f"Error parsing {filename}") from exc
return {
"call_duration": parse_call_duration(xml),
**format_datetime(parse_call_datetime(xml)),
**parse_messages(xml),
}
def parse_call_duration(xml: ElementTree.Element) -> Optional[str]:
"""
Parse a call's duration from HTML.
<abbr class="duration" title="PT2M23S">(00:02:23)</abbr>
"""
element = xml.find(".//*[@class='duration']")
if element is None or element.text is None:
return None
return element.text.strip("()")
def parse_call_datetime(xml: ElementTree.Element) -> Optional[datetime]:
"""
Parse a call's timestamp from HTML.
<abbr class="published" title="2020-06-14T12:40:38.000-04:00">
Jun 14, 2020, 12:40:38 PM Eastern Time
</abbr>
"""
element = xml.find(".//*[@class='published']")
if element is None:
return None
return datetime.fromisoformat(element.get("title", ""))
def format_datetime(dt: Optional[datetime]) -> CallDict:
"""Convert a datetime object to a dictionary of date and time."""
if dt is None:
return {}
return {
"date": dt.strftime("%Y-%m-%d"),
"time": dt.strftime("%I:%M %p"),
}
def parse_messages(xml: ElementTree.Element) -> CallDict:
"""
Parse message metadata from HTML.
<div class="hChatLog hfeed">
<div class="message">
<abbr class="dt" title="2020-06-23T21:10:00.971-04:00">
Jun 23, 2020, 9:10:00 PM Eastern Time
</abbr>
<!-- ... -->
</div>
<!-- ... -->
</div>
"""
messages = xml.findall(".//*[@class='message']")
if not messages:
return {}
first_datetime = parse_message_datetime(messages[0])
last_datetime = parse_message_datetime(messages[-1])
if not (first_datetime and last_datetime):
return {}
return {
**format_datetime(first_datetime),
"message_days": (last_datetime - first_datetime).days,
"message_count": len(messages),
}
def parse_message_datetime(xml: ElementTree.Element) -> Optional[datetime]:
"""
Parse a message's timestamp from HTML.
<abbr class="dt" title="2020-06-23T21:10:00.971-04:00">
"""
element = xml.find(".//*[@class='dt']")
if element is None:
return None
return datetime.fromisoformat(element.get("title", ""))
def write_csv(
calls: List[CallDict],
fieldnames: List[str],
csvfile: IO[str],
) -> None:
"""Write call history metadata as comma-separated values."""
writer = csv.DictWriter(
csvfile,
fieldnames=fieldnames,
extrasaction="ignore",
# Workaround unwanted CRLf: https://stackoverflow.com/a/17725590/3188289
# Could maybe use os.linesep instead of "\n"?
lineterminator="\n",
)
writer.writeheader()
writer.writerows(calls)
if __name__ == "__main__":
main()