first cut

.gitignore

@@ -0,0 +1,8 @@
+/*.json
+/email_examples
+__pycache__
+/maigic.db
+/secrets.sh
+/secret.sh
+/secrets
+/credentials.json

.python-version

@@ -0,0 +1 @@
+3.12

README.md

@@ -0,0 +1 @@
+Collaboration doc for this initiative: https://docs.google.com/document/d/1duhM5Ufkq_doBnMvHo65S7cplmUJw4KqQvDRfZlmj7s/edit

assistant.py

@@ -0,0 +1,203 @@
+import json
+from pprint import pformat
+from typing import Any
+from src.db.models import DB, Chat, Item, ItemType
+from src.gmail.client import Gmail
+from src.gmail.models import GMailMessage
+import logging
+
+from src.gpt.gpt import GPT, Line, Role
+from pydantic import BaseModel
+
+LOGGER = logging.getLogger(__file__)
+
+
+GMAIL_PRIMER = Line(
+    role=Role.SYSTEM,
+    # FIXME: this content needs to be tailored to the custome
+    content="""
+        I am Kamen Yotov.
+        My emails are [email protected], [email protected], [email protected].
+        
+        When I first receive an email, I want you to analyze it respond with the analysis in json as follows:
+
+        * "author"
+        * "time_received" in iso format
+        * "categories"
+            ** "urgent": bool
+            ** "important": bool
+            ** "spam": bool
+        * "summary" : summary of the content
+        * "action": proposed next action
+
+        Then answer my follow up questions in markdown supported by slack api.
+    """,
+)
+
+
+class ItemKey(BaseModel):
+    type: ItemType
+    id: str
+
+    @classmethod
+    def from_item(cls, item: Item) -> "ItemKey":
+        return ItemKey(type=item.type, id=item.id)
+
+
+class Conversation:
+    def __init__(self, db: DB, gpt: GPT, item_key: ItemKey) -> None:
+        self.db = db
+        self.gpt = gpt
+        self.item_key = item_key
+
+        self.item = (
+            db.session.query(Item)
+            .where((Item.type == item_key.type) & (Item.id == item_key.id))
+            .one()
+        )
+
+        self.thread = list(
+            db.session.query(Chat)
+            .where((Chat.type == item_key.type) & (Chat.id == item_key.id))
+            .order_by(Chat.seq)
+            .all()
+        )
+
+    def primer_lines(self) -> list[Line]:
+        return [
+            GMAIL_PRIMER,
+            Line(role=Role.USER, content=self.item.content),
+        ]
+
+    def handle_user_reply(self, message: str = "") -> str:
+        lines = self.primer_lines() + [
+            Line(role=x.role, content=x.content) for x in self.thread
+        ]
+
+        if len(lines) == 2:  # this conversation has not started yet!
+            assert message == ""  # there is no message by user yet!
+        else:
+            assert len(lines) >= 3, lines  # there should be at least 1 reply ...
+            assert lines[-1].role == Role.ASSISTANT  # ... by the assistant
+
+            assert message
+            lines.append(Line(role=Role.USER, content=message))
+
+        r = self.gpt.complete(lines)
+
+        # TODO: this is where we need to differentiate if it is a tool/function call...
+        #       for now assert it is not!
+        assert len(r.choices) == 1
+        choice = r.choices[0]
+        assert choice.finish_reason == "stop"
+        result = choice.message.content
+        assert result
+
+        self.db.session.add(
+            Chat(
+                type=self.item.type,
+                id=self.item.id,
+                role=Role.USER,
+                content=message,
+            )
+        )
+        self.db.session.add(
+            Chat(
+                type=self.item.type,
+                id=self.item.id,
+                role=Role.ASSISTANT,
+                content=result,
+            )
+        )
+        self.db.session.commit()
+
+        return result
+
+
+class Assistant:
+    def __init__(self) -> None:
+        # self.db = DB()
+        # self.gpt = GPT()
+        pass
+
+    def fetch_emails(self) -> int:
+        try:
+            gmail = Gmail()
+            db = DB()
+            new_count = 0
+            # TODO(kamen): figure out how to query for new emails only?
+            #   maybe based on date of the last email we know about?
+            #   just ask for emails that are newer than that date?
+            for message in gmail.query():
+                if (
+                    found := db.session.query(Item)
+                    .where((Item.type == ItemType.gmail) & (Item.id == message["id"]))
+                    .first()
+                ):
+                    LOGGER.debug("skipping %s already in database", found.id)
+                else:
+                    mm = gmail.get_message(message["id"])
+                    m = GMailMessage(mm)
+                    db.session.add(
+                        Item(type=ItemType.gmail, id=mm["id"], content=m.as_md())
+                    )
+                    db.session.commit()
+                    LOGGER.debug("added %s to database", mm["id"])
+                    new_count += 1
+            return new_count
+        except Exception:
+            LOGGER.exception("fetch_emails failed")
+            raise
+
+    def handle_one(self) -> dict[str, Any]:
+        try:
+            db = DB()
+            gpt = GPT()
+
+            if (
+                new_item := db.session.query(Item)
+                .where(Item.slack_thread.is_(None))
+                .first()
+            ):
+                convo = Conversation(db, gpt, ItemKey.from_item(new_item))
+
+                LOGGER.info("handling %s", new_item.id)
+
+                result = convo.handle_user_reply()
+                result = result.removeprefix("```json").removesuffix("```")
+                result = json.loads(result)
+                result["id"] = new_item.id
+
+                LOGGER.info(pformat(result))
+                return result
+        except:
+            LOGGER.exception("")
+            raise
+
+    def handle_reply(self, thread_ts: str, reply: str) -> str:
+        try:
+            db = DB()
+            gpt = GPT()
+
+            if (
+                item := db.session.query(Item)
+                .where(Item.slack_thread == thread_ts)
+                .first()
+            ):
+                convo = Conversation(db, gpt, ItemKey.from_item(item))
+
+                LOGGER.info("handling reply on %s for %s", thread_ts, item.id)
+                result = convo.handle_user_reply(reply)
+                LOGGER.info(pformat(result))
+                return result
+            else:
+                raise RuntimeError(f"item not found for {thread_ts}")
+        except:
+            LOGGER.exception("")
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+    a = Assistant()
+    a.fetch_emails()
+    a.handle_one()

pyproject.toml

@@ -0,0 +1,31 @@
+[project]
+name = "maigic"
+version = "0.1.0"
+description = "Add your description here"
+readme = "README.md"
+requires-python = ">=3.12"
+dependencies = [
+    "google-api-python-client>=2.146.0",
+    "google-auth>=2.35.0",
+    "google-auth-httplib2>=0.2.0",
+    "google-auth-oauthlib>=1.2.1",
+    "google-apps-chat>=0.1.11",
+    "openai>=1.50.2",
+    "pydantic>=2.9.2",
+    "html2text>=2024.2.26",
+    "slack-sdk>=3.33.1",
+    "sqlalchemy>=2.0.35",
+    "slack-bolt>=1.20.1",
+    "trio>=0.26.2",
+    "trio-asyncio>=0.15.0",
+    "aiohttp>=3.10.8",
+    "asyncio>=3.4.3",
+    "humanreadable>=0.4.0",
+]
+
+[tool.uv]
+dev-dependencies = [
+    "black>=24.8.0",
+    "mypy>=1.11.2",
+    "ruff>=0.6.8",
+]