A Python script to effortlessly extract and format your ChatGPT conversations data export from JSON files to well-structured markdown files, with YAML metadata headers. And it all happens locally !
"""Module for handling user configuration and updating the models."""

import json
from typing import Any, Dict

from models.conversation import Conversation
from models.conversation_set import ConversationSet
from models.message import Message
from models.node import Node
from utils.utils import default_output_folder, get_openai_zip_filepath
from views.prompt_user import prompt_user


def get_user_configs() -> Dict[str, Any]:
    """Loads the default configs and calls the prompt_user function with those defaults.
    Returns the new configuration."""

    with open("config.json", "r", encoding="utf-8") as file:
        default_configs = json.load(file)

    if not default_configs["zip_file"]:
        default_configs["zip_file"] = get_openai_zip_filepath()

    if not default_configs["output_folder"]:
        default_configs["output_folder"] = default_output_folder()

    return prompt_user(default_configs)


def update_config_file(user_configs: Dict[str, Any]) -> None:
    """Update the config file with the new configuration options."""
    with open("config.json", "w", encoding="utf-8") as file:
        json.dump(user_configs, file, indent=2)


def set_model_configs(configs: Dict[str, Any]) -> None:
    """Set the configuration for all models."""
    Conversation.configuration = configs.get("conversation", {})
    ConversationSet.configuration = configs.get("conversation_set", {})
    Message.configuration = configs.get("message", {})
    Node.configuration = configs.get("node", {})

controllers/configuration.py
mohamed-chs/chatgpt-history-export-to-md
"""Module for all the data visualizations.

Should ideally only return matplotlib objects, and not deal with the filesystem."""

from pathlib import Path
from typing import Any, List, Set

import matplotlib.dates as mdates
import matplotlib.pyplot as plt
import nltk  # type: ignore
import pandas as pd
from nltk.corpus import stopwords  # type: ignore
from wordcloud import WordCloud  # type: ignore

from models.conversation_set import ConversationSet


# Ensure that the stopwords are downloaded
def load_nltk_stopwords() -> Set[str]:
    """Loads the nltk stopwords. Returns a set of stopwords."""

    try:
        nltk.data.find("corpora/stopwords")  # type: ignore
    except LookupError:
        nltk.download("stopwords")  # type: ignore

    languages = [
        "arabic",
        "english",
        "french",
        "german",
        "spanish",
        "portuguese",
    ]  # add more languages here ...

    stop_words = set(
        word for lang in languages for word in stopwords.words(lang)  # type: ignore
    )

    return stop_words


def wordcloud_from_text(
    text: str,
    **kwargs: Any,
) -> WordCloud:
    """Creates a wordcloud from the given text. Returns a WordCloud object."""

    font_path = kwargs.get("font_path", "assets/fonts/ArchitectsDaughter-Regular.ttf")
    colormap = kwargs.get("colormap", "prism")
    width = kwargs.get("width", 1000)
    height = kwargs.get("height", 1000)

    background_color = kwargs.get("background_color", None)
    if background_color is None:
        mode = kwargs.get("mode", "RGBA")
    else:
        mode = kwargs.get("mode", "RGB")

    include_numbers = kwargs.get("include_numbers", False)

    custom_stopwords: List[str] = kwargs.get("stopwords", [])

    default_stopwords = load_nltk_stopwords()

    stop_words = default_stopwords.union(set(custom_stopwords))

    # TODO: add more arguments here ...

    wordcloud = WordCloud(
        font_path=font_path,
        colormap=colormap,
        stopwords=stop_words,
        width=width,
        height=height,
        background_color=background_color,
        mode=mode,
        include_numbers=include_numbers,
    ).generate(  # type: ignore
        text
    )

    return wordcloud


def wordcloud_from_conversation_set(
    conversation_set: ConversationSet, **kwargs: Any
) -> WordCloud:
    """Creates a wordcloud from the given conversation set. Returns a WordCloud object."""
    text = (
        conversation_set.all_user_text() + "\n" + conversation_set.all_assistant_text()
    )

    return wordcloud_from_text(text, **kwargs)


def create_save_graph(all_timestamps: List[float], file_path: Path) -> None:
    """Creates and saves a graph from the given timestamps."""

    df = pd.DataFrame(all_timestamps, columns=["timestamp"])  # type: ignore
    df["datetime"] = pd.to_datetime(df["timestamp"], unit="s")  # type: ignore

    daily_counts = df.groupby(df["datetime"].dt.date).size()  # type: ignore

    plt.figure(figsize=(15, 7))  # type: ignore

    daily_counts.plot(
        kind="line",
        marker="o",
        linestyle="-",
        linewidth=2,
        markersize=8,
        color="royalblue",
        markeredgecolor="white",
        markeredgewidth=0.5,
    )

    plt.title("ChatGPT Prompts per Day", fontsize=20, fontweight="bold", pad=20)  # type: ignore
    plt.xlabel("Month", fontsize=16, labelpad=15)  # type: ignore
    plt.ylabel("Number of Prompts", fontsize=16, labelpad=15)  # type: ignore
    plt.xticks(fontsize=14)  # type: ignore
    plt.yticks(fontsize=14)  # type: ignore

    ax = plt.gca()  # type: ignore
    ax.xaxis.set_major_locator(mdates.MonthLocator())  # type: ignore
    ax.xaxis.set_major_formatter(mdates.DateFormatter("%B"))  # type: ignore

    plt.xticks(rotation=45)  # type: ignore

    plt.grid(True, linestyle="--", linewidth=0.5, alpha=0.7)  # type: ignore

    plt.tight_layout()  # type: ignore

    plt.savefig(file_path)  # type: ignore

    # close the plot
    plt.close()  # type: ignore

