108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Literal
|
|
|
|
from pydantic import BaseModel, Field, model_validator
|
|
|
|
|
|
class EmailData(BaseModel):
|
|
subject: str
|
|
body: str
|
|
|
|
|
|
class EmailAddress(BaseModel):
|
|
name: str | None = None
|
|
address: str | None = None
|
|
|
|
|
|
class Recipient(BaseModel):
|
|
emailAddress: EmailAddress | None = None
|
|
|
|
|
|
class EmailBody(BaseModel):
|
|
contentType: str | None = None
|
|
content: str | None = None
|
|
|
|
|
|
class Flag(BaseModel):
|
|
flagStatus: str | None = None
|
|
|
|
|
|
class ClassifyRequest(BaseModel):
|
|
email_data: EmailData | None = None
|
|
provider: Literal["openai", "anthropic"] | None = None
|
|
model: str | None = None
|
|
base_url: str | None = None
|
|
api_key: str | None = Field(default=None, exclude=True)
|
|
temperature: float | None = None
|
|
|
|
id: str | None = None
|
|
internetMessageId: str | None = None
|
|
conversationId: str | None = None
|
|
subject: str | None = None
|
|
bodyPreview: str | None = None
|
|
body: EmailBody | None = None
|
|
sender: Recipient | None = None
|
|
from_: Recipient | None = Field(default=None, alias="from")
|
|
toRecipients: list[Recipient] = Field(default_factory=list)
|
|
ccRecipients: list[Recipient] = Field(default_factory=list)
|
|
bccRecipients: list[Recipient] = Field(default_factory=list)
|
|
replyTo: list[Recipient] = Field(default_factory=list)
|
|
receivedDateTime: str | None = None
|
|
sentDateTime: str | None = None
|
|
hasAttachments: bool | None = None
|
|
importance: str | None = None
|
|
isRead: bool | None = None
|
|
flag: Flag | None = None
|
|
from_address: str | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def populate_email_data(self) -> "ClassifyRequest":
|
|
subject = self.email_data.subject if self.email_data else self.subject
|
|
body = self.email_data.body if self.email_data else (self.body.content if self.body and self.body.content else None)
|
|
if not subject or not body:
|
|
raise ValueError("Request must include either email_data or Outlook subject/body.content fields")
|
|
self.email_data = EmailData(subject=subject, body=body)
|
|
if not self.from_address:
|
|
self.from_address = (
|
|
(self.from_.emailAddress.address if self.from_ and self.from_.emailAddress else None)
|
|
or (self.sender.emailAddress.address if self.sender and self.sender.emailAddress else None)
|
|
)
|
|
return self
|
|
|
|
model_config = {"populate_by_name": True}
|
|
|
|
|
|
class ClassificationDetails(BaseModel):
|
|
summary: str | None = None
|
|
suggested_title: str | None = None
|
|
suggested_notes: str | None = None
|
|
deadline: str | None = None
|
|
people: list[str] = Field(default_factory=list)
|
|
organizations: list[str] = Field(default_factory=list)
|
|
attachments_referenced: list[str] = Field(default_factory=list)
|
|
next_steps: list[str] = Field(default_factory=list)
|
|
key_points: list[str] = Field(default_factory=list)
|
|
source_signals: list[str] = Field(default_factory=list)
|
|
dedupe_key: str | None = None
|
|
|
|
|
|
class DedupeResult(BaseModel):
|
|
status: Literal["new", "duplicate", "updated"]
|
|
seen_count: int = 1
|
|
matched_on: Literal["none", "id", "conversation", "fingerprint"] = "none"
|
|
message_id: str | None = None
|
|
conversation_id: str | None = None
|
|
fingerprint: str
|
|
|
|
|
|
class ClassificationResult(BaseModel):
|
|
needs_action: bool
|
|
category: Literal["action_required", "question", "fyi", "newsletter", "promotional", "automated", "alert", "uncategorized"]
|
|
priority: Literal["high", "medium", "low"]
|
|
task_description: str | None = None
|
|
reasoning: str
|
|
confidence: float
|
|
details: ClassificationDetails | None = None
|
|
dedupe: DedupeResult | None = None
|