ndbeals-keep-exporter/keep_exporter/export.py

#!/usr/bin/env python3
"""Library functions that do the heavy lifting of this package."""
# pylint: disable=protected-access
import datetime
import mimetypes
import pathlib
from typing import Dict, List, NamedTuple, Optional, Set, Tuple, Union, ValuesView

import click

# import click_config_file
import frontmatter
import gkeepapi
from gkeepapi.node import NodeAudio, NodeDrawing, NodeImage
from mdutils.mdutils import MdUtils
from pathvalidate import sanitize_filename

mimetypes.add_type("audio/3gpp", ".3gp")


def all_note_media(
    note: gkeepapi._node.Note,
) -> List[Union[NodeImage, NodeDrawing, NodeAudio]]:
    """
    Returns a filtered list of only "media" blobs associate with the note.
    Currently NodeDrawing, NodeImage, and NodeMedia.
    There are other blob types, but they don't seem actionable as media.
    """
    return note.images + note.drawings + note.audio


def download_media(
    keep: gkeepapi.Keep,
    note: gkeepapi._node.Note,
    mediapath: pathlib.Path,
    skip_existing: bool,
) -> Tuple[List[pathlib.Path], int]:
    """Downloads media files from keep

    Args:
        keep (gkeepapi.Keep): keep instance
        note (gkeepapi._node.Note): specific note to download media from
        mediapath (pathlib.Path): path to put media
        skip_existing (bool): skip existing media?

    Returns:
        Tuple[List[pathlib.Path], int]: List of paths where media was put
    """

    note_media = all_note_media(note)
    if not note_media:
        return ([], 0)

    ret = []
    downloaded_media = 0

    for media in note_media:
        meta = media.blob.save()
        # ocr = meta["extracted_text"]  # TODO save ocr as metadata? in markdown or image?

        if meta.get("type", "") == "DRAWING":
            extension = mimetypes.guess_extension(
                meta.get("drawingInfo", {})
                .get("snapshotData", {})
                .get("mimetype", "image/png")
            )  # All drawings seem to be pngs
        elif meta.get("type") == "IMAGE":
            extension = mimetypes.guess_extension(meta.get("mimetype", "image/jpeg"))
            # .jpe just feels weird, but it's my default in testing
            if extension == ".jpe":
                extension = ".jpg"
        else:  # 'AUDIO'
            extension = mimetypes.guess_extension(meta.get("mimetype", "audio/3gpp"))

        # nest media files under folders named by the note's ID
        # this simplifies figuring out the note media files came from
        note_media_path = mediapath / note.id
        note_media_path.mkdir(exist_ok=True)

        media_file = (
            note_media_path / f"{sanitize_filename(media.id,max_len=135)}{extension}"
        )

        # checking size isn't perfect, and drawings don't have a size,
        # but it doesn't seem right to always re-download images that likely
        # haven't changed
        if (
            skip_existing
            and media_file.exists()
            and hasattr(media.blob, "byte_size")
            and media_file.stat().st_size == media.blob.byte_size
        ):
            click.echo(
                f"Media file f{media_file} exists and is same size as in Google Keep. Skipping."
            )
        else:
            print(f"Downloading media {meta.get('type')} {media.id} for note {note.id}")

            url = keep._media_api.get(media)
            media_data = keep._media_api._session.get(url)
            with media_file.open("wb") as file_handle:
                file_handle.write(media_data.content)

            downloaded_media += 1

        ret.append(media_file)

    return (ret, downloaded_media)


def build_frontmatter(note: gkeepapi._node.Note, markdown: str) -> frontmatter.Post:
    """Builds the frontmatter header put at the top of notes

    Args:
        note (gkeepapi._node.Note): Note to build header from
        markdown (str): markdown body

    Returns:
        frontmatter.Post: Completed header + body
    """
    metadata = {
        "title": note.title,
        "url": note.url,
        "pinned": note.pinned,
        "archived": note.archived,
        "trashed": note.trashed,
        # "deleted": note.deleted, # unused, when google keep actually ~deletes~ a note, the api no longer returns it
        "tags": [label.name for label in note.labels.all()],
        "color": note.color.name,
        "timestamps": {
            "created": note.timestamps.created.timestamp(),
            "edited": note.timestamps.edited.timestamp(),
            "updated": note.timestamps.updated.timestamp(),
        },
        "google_keep_id": note.id,
        "type": note.type.name,
        "parent_id": note.parent_id,
        "sort": note.sort,
    }

    # gkeepapi appears to be treating "0" as a timestamp rather than null.
    # Sometimes the data structure does not have the key at all instead of 0.
    if note.timestamps.trashed and note.timestamps.trashed.year > 1970:
        metadata["timestamps"]["trashed"] = note.timestamps.trashed.timestamp()
    if note.timestamps.deleted and note.timestamps.deleted.year > 1970:
        metadata["timestamps"]["deleted"] = note.timestamps.deleted.timestamp()

    return frontmatter.Post(markdown, handler=None, **metadata)