controllers/data_analysis.py
mohamed-chs/chatgpt-history-export-to-md
"""Module for various processes that are used in the controllers.

Should ideally be the only module that deals with the filesystem.
(besides utils.py, but that doesn't save anything to disk)"""


import json
import os
from pathlib import Path
from typing import Any
from zipfile import ZipFile

from tqdm import tqdm

from models.conversation import Conversation
from models.conversation_set import ConversationSet

from .data_analysis import wordcloud_from_conversation_set


def load_conversations_from_openai_zip(zip_filepath: Path) -> ConversationSet:
    """Load the conversations from the OpenAI zip export file."""

    with ZipFile(zip_filepath, "r") as file:
        file.extractall(zip_filepath.with_suffix(""))

    conversations_path = zip_filepath.with_suffix("") / "conversations.json"

    with open(conversations_path, "r", encoding="utf-8") as file:
        conversations = json.load(file)

    return ConversationSet(conversations)


def load_conversations_from_bookmarklet_json(json_filepath: Path) -> ConversationSet:
    """Load the conversations from the bookmarklet json export file."""

    with open(json_filepath, "r", encoding="utf-8") as file:
        conversations = json.load(file)

    return ConversationSet(conversations)


def save_conversation_to_file(conversation: Conversation, file_path: Path) -> None:
    """Save a conversation to a file, with added modification time."""
    base_file_name = file_path.stem

    counter = 0
    while file_path.exists():
        counter += 1
        file_path = file_path.with_name(
            f"{base_file_name} ({counter}){file_path.suffix}"
        )

    with open(file_path, "w", encoding="utf-8") as file:
        file.write(conversation.file_text_content())
    os.utime(file_path, (conversation.update_time, conversation.update_time))


def save_conversation_set_to_dir(
    conversation_set: ConversationSet, dir_path: Path
) -> None:
    """Save a conversation set to a directory."""
    for conversation in tqdm(
        conversation_set.conversation_list, desc="Writing Markdown 📄 files"
    ):
        file_path = dir_path / f"{conversation.file_name()}.md"
        save_conversation_to_file(conversation, file_path)


def create_n_save_wordclouds(
    conversation_set: ConversationSet, folder_path: Path, **kwargs: Any
) -> None:
    """Create the wordclouds for the conversations in the conversation set."""

    weeks_dict = conversation_set.grouped_by_week()
    months_dict = conversation_set.grouped_by_month()
    years_dict = conversation_set.grouped_by_year()

    for week in tqdm(weeks_dict.keys(), desc="Creating weekly wordclouds 🔡☁️ "):
        wordcloud_from_conversation_set(
            weeks_dict[week], **kwargs
        ).to_file(  # type: ignore
            folder_path / f"{week.strftime('%Y week %W')}.png"
        )

    for month in tqdm(months_dict.keys(), desc="Creating monthly wordclouds 🔡☁️ "):
        wordcloud_from_conversation_set(
            months_dict[month], **kwargs
        ).to_file(  # type: ignore
            folder_path / f"{month.strftime('%Y %B')}.png"
        )

    for year in tqdm(years_dict.keys(), desc="Creating yearly wordclouds 🔡☁️ "):
        wordcloud_from_conversation_set(
            years_dict[year], **kwargs
        ).to_file(  # type: ignore
            folder_path / f"{year.strftime('%Y')}.png"
        )


def save_custom_instructions_to_file(
    conversation_set: ConversationSet, file_path: Path
) -> None:
    """Create JSON file for custom instructions in the conversation set."""

    with open(file_path, "w", encoding="utf-8") as file:
        json.dump(conversation_set.all_custom_instructions(), file, indent=2)

controllers/file_system.py
mohamed-chs/chatgpt-history-export-to-md
"""Main file for testing the program."""

import os
import shutil
from pathlib import Path

from controllers.configuration import (
    get_user_configs,
    set_model_configs,
    update_config_file,
)
from controllers.data_analysis import create_save_graph
from controllers.file_system import (
    create_n_save_wordclouds,
    load_conversations_from_bookmarklet_json,
    load_conversations_from_openai_zip,
    save_conversation_set_to_dir,
    save_custom_instructions_to_file,
)
from utils.utils import get_bookmarklet_json_filepath

# DEBUG / PROFILING
pid = os.getpid()
print("PID:", pid)


