Sync chats and artifacts

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "claudesync"
-version = "0.3.8"
+version = "0.3.9"
 authors = [
     {name = "Jahziah Wagner", email = "[email protected]"},
 ]
@@ -26,6 +26,7 @@ dependencies = [
     "pytest",
     "pytest-cov",
     "click_completion",
+    "tqdm",
 ]
 
 [project.urls]

requirements.txt

@@ -5,4 +5,5 @@ crontab>=1.0.1
 setuptools>=65.5.1
 pytest>=8.2.2
 pytest-cov>=5.0.0
-click_completion>=0.5.2
+click_completion>=0.5.2
+tqdm>=4.66.4

src/claudesync/chat_sync.py

@@ -0,0 +1,157 @@
+import os
+import json
+import logging
+from tqdm import tqdm
+from .config_manager import ConfigManager
+from .exceptions import ConfigurationError
+
+logger = logging.getLogger(__name__)
+
+
+def sync_chats(provider, config):
+    """
+    Synchronize chats and their artifacts from the remote source.
+
+    This function fetches all chats for the active organization, saves their metadata,
+    messages, and extracts any artifacts found in the assistant's messages.
+
+    Args:
+        provider: The API provider instance.
+        config: The configuration manager instance.
+
+    Raises:
+        ConfigurationError: If required configuration settings are missing.
+    """
+    # Get the configured destinations for chats and artifacts
+    chat_destination = config.get("chat_destination")
+    artifact_destination = config.get("artifact_destination")
+    if not chat_destination or not artifact_destination:
+        raise ConfigurationError(
+            "Chat or artifact destination not set. Use 'claudesync config set chat_destination <path>' and "
+            "'claudesync config set artifact_destination <path>' to set them."
+        )
+
+    # Get the active organization ID
+    organization_id = config.get("active_organization_id")
+    if not organization_id:
+        raise ConfigurationError(
+            "No active organization set. Please select an organization."
+        )
+
+    # Fetch all chats for the organization
+    logger.info(f"Fetching chats for organization {organization_id}")
+    chats = provider.get_chat_conversations(organization_id)
+    logger.info(f"Found {len(chats)} chats")
+
+    # Process each chat
+    for chat in tqdm(chats, desc="Syncing chats"):
+        logger.info(f"Processing chat {chat['uuid']}")
+        chat_folder = os.path.join(chat_destination, chat["uuid"])
+        os.makedirs(chat_folder, exist_ok=True)
+
+        # Save chat metadata
+        with open(os.path.join(chat_folder, "metadata.json"), "w") as f:
+            json.dump(chat, f, indent=2)
+
+        # Fetch full chat conversation
+        logger.info(f"Fetching full conversation for chat {chat['uuid']}")
+        full_chat = provider.get_chat_conversation(organization_id, chat["uuid"])
+
+        # Process each message in the chat
+        for message in full_chat["chat_messages"]:
+            # Save the message
+            message_file = os.path.join(chat_folder, f"{message['uuid']}.json")
+            with open(message_file, "w") as f:
+                json.dump(message, f, indent=2)
+
+            # Handle artifacts in assistant messages
+            if message["sender"] == "assistant":
+                artifacts = extract_artifacts(message["text"])
+                if artifacts:
+                    logger.info(
+                        f"Found {len(artifacts)} artifacts in message {message['uuid']}"
+                    )
+                    for artifact in artifacts:
+                        # Save each artifact
+                        artifact_file = os.path.join(
+                            artifact_destination,
+                            f"{artifact['identifier']}.{get_file_extension(artifact['type'])}",
+                        )
+                        os.makedirs(os.path.dirname(artifact_file), exist_ok=True)
+                        with open(artifact_file, "w") as f:
+                            f.write(artifact["content"])
+
+    logger.info(f"Chats synchronized to {chat_destination}")
+    logger.info(f"Artifacts synchronized to {artifact_destination}")
+
+
+def get_file_extension(artifact_type):
+    """
+    Get the appropriate file extension for a given artifact type.
+
+    Args:
+        artifact_type (str): The MIME type of the artifact.
+
+    Returns:
+        str: The corresponding file extension.
+    """
+    type_to_extension = {
+        "text/html": "html",
+        "application/vnd.ant.code": "txt",
+        "image/svg+xml": "svg",
+        "application/vnd.ant.mermaid": "mmd",
+        "application/vnd.ant.react": "jsx",
+    }
+    return type_to_extension.get(artifact_type, "txt")
+
+
+def extract_artifacts(text):
+    """
+    Extract artifacts from the given text.
+
+    This function searches for antArtifact tags in the text and extracts
+    the artifact information, including identifier, type, and content.
+
+    Args:
+        text (str): The text to search for artifacts.
+
+    Returns:
+        list: A list of dictionaries containing artifact information.
+    """
+    artifacts = []
+    start_tag = '<antArtifact undefined isClosed="true" />'
+
+    while start_tag in text:
+        start = text.index(start_tag)
+        end = text.index(end_tag, start) + len(end_tag)
+
+        artifact_text = text[start:end]
+        identifier = extract_attribute(artifact_text, "identifier")
+        artifact_type = extract_attribute(artifact_text, "type")
+        content = artifact_text[
+            artifact_text.index(">") + 1 : artifact_text.rindex("<")
+        ]
+
+        artifacts.append(
+            {"identifier": identifier, "type": artifact_type, "content": content}
+        )
+
+        text = text[end:]
+
+    return artifacts
+
+
+def extract_attribute(text, attribute):
+    """
+    Extract the value of a specific attribute from an XML-like tag.
+
+    Args:
+        text (str): The XML-like tag text.
+        attribute (str): The name of the attribute to extract.
+
+    Returns:
+        str: The value of the specified attribute.
+    """
+    start = text.index(f'{attribute}="') + len(f'{attribute}="')
+    end = text.index('"', start)
+    return text[start:end]