def build_markdown(note: gkeepapi._node.Note, images: List[pathlib.Path]) -> str:
    """builds the markdown body from a given note

    Args:
        note (gkeepapi._node.Note): Note to build body out of
        images (List[pathlib.Path]): Images embedded in body

    Returns:
        str: string representation of the body contents
    """
    doc = MdUtils(
        ""
    )  # mdutils requires a string file name. We're not using it to write files, ignore it.

    doc.new_header(1, note.title)
    doc.new_header(2, "Note")

    text = note.text
    text = text.replace("☑ ", "- [X] ")
    text = text.replace("☐ ", "- [ ] ")

    doc.new_paragraph(text)

    if note.annotations.links:
        doc.new_line()
        doc.new_line()
        doc.new_header(2, "Links")
        doc.new_list(
            [
                doc.new_inline_link(link=link.url, text=link.title)
                for link in note.annotations.links
            ]
        )

    if images:
        doc.new_line()
        doc.new_header(2, "Attached Media")

        for image in images:
            doc.new_line(doc.new_inline_image("", image.name))

    return doc.file_data_text


def write_note(target_path, header, note, markdown):
    """Writes built notes to disk, optionally builds a header/frontmatter too."""
    with target_path.open("wb+") as file_handle:
        if header:
            front_matter = build_frontmatter(note, markdown)
            frontmatter.dump(
                front_matter, file_handle, sort_keys=False
            )  # don't sort frontmatter keys
        else:
            file_handle.write(markdown.encode("utf-8"))


LocalMedia = NamedTuple(
    "LocalMedia",
    [
        ("path", pathlib.Path),
        ("google_keep_note_id", str),
        ("google_keep_media_id", str),
    ],
)


class LocalNote:
    def __init__(
        self,
        google_keep_id: str,
        path: Optional[pathlib.Path] = None,
        timestamp_updated: Optional[datetime.datetime] = None,
        local_media: Dict[str, LocalMedia] = None,
    ):
        self.google_keep_id = google_keep_id
        self.path = path
        self.timestamp_updated = timestamp_updated

        if not local_media:
            self.local_media: Dict[str, LocalMedia] = {}
        else:
            self.local_media = local_media


def index_existing_files(directory: pathlib.Path) -> Dict[str, LocalNote]:
    """
    Scans the output folder looking for existing markdown files
    and media files and builds an index by google_keep_id of those files
    using the metadata in the markdown frontmatter and the filenames
    of the media files.
    """
    index: Dict[str, LocalNote] = {}

    keep_notes = 0
    unknown_notes = 0
    errors = 0
    media = 0

    for file in directory.rglob("*"):
        if not file.is_file():
            continue

        # markdown file
        if file.name.endswith(".md"):
            try:
                with file.open("rt") as file_handle:
                    note_frontmatter = frontmatter.load(file_handle)

                    google_keep_id: str = note_frontmatter.metadata.get("google_keep_id")
                    if google_keep_id:
                        if google_keep_id in index and index[google_keep_id].path:
                            click.echo(
                                f"Same Google Keep ID {google_keep_id} in multiple files:\n"
                                f"    {file}\n"
                                f"    {index[google_keep_id].path}\n"
                                f"Only the last file will be updated."
                            )

                        keep_notes += 1
                        index.setdefault(google_keep_id, LocalNote(google_keep_id))

                        updated: datetime.datetime = datetime.datetime.fromtimestamp(
                            note_frontmatter.metadata.get("timestamps", {}).get("updated")
                        )

                        index[google_keep_id].timestamp_updated = updated
                        index[google_keep_id].path = file
                    else:
                        unknown_notes += 1

            except IOError as ex:
                errors = 0
                click.echo(
                    f"Unable to open Markdown file {file}. Skipping: {str(ex)}",
                    err=True,
                )

        # media file
        else:
            media += 1

            google_keep_id = file.parent.name
            media_id = ".".join(file.name.split(".")[0:2])

            index.setdefault(google_keep_id, LocalNote(google_keep_id))
            index[google_keep_id].local_media[media_id] = LocalMedia(
                file, google_keep_id, media_id
            )

    click.echo(
        f"Indexed local files: {keep_notes} Google Keep notes, {unknown_notes} unknown markdown files, {media} media files, {errors} errors"
    )

    return index