def main() -> None:
    """Main function."""

    print(
        "Welcome to ChatGPT Data Visualizer ✨📊!\n\n"
        "Follow the instructions in the command line.\n\n"
        "Press 'ENTER' to select the default options.\n\n"
        "If you encounter any issues 🐛, please report 🚨 them here:\n\n"
        "➡️ https://github.com/mohamed-chs/chatgpt-history-export-to-md/issues/new/choose 🔗\n\n"
    )

    configs_dict = get_user_configs()

    print("\n\nAnd we're off! 🚀🚀🚀\n")

    set_model_configs(configs_dict)

    print("Loading data 📂 ...\n")

    zip_filepath = Path(configs_dict["zip_file"])

    all_conversations_set = load_conversations_from_openai_zip(zip_filepath)

    bookmarklet_json_filepath = get_bookmarklet_json_filepath()
    if bookmarklet_json_filepath:
        print("Found bookmarklet download, loading 📂 ...\n")
        bookmarklet_conversations_set = load_conversations_from_bookmarklet_json(
            bookmarklet_json_filepath
        )
        all_conversations_set.update(bookmarklet_conversations_set)

    output_folder = Path(configs_dict["output_folder"])

    # overwrite the output folder if it already exists (might change this in the future)
    if output_folder.exists() and output_folder.is_dir():
        shutil.rmtree(output_folder)

    output_folder.mkdir(parents=True, exist_ok=True)

    markdown_folder = output_folder / "Markdown"
    markdown_folder.mkdir(parents=True, exist_ok=True)

    save_conversation_set_to_dir(all_conversations_set, markdown_folder)

    print(f"\nDone ✅ ! Check the output 📄 here : {markdown_folder.as_uri()} 🔗\n")

    graph_folder = output_folder / "Graphs"
    graph_folder.mkdir(parents=True, exist_ok=True)

    print("Creating graph 📈 of prompts per day ...\n")

    graph_path = graph_folder / "all messages.png"

    create_save_graph(all_conversations_set.all_user_message_timestamps(), graph_path)

    print(f"\nDone ✅ ! Check the output 📈 here : {graph_folder.as_uri()} 🔗\n")
    print("(more graphs 📈 will be added in the future ...)\n")

    wordcloud_folder = output_folder / "Word Clouds"
    wordcloud_folder.mkdir(parents=True, exist_ok=True)

    font_path = f"assets/fonts/{configs_dict['wordcloud']['font']}.ttf"
    colormap = configs_dict["wordcloud"]["colormap"]

    create_n_save_wordclouds(
        all_conversations_set, wordcloud_folder, font_path=font_path, colormap=colormap
    )

    print(f"\nDone ✅ ! Check the output 🔡☁️ here : {wordcloud_folder.as_uri()} 🔗\n")

    print("Writing custom instructions 📝 ...\n")

    custom_instructions_filepath = output_folder / "custom_instructions.json"

    save_custom_instructions_to_file(
        all_conversations_set, custom_instructions_filepath
    )

    print(
        f"\nDone ✅ ! Check the output 📝 here : {custom_instructions_filepath.as_uri()} 🔗\n"
    )

    update_config_file(configs_dict)
    print("(Settings ⚙️ have been updated and saved to 'config.json')\n")

    print(
        "ALL DONE 🎉🎉🎉 !\n\n"
        f"Explore the full gallery 🖼️ at: {output_folder.as_uri()} 🔗\n\n"
        "I hope you enjoy the outcome 🤞.\n\n"
        "If you appreciate it, kindly give the project a star 🌟 on GitHub :\n\n"
        "➡️ https://github.com/mohamed-chs/chatgpt-history-export-to-md 🔗\n\n"
    )


if __name__ == "__main__":
    main()

main.py
mohamed-chs/chatgpt-history-export-to-md
"""
Just a placeholder for now.
a bunch of classes and functions to handle conversations, messages, stats, etc.

object path : conversations.json -> conversation (one of the list items)
"""

import re
from datetime import datetime, timedelta
from time import ctime
from typing import Any, Dict, List, Optional

from utils.utils import ensure_closed_code_blocks, replace_latex_delimiters

from .node import Node