src/claudesync/cli/chat.py

@@ -0,0 +1,116 @@
+import click
+import logging
+import json
+from tqdm import tqdm
+from ..exceptions import ProviderError
+from ..utils import handle_errors, validate_and_get_provider
+from ..chat_sync import sync_chats
+
+logger = logging.getLogger(__name__)
+
+
+@click.group()
+def chat():
+    """Manage and synchronize chats."""
+    pass
+
+
+@chat.command()
+@click.pass_obj
+@handle_errors
+def sync(config):
+    """Synchronize chats and their artifacts from the remote source."""
+    provider = validate_and_get_provider(config)
+    sync_chats(provider, config)
+
+
+@chat.command()
+@click.pass_obj
+@handle_errors
+def list(config):
+    """List all chats."""
+    provider = validate_and_get_provider(config)
+    organization_id = config.get("active_organization_id")
+    chats = provider.get_chat_conversations(organization_id)
+    for chat in chats:
+        project_name = chat.get("project", {}).get("name", "N/A")
+        click.echo(
+            f"UUID: {chat.get('uuid', 'Unknown')}, "
+            f"Name: {chat.get('name', 'Unnamed')}, "
+            f"Project: {project_name}, "
+            f"Updated: {chat.get('updated_at', 'Unknown')}"
+        )
+
+
+@chat.command()
+@click.option("-a", "--all", "delete_all", is_flag=True, help="Delete all chats")
+@click.pass_obj
+@handle_errors
+def delete(config, delete_all):
+    """Delete chats. Use -a to delete all chats, or select a chat to delete."""
+    provider = validate_and_get_provider(config)
+    organization_id = config.get("active_organization_id")
+
+    def delete_chats(uuids):
+        try:
+            result = provider.delete_chat(organization_id, uuids)
+            return len(result)
+        except ProviderError as e:
+            logger.error(f"Error deleting chats: {str(e)}")
+            click.echo(f"Error occurred while deleting chats: {str(e)}")
+            return 0, len(uuids)
+
+    if delete_all:
+        if click.confirm("Are you sure you want to delete all chats?"):
+            total_deleted = 0
+            with click.progressbar(length=100, label="Deleting chats") as bar:
+                while True:
+                    chats = provider.get_chat_conversations(organization_id)
+                    if not chats:
+                        break
+                    uuids_to_delete = [chat["uuid"] for chat in chats[:50]]
+                    deleted = delete_chats(uuids_to_delete)
+                    total_deleted += deleted
+                    bar.update(len(uuids_to_delete))
+            click.echo(f"Chat deletion complete. Total chats deleted: {total_deleted}")
+    else:
+        chats = provider.get_chat_conversations(organization_id)
+        if not chats:
+            click.echo("No chats found.")
+            return
+
+        click.echo("Available chats:")
+        for idx, chat in enumerate(chats, 1):
+            project_name = chat.get("project", {}).get("name", "N/A")
+            click.echo(
+                f"{idx}. Name: {chat.get('name', 'Unnamed')}, Project: {project_name}, Updated: {chat.get('updated_at', 'Unknown')}"
+            )
+
+        while True:
+            selection = click.prompt(
+                "Enter the number of the chat to delete (or 'q' to quit)", type=str
+            )
+            if selection.lower() == "q":
+                return
+
+            try:
+                selection = int(selection)
+                if 1 <= selection <= len(chats):
+                    selected_chat = chats[selection - 1]
+                    if click.confirm(
+                        f"Are you sure you want to delete the chat '{selected_chat.get('name', 'Unnamed')}'?"
+                    ):
+                        deleted, failed = delete_chats([selected_chat["uuid"]])
+                        if deleted:
+                            click.echo(
+                                f"Successfully deleted chat: {selected_chat.get('name', 'Unnamed')}"
+                            )
+                        else:
+                            click.echo(
+                                f"Failed to delete chat: {selected_chat.get('name', 'Unnamed')}"
+                            )
+                    break
+                else:
+                    click.echo("Invalid selection. Please try again.")
+            except ValueError:
+                click.echo("Invalid input. Please enter a number or 'q' to quit.")