def try_rename_note(note: LocalNote, target_file: pathlib.Path) -> pathlib.Path:
    """
    Attempts to rename an existing note to the new canonical filename,
    but accepts failures to rename. Returns the path the note now exists
    in, either the old path or the new renamed path.
    """
    if not note.path:
        return target_file

    click.echo(f"Renaming [{note.path}] to [{target_file}]")

    try:
        note.path.rename(target_file)
        return target_file
    # pylint: disable=broad-except
    except Exception as ex:
        click.echo(f"Unable to rename note. Using existing name: {ex}", err=True)
        return note.path


def build_note_unique_path(
    notepath: pathlib.Path,
    note: gkeepapi._node.Note,
    date_format: str,
    local_index: Dict[str, LocalNote],
) -> pathlib.Path:

    title = note.title.strip()
    if len(title) < 1:
        title = "untitled"

    date_str = note.timestamps.created.strftime(date_format)
    filename = f'{sanitize_filename(f"{date_str} - " + title,max_len=135)}.md'
    target_path = notepath / filename

    local_note = local_index.get(note.id)
    local_path = local_note.path if local_note else None

    if local_path:
        # if the note filename matches the current filename for that note, then we're good
        if local_path == target_path:
            return local_path

        # if re-naming would result in having to de-dupe the target filename, keep the
        # exising filename - initial pass at fixing this just resulted in bouncing between
        # two different filenames each pass
        if target_path.exists():
            click.echo(
                f"Note {note.id} will not be renamed. Target file [{target_path}] exists."
            )
            return local_path

    # otherwise, if the file already exists avoid overwriting it
    # put the unique note ID and an incrementing index at the end of the filename
    dedupe_index = 1
    while target_path.exists():
        filename = f'{sanitize_filename(f"{date_str} - " + title,max_len=135)}.{note.id}.{dedupe_index}.md'
        target_path = notepath / filename
        dedupe_index += 1

    return target_path


def delete_local_only_files(
    local_index: Dict[str, LocalNote],
    keep_notes: Dict[str, List[gkeepapi.node.Note]],
    delete_local: bool,
) -> Tuple[int, int]:
    """
    Checks the local index for any notes or media that exist only locally
    and were not returned in the Google Keep API call.
    """
    deleted_notes, deleted_media = 0, 0

    local_only_note_ids = set(local_index.keys()).difference(set(keep_notes.keys()))

    if local_only_note_ids:
        if not delete_local:
            click.echo(
                f"{len(local_only_note_ids)} notes exist locally, but not in Google Keep. Add argument [--delete-local] to delete."
            )
        else:
            click.echo(
                f"{len(local_only_note_ids)} notes exist locally, but not in Google Keep. Trashing local files."
            )
            for note_id in local_only_note_ids:
                note_path = local_index[note_id].path
                deleted_notes += 1
                if note_path:
                    click.echo(
                        f"    Deleting unknown local note [{note_id}] file [{local_index[note_id].path}]"
                    )
                    note_path.unlink()

    local_only_media: Set[Tuple[str, str]] = set(
        [
            (local_media.google_keep_note_id, local_media.google_keep_media_id)
            for local_note in local_index.values()
            for local_media in local_note.local_media.values()
            if local_media.google_keep_note_id and local_media.google_keep_media_id
        ]
    )

    notes: ValuesView[gkeepapi._node.Note] = keep_notes.values()
    keep_media: Set[Tuple[str, str]] = set(
        [
            (keep_note.id, keep_media.id)
            for keep_note in notes
            for keep_media in all_note_media(keep_note)
        ]
    )

    local_only_media_ids = local_only_media.difference(keep_media)
    if not local_only_media_ids:
        return (deleted_notes, 0)

    if not delete_local:
        click.echo(
            f"{len(local_only_note_ids)} media files exist locally, but not in Google Keep. Add argument [--delete-local] to delete."
        )
        return (deleted_notes, 0)

    for (note_id, media_id) in local_only_media_ids:
        media = local_index[note_id].local_media[media_id]

        click.echo(
            f"    Deleting media [{media_id}] for note [{note_id}] file [{media.path}]"
        )
        media.path.unlink()
        deleted_media += 1

    return (deleted_notes, deleted_media)