class Conversation:
    """Stores the conversation object from the conversations.json file."""

    configuration: Dict[str, Any] = {}

    def __init__(self, conversation: Dict[str, Any]):
        self.title = conversation.get("title", None)
        self.create_time = conversation.get("create_time", None)
        self.update_time = conversation.get("update_time", None)
        self.mapping = Node.nodes_from_mapping(conversation.get("mapping", None))
        self.moderation_results = conversation.get("moderation_results", None)
        self.current_node = self.mapping[conversation.get("current_node", None)]
        self.plugin_ids = conversation.get("plugin_ids", None)
        self.conversation_id = conversation.get("conversation_id", None)
        self.conversation_template_id = conversation.get(
            "conversation_template_id", None
        )
        self.id = conversation.get("id", None)

    def chat_link(self) -> str:
        """Chat URL.

        Links to the original chat, not a 'shared' one. Needs user's login to chat.openai.com.
        """
        return f"https://chat.openai.com/c/{self.conversation_id}"

    def _main_branch_nodes(self) -> List[Node]:
        """List of all nodes that have a message in the current 'main' branch.

        the 'current_node' represents the last node in the main branch."""

        nodes: List[Node] = []
        curr_node = self.current_node
        curr_parent = curr_node.parent

        while curr_parent:
            if curr_node.message:
                nodes.append(curr_node)
            curr_node = curr_parent
            curr_parent = curr_node.parent

        nodes.reverse()

        return nodes

    def _all_message_nodes(self) -> List[Node]:
        """List of all nodes that have a message in the conversation, including all branches."""

        nodes: List[Node] = []
        for _, node in self.mapping.items():
            if node.message:
                nodes.append(node)

        return nodes

    def has_multiple_branches(self) -> bool:
        """Check if the conversation has multiple branches."""
        return len(self._all_message_nodes()) > len(self._main_branch_nodes())

    def leaf_count(self) -> int:
        """Number of leaves in the conversation."""
        return sum(1 for node in self._all_message_nodes() if not node.children)

    def content_types(self) -> List[str]:
        """List of all content types in the conversation. (all branches)

        (e.g. text, code, execution_output, etc.)"""
        return list(
            set(
                node.message.content_type()
                for node in self._all_message_nodes()
                if node.message
            )
        )

    def message_count(self) -> int:
        """Number of 'user' and 'assistant' messages in the conversation. (all branches)"""
        return sum(
            1
            for node in self._all_message_nodes()
            if node.message and node.message.author_role() in ("user", "assistant")
        )

    def user_message_timestamps(self) -> List[float]:
        """List of all 'user' message timestamps in the conversation.
        (all branches) Useful for generating time series plots."""
        return [node.message.create_time for node in self._user_nodes() if node.message]

    def assistant_message_timestamps(self) -> List[float]:
        """List of all 'assistant' message timestamps in the conversation.
        (all branches) Useful for generating time series plots."""
        return [
            node.message.create_time for node in self._assistant_nodes() if node.message
        ]

    def _system_nodes(self) -> List[Node]:
        """List of all 'system' nodes in the conversation. (all branches)"""
        return [
            node
            for node in self._all_message_nodes()
            if node.message and node.message.author_role() == "system"
        ]

    def _user_nodes(self) -> List[Node]:
        """List of all 'user' nodes in the conversation. (all branches)"""
        return [
            node
            for node in self._all_message_nodes()
            if node.message and node.message.author_role() == "user"
        ]

    def _assistant_nodes(self) -> List[Node]:
        """List of all 'assistant' nodes in the conversation. (all branches)"""
        return [
            node
            for node in self._all_message_nodes()
            if node.message and node.message.author_role() == "assistant"
        ]

    def _tool_nodes(self) -> List[Node]:
        """List of all 'tool' nodes in the conversation. (all branches)"""
        return [
            node
            for node in self._all_message_nodes()
            if node.message and node.message.author_role() == "tool"
        ]

    def entire_user_text(self) -> str:
        """Entire raw 'user' text in the conversation. (all branches)

        Useful for generating word clouds."""
        return "\n".join(
            node.message.content_text() for node in self._user_nodes() if node.message
        )

    def entire_assistant_text(self) -> str:
        """Entire raw 'assistant' text in the conversation. (all branches)

        Useful for generating word clouds."""
        return "\n".join(
            node.message.content_text()
            for node in self._assistant_nodes()
            if node.message
        )

    def model_slug(self) -> Optional[str]:
        """ChatGPT model used for the conversation."""
        assistant_nodes = self._assistant_nodes()
        if not assistant_nodes:
            return None

        message = assistant_nodes[0].message
        if not message:
            return None

        return message.model_slug()

    def used_plugins(self) -> List[str]:
        """List of all ChatGPT plugins used in the conversation."""
        return list(
            set(
                node.message.metadata["invoked_plugin"]["namespace"]
                for node in self._tool_nodes()
                if node.message and node.message.metadata.get("invoked_plugin")
            )
        )

    def custom_instructions(self) -> Optional[Dict[str, str]]:
        """Custom instructions used for the conversation."""
        if len(self._system_nodes()) < 2:
            return None
        context_message = self._system_nodes()[1].message
        if context_message and context_message.metadata.get(
            "is_user_system_message", None
        ):
            return context_message.metadata.get("user_context_message_data", None)
        return None

        # TODO: check if this is the same for conversations from the bookmarklet download

    def yaml_header(self) -> str:
        """YAML metadata header for the conversation."""
        yaml_config = self.configuration.get("yaml", {})

        yaml_map = {
            "title": self.title,
            "chat_link": self.chat_link(),
            "create_time": ctime(self.create_time),
            "update_time": ctime(self.update_time),
            "model": self.model_slug(),
            "used_plugins": self.used_plugins(),
            "message_count": self.message_count(),
            "content_types": self.content_types(),
            "custom_instructions": self.custom_instructions(),
        }

        yaml = "---\n"

        for key, value in yaml_map.items():
            if yaml_config.get(key, True):
                yaml += f"{key}: {value}\n"

        yaml += "---\n"

        return yaml

    def _branch_indicator(self, node: Node) -> str:
        """Get the branch indicator for the given node."""
        if node in self._main_branch_nodes():
            return "(main branch ⎇)"
        return "(other branch ⎇)"

        # TODO: placeholder for now, to be implemented later

    def file_text_content(self) -> str:
        """Returns the full markdown text content of the conversation."""
        markdown_config = self.configuration.get("markdown", {})
        latex_delimiters = markdown_config.get("latex_delimiters", "default")

        markdown = self.yaml_header()

        for node in self._all_message_nodes():
            if node.message:
                content = ensure_closed_code_blocks(node.message.content_text())
                # prevent empty messages from taking up white space
                content = f"\n{content}\n" if content else ""
                if latex_delimiters == "dollar sign":
                    content = replace_latex_delimiters(content)
                markdown += (
                    f"\n{self._branch_indicator(node)}\n"
                    f"{node.header()}{content}{node.footer()}\n---\n"
                )
        return markdown

    def stats(self) -> Dict[str, Any]:
        """Get diverse insightful stats on the conversation."""
        return {}

        # TODO: add stats

    def file_name(self) -> str:
        """Sanitized title of the conversation, compatible with file names."""
        file_anti_pattern = re.compile(r'[<>:"/\\|?*\n\r\t\f\v]')
        return file_anti_pattern.sub("_", self.title) if self.title else "untitled"

    def start_of_year(self) -> datetime:
        """Returns the first of January of the year the conversation was created in,
        as a datetime object."""
        return datetime(datetime.fromtimestamp(self.create_time).year, 1, 1)

    def start_of_month(self) -> datetime:
        """Returns the first of the month the conversation was created in,
        as a datetime object."""
        return datetime(
            datetime.fromtimestamp(self.create_time).year,
            datetime.fromtimestamp(self.create_time).month,
            1,
        )

    def start_of_week(self) -> datetime:
        """Returns the monday of the week the conversation was created in,
        as a datetime object."""
        start_of_week = datetime.fromtimestamp(self.create_time) - timedelta(
            days=datetime.fromtimestamp(self.create_time).weekday()
        )
        return start_of_week.replace(hour=0, minute=0, second=0, microsecond=0)