src/claudesync/cli/main.py

@@ -2,12 +2,18 @@
 import click_completion
 import click_completion.core
 
+from claudesync.cli.chat import chat
 from claudesync.config_manager import ConfigManager
 from .api import api
 from .organization import organization
 from .project import project
 from .sync import ls, sync, schedule
 from .config import config
+import logging
+
+logging.basicConfig(
+    level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
+)
 
 click_completion.init()
 
@@ -55,6 +61,7 @@ def status(config):
 cli.add_command(sync)
 cli.add_command(schedule)
 cli.add_command(config)
+cli.add_command(chat)
 
 if __name__ == "__main__":
     cli()

src/claudesync/cli/sync.py

@@ -7,6 +7,7 @@
 from claudesync.utils import get_local_files
 from ..utils import handle_errors, validate_and_get_provider
 from ..syncmanager import SyncManager
+from ..chat_sync import sync_chats
 
 
 @click.command()
@@ -34,21 +35,21 @@ def ls(config):
 @click.pass_obj
 @handle_errors
 def sync(config):
-    """Synchronize local files with the active remote project."""
+    """Synchronize both projects and chats."""
     provider = validate_and_get_provider(config)
-    local_path = config.get("local_path")
-
-    validate_local_path(local_path)
 
+    # Sync projects
     sync_manager = SyncManager(provider, config)
     remote_files = provider.list_files(
         sync_manager.active_organization_id, sync_manager.active_project_id
     )
-    local_files = get_local_files(local_path)
-
+    local_files = get_local_files(config.get("local_path"))
     sync_manager.sync(local_files, remote_files)
+    click.echo("Project sync completed successfully.")
 
-    click.echo("Sync completed successfully.")
+    # Sync chats
+    sync_chats(provider, config)
+    click.echo("Chat sync completed successfully.")
 
 
 def validate_local_path(local_path):