models/conversation.py
mohamed-chs/chatgpt-history-export-to-md
"""ConversationSet class to model a set of conversations.

Can be the set of all conversations, or a set of conversations in a week, month, or year.

gonna be useful for updating the data in the future

TODO: turn the conversations list into a dict, with the id as the key"""

from datetime import datetime
from time import ctime
from typing import Any, Dict, List

from .conversation import Conversation


class ConversationSet:
    """Stores a set of conversations."""

    configuration: Dict[str, Any] = {}

    def __init__(self, conversations: List[Dict[str, Any]]):
        conversation_dict: Dict[str, Conversation] = {
            conversation["conversation_id"]: Conversation(conversation)
            for conversation in conversations
        }

        self.conversation_dict = conversation_dict

        self.conversation_list = list(self.conversation_dict.values())

    def add_conversation(self, conversation: Conversation) -> None:
        """Add a conversation to the dictionary and list."""
        self.conversation_dict[conversation.conversation_id] = conversation
        self.conversation_list.append(conversation)

    def last_updated(self) -> float:
        """Returns the timestamp of the last updated conversation in the list."""
        return max(conversation.update_time for conversation in self.conversation_list)

    def update(self, conversation_set: "ConversationSet") -> None:
        """Update the conversation set with the new one."""
        if conversation_set.last_updated() <= self.last_updated():
            return
        self.conversation_dict.update(conversation_set.conversation_dict)
        self.conversation_list = list(self.conversation_dict.values())

    def grouped_by_week(self) -> Dict[datetime, "ConversationSet"]:
        """Get a dictionary of conversations in the list grouped by the start of the week."""
        grouped: Dict[datetime, "ConversationSet"] = {}
        for conversation in self.conversation_list:
            week_start = conversation.start_of_week()
            if week_start not in grouped:
                grouped[week_start] = ConversationSet([])
            grouped[week_start].add_conversation(conversation)
        return grouped

    def grouped_by_month(self) -> Dict[datetime, "ConversationSet"]:
        """Get a dictionary of conversations in the list grouped by the start of the month."""
        grouped: Dict[datetime, "ConversationSet"] = {}
        for conversation in self.conversation_list:
            month_start = conversation.start_of_month()
            if month_start not in grouped:
                grouped[month_start] = ConversationSet([])
            grouped[month_start].add_conversation(conversation)
        return grouped

    def grouped_by_year(self) -> Dict[datetime, "ConversationSet"]:
        """Get a dictionary of conversations in the list grouped by the start of the year."""
        grouped: Dict[datetime, "ConversationSet"] = {}
        for conversation in self.conversation_list:
            year_start = conversation.start_of_year()
            if year_start not in grouped:
                grouped[year_start] = ConversationSet([])
            grouped[year_start].add_conversation(conversation)
        return grouped

    def all_custom_instructions(self) -> List[Dict[str, Any]]:
        """Get a list of all custom instructions, in all conversations in the set."""
        custom_instructions: List[Dict[str, Any]] = []

        for conversation in self.conversation_list:
            if not conversation.custom_instructions():
                continue

            instructions_info = {
                "chat_title": conversation.title,
                "chat_link": conversation.chat_link(),
                "time": ctime(conversation.create_time),
                "custom_instructions": conversation.custom_instructions(),
            }

            custom_instructions.append(instructions_info)

        return custom_instructions

    def all_user_message_timestamps(self) -> List[float]:
        """Get a list of all message timestamps, in all conversations in the list."""
        timestamps: List[float] = []

        for conversation in self.conversation_list:
            timestamps.extend(conversation.user_message_timestamps())

        return timestamps

    def all_assistant_message_timestamps(self) -> List[float]:
        """Get a list of all message timestamps, in all conversations in the list."""
        timestamps: List[float] = []

        for conversation in self.conversation_list:
            timestamps.extend(conversation.assistant_message_timestamps())

        return timestamps

    def all_user_text(self) -> str:
        """Get a string of all user text, in all conversations in the list."""
        return "\n".join(
            conversation.entire_user_text() for conversation in self.conversation_list
        )

    def all_assistant_text(self) -> str:
        """Get a string of all assistant text, in all conversations in the list."""
        return "\n".join(
            conversation.entire_assistant_text()
            for conversation in self.conversation_list
        )

models/conversation_set.py
mohamed-chs/chatgpt-history-export-to-md
"""Represents a single message in a conversation. It's contained in a Node object.
(it could be written in node.py, but I think it's better to separate them)

object path : conversations.json -> conversation -> mapping -> mapping node -> message
"""

from typing import Any, Dict, Optional


class Message:
    """A single message in a conversation."""

    configuration: Dict[str, Any] = {}

    def __init__(self, message: Dict[str, Any]):
        self.id = message.get("id", None)
        self.author = message.get("author", None)
        self.create_time = message.get("create_time", None)
        self.update_time = message.get("update_time", None)
        self.content = message.get("content", None)
        self.status = message.get("status", None)
        self.end_turn = message.get("end_turn", None)
        self.weight = message.get("weight", None)
        self.metadata = message.get("metadata", None)
        self.recipient = message.get("recipient", None)

    def author_role(self) -> str:
        """The role of the author of the message.

        'user', 'assistant', 'system' or 'tool'."""
        return self.author["role"]

    def author_header(self) -> str:
        """Get the title header of the message based on the configs."""
        author_config: Dict[str, Any] = self.configuration.get("author_headers", {})
        if self.author_role() == "system":
            return author_config.get("system", "### System")
        return author_config.get(self.author_role(), f"# {self.author_role().title()}")

    def content_text(self) -> str:
        """get the text content of the message."""
        if "parts" in self.content:
            return self.content["parts"][0]
        if "text" in self.content:
            return f"```python\n{self.content['text']}\n```"
        return ""

    def content_type(self) -> str:
        """get the content type of the message."""
        return self.content["content_type"]

    def model_slug(self) -> Optional[str]:
        """get the model used for the message."""
        return self.metadata.get("model_slug", None)

models/message.py
mohamed-chs/chatgpt-history-export-to-md
"""
Node class and methods for the node object in a conversation
object path : conversations.json -> conversation -> mapping -> mapping node

will implement methods to handle conversation branches, like
counting the number of branches,
get the branch of a given node,
and some other version control stuff
"""

from typing import Any, Dict, List, Optional

from .message import Message


class Node:
    """Node class for representing a node in a conversation tree."""

    configuration: Dict[str, Any] = {}

    def __init__(
        self,
        id: str,
        message: Optional[Message],
        parent: Optional["Node"],
        children: Optional[List["Node"]],
    ):
        self.id = id
        self.message = message
        self.parent = parent
        if children is None:
            self.children: List["Node"] = []

    def add_child(self, node: "Node"):
        """Add a child to the node."""
        self.children.append(node)
        node.parent = self

    @staticmethod
    def nodes_from_mapping(mapping: Dict[str, Any]) -> Dict[str, "Node"]:
        """Returns a dictionary of connected Node objects, based on the mapping."""
        nodes: Dict[str, Node] = {}

        # First pass: Create nodes
        for key, value in mapping.items():
            message = Message(value["message"]) if value.get("message") else None
            nodes[key] = Node(id=key, message=message, parent=None, children=None)

        # Second pass: Connect nodes
        for key, value in mapping.items():
            for child_id in value["children"]:
                nodes[key].add_child(nodes[child_id])

        return nodes

    def header(self) -> Optional[str]:
        """Get the header of the node message, containing a link to its parent."""
        if self.message is None:
            return None

        parent_link = (
            f"[parent ⬆️](#{self.parent.id})\n"
            if self.parent and self.parent.message
            else ""
        )
        return f"###### {self.id}\n{parent_link}{self.message.author_header()}\n"

    def footer(self) -> Optional[str]:
        """Get the footer of the node message, containing links to its children."""
        if self.message is None:
            return None

        if len(self.children) == 0:
            return ""
        if len(self.children) == 1:
            return f"\n[child ⬇️](#{self.children[0].id})\n"

        footer = "\n" + " | ".join(
            [f"[child {i+1} ⬇️](#{child.id})" for i, child in enumerate(self.children)]
        )
        return footer + "\n"

    # just for testing, and fun. It's not really ideal for the real thing
    def show(self, level: int = 0) -> str:
        """Return a tree representation of the node and its descendants."""

        lines = [
            (
                f"{'--'*level}lvl{level} : {self.id[:10]}... ,\n"
                f"{'--'*level} : "
                f"{self.message.author_header() if self.message else None} ,\n"
                f"{'--'*level} : "
                f"{self.message.content_text()[:50] if self.message else None}... \n\n"
            )
        ]
        for child in self.children:
            lines.append(child.show(level + 1))

        return "".join(lines)

models/node.py
mohamed-chs/chatgpt-history-export-to-md
import os
import sys
import venv

BASE_DIR = os.path.abspath(os.path.dirname(__file__))
VENV_DIR = os.path.join(BASE_DIR, ".venv")


def create_virtual_environment():
    # Create virtual environment
    env_builder = venv.EnvBuilder(with_pip=True)
    env_builder.create(VENV_DIR)


def pip_install_requirements():
    pip_exe = (
        os.path.join(VENV_DIR, "bin", "pip")
        if sys.platform != "win32"
        else os.path.join(VENV_DIR, "Scripts", "pip.exe")
    )

    # Install requirements
    os.system(f"{pip_exe} install -r requirements.txt")


if __name__ == "__main__":
    print("Creating virtual environment...\n")
    create_virtual_environment()
    print("Installing requirements... (This may take a minute..)\n")
    pip_install_requirements()

    # Post-setup instructions
    print("\nSetup completed successfully!")
    print(
        "\nTo activate the virtual environment, use the following command based on your platform:"
    )
    if sys.platform == "win32":
        print("\nFor Command Prompt:\n.venv\\Scripts\\activate.bat")
        print("\nFor PowerShell:\n.venv\\Scripts\\Activate.ps1")
    else:
        print("\nsource .venv/bin/activate")

setup.py
mohamed-chs/chatgpt-history-export-to-md
"""Utility functions for the project."""

import re
from pathlib import Path
from typing import List, Optional
from zipfile import ZipFile


def ensure_closed_code_blocks(string: str) -> str:
    """Ensure that all code blocks are closed."""
    # A code block can be opened with triple backticks, possibly followed by a language name.
    # It can only be closed however with triple backticks, with nothing else on the line.

    open_code_block = False

    lines = string.split("\n")

    for line in lines:
        if line.startswith("```") and not open_code_block:
            open_code_block = True
            continue

        if line == "```" and open_code_block:
            open_code_block = False

    if open_code_block:
        string += "\n```"

    return string


def replace_latex_delimiters(string: str) -> str:
    """Replace all the LaTeX bracket delimiters in the string with dollar sign ones."""

    string = re.sub(r"\\\[", "$$", string)
    string = re.sub(r"\\\]", "$$", string)
    string = re.sub(r"\\\(", "$", string)
    string = re.sub(r"\\\)", "$", string)

    return string


def get_font_names() -> List[str]:
    """Returns a list of all the font names in the assets/fonts folder."""

    fonts_path = Path("assets/fonts")
    font_names = [font.stem for font in fonts_path.iterdir() if font.is_file()]
    return font_names


def get_colormap_names() -> List[str]:
    """Returns a list of all the colormap names in the assets/colormaps.txt file."""

    colormaps_path = Path("assets/colormaps.txt")

    with open(colormaps_path, "r", encoding="utf-8") as file:
        colormaps_list = file.read().splitlines()

    return colormaps_list


def validate_zip_file(path_str: str) -> bool:
    """Returns True if the given path is a zip file with a 'conversations.json' file."""

    if not Path(path_str).is_file() or Path(path_str).suffix != ".zip":
        return False
    with ZipFile(path_str) as zip_ref:
        if "conversations.json" not in zip_ref.namelist():
            return False
    return True


def default_output_folder() -> str:
    """Returns the default output folder path.

    (put the function in a separate file to isolate file system operations)"""

    return str(Path.home() / "Documents" / "ChatGPT Data")


def get_openai_zip_filepath() -> str:
    """Returns the path to the most recent zip file in the Downloads folder,
    excluding those containing 'bookmarklet'."""

    downloads_folder = Path.home() / "Downloads"

    # Filter out zip files with names that contain "bookmarklet"
    zip_files = (
        x for x in downloads_folder.glob("*.zip") if "bookmarklet" not in x.name
    )

    # Most recent zip file in downloads folder, excluding those containing "bookmarklet"
    default_zip_filepath: Path = max(zip_files, key=lambda x: x.stat().st_ctime)

    return str(default_zip_filepath)


def get_bookmarklet_json_filepath() -> Optional[Path]:
    """Returns the path to the most recent json file in the Downloads folder,
    containing 'bookmarklet'."""

    downloads_folder = Path.home() / "Downloads"

    # Filter out json files with names that do not contain "bookmarklet"
    bookmarklet_json_files = [
        x for x in downloads_folder.glob("*.json") if "bookmarklet" in x.name
    ]

    if not bookmarklet_json_files:
        return None

    # Most recent json file in downloads folder, containing "bookmarklet"
    bookmarklet_json_filepath: Path = max(
        bookmarklet_json_files, key=lambda x: x.stat().st_ctime
    )

    return bookmarklet_json_filepath

utils/utils.py
mohamed-chs/chatgpt-history-export-to-md
"""Module for the questionary user interface."""

from typing import Any, Dict

import questionary

from utils.utils import get_colormap_names, get_font_names, validate_zip_file


def validate_header(text: str) -> bool:
    """Returns True if the given text is a valid markdown header."""
    return text.startswith("#")


def prompt_user(default_configs: Dict[str, Any]) -> Dict[str, Any]:
    """Prompts the user for input and returns the choices as a dictionary."""

    custom_style = questionary.Style(
        [
            ("qmark", "fg:#34eb9b bold"),
            ("question", "bold fg:#e0e0e0"),
            ("answer", "fg:#34ebeb bold"),
            ("pointer", "fg:#e834eb bold"),
            ("highlighted", "fg:#349ceb bold"),
            ("selected", "fg:#34ebeb"),
            ("separator", "fg:#eb3434"),
            ("instruction", "fg:#eb9434"),
            ("text", "fg:#b2eb34"),
            ("disabled", "fg:#858585 italic"),
        ]
    )

    user_configs: Dict[str, Any] = {}

    # ------------------------ zip file and output folder ------------------------

    user_configs["zip_file"] = questionary.path(
        "Enter the path to the zip file :",
        default=default_configs["zip_file"],
        validate=validate_zip_file,
        style=custom_style,
    ).ask()

    user_configs["output_folder"] = questionary.path(
        "Enter the path to the output folder :",
        default=default_configs["output_folder"],
        style=custom_style,
    ).ask()

    # --------------------- message -----------------------

    user_configs["message"] = {}

    # ------------------------ author headers ------------------------

    user_configs["message"]["author_headers"] = {}

    for author_role in default_configs["message"]["author_headers"].keys():
        user_configs["message"]["author_headers"][author_role] = questionary.text(
            f"Enter the message header (#) for messages from '{author_role}' :",
            default=default_configs["message"]["author_headers"][author_role],
            validate=validate_header,
            style=custom_style,
        ).ask()

    # ------------------ conversation ---------------------

    user_configs["conversation"] = {}

    # ------------------------ markdown ------------------------

    user_configs["conversation"]["markdown"] = {}

    user_configs["conversation"]["markdown"]["latex_delimiters"] = questionary.select(
        "Select the LaTeX math delimiters you want to use :",
        choices=["default", "dollar sign"],
        default=default_configs["conversation"]["markdown"]["latex_delimiters"],
        style=custom_style,
    ).ask()

    # ------------------------ yaml header ------------------------

    user_configs["conversation"]["yaml"] = {}

    yaml_choices = [
        questionary.Choice(header, checked=value)
        for header, value in default_configs["conversation"]["yaml"].items()
    ]

    selected_headers = questionary.checkbox(
        "Select the YAML metadata headers you want to include :",
        choices=yaml_choices,
        style=custom_style,
    ).ask()

    for header in default_configs["conversation"]["yaml"].keys():
        user_configs["conversation"]["yaml"][header] = header in selected_headers

    # ------------------------ word clouds ------------------------

    user_configs["wordcloud"] = {}

    font_names = get_font_names()

    user_configs["wordcloud"]["font"] = questionary.select(
        "Select the font you want to use for the word clouds :",
        choices=font_names,
        default=default_configs["wordcloud"]["font"]
        if default_configs["wordcloud"]["font"]
        else None,
        style=custom_style,
    ).ask()

    colormaps_list = get_colormap_names()

    user_configs["wordcloud"]["colormap"] = questionary.select(
        "Select the color theme you want to use for the word clouds :",
        choices=colormaps_list,
        default=default_configs["wordcloud"]["colormap"]
        if default_configs["wordcloud"]["colormap"]
        else None,
        style=custom_style,
    ).ask()

    # ------------------ node ---------------------

    user_configs["node"] = {}

    # -------------------- conversation set --------------------

    user_configs["conversation_set"] = {}

    # ------------- return -----------------

    return user_configs

views/prompt_user.py
mohamed-chs/chatgpt-history-export-to-md