diff --git a/SCHEMA.md b/SCHEMA.md new file mode 100644 index 0000000..a5e155e --- /dev/null +++ b/SCHEMA.md @@ -0,0 +1,74 @@ +# Howl Database Schema — Seed Data Guide + +This file describes what data to insert and how. The daemon populates `email_log` automatically — only the three sections below need to be seeded. + +--- + +## `customers` + `customer_emails` + +Insert the `customers` row first, then insert one or more `customer_emails` rows using the returned UUID. + +### `customers` +| Column | Type | Required | Notes | +|--------|------|----------|-------| +| `name` | text | yes | Full name | +| `company` | text | no | Company name | +| `phone` | text | no | | +| `notes` | text | no | Free-form notes | +| `is_active` | bool | no | Defaults to `true` | + +### `customer_emails` +| Column | Type | Required | Notes | +|--------|------|----------|-------| +| `customer_id` | UUID | yes | FK → `customers.id` | +| `email_address` | text | yes | Must be globally unique across this table | +| `label` | text | no | e.g. `"work"`, `"billing"` | +| `is_primary` | bool | no | Defaults to `false` — set `true` on the main address | + +--- + +## `vendors` + `vendor_emails` + +Same insert pattern as customers — parent row first, then email rows. + +### `vendors` +| Column | Type | Required | Notes | +|--------|------|----------|-------| +| `name` | text | yes | | +| `company` | text | no | | +| `service_category` | text | no | e.g. `"logistics"`, `"IT"`, `"legal"` | +| `phone` | text | no | | +| `notes` | text | no | | +| `is_active` | bool | no | Defaults to `true` | + +### `vendor_emails` +| Column | Type | Required | Notes | +|--------|------|----------|-------| +| `vendor_id` | UUID | yes | FK → `vendors.id` | +| `email_address` | text | yes | Must be globally unique across this table | +| `label` | text | no | | +| `is_primary` | bool | no | Defaults to `false` | + +--- + +## `whitelist` + +Standalone rows — no FK dependencies. At least one of `email_address` or `domain` must be set per row. + +| Column | Type | Required | Notes | +|--------|------|----------|-------| +| `email_address` | text | one of these | Exact address match e.g. `"alice@example.com"` | +| `domain` | text | one of these | Entire domain match e.g. `"partnerco.com"` | +| `description` | text | no | Why this entry is whitelisted | +| `added_by` | text | no | Who added it | +| `expires_at` | timestamptz | no | Leave null for permanent entries | +| `is_active` | bool | no | Defaults to `true` | + +--- + +## Notes + +- **`email_log`** — do not seed this table. The daemon writes to it automatically as it processes emails. +- Email addresses are unique within their respective table (`customer_emails`, `vendor_emails`). There is no DB-level constraint preventing the same address appearing in both, but it should not happen logically. +- All `id` columns are UUID and auto-generated — do not supply them unless you have a specific reason to. +- `DATABASE_URL` is in `.env` — use the `postgresql+asyncpg://` form for async code, or `postgresql+psycopg://` for sync/migration code. diff --git a/howl/config.py b/howl/config.py index 46bd6ab..0324ece 100644 --- a/howl/config.py +++ b/howl/config.py @@ -2,7 +2,7 @@ from __future__ import annotations from typing import Literal -from pydantic import Field, field_validator +from pydantic import field_validator from pydantic_settings import BaseSettings, SettingsConfigDict @@ -16,14 +16,18 @@ class Settings(BaseSettings): graph_auth_mode: Literal["client_credentials", "delegated"] = "client_credentials" msal_token_cache_path: str = ".msal_cache.bin" graph_mailbox: str + graph_mailboxes: str = "" # comma-separated list of additional mailboxes for the scanner graph_poll_interval_seconds: int = 300 graph_batch_size: int = 50 graph_max_body_chars: int = 4000 - # Anthropic - anthropic_api_key: str + # LLM + llm_provider: Literal["anthropic", "ollama"] = "anthropic" + anthropic_api_key: str = "" anthropic_model: str = "claude-sonnet-4-6" anthropic_max_tokens: int = 1024 + ollama_base_url: str = "http://localhost:11434" + ollama_model: str = "qwen2.5:14b" llm_confidence_threshold: float = 0.60 # PostgreSQL @@ -42,6 +46,12 @@ class Settings(BaseSettings): folder_review: str = "Needs Review" folder_spam: str = "Junk Email" folder_escalate: str = "Escalate" + folder_infosec: str = "Infosec" + + + # Web UI + web_host: str = "127.0.0.1" + web_port: int = 8080 # Logging log_level: str = "INFO" diff --git a/howl/db/models.py b/howl/db/models.py index 273f504..f42ace1 100644 --- a/howl/db/models.py +++ b/howl/db/models.py @@ -9,6 +9,7 @@ from sqlalchemy import ( CheckConstraint, DateTime, Enum, + ForeignKey, Index, Integer, Numeric, @@ -59,6 +60,7 @@ class CustomerEmail(Base): id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) customer_id: Mapped[uuid.UUID] = mapped_column( UUID(as_uuid=True), + ForeignKey("customers.id", ondelete="CASCADE"), nullable=False, ) email_address: Mapped[str] = mapped_column(Text, nullable=False) @@ -99,7 +101,11 @@ class VendorEmail(Base): __table_args__ = (UniqueConstraint("email_address", name="uq_vendor_email"),) id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) - vendor_id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), nullable=False) + vendor_id: Mapped[uuid.UUID] = mapped_column( + UUID(as_uuid=True), + ForeignKey("vendors.id", ondelete="CASCADE"), + nullable=False, + ) email_address: Mapped[str] = mapped_column(Text, nullable=False) label: Mapped[Optional[str]] = mapped_column(Text) is_primary: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) @@ -133,9 +139,22 @@ class Whitelist(Base): # --------------------------------------------------------------------------- -# Email Log +# Purge Rules # --------------------------------------------------------------------------- + +class PurgeRule(Base): + __tablename__ = "purge_rules" + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + email_address: Mapped[str] = mapped_column(Text, nullable=False, unique=True) + display_name: Mapped[Optional[str]] = mapped_column(Text) + older_than_days: Mapped[int] = mapped_column(Integer, nullable=False, default=30) + is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + notes: Mapped[Optional[str]] = mapped_column(Text) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + + SENDER_TYPE_ENUM = Enum( "customer", "vendor", "whitelist", "unknown", name="sender_type", create_type=False ) @@ -158,6 +177,33 @@ PROCESSING_STATUS_ENUM = Enum( ) +# --------------------------------------------------------------------------- +# Sender Profiles +# --------------------------------------------------------------------------- + + +class SenderProfile(Base): + __tablename__ = "sender_profiles" + + id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4) + email_address: Mapped[str] = mapped_column(Text, nullable=False, unique=True) + display_name: Mapped[Optional[str]] = mapped_column(Text) + sender_type: Mapped[str] = mapped_column(SENDER_TYPE_ENUM, nullable=False, default="unknown") + email_types: Mapped[list] = mapped_column(JSONB, nullable=False, default=list) + processing_instructions: Mapped[Optional[str]] = mapped_column(Text) + notes: Mapped[Optional[str]] = mapped_column(Text) + is_active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + +# --------------------------------------------------------------------------- +# Email Log +# --------------------------------------------------------------------------- + + class EmailLog(Base): __tablename__ = "email_log" diff --git a/howl/db/queries.py b/howl/db/queries.py index bcfcd93..9986ba7 100644 --- a/howl/db/queries.py +++ b/howl/db/queries.py @@ -9,7 +9,7 @@ import structlog from sqlalchemy import select, text, update from sqlalchemy.ext.asyncio import AsyncSession -from howl.db.models import EmailLog +from howl.db.models import EmailLog, SenderProfile log = structlog.get_logger(__name__) @@ -164,3 +164,16 @@ async def get_recent_logs(session: AsyncSession, limit: int = 20) -> list[EmailL select(EmailLog).order_by(EmailLog.created_at.desc()).limit(limit) ) return list(result.scalars().all()) + + +async def lookup_sender_profile( + session: AsyncSession, email_address: str +) -> Optional[SenderProfile]: + """Return the active sender profile for an address, or None.""" + from sqlalchemy import func as _func + result = await session.execute( + select(SenderProfile) + .where(_func.lower(SenderProfile.email_address) == email_address.lower()) + .where(SenderProfile.is_active == True) + ) + return result.scalar_one_or_none() diff --git a/howl/graph/client.py b/howl/graph/client.py index 185ee94..a66eae0 100644 --- a/howl/graph/client.py +++ b/howl/graph/client.py @@ -89,13 +89,17 @@ class GraphClient: folder: str = "Inbox", top: int = 50, skip_token: Optional[str] = None, + received_after: Optional[datetime] = None, ) -> tuple[list[Message], Optional[str]]: """ Return a batch of unread messages and the next skip token (or None if done). """ url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/{folder}/messages" + filter_parts = ["isRead eq false"] + if received_after is not None: + filter_parts.append(f"receivedDateTime ge {received_after.strftime('%Y-%m-%dT%H:%M:%SZ')}") params: dict = { - "$filter": "isRead eq false", + "$filter": " and ".join(filter_parts), "$top": str(top), "$select": ( "id,conversationId,subject,sender,receivedDateTime," @@ -197,6 +201,119 @@ class GraphClient: log.info("folder_created", mailbox=mailbox, folder_name=folder_name) return data["id"] + @retry( + retry=retry_if_exception(_is_retryable), + stop=stop_after_attempt(4), + wait=wait_exponential(multiplier=1, min=2, max=30), + reraise=True, + ) + def delete_message(self, mailbox: str, message_id: str) -> None: + url = f"{GRAPH_BASE}/users/{mailbox}/messages/{message_id}" + response = self._http.delete(url, headers=self._headers()) + response.raise_for_status() + + def iter_messages_in_range( + self, + mailbox: str, + received_after: datetime, + top: int = 100, + ): + """ + Yield Message objects received on or after `received_after`, most-recent first. + Follows @odata.nextLink directly to handle both $skip and $skipToken pagination. + Stops as soon as a page contains a message older than the cutoff. + """ + url = f"{GRAPH_BASE}/users/{mailbox}/messages" + params: dict = { + "$top": str(top), + "$select": "id,subject,sender,receivedDateTime,hasAttachments,importance", + "$orderby": "receivedDateTime desc", + } + + while url: + if params: + data = self._get(url, params) + params = {} # subsequent requests use nextLink which already has params + else: + data = self._get(url) + + raw = data.get("value", []) + done = False + for r in raw: + msg = _parse_message(r) + if msg.received_at is not None and msg.received_at < received_after: + done = True + break + yield msg + + if done: + break + url = data.get("@odata.nextLink", "") + + @retry( + retry=retry_if_exception(_is_retryable), + stop=stop_after_attempt(4), + wait=wait_exponential(multiplier=1, min=2, max=30), + reraise=True, + ) + def get_messages_from_sender( + self, + mailbox: str, + sender_address: str, + limit: int = 5, + ) -> list[Message]: + """Fetch recent messages from a specific sender, most recent first.""" + safe_addr = sender_address.replace("'", "''") + url = f"{GRAPH_BASE}/users/{mailbox}/mailFolders/AllItems/messages" + params: dict = { + "$filter": f"from/emailAddress/address eq '{safe_addr}'", + "$top": str(limit), + "$select": "id,conversationId,subject,sender,receivedDateTime,hasAttachments,importance", + } + data = self._get(url, params) + # Sort by received date descending (Graph doesn't guarantee order with $filter alone) + messages = [_parse_message(m) for m in data.get("value", [])] + messages.sort(key=lambda m: m.received_at or datetime.min, reverse=True) + return messages + + def collect_purgeable_messages( + self, + mailbox: str, + rules: list[tuple[str, datetime]], + ) -> list["Message"]: + """ + Single-pass full mailbox scan. Returns all messages whose sender matches + one of the (address, received_before) rule pairs. + Much more efficient than one scan per sender. + """ + # Build a lookup: addr_lower -> received_before cutoff + rule_map: dict[str, datetime] = {addr.lower(): cutoff for addr, cutoff in rules} + + to_delete: list[Message] = [] + url = f"{GRAPH_BASE}/users/{mailbox}/messages" + params: dict = { + "$top": "100", + "$select": "id,subject,sender,receivedDateTime", + "$orderby": "receivedDateTime desc", + } + + while url: + if params: + data = self._get(url, params) + params = {} + else: + data = self._get(url) + + for r in data.get("value", []): + msg = _parse_message(r) + cutoff = rule_map.get(msg.sender_address.lower()) + if cutoff and msg.received_at and msg.received_at < cutoff: + to_delete.append(msg) + + url = data.get("@odata.nextLink", "") + + return to_delete + def close(self) -> None: self._http.close() diff --git a/howl/llm/client.py b/howl/llm/client.py index 82d5e17..adbeb42 100644 --- a/howl/llm/client.py +++ b/howl/llm/client.py @@ -1,23 +1,182 @@ from __future__ import annotations +import json +from typing import TYPE_CHECKING, Any, Optional + +import httpx import structlog -from anthropic import Anthropic from tenacity import retry, retry_if_exception_type, stop_after_attempt, wait_exponential from howl.config import Settings from howl.db.queries import SenderMatch from howl.graph.client import Message -from howl.llm.prompts import SYSTEM_PROMPT, build_prompt -from howl.llm.schemas import CLASSIFY_EMAIL_TOOL, EmailClassification +from howl.llm.prompts import PROFILE_RECOMMEND_SYSTEM_PROMPT, SYSTEM_PROMPT, build_profile_recommend_prompt, build_prompt +from howl.llm.schemas import CLASSIFY_EMAIL_TOOL, RECOMMEND_PROFILE_TOOL, EmailClassification, ProfileRecommendation + +if TYPE_CHECKING: + from datetime import datetime + + from howl.db.models import SenderProfile log = structlog.get_logger(__name__) -class LLMClient: +# --------------------------------------------------------------------------- +# Anthropic-format tool → OpenAI-format tool +# --------------------------------------------------------------------------- + +def _to_openai_tool(tool: dict) -> dict: + return { + "type": "function", + "function": { + "name": tool["name"], + "description": tool.get("description", ""), + "parameters": tool["input_schema"], + }, + } + + +# --------------------------------------------------------------------------- +# Provider backends +# --------------------------------------------------------------------------- + +class _AnthropicBackend: def __init__(self, settings: Settings) -> None: + from anthropic import Anthropic + self._client = Anthropic(api_key=settings.anthropic_api_key) self._model = settings.anthropic_model self._max_tokens = settings.anthropic_max_tokens + + def call( + self, + system: str, + user_prompt: str, + tool: dict, + tool_name: str, + ) -> tuple[dict, dict]: + """Returns (tool_input_dict, raw_response_dict).""" + response = self._client.messages.create( + model=self._model, + max_tokens=self._max_tokens, + system=system, + tools=[tool], + tool_choice={"type": "tool", "name": tool_name}, + messages=[{"role": "user", "content": user_prompt}], + ) + + tool_use_block = next( + (block for block in response.content if block.type == "tool_use"), + None, + ) + if tool_use_block is None: + raise ValueError(f"LLM did not call the {tool_name} tool") + + raw_response = { + "model": response.model, + "stop_reason": response.stop_reason, + "usage": { + "input_tokens": response.usage.input_tokens, + "output_tokens": response.usage.output_tokens, + }, + "tool_input": tool_use_block.input, + } + return tool_use_block.input, raw_response + + +class _OllamaBackend: + """Uses Ollama's native /api/chat with JSON mode instead of tool use, + which is more reliable across quantised models.""" + + def __init__(self, settings: Settings) -> None: + self._base_url = settings.ollama_base_url.rstrip("/") + self._model = settings.ollama_model + self._http = httpx.Client(timeout=180) + + @staticmethod + def _schema_instruction(tool: dict) -> str: + """Convert a tool definition into a JSON-mode prompt suffix.""" + schema = tool["input_schema"] + props = schema.get("properties", {}) + required = schema.get("required", []) + + lines = [ + "", + "Respond with a JSON object containing exactly these fields:", + ] + for name, prop in props.items(): + req = " (required)" if name in required else " (optional)" + desc = prop.get("description", "") + ptype = prop.get("type", "") + enum = prop.get("enum") + items_enum = prop.get("items", {}).get("enum") if prop.get("type") == "array" else None + + constraint = "" + if enum: + constraint = f", one of: {', '.join(enum)}" + elif items_enum: + constraint = f", array of values from: {', '.join(items_enum)}" + + lines.append(f"- {name} ({ptype}{constraint}){req}: {desc}") + + return "\n".join(lines) + + def call( + self, + system: str, + user_prompt: str, + tool: dict, + tool_name: str, + ) -> tuple[dict, dict]: + """Returns (tool_input_dict, raw_response_dict).""" + schema_suffix = self._schema_instruction(tool) + + payload: dict[str, Any] = { + "model": self._model, + "messages": [ + {"role": "system", "content": system + schema_suffix}, + {"role": "user", "content": user_prompt}, + ], + "format": "json", + "stream": False, + } + + resp = self._http.post(f"{self._base_url}/api/chat", json=payload) + resp.raise_for_status() + data = resp.json() + + content = data.get("message", {}).get("content", "") + try: + tool_input = json.loads(content) + except (json.JSONDecodeError, TypeError): + raise ValueError( + f"Ollama returned invalid JSON for {tool_name}. " + f"Response: {content[:500]}" + ) + + raw_response = { + "model": data.get("model", self._model), + "stop_reason": data.get("done_reason", ""), + "usage": { + "input_tokens": data.get("prompt_eval_count", 0), + "output_tokens": data.get("eval_count", 0), + }, + "tool_input": tool_input, + } + return tool_input, raw_response + + +# --------------------------------------------------------------------------- +# Public LLMClient +# --------------------------------------------------------------------------- + +class LLMClient: + def __init__(self, settings: Settings) -> None: + self._provider = settings.llm_provider + if self._provider == "ollama": + self._backend = _OllamaBackend(settings) + else: + self._backend = _AnthropicBackend(settings) self._max_body_chars = settings.graph_max_body_chars @retry( @@ -31,52 +190,72 @@ class LLMClient: message: Message, body: str, sender_match: SenderMatch, + sender_profile: Optional["SenderProfile"] = None, ) -> tuple[EmailClassification, dict]: """ - Send the email to Claude for classification via tool use. + Send the email for classification via tool use. Returns: (EmailClassification, raw_response_dict) """ - user_prompt = build_prompt(message, body, sender_match, self._max_body_chars) + user_prompt = build_prompt(message, body, sender_match, self._max_body_chars, sender_profile) - response = self._client.messages.create( - model=self._model, - max_tokens=self._max_tokens, + tool_input, raw_response = self._backend.call( system=SYSTEM_PROMPT, - tools=[CLASSIFY_EMAIL_TOOL], - tool_choice={"type": "tool", "name": "classify_email"}, - messages=[{"role": "user", "content": user_prompt}], + user_prompt=user_prompt, + tool=CLASSIFY_EMAIL_TOOL, + tool_name="classify_email", ) - # Extract the tool use block - tool_use_block = next( - (block for block in response.content if block.type == "tool_use"), - None, - ) - - if tool_use_block is None: - raise ValueError("Claude did not call the classify_email tool") - - raw_input = tool_use_block.input - classification = EmailClassification.model_validate(raw_input) - - raw_response = { - "model": response.model, - "stop_reason": response.stop_reason, - "usage": { - "input_tokens": response.usage.input_tokens, - "output_tokens": response.usage.output_tokens, - }, - "tool_input": raw_input, - } + classification = EmailClassification.model_validate(tool_input) log.debug( "email_classified", classification=classification.classification, action=classification.action, confidence=classification.confidence, - model=response.model, + model=raw_response.get("model"), ) return classification, raw_response + + @retry( + retry=retry_if_exception_type(Exception), + stop=stop_after_attempt(3), + wait=wait_exponential(multiplier=1, min=2, max=15), + reraise=True, + ) + def recommend_profile( + self, + sender_address: str, + sender_name: str, + sample_emails: list[tuple[str, str, "Optional[datetime]", bool]], + ) -> tuple[ProfileRecommendation, dict]: + """ + Analyse sample emails from a sender and recommend profile settings. + + Returns: + (ProfileRecommendation, raw_response_dict) + """ + user_prompt = build_profile_recommend_prompt( + sender_address, sender_name, sample_emails, self._max_body_chars, + ) + + tool_input, raw_response = self._backend.call( + system=PROFILE_RECOMMEND_SYSTEM_PROMPT, + user_prompt=user_prompt, + tool=RECOMMEND_PROFILE_TOOL, + tool_name="recommend_sender_profile", + ) + + recommendation = ProfileRecommendation.model_validate(tool_input) + + log.debug( + "profile_recommended", + sender=sender_address, + sender_type=recommendation.sender_type, + email_types=recommendation.email_types, + model=raw_response.get("model"), + ) + + return recommendation, raw_response diff --git a/howl/llm/prompts.py b/howl/llm/prompts.py index 3f77b49..e5cddbc 100644 --- a/howl/llm/prompts.py +++ b/howl/llm/prompts.py @@ -1,8 +1,14 @@ from __future__ import annotations +from datetime import datetime +from typing import TYPE_CHECKING, Optional + from howl.db.queries import SenderMatch from howl.graph.client import Message +if TYPE_CHECKING: + from howl.db.models import SenderProfile + SYSTEM_PROMPT = """\ You are an email triage assistant. Your job is to classify incoming emails and \ recommend the appropriate action for each one. @@ -20,7 +26,10 @@ Classification guidelines: - customer_inquiry: A customer asking a question, requesting support, or following up - vendor_invoice: An invoice, bill, or payment request from a vendor - vendor_notification: Order confirmations, shipping notifications, account updates from vendors -- newsletter: Marketing emails, product announcements, digests (even from known contacts) +- vendor_marketing: Promotional or marketing email from a known vendor (product announcements, offers, campaigns) +- vendor_event: Event invitations, webinars, trade shows, or conference announcements from a known vendor +- infosec_advisory: Security advisories, vulnerability disclosures, threat intelligence, or cybersecurity alerts from any sender +- newsletter: Marketing emails, product announcements, digests from unknown or non-vendor senders - spam: Unsolicited promotional or bulk email from unknown senders - phishing: Emails attempting to deceive the recipient (fake invoices, credential requests, etc.) - support_request: A request for help or technical support @@ -28,14 +37,15 @@ Classification guidelines: - unknown: Cannot determine the nature of the email Action guidelines: -- move_customer: Use for customer emails that are inquiries, requests, or correspondence -- move_vendor: Use for vendor invoices, notifications, and correspondence +- inbox_keep: Use for customer emails (inquiries, correspondence) and routine vendor emails (invoices, notifications) +- move_infosec: Use for infosec_advisory emails — security alerts, vulnerability disclosures, threat intelligence +- move_vendor: Use for vendor_marketing emails +- move_review: Use for vendor_event emails, or when uncertain and human judgement is needed - move_whitelist: Use for trusted senders that are neither customers nor vendors - flag_follow_up: Use when the email needs a response but is not urgent - move_spam: Use only for high-confidence spam or newsletters from unknown senders -- move_review: Use when uncertain or when the email needs human judgement -- inbox_keep: Keep in inbox (use sparingly — prefer a more specific action) - escalate: Use for urgent, time-critical, or potentially problematic emails +- move_customer: Do not use — customer emails stay in the inbox Be conservative: when uncertain, recommend move_review and set requires_human_review=true \ rather than taking an irreversible action. Never recommend moving email from a known \ @@ -48,6 +58,7 @@ def build_prompt( body: str, sender_match: SenderMatch, max_body_chars: int = 4000, + sender_profile: Optional["SenderProfile"] = None, ) -> str: """Build the user-turn prompt string for a single email.""" lines: list[str] = [] @@ -60,6 +71,12 @@ def build_prompt( if sender_match.notes: lines.append(f"Notes: {sender_match.notes}") + if sender_profile: + if sender_profile.email_types: + lines.append(f"Known Email Types: {', '.join(sender_profile.email_types)}") + if sender_profile.processing_instructions: + lines.append(f"Processing Instructions: {sender_profile.processing_instructions}") + lines.append(f"Subject: {message.subject or '(no subject)'}") lines.append(f"Received: {message.received_at.isoformat() if message.received_at else 'unknown'}") lines.append(f"Has Attachments: {'yes' if message.has_attachments else 'no'}") @@ -75,3 +92,50 @@ def build_prompt( lines.append("---") return "\n".join(lines) + + +PROFILE_RECOMMEND_SYSTEM_PROMPT = """\ +You are analysing sample emails from a single sender to recommend sender profile \ +settings for an email triage system. + +Based on patterns in the subject lines, body text, and metadata, recommend: +1. **sender_type**: Is this sender a customer, vendor, whitelisted/trusted contact, or unknown? +2. **email_types**: What kinds of emails does this sender typically send? Choose from: \ +invoice, statement, notification, shipping, marketing, newsletter, report, support, event, other. +3. **processing_instructions**: Write clear, specific instructions for how the AI triage \ +system should handle emails from this sender. Describe what the sender typically sends, \ +notable patterns (e.g. attachments, recurring schedule, key identifiers), and the \ +recommended action. + +Be specific and practical in your processing instructions — they will be injected into \ +every future email classification prompt for this sender.\ +""" + + +def build_profile_recommend_prompt( + sender_address: str, + sender_name: str, + sample_emails: list[tuple[str, str, Optional[datetime], bool]], + max_body_chars: int = 1500, +) -> str: + """Build the user prompt for profile recommendation. + + sample_emails: list of (subject, body, received_at, has_attachments) + """ + lines: list[str] = [] + lines.append(f"Sender: {sender_name or '(no name)'} <{sender_address}>") + lines.append(f"Sample count: {len(sample_emails)}") + lines.append("") + + for i, (subject, body, received_at, has_attachments) in enumerate(sample_emails, 1): + lines.append(f"--- Email {i} ---") + lines.append(f"Subject: {subject or '(no subject)'}") + if received_at: + lines.append(f"Received: {received_at.isoformat()}") + lines.append(f"Has Attachments: {'yes' if has_attachments else 'no'}") + truncated = len(body) > max_body_chars + lines.append(f"Body{' (truncated)' if truncated else ''}:") + lines.append(body[:max_body_chars]) + lines.append("") + + return "\n".join(lines) diff --git a/howl/llm/schemas.py b/howl/llm/schemas.py index f282200..cb05b1b 100644 --- a/howl/llm/schemas.py +++ b/howl/llm/schemas.py @@ -11,6 +11,7 @@ EmailActionLiteral = Literal[ "move_customer", "move_vendor", "move_whitelist", + "move_infosec", "move_spam", "move_review", "escalate", @@ -20,6 +21,9 @@ ClassificationLiteral = Literal[ "customer_inquiry", "vendor_invoice", "vendor_notification", + "vendor_marketing", + "vendor_event", + "infosec_advisory", "newsletter", "spam", "phishing", @@ -44,6 +48,13 @@ class EmailClassification(BaseModel): return round(v, 3) +class ProfileRecommendation(BaseModel): + sender_type: Literal["customer", "vendor", "whitelist", "unknown"] + email_types: list[str] = Field(default_factory=list) + processing_instructions: str + reasoning: str + + # Tool definition passed to Claude CLASSIFY_EMAIL_TOOL = { "name": "classify_email", @@ -60,6 +71,9 @@ CLASSIFY_EMAIL_TOOL = { "customer_inquiry", "vendor_invoice", "vendor_notification", + "vendor_marketing", + "vendor_event", + "infosec_advisory", "newsletter", "spam", "phishing", @@ -83,6 +97,7 @@ CLASSIFY_EMAIL_TOOL = { "move_customer", "move_vendor", "move_whitelist", + "move_infosec", "move_spam", "move_review", "escalate", @@ -120,3 +135,54 @@ CLASSIFY_EMAIL_TOOL = { ], }, } + + +RECOMMEND_PROFILE_TOOL = { + "name": "recommend_sender_profile", + "description": ( + "Based on sample emails from a sender, recommend sender profile settings " + "including sender type, typical email types, and processing instructions." + ), + "input_schema": { + "type": "object", + "properties": { + "sender_type": { + "type": "string", + "enum": ["customer", "vendor", "whitelist", "unknown"], + "description": "The type of sender based on the email patterns observed.", + }, + "email_types": { + "type": "array", + "items": { + "type": "string", + "enum": [ + "invoice", + "statement", + "notification", + "shipping", + "marketing", + "newsletter", + "report", + "support", + "event", + "other", + ], + }, + "description": "The types of emails this sender typically sends.", + }, + "processing_instructions": { + "type": "string", + "description": ( + "Specific instructions for how the email triage system should handle " + "emails from this sender. Describe what the sender typically sends, " + "key patterns to look for, and recommended handling." + ), + }, + "reasoning": { + "type": "string", + "description": "Brief explanation of why you chose these settings based on the sample emails.", + }, + }, + "required": ["sender_type", "email_types", "processing_instructions", "reasoning"], + }, +} diff --git a/howl/main.py b/howl/main.py index 7e40a90..3a67fa1 100644 --- a/howl/main.py +++ b/howl/main.py @@ -1,6 +1,7 @@ from __future__ import annotations import asyncio +from datetime import datetime, timedelta, timezone from typing import Optional import structlog @@ -72,6 +73,305 @@ def dry_run( console.print("[green]Dry run complete. Check the email_log table for results.[/green]") +@app.command(name="run-once") +def run_once( + mailbox: str = typer.Option("lorentz@wulfconsulting.com", "--mailbox", "-m", help="Mailbox to process"), + days: int = typer.Option(10, "--days", "-d", help="Only process emails from the last N days"), + dry: bool = typer.Option(False, "--dry-run", help="Analyse without taking any actions"), +) -> None: + """ + Process all unread emails from the last N days in a single pass, then exit. + """ + import os + os.environ["GRAPH_MAILBOX"] = mailbox + if dry: + os.environ["DRY_RUN"] = "true" + + settings, graph, processor = _bootstrap() + + since = datetime.now(timezone.utc) - timedelta(days=days) + console.print(f"[cyan]Processing unread emails in {mailbox} since {since.strftime('%Y-%m-%d')}[/cyan]") + if settings.dry_run: + console.print("[yellow]DRY RUN mode — no email actions will be taken[/yellow]") + + async def _run(): + processed = 0 + skip_token = None + while True: + messages, skip_token = graph.list_unread_messages( + mailbox, top=settings.graph_batch_size, skip_token=skip_token, received_after=since + ) + for message in messages: + await processor.process_message(message) + processed += 1 + if not skip_token: + break + console.print(f"[green]Done — processed {processed} message(s).[/green]") + + asyncio.run(_run()) + + +@app.command() +def purge( + mailbox: str = typer.Option("", "--mailbox", "-m", help="Mailbox to purge (defaults to GRAPH_MAILBOX)"), +) -> None: + """ + Delete emails from senders in the purge_rules table older than their configured threshold. + Manage rules with: purge-add, purge-remove, purge-list. + """ + settings, graph, _ = _bootstrap() + target_mailbox = mailbox or settings.graph_mailbox + + async def _run(): + from howl.db import engine as db_engine + from howl.db.models import PurgeRule + from sqlalchemy import select + + async with db_engine.async_session() as session: + result = await session.execute( + select(PurgeRule).where(PurgeRule.is_active == True) + ) + rules = list(result.scalars().all()) + + if not rules: + console.print("[yellow]No active purge rules. Use 'purge-add' to add senders.[/yellow]") + return + + console.print(f"[cyan]Scanning {target_mailbox} for messages matching {len(rules)} rule(s)...[/cyan]") + rule_pairs = [ + (rule.email_address, datetime.now(timezone.utc) - timedelta(days=rule.older_than_days)) + for rule in rules + ] + to_delete = graph.collect_purgeable_messages(target_mailbox, rule_pairs) + console.print(f"[dim]Found {len(to_delete)} message(s) to delete.[/dim]") + + # Count per sender for the summary + from collections import Counter + counts: Counter = Counter() + for msg in to_delete: + counts[msg.sender_address.lower()] += 1 + graph.delete_message(target_mailbox, msg.id) + + rule_by_addr = {r.email_address: r for r in rules} + for addr, n in counts.most_common(): + rule = rule_by_addr.get(addr) + label = (rule.display_name or addr) if rule else addr + console.print(f" {label} <{addr}>: [red]{n} deleted[/red]") + + if not counts: + console.print("[green]Nothing to delete — all clean.[/green]") + else: + console.print(f"[green]Done — {sum(counts.values())} message(s) deleted total.[/green]") + + asyncio.run(_run()) + + +@app.command(name="purge-list") +def purge_list() -> None: + """Show all purge rules.""" + settings, _, _ = _bootstrap() + + async def _run(): + from howl.db import engine as db_engine + from howl.db.models import PurgeRule + from sqlalchemy import select + from rich.table import Table + + async with db_engine.async_session() as session: + result = await session.execute(select(PurgeRule).order_by(PurgeRule.created_at)) + rules = list(result.scalars().all()) + + table = Table(title="Purge Rules") + table.add_column("Sender", width=35) + table.add_column("Display Name", width=25) + table.add_column("Delete After", width=12) + table.add_column("Active", width=6) + table.add_column("Notes", width=30) + for r in rules: + table.add_row( + r.email_address, + r.display_name or "-", + f"{r.older_than_days}d", + "[green]yes[/green]" if r.is_active else "[dim]no[/dim]", + r.notes or "-", + ) + console.print(table) + + asyncio.run(_run()) + + +@app.command(name="purge-add") +def purge_add( + email: str = typer.Argument(..., help="Sender email address"), + name: str = typer.Option("", "--name", "-n", help="Display name"), + days: int = typer.Option(30, "--days", "-d", help="Delete emails older than N days"), + notes: str = typer.Option("", "--notes", help="Optional notes"), +) -> None: + """Add a sender to the purge rules.""" + settings, _, _ = _bootstrap() + + async def _run(): + from howl.db import engine as db_engine + from howl.db.models import PurgeRule + + async with db_engine.async_session() as session: + rule = PurgeRule( + email_address=email.lower().strip(), + display_name=name or None, + older_than_days=days, + notes=notes or None, + ) + session.add(rule) + console.print(f"[green]Added purge rule for {email} (>{days}d)[/green]") + + asyncio.run(_run()) + + +@app.command(name="purge-analyze") +def purge_analyze( + mailbox: str = typer.Option("lorentz@wulfconsulting.com", "--mailbox", "-m"), + days: int = typer.Option(120, "--days", "-d", help="Look-back window in days"), + min_count: int = typer.Option(2, "--min", help="Minimum email count to surface a sender"), +) -> None: + """ + Scan the last N days of mail, rank high-volume/automated senders, and + walk through them one at a time for purge-rule approval. + """ + import re + from collections import defaultdict + + settings, graph, _ = _bootstrap() + since = datetime.now(timezone.utc) - timedelta(days=days) + + # --- fetch all messages in range --- + console.print(f"[cyan]Scanning {mailbox} for the last {days} days...[/cyan]") + sender_counts: dict[str, int] = defaultdict(int) + sender_names: dict[str, str] = {} + sender_subjects: dict[str, list[str]] = defaultdict(list) + + total = 0 + for msg in graph.iter_messages_in_range(mailbox, since): + addr = msg.sender_address.lower() + sender_counts[addr] += 1 + if addr not in sender_names and msg.sender_name: + sender_names[addr] = msg.sender_name + if msg.subject and len(sender_subjects[addr]) < 3: + sender_subjects[addr].append(msg.subject) + total += 1 + if total % 100 == 0: + console.print(f"[dim] ...{total} messages scanned[/dim]") + + console.print(f"[dim]Scanned {total} messages from {len(sender_counts)} unique senders.[/dim]\n") + + # --- load exclusions: already purged + known customers/vendors --- + async def _load_exclusions() -> set[str]: + from howl.db import engine as db_engine + from howl.db.models import CustomerEmail, PurgeRule, VendorEmail + from sqlalchemy import select + + excluded: set[str] = set() + async with db_engine.async_session() as session: + for model in (PurgeRule, CustomerEmail, VendorEmail): + result = await session.execute(select(model.email_address)) + for (addr,) in result: + excluded.add(addr.lower()) + return excluded + + excluded = asyncio.run(_load_exclusions()) + + # --- score and rank candidates --- + _AUTO_PATTERNS = re.compile( + r"(no.?reply|noreply|alert|notification|notification|mailer|postmaster|" + r"donotreply|do-not-reply|bounce|automated|info@|newsletter|digest|" + r"updates@|support@|billing@|invoice|statement|confirm|accounts@|" + r"lockbox|daemon|robot|system|admin@)", + re.IGNORECASE, + ) + + def score(addr: str, count: int) -> int: + s = count * 10 + if _AUTO_PATTERNS.search(addr): + s += 50 + return s + + candidates = [ + (addr, count) + for addr, count in sender_counts.items() + if addr not in excluded and count >= min_count + ] + candidates.sort(key=lambda x: score(x[0], x[1]), reverse=True) + + if not candidates: + console.print("[green]No new candidates found.[/green]") + return + + console.print(f"Found [bold]{len(candidates)}[/bold] candidates. Reviewing top senders — y/n/q to quit.\n") + + # Collect approvals synchronously, insert all at the end in one async pass + approved: list[tuple[str, Optional[str]]] = [] + + for addr, count in candidates: + name = sender_names.get(addr, "") + subjects = sender_subjects.get(addr, []) + console.print(f"[bold]{name or addr}[/bold] [dim]<{addr}>[/dim]") + console.print(f" {count} emails in last {days} days") + for s in subjects: + console.print(f" [dim]· {s[:80]}[/dim]") + + answer = typer.prompt(" Add to purge list? [y/n/q]", default="n").strip().lower() + if answer == "q": + console.print("[dim]Stopped.[/dim]") + break + elif answer == "y": + approved.append((addr, name or None)) + console.print(f" [green]Queued.[/green]") + console.print() + + if approved: + async def _insert_approved() -> None: + from howl.config import get_settings as _gs + from howl.db.engine import init_engine as _init + from howl.db.models import PurgeRule + from howl.db import engine as db_engine + + _init(_gs()) + async with db_engine.async_session() as session: + for addr, name in approved: + session.add(PurgeRule( + email_address=addr, + display_name=name, + older_than_days=30, + )) + + asyncio.run(_insert_approved()) + console.print(f"[green]Added {len(approved)} rule(s). Run 'purge' to action them.[/green]") + else: + console.print("[dim]Nothing added.[/dim]") + + +@app.command(name="purge-remove") +def purge_remove( + email: str = typer.Argument(..., help="Sender email address to remove"), +) -> None: + """Deactivate a purge rule (does not delete the row).""" + settings, _, _ = _bootstrap() + + async def _run(): + from howl.db import engine as db_engine + from howl.db.models import PurgeRule + from sqlalchemy import select, update + + async with db_engine.async_session() as session: + await session.execute( + update(PurgeRule) + .where(PurgeRule.email_address == email.lower().strip()) + .values(is_active=False) + ) + console.print(f"[yellow]Deactivated purge rule for {email}[/yellow]") + + asyncio.run(_run()) + + @app.command() def status( limit: int = typer.Option(20, "--limit", "-n", help="Number of recent log entries to show"), @@ -120,5 +420,35 @@ def status( asyncio.run(_run()) +@app.command() +def web( + host: Optional[str] = typer.Option(None, "--host", help="Bind host (default: WEB_HOST from config)"), + port: Optional[int] = typer.Option(None, "--port", help="Bind port (default: WEB_PORT from config)"), + reload: bool = typer.Option(False, "--reload", help="Enable auto-reload (dev mode)"), +) -> None: + """Start the Howl web UI.""" + from howl.config import get_settings + + import uvicorn + + settings = get_settings() + bind_host = host or settings.web_host + bind_port = port or settings.web_port + + console.print(f"[cyan]Starting Howl web UI on http://{bind_host}:{bind_port}[/cyan]") + + if reload: + uvicorn.run( + "howl.web.app:create_app", + factory=True, + host=bind_host, + port=bind_port, + reload=True, + ) + else: + from howl.web.app import create_app + uvicorn.run(create_app(), host=bind_host, port=bind_port) + + if __name__ == "__main__": app() diff --git a/howl/pipeline/actions.py b/howl/pipeline/actions.py index 241e03b..3bf3322 100644 --- a/howl/pipeline/actions.py +++ b/howl/pipeline/actions.py @@ -46,6 +46,10 @@ class ActionExecutor: folder_id = self._graph.get_or_create_folder(mailbox, self._settings.folder_spam) self._graph.move_message(mailbox, message_id, folder_id) + elif final_action == "move_infosec": + folder_id = self._graph.get_or_create_folder(mailbox, self._settings.folder_infosec) + self._graph.move_message(mailbox, message_id, folder_id) + elif final_action == "move_review": folder_id = self._graph.get_or_create_folder(mailbox, self._settings.folder_review) self._graph.move_message(mailbox, message_id, folder_id) diff --git a/howl/pipeline/classifier.py b/howl/pipeline/classifier.py index 0a97bae..7f0f3ec 100644 --- a/howl/pipeline/classifier.py +++ b/howl/pipeline/classifier.py @@ -24,7 +24,78 @@ def decide( final_action = llm_action overridden = False - # Rule 1: Whitelist senders are never moved to spam/junk. + # Rule 1a: Vendor invoices are always flagged for follow-up. + if classification.classification == "vendor_invoice" and final_action not in ( + "escalate", + "move_review", + ): + if final_action != "flag_follow_up": + final_action = "flag_follow_up" + overridden = True + log.info( + "classifier_override", + rule="vendor_invoice_flag_follow_up", + original_action=llm_action, + final_action=final_action, + ) + + # Rule 1b: Infosec advisories always go to the Infosec folder. + elif classification.classification == "infosec_advisory": + if final_action != "move_infosec": + final_action = "move_infosec" + overridden = True + log.info( + "classifier_override", + rule="infosec_advisory_to_folder", + original_action=llm_action, + final_action=final_action, + ) + + # Rule 1c: Vendor marketing always goes to the vendor folder. + elif classification.classification == "vendor_marketing": + if final_action != "move_vendor": + final_action = "move_vendor" + overridden = True + log.info( + "classifier_override", + rule="vendor_marketing_to_folder", + original_action=llm_action, + final_action=final_action, + ) + + # Rule 1d: Vendor events always go to review. + elif classification.classification == "vendor_event": + if final_action != "move_review": + final_action = "move_review" + overridden = True + log.info( + "classifier_override", + rule="vendor_event_to_review", + original_action=llm_action, + final_action=final_action, + ) + + # Rule 1e: All other customer and vendor emails stay in the inbox. + elif sender_match.sender_type == "customer" and final_action == "move_customer": + final_action = "inbox_keep" + overridden = True + log.info( + "classifier_override", + rule="customer_stays_in_inbox", + original_action=llm_action, + final_action=final_action, + ) + elif sender_match.sender_type == "vendor" and final_action == "move_vendor": + final_action = "inbox_keep" + overridden = True + log.info( + "classifier_override", + rule="vendor_stays_in_inbox", + original_action=llm_action, + final_action=final_action, + ) + + # Rule 2: Whitelist senders are never moved to spam/junk. if sender_match.sender_type == "whitelist" and llm_action in ("move_spam",): final_action = "inbox_keep" overridden = True diff --git a/howl/pipeline/processor.py b/howl/pipeline/processor.py index 0c09d6c..213714f 100644 --- a/howl/pipeline/processor.py +++ b/howl/pipeline/processor.py @@ -2,6 +2,7 @@ from __future__ import annotations from datetime import datetime, timezone +import httpx import structlog from howl.config import Settings @@ -60,19 +61,35 @@ class EmailProcessor: # DB sender lookup async with db_engine.async_session() as session: sender_match = await queries.lookup_sender(session, message.sender_address) + sender_profile = await queries.lookup_sender_profile(session, message.sender_address) log.info( "sender_matched", sender_type=sender_match.sender_type, entity=sender_match.entity_name, + has_profile=sender_profile is not None, message_id=message.id, ) # Fetch full email body - body = self._graph.get_message_body(self._settings.graph_mailbox, message.id) + try: + body = self._graph.get_message_body(self._settings.graph_mailbox, message.id) + except httpx.HTTPStatusError as exc: + if exc.response.status_code == 404: + log.warning("message_not_found_skipping", message_id=message.id) + async with db_engine.async_session() as session: + await queries.update_email_log( + session, + log_entry.id, + status="skipped", + error_message="message_not_found", + processing_completed_at=datetime.now(timezone.utc), + ) + return + raise # LLM classification - classification, raw_response = self._llm.classify(message, body, sender_match) + classification, raw_response = self._llm.classify(message, body, sender_match, sender_profile) # Business rule override final_action, overridden = classifier.decide( @@ -102,8 +119,10 @@ class EmailProcessor: # Execute action self._executor.execute(log_entry, final_action) - # Mark as read (no-op in dry_run — action executor already skipped) - if not self._settings.dry_run: + # Mark as read only when the message wasn't moved — after a move the + # shorthand /messages/{id} URL returns 404 on the original mailbox path. + _moves = {"move_customer", "move_vendor", "move_spam", "move_review", "move_infosec", "escalate"} + if not self._settings.dry_run and final_action not in _moves: self._graph.mark_as_read(self._settings.graph_mailbox, message.id) # Mark log as completed @@ -136,7 +155,7 @@ class EmailProcessor: retry_count=log_entry.retry_count + 1, ) - async def poll_once(self) -> int: + async def poll_once(self, received_after: datetime | None = None) -> int: """ Fetch one batch of unread messages and process each one. Returns the number of messages processed. @@ -148,7 +167,7 @@ class EmailProcessor: while True: messages, skip_token = self._graph.list_unread_messages( - mailbox, top=batch_size, skip_token=skip_token + mailbox, top=batch_size, skip_token=skip_token, received_after=received_after ) for message in messages: await self.process_message(message) @@ -190,8 +209,10 @@ class EmailProcessor: """Re-run pipeline for a failed message, updating existing log row.""" try: sender_match_result = None + sender_profile = None async with db_engine.async_session() as session: sender_match_result = await queries.lookup_sender(session, message.sender_address) + sender_profile = await queries.lookup_sender_profile(session, message.sender_address) await queries.update_email_log( session, log_entry.id, status="processing", @@ -199,11 +220,12 @@ class EmailProcessor: ) body = self._graph.get_message_body(self._settings.graph_mailbox, message.id) - classification, raw_response = self._llm.classify(message, body, sender_match_result) + classification, raw_response = self._llm.classify(message, body, sender_match_result, sender_profile) final_action, overridden = classifier.decide(sender_match_result, classification, self._settings) self._executor.execute(log_entry, final_action) - if not self._settings.dry_run: + _moves = {"move_customer", "move_vendor", "move_spam", "move_review", "move_infosec", "escalate"} + if not self._settings.dry_run and final_action not in _moves: self._graph.mark_as_read(self._settings.graph_mailbox, message.id) async with db_engine.async_session() as session: diff --git a/howl/web/__init__.py b/howl/web/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/howl/web/app.py b/howl/web/app.py new file mode 100644 index 0000000..ddcd58a --- /dev/null +++ b/howl/web/app.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from contextlib import asynccontextmanager +from pathlib import Path + +from fastapi import FastAPI +from fastapi.staticfiles import StaticFiles + +from howl.web.deps import templates # noqa: F401 — registers template dir +from howl.web.routes import analysis, dashboard, fetch, log, purge, runner, senders + +STATIC_DIR = Path(__file__).parent / "static" + + +@asynccontextmanager +async def _lifespan(app: FastAPI): + from howl.config import get_settings + from howl.db.engine import dispose_engine, init_engine + + settings = get_settings() + init_engine(settings) + yield + await dispose_engine() + + +def create_app() -> FastAPI: + app = FastAPI(title="Howl", lifespan=_lifespan) + + STATIC_DIR.mkdir(exist_ok=True) + app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") + + app.include_router(dashboard.router) + app.include_router(fetch.router) + app.include_router(purge.router) + app.include_router(log.router) + app.include_router(analysis.router) + app.include_router(runner.router) + app.include_router(senders.router) + + return app diff --git a/howl/web/deps.py b/howl/web/deps.py new file mode 100644 index 0000000..8ebc2c0 --- /dev/null +++ b/howl/web/deps.py @@ -0,0 +1,78 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import AsyncGenerator + +from fastapi.templating import Jinja2Templates +from sqlalchemy.ext.asyncio import AsyncSession + +from howl.db.engine import async_session + +TEMPLATES_DIR = Path(__file__).parent / "templates" +templates = Jinja2Templates(directory=str(TEMPLATES_DIR)) + +# Add tojson filter for use in templates (e.g. raw LLM response display) +templates.env.filters["tojson"] = lambda v, indent=None: json.dumps(v, indent=indent, default=str) + +# Human-readable labels for display +ACTION_LABELS: dict[str, str] = { + "inbox_keep": "Keep in Inbox", + "flag_follow_up": "Flagged", + "move_customer": "→ Customers", + "move_vendor": "→ Vendors", + "move_whitelist": "→ Whitelist", + "move_infosec": "→ Infosec", + "move_spam": "→ Spam", + "move_review": "→ Needs Review", + "escalate": "Escalated", +} + +CLASSIFICATION_LABELS: dict[str, str] = { + "customer_inquiry": "Customer Inquiry", + "vendor_invoice": "Vendor Invoice", + "vendor_notification": "Vendor Notification", + "vendor_marketing": "Vendor Marketing", + "vendor_event": "Vendor Event", + "infosec_advisory": "Infosec Advisory", + "newsletter": "Newsletter", + "spam": "Spam", + "phishing": "Phishing", + "support_request": "Support Request", + "urgent_action_required": "Urgent Action", + "unknown": "Unknown", +} + +SENDER_TYPE_LABELS: dict[str, str] = { + "customer": "Customer", + "vendor": "Vendor", + "whitelist": "Whitelist", + "unknown": "Unknown", +} + +STATUS_LABELS: dict[str, str] = { + "pending": "Pending", + "processing": "Processing", + "completed": "Completed", + "failed": "Failed", + "skipped": "Skipped", +} + + +EMAIL_TYPE_OPTIONS: list[tuple[str, str]] = [ + ("invoice", "Invoice / Bill"), + ("statement", "Account Statement"), + ("notification", "Notification / Confirmation"), + ("shipping", "Shipping Update"), + ("marketing", "Marketing / Promotional"), + ("newsletter", "Newsletter / Digest"), + ("report", "Report / Export"), + ("support", "Support Communication"), + ("event", "Event / Webinar Invite"), + ("other", "Other"), +] + + +async def get_db() -> AsyncGenerator[AsyncSession, None]: + async with async_session() as session: + yield session diff --git a/howl/web/routes/__init__.py b/howl/web/routes/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/howl/web/routes/analysis.py b/howl/web/routes/analysis.py new file mode 100644 index 0000000..db26bea --- /dev/null +++ b/howl/web/routes/analysis.py @@ -0,0 +1,472 @@ +from __future__ import annotations + +import asyncio +import json +from collections import defaultdict +from datetime import datetime, timedelta, timezone +from typing import Optional + +from fastapi import APIRouter, Depends, Query, Request +from fastapi.responses import HTMLResponse +from sqlalchemy import func, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from howl.db.models import CustomerEmail, EmailLog, PurgeRule, SenderProfile, VendorEmail, Whitelist +from howl.web.deps import ( + CLASSIFICATION_LABELS, + EMAIL_TYPE_OPTIONS, + SENDER_TYPE_LABELS, + get_db, + templates, +) + +router = APIRouter(prefix="/analysis") + +PRESETS = {"7": 7, "14": 14, "30": 30, "60": 60, "90": 90, "180": 180, "365": 365} +_SENDER_TYPES = ["customer", "vendor", "whitelist", "unknown"] + + +def _available_mailboxes() -> list[str]: + """Return the list of mailboxes available for scanning.""" + from howl.config import get_settings + + settings = get_settings() + if settings.graph_mailboxes: + return [m.strip() for m in settings.graph_mailboxes.split(",") if m.strip()] + return [settings.graph_mailbox] + + +def _resolve_lookback(days: Optional[str], custom_days: Optional[str]) -> tuple[int, str]: + custom_days_int = int(custom_days) if custom_days and custom_days.strip().isdigit() else None + if days == "custom" and custom_days_int: + return max(1, min(custom_days_int, 3650)), "custom" + if days in PRESETS: + return PRESETS[days], days + return 180, "180" + + +# --------------------------------------------------------------------------- +# Shared helpers +# --------------------------------------------------------------------------- + +async def _enrich_senders( + raw_senders: list[dict], + db: AsyncSession, +) -> list[dict]: + """Cross-reference sender data with DB tables for type, purge, and profile status.""" + all_addresses = [s["address"].lower() for s in raw_senders] + + customer_result = await db.execute( + select(CustomerEmail.email_address).where( + func.lower(CustomerEmail.email_address).in_(all_addresses) + ) + ) + customer_addrs = {r[0].lower() for r in customer_result} + + vendor_result = await db.execute( + select(VendorEmail.email_address).where( + func.lower(VendorEmail.email_address).in_(all_addresses) + ) + ) + vendor_addrs = {r[0].lower() for r in vendor_result} + + whitelist_result = await db.execute( + select(Whitelist.email_address).where( + Whitelist.email_address.isnot(None), + func.lower(Whitelist.email_address).in_(all_addresses), + Whitelist.is_active == True, + ) + ) + whitelist_addrs = {r[0].lower() for r in whitelist_result} + + purge_result = await db.execute( + select(PurgeRule.email_address).where(PurgeRule.is_active == True) + ) + purge_addrs = {r[0].lower() for r in purge_result} + + profile_result = await db.execute( + select(SenderProfile.email_address).where(SenderProfile.is_active == True) + ) + profile_addrs = {r[0].lower() for r in profile_result} + + def _sender_type(addr: str) -> str: + a = addr.lower() + if a in customer_addrs: + return "customer" + if a in vendor_addrs: + return "vendor" + if a in whitelist_addrs: + return "whitelist" + return "unknown" + + return [ + { + "address": s["address"], + "name": s["name"], + "email_count": s["count"], + "first_seen": s["first_seen"], + "last_seen": s["last_seen"], + "sender_type": _sender_type(s["address"]), + "sender_type_label": SENDER_TYPE_LABELS.get(_sender_type(s["address"]), "Unknown"), + "has_purge_rule": s["address"].lower() in purge_addrs, + "has_profile": s["address"].lower() in profile_addrs, + } + for s in raw_senders + ] + + +# --------------------------------------------------------------------------- +# Page shell — loads instantly, live scan triggered by htmx +# --------------------------------------------------------------------------- + +@router.get("", response_class=HTMLResponse) +async def analysis_page( + request: Request, + days: Optional[str] = Query(None), + custom_days: Optional[str] = Query(None), + tab: str = Query("analysis"), + mailbox: Optional[str] = Query(None), + db: AsyncSession = Depends(get_db), +): + # Default period: 180d for analysis, 30d for scanner + if days is None: + days = "30" if tab == "scanner" else "180" + lookback, days_param = _resolve_lookback(days, custom_days) + since = datetime.now(timezone.utc) - timedelta(days=lookback) + + mailboxes = _available_mailboxes() + selected_mailbox = mailbox if mailbox in mailboxes else mailboxes[0] + + # Chart data only needed for the analysis tab + classification_data: list[dict] = [] + daily_rows: list = [] + + if tab == "analysis": + cls_result = await db.execute( + select(EmailLog.llm_classification, func.count().label("n")) + .where(EmailLog.llm_classification.isnot(None), EmailLog.received_at >= since) + .group_by(EmailLog.llm_classification) + .order_by(func.count().desc()) + ) + classification_data = [ + { + "label": CLASSIFICATION_LABELS.get(r.llm_classification, r.llm_classification), + "count": r.n, + } + for r in cls_result + ] + + daily_result = await db.execute( + text( + """ + SELECT DATE(received_at AT TIME ZONE 'UTC') AS day, COUNT(*) AS n + FROM email_log + WHERE received_at >= :since + GROUP BY day ORDER BY day + """ + ), + {"since": since}, + ) + daily_rows = daily_result.all() + + resp = templates.TemplateResponse( + request, + "analysis.html", + { + "active_page": "analysis", + "active_tab": tab, + "lookback_days": lookback, + "lookback_param": days_param, + "custom_days": custom_days, + "since": since, + "classification_labels_json": json.dumps([d["label"] for d in classification_data]), + "classification_counts_json": json.dumps([d["count"] for d in classification_data]), + "daily_labels_json": json.dumps([str(r.day) for r in daily_rows]), + "daily_counts_json": json.dumps([r.n for r in daily_rows]), + "mailboxes": mailboxes, + "selected_mailbox": selected_mailbox, + }, + ) + resp.headers["Cache-Control"] = "no-store" + return resp + + +# --------------------------------------------------------------------------- +# Live scan — queries Graph API directly, returns table fragment +# --------------------------------------------------------------------------- + +@router.get("/live", response_class=HTMLResponse) +async def analysis_live( + request: Request, + days: Optional[str] = Query("180"), + custom_days: Optional[str] = Query(None), + db: AsyncSession = Depends(get_db), +): + lookback, _ = _resolve_lookback(days, custom_days) + since = datetime.now(timezone.utc) - timedelta(days=lookback) + + try: + raw_senders = await asyncio.to_thread(_scan_mailbox, since) + except Exception as exc: + return templates.TemplateResponse( + request, + "analysis_live_error.html", + {"error": str(exc)}, + ) + + if not raw_senders: + return HTMLResponse( + '' + "No messages found in this period." + ) + + senders = await _enrich_senders(raw_senders, db) + + resp = templates.TemplateResponse( + request, + "analysis_live_rows.html", + {"senders": senders}, + ) + resp.headers["Cache-Control"] = "no-store" + return resp + + +# --------------------------------------------------------------------------- +# Scanner — individual messages in descending chronological order +# --------------------------------------------------------------------------- + +@router.get("/scanner", response_class=HTMLResponse) +async def scanner_live( + request: Request, + days: Optional[str] = Query("30"), + custom_days: Optional[str] = Query(None), + mailbox: Optional[str] = Query(None), + db: AsyncSession = Depends(get_db), +): + lookback, _ = _resolve_lookback(days, custom_days) + since = datetime.now(timezone.utc) - timedelta(days=lookback) + + mailboxes = _available_mailboxes() + selected_mailbox = mailbox if mailbox in mailboxes else mailboxes[0] + + try: + raw_messages = await asyncio.to_thread(_scan_mailbox_messages, since, selected_mailbox) + except Exception as exc: + return templates.TemplateResponse( + request, + "analysis_live_error.html", + {"error": str(exc)}, + ) + + if not raw_messages: + return HTMLResponse( + '' + "No messages found in this period." + ) + + # Enrich with sender profile/type info + all_addresses = list({m["sender_address"].lower() for m in raw_messages}) + + customer_result = await db.execute( + select(CustomerEmail.email_address).where( + func.lower(CustomerEmail.email_address).in_(all_addresses) + ) + ) + customer_addrs = {r[0].lower() for r in customer_result} + + vendor_result = await db.execute( + select(VendorEmail.email_address).where( + func.lower(VendorEmail.email_address).in_(all_addresses) + ) + ) + vendor_addrs = {r[0].lower() for r in vendor_result} + + whitelist_result = await db.execute( + select(Whitelist.email_address).where( + Whitelist.email_address.isnot(None), + func.lower(Whitelist.email_address).in_(all_addresses), + Whitelist.is_active == True, + ) + ) + whitelist_addrs = {r[0].lower() for r in whitelist_result} + + profile_result = await db.execute( + select(SenderProfile.email_address).where(SenderProfile.is_active == True) + ) + profile_addrs = {r[0].lower() for r in profile_result} + + def _sender_type(addr: str) -> str: + a = addr.lower() + if a in customer_addrs: + return "customer" + if a in vendor_addrs: + return "vendor" + if a in whitelist_addrs: + return "whitelist" + return "unknown" + + messages = [ + { + **m, + "sender_type": _sender_type(m["sender_address"]), + "sender_type_label": SENDER_TYPE_LABELS.get(_sender_type(m["sender_address"]), "Unknown"), + "has_profile": m["sender_address"].lower() in profile_addrs, + } + for m in raw_messages + ] + + resp = templates.TemplateResponse( + request, + "analysis_scanner_rows.html", + {"messages": messages, "mailbox": selected_mailbox}, + ) + resp.headers["Cache-Control"] = "no-store" + return resp + + +# --------------------------------------------------------------------------- +# LLM profile recommendation +# --------------------------------------------------------------------------- + +@router.post("/recommend-profile", response_class=HTMLResponse) +async def recommend_profile( + request: Request, + email: str = Query(...), + name: str = Query(""), + mailbox: Optional[str] = Query(None), +): + """Fetch sample emails for a sender and ask Claude to recommend profile settings.""" + sender_types = _SENDER_TYPES + form_ctx = { + "profile": None, + "prefill_email": email, + "prefill_name": name, + "sender_types": sender_types, + "sender_type_labels": SENDER_TYPE_LABELS, + "email_type_options": EMAIL_TYPE_OPTIONS, + "prefill_sender_type": None, + "prefill_email_types": None, + "prefill_processing_instructions": None, + "prefill_reasoning": None, + } + + try: + recommendation = await asyncio.to_thread(_fetch_and_recommend, email, name, mailbox) + except Exception as exc: + form_ctx["prefill_error"] = f"AI recommendation failed: {exc}. Fill in the profile manually." + return templates.TemplateResponse(request, "sender_profile_form.html", form_ctx) + + if recommendation is None: + form_ctx["prefill_error"] = "No emails found for this sender. Fill in the profile manually." + return templates.TemplateResponse(request, "sender_profile_form.html", form_ctx) + + form_ctx.update({ + "prefill_sender_type": recommendation.sender_type, + "prefill_email_types": recommendation.email_types, + "prefill_processing_instructions": recommendation.processing_instructions, + "prefill_reasoning": recommendation.reasoning, + }) + return templates.TemplateResponse(request, "sender_profile_form.html", form_ctx) + + +def _fetch_and_recommend(email: str, name: str, mailbox: str | None = None): + """Synchronous helper: fetch sample emails from Graph, send to LLM.""" + from howl.config import get_settings + from howl.graph.auth import build_token_provider + from howl.graph.client import GraphClient + from howl.llm.client import LLMClient + + settings = get_settings() + token_provider = build_token_provider(settings) + graph = GraphClient(token_provider) + target = mailbox or settings.graph_mailbox + + try: + messages = graph.get_messages_from_sender(target, email, limit=5) + if not messages: + return None + + sample_emails = [] + for msg in messages: + body = graph.get_message_body(target, msg.id) + sample_emails.append((msg.subject or "", body, msg.received_at, msg.has_attachments)) + + llm = LLMClient(settings) + recommendation, _ = llm.recommend_profile(email, name, sample_emails) + return recommendation + finally: + graph.close() + + +# --------------------------------------------------------------------------- +# Synchronous Graph scan (runs in thread via asyncio.to_thread) +# --------------------------------------------------------------------------- + +def _scan_mailbox(since: datetime) -> list[dict]: + from howl.config import get_settings + from howl.graph.auth import build_token_provider + from howl.graph.client import GraphClient + + settings = get_settings() + token_provider = build_token_provider(settings) + graph = GraphClient(token_provider) + + counts: dict[str, int] = defaultdict(int) + names: dict[str, str] = {} + first_seen: dict[str, datetime] = {} + last_seen: dict[str, datetime] = {} + + try: + for msg in graph.iter_messages_in_range(settings.graph_mailbox, since): + addr = msg.sender_address.lower() + counts[addr] += 1 + if addr not in names and msg.sender_name: + names[addr] = msg.sender_name + if msg.received_at: + if addr not in first_seen or msg.received_at < first_seen[addr]: + first_seen[addr] = msg.received_at + if addr not in last_seen or msg.received_at > last_seen[addr]: + last_seen[addr] = msg.received_at + finally: + graph.close() + + return sorted( + [ + { + "address": addr, + "name": names.get(addr, ""), + "count": count, + "first_seen": first_seen.get(addr), + "last_seen": last_seen.get(addr), + } + for addr, count in counts.items() + ], + key=lambda x: -x["count"], + ) + + +def _scan_mailbox_messages(since: datetime, mailbox: str | None = None) -> list[dict]: + """Return individual messages (most recent first) from Graph API.""" + from howl.config import get_settings + from howl.graph.auth import build_token_provider + from howl.graph.client import GraphClient + + settings = get_settings() + token_provider = build_token_provider(settings) + graph = GraphClient(token_provider) + target = mailbox or settings.graph_mailbox + + messages: list[dict] = [] + try: + for msg in graph.iter_messages_in_range(target, since): + messages.append({ + "sender_address": msg.sender_address, + "sender_name": msg.sender_name or "", + "subject": msg.subject or "(no subject)", + "received_at": msg.received_at, + "has_attachments": msg.has_attachments, + }) + finally: + graph.close() + + return messages diff --git a/howl/web/routes/dashboard.py b/howl/web/routes/dashboard.py new file mode 100644 index 0000000..5f2bf74 --- /dev/null +++ b/howl/web/routes/dashboard.py @@ -0,0 +1,92 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from fastapi import APIRouter, Depends, Request +from fastapi.responses import HTMLResponse +from sqlalchemy import func, select, text +from sqlalchemy.ext.asyncio import AsyncSession + +from howl.db.models import EmailLog +from howl.web.deps import ACTION_LABELS, get_db, templates + +router = APIRouter() + + +@router.get("/", response_class=HTMLResponse) +async def dashboard(request: Request, db: AsyncSession = Depends(get_db)): + now = datetime.now(timezone.utc) + today_start = now.replace(hour=0, minute=0, second=0, microsecond=0) + week_start = now - timedelta(days=7) + + # Total counts + total_all = await db.scalar(select(func.count()).select_from(EmailLog)) or 0 + total_today = ( + await db.scalar( + select(func.count()).select_from(EmailLog).where(EmailLog.received_at >= today_start) + ) + or 0 + ) + total_week = ( + await db.scalar( + select(func.count()).select_from(EmailLog).where(EmailLog.received_at >= week_start) + ) + or 0 + ) + + # Failed count + failed_count = ( + await db.scalar( + select(func.count()).select_from(EmailLog).where(EmailLog.status == "failed") + ) + or 0 + ) + + # Average confidence + avg_conf = await db.scalar( + select(func.avg(EmailLog.llm_confidence)).where(EmailLog.llm_confidence.isnot(None)) + ) + avg_conf = round(float(avg_conf) * 100, 1) if avg_conf else None + + # Action breakdown + action_result = await db.execute( + select(EmailLog.final_action, func.count().label("n")) + .where(EmailLog.final_action.isnot(None)) + .group_by(EmailLog.final_action) + .order_by(func.count().desc()) + ) + action_counts = [ + {"action": row.final_action, "label": ACTION_LABELS.get(row.final_action, row.final_action), "count": row.n} + for row in action_result + ] + + # Override rate + override_count = ( + await db.scalar( + select(func.count()).select_from(EmailLog).where(EmailLog.action_overridden == True) + ) + or 0 + ) + + # Recent 10 log entries + recent_result = await db.execute( + select(EmailLog).order_by(EmailLog.received_at.desc().nullslast()).limit(10) + ) + recent = list(recent_result.scalars()) + + return templates.TemplateResponse( + request, + "dashboard.html", + { + "active_page": "dashboard", + "total_all": total_all, + "total_today": total_today, + "total_week": total_week, + "failed_count": failed_count, + "avg_conf": avg_conf, + "action_counts": action_counts, + "override_count": override_count, + "recent": recent, + "action_labels": ACTION_LABELS, + }, + ) diff --git a/howl/web/routes/fetch.py b/howl/web/routes/fetch.py new file mode 100644 index 0000000..6f73200 --- /dev/null +++ b/howl/web/routes/fetch.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +from fastapi import APIRouter, Request +from fastapi.responses import HTMLResponse + +from howl.web.deps import templates + +router = APIRouter(prefix="/fetch") + + +@router.post("", response_class=HTMLResponse) +async def fetch_mail(request: Request): + since = datetime.now(timezone.utc) - timedelta(days=14) + graph = None + try: + processor, graph = _build_processor() + processed = await processor.poll_once(received_after=since) + except Exception as exc: + return templates.TemplateResponse( + request, + "fetch_result.html", + {"error": str(exc), "processed": 0}, + ) + finally: + if graph is not None: + graph.close() + return templates.TemplateResponse( + request, + "fetch_result.html", + {"error": None, "processed": processed}, + ) + + +def _build_processor(): + from howl.config import get_settings + from howl.graph.auth import build_token_provider + from howl.graph.client import GraphClient + from howl.llm.client import LLMClient + from howl.pipeline.processor import EmailProcessor + + settings = get_settings() + token_provider = build_token_provider(settings) + graph = GraphClient(token_provider) + llm = LLMClient(settings) + return EmailProcessor(graph, llm, settings), graph diff --git a/howl/web/routes/log.py b/howl/web/routes/log.py new file mode 100644 index 0000000..20ebcc8 --- /dev/null +++ b/howl/web/routes/log.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import uuid +from typing import Optional + +from fastapi import APIRouter, Depends, Query, Request +from fastapi.responses import HTMLResponse +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from howl.db.models import EmailLog +from howl.db.queries import lookup_sender_profile +from howl.web.deps import ( + ACTION_LABELS, + CLASSIFICATION_LABELS, + SENDER_TYPE_LABELS, + STATUS_LABELS, + get_db, + templates, +) + +router = APIRouter(prefix="/log") + +PER_PAGE = 20 + +STATUSES = ["pending", "processing", "completed", "failed", "skipped"] +SENDER_TYPES = ["customer", "vendor", "whitelist", "unknown"] +CLASSIFICATIONS = list(CLASSIFICATION_LABELS.keys()) + + +@router.get("", response_class=HTMLResponse) +async def log_list( + request: Request, + page: int = Query(1, ge=1), + status: Optional[str] = Query(None), + sender_type: Optional[str] = Query(None), + classification: Optional[str] = Query(None), + q: Optional[str] = Query(None), + db: AsyncSession = Depends(get_db), +): + stmt = select(EmailLog) + + if status and status in STATUSES: + stmt = stmt.where(EmailLog.status == status) + if sender_type and sender_type in SENDER_TYPES: + stmt = stmt.where(EmailLog.sender_type == sender_type) + if classification and classification in CLASSIFICATIONS: + stmt = stmt.where(EmailLog.llm_classification == classification) + if q: + like = f"%{q}%" + stmt = stmt.where( + EmailLog.sender_address.ilike(like) | EmailLog.subject.ilike(like) + ) + + from sqlalchemy import func, select as _select + + # Total count for pagination + count_stmt = _select(func.count()).select_from(stmt.subquery()) + total = await db.scalar(count_stmt) or 0 + + stmt = stmt.order_by(EmailLog.received_at.desc().nullslast()) + stmt = stmt.offset((page - 1) * PER_PAGE).limit(PER_PAGE) + result = await db.execute(stmt) + entries = list(result.scalars()) + + total_pages = max(1, (total + PER_PAGE - 1) // PER_PAGE) + + return templates.TemplateResponse( + request, + "log.html", + { + "active_page": "log", + "entries": entries, + "page": page, + "total": total, + "total_pages": total_pages, + "per_page": PER_PAGE, + "filter_status": status or "", + "filter_sender_type": sender_type or "", + "filter_classification": classification or "", + "filter_q": q or "", + "statuses": STATUSES, + "sender_types": SENDER_TYPES, + "classifications": CLASSIFICATIONS, + "action_labels": ACTION_LABELS, + "classification_labels": CLASSIFICATION_LABELS, + "sender_type_labels": SENDER_TYPE_LABELS, + "status_labels": STATUS_LABELS, + }, + ) + + +@router.get("/{entry_id}", response_class=HTMLResponse) +async def log_detail(request: Request, entry_id: uuid.UUID, db: AsyncSession = Depends(get_db)): + from howl.web.deps import EMAIL_TYPE_OPTIONS + entry = await db.get(EmailLog, entry_id) + if not entry: + return HTMLResponse("Not found", status_code=404) + sender_profile = await lookup_sender_profile(db, entry.sender_address) + return templates.TemplateResponse( + request, + "log_detail.html", + { + "active_page": "log", + "entry": entry, + "profile": sender_profile, + "email": entry.sender_address, + "action_labels": ACTION_LABELS, + "classification_labels": CLASSIFICATION_LABELS, + "sender_type_labels": SENDER_TYPE_LABELS, + "status_labels": STATUS_LABELS, + "email_type_options": EMAIL_TYPE_OPTIONS, + }, + ) diff --git a/howl/web/routes/purge.py b/howl/web/routes/purge.py new file mode 100644 index 0000000..71aa80b --- /dev/null +++ b/howl/web/routes/purge.py @@ -0,0 +1,110 @@ +from __future__ import annotations + +import uuid +from typing import Annotated, Optional + +from fastapi import APIRouter, Depends, Form, Request +from fastapi.responses import HTMLResponse +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from howl.db.models import PurgeRule +from howl.web.deps import get_db, templates + +router = APIRouter(prefix="/purge") + + +@router.get("", response_class=HTMLResponse) +async def purge_list(request: Request, db: AsyncSession = Depends(get_db)): + result = await db.execute(select(PurgeRule).order_by(PurgeRule.created_at.desc())) + rules = list(result.scalars()) + return templates.TemplateResponse( + request, "purge.html", {"active_page": "purge", "rules": rules} + ) + + +@router.get("/new", response_class=HTMLResponse) +async def purge_new_form(request: Request): + """Return an empty add-rule form row fragment.""" + return templates.TemplateResponse(request, "purge_form_row.html", {"rule": None}) + + +@router.post("", response_class=HTMLResponse) +async def purge_create( + request: Request, + email_address: Annotated[str, Form()], + display_name: Annotated[str, Form()] = "", + older_than_days: Annotated[int, Form()] = 30, + notes: Annotated[str, Form()] = "", + db: AsyncSession = Depends(get_db), +): + # Check for duplicate + existing = await db.scalar( + select(PurgeRule).where(PurgeRule.email_address == email_address.lower().strip()) + ) + if existing: + # Re-activate if it was deactivated + existing.is_active = True + existing.display_name = display_name.strip() or None + existing.older_than_days = older_than_days + existing.notes = notes.strip() or None + rule = existing + else: + rule = PurgeRule( + email_address=email_address.lower().strip(), + display_name=display_name.strip() or None, + older_than_days=older_than_days, + notes=notes.strip() or None, + ) + db.add(rule) + await db.flush() + return templates.TemplateResponse(request, "purge_rule_row.html", {"rule": rule}) + + +@router.get("/{rule_id}/edit", response_class=HTMLResponse) +async def purge_edit_form(request: Request, rule_id: uuid.UUID, db: AsyncSession = Depends(get_db)): + rule = await db.get(PurgeRule, rule_id) + if not rule: + return HTMLResponse("", status_code=404) + return templates.TemplateResponse(request, "purge_form_row.html", {"rule": rule}) + + +@router.get("/{rule_id}/row", response_class=HTMLResponse) +async def purge_row(request: Request, rule_id: uuid.UUID, db: AsyncSession = Depends(get_db)): + """Return the read-only row HTML for a rule (used by cancel button).""" + rule = await db.get(PurgeRule, rule_id) + if not rule: + return HTMLResponse("", status_code=404) + return templates.TemplateResponse(request, "purge_rule_row.html", {"rule": rule}) + + +@router.put("/{rule_id}", response_class=HTMLResponse) +async def purge_update( + request: Request, + rule_id: uuid.UUID, + email_address: Annotated[str, Form()], + display_name: Annotated[str, Form()] = "", + older_than_days: Annotated[int, Form()] = 30, + notes: Annotated[str, Form()] = "", + is_active: Annotated[Optional[str], Form()] = None, + db: AsyncSession = Depends(get_db), +): + rule = await db.get(PurgeRule, rule_id) + if not rule: + return HTMLResponse("", status_code=404) + rule.email_address = email_address.lower().strip() + rule.display_name = display_name.strip() or None + rule.older_than_days = older_than_days + rule.notes = notes.strip() or None + rule.is_active = is_active == "on" + await db.flush() + return templates.TemplateResponse(request, "purge_rule_row.html", {"rule": rule}) + + +@router.delete("/{rule_id}", response_class=HTMLResponse) +async def purge_delete(rule_id: uuid.UUID, db: AsyncSession = Depends(get_db)): + await db.execute( + update(PurgeRule).where(PurgeRule.id == rule_id).values(is_active=False) + ) + # Return empty row so htmx swaps it out + return HTMLResponse(''.format(rule_id)) diff --git a/howl/web/routes/runner.py b/howl/web/routes/runner.py new file mode 100644 index 0000000..4f8609a --- /dev/null +++ b/howl/web/routes/runner.py @@ -0,0 +1,140 @@ +from __future__ import annotations + +import asyncio +from collections import Counter +from datetime import datetime, timedelta, timezone +from typing import Annotated, Optional + +from fastapi import APIRouter, Depends, Form, Request +from fastapi.responses import HTMLResponse +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from howl.db.models import PurgeRule +from howl.web.deps import get_db, templates + +router = APIRouter(prefix="/runner") + + +# --------------------------------------------------------------------------- +# Page +# --------------------------------------------------------------------------- + +@router.get("", response_class=HTMLResponse) +async def runner_page(request: Request, db: AsyncSession = Depends(get_db)): + result = await db.execute( + select(PurgeRule).where(PurgeRule.is_active == True).order_by(PurgeRule.email_address) + ) + active_rules = list(result.scalars()) + return templates.TemplateResponse( + request, + "runner.html", + {"active_page": "runner", "active_rules": active_rules}, + ) + + +# --------------------------------------------------------------------------- +# Preview / Execute +# --------------------------------------------------------------------------- + +@router.post("/go", response_class=HTMLResponse) +async def runner_go( + request: Request, + mode: Annotated[str, Form()], # "preview" | "execute" + source: Annotated[str, Form()], # "table" | "custom" + custom_address: Annotated[str, Form()] = "", + custom_days: Annotated[int, Form()] = 30, + db: AsyncSession = Depends(get_db), +): + # Build the list of (email_address, cutoff_datetime) rules to apply + now = datetime.now(timezone.utc) + + if source == "table": + result = await db.execute( + select(PurgeRule).where(PurgeRule.is_active == True) + ) + rules = [ + (r.email_address, now - timedelta(days=r.older_than_days)) + for r in result.scalars() + ] + rule_labels = { + r.email_address: f"{r.display_name or r.email_address} (>{r.older_than_days}d)" + for r in (await db.execute(select(PurgeRule).where(PurgeRule.is_active == True))).scalars() + } + else: + addr = custom_address.strip().lower() + if not addr: + return templates.TemplateResponse( + request, "runner_results.html", + {"error": "Email address is required for custom mode.", "mode": mode, "rows": [], "total": 0}, + ) + rules = [(addr, now - timedelta(days=custom_days))] + rule_labels = {addr: f"{addr} (>{custom_days}d)"} + + if not rules: + return templates.TemplateResponse( + request, "runner_results.html", + {"error": "No active purge rules found. Add rules on the Purge Rules page first.", "mode": mode, "rows": [], "total": 0}, + ) + + dry_run = (mode == "preview") + + try: + rows, total = await asyncio.to_thread(_run_purge, rules, rule_labels, dry_run) + except Exception as exc: + return templates.TemplateResponse( + request, "runner_results.html", + {"error": str(exc), "mode": mode, "rows": [], "total": 0}, + ) + + return templates.TemplateResponse( + request, "runner_results.html", + {"error": None, "mode": mode, "rows": rows, "total": total, "dry_run": dry_run}, + ) + + +# --------------------------------------------------------------------------- +# Synchronous Graph work (runs in thread) +# --------------------------------------------------------------------------- + +def _run_purge( + rules: list[tuple[str, datetime]], + rule_labels: dict[str, str], + dry_run: bool, +) -> tuple[list[dict], int]: + from howl.config import get_settings + from howl.graph.auth import build_token_provider + from howl.graph.client import GraphClient + + settings = get_settings() + token_provider = build_token_provider(settings) + graph = GraphClient(token_provider) + + try: + messages = graph.collect_purgeable_messages(settings.graph_mailbox, rules) + + counts: Counter[str] = Counter() + subjects: dict[str, list[str]] = {} + + for msg in messages: + addr = msg.sender_address.lower() + counts[addr] += 1 + if addr not in subjects: + subjects[addr] = [] + if len(subjects[addr]) < 3 and msg.subject: + subjects[addr].append(msg.subject) + if not dry_run: + graph.delete_message(settings.graph_mailbox, msg.id) + + rows = [ + { + "address": addr, + "label": rule_labels.get(addr, addr), + "count": count, + "subjects": subjects.get(addr, []), + } + for addr, count in counts.most_common() + ] + return rows, sum(counts.values()) + finally: + graph.close() diff --git a/howl/web/routes/senders.py b/howl/web/routes/senders.py new file mode 100644 index 0000000..a5dae3b --- /dev/null +++ b/howl/web/routes/senders.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import uuid +from typing import Annotated, Optional + +from fastapi import APIRouter, Depends, Form, Query, Request +from fastapi.responses import HTMLResponse +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from howl.db.models import SenderProfile +from howl.web.deps import EMAIL_TYPE_OPTIONS, SENDER_TYPE_LABELS, get_db, templates + +router = APIRouter(prefix="/senders") + +_SENDER_TYPES = list(SENDER_TYPE_LABELS.keys()) + + +@router.get("", response_class=HTMLResponse) +async def senders_list(request: Request, db: AsyncSession = Depends(get_db)): + result = await db.execute( + select(SenderProfile) + .where(SenderProfile.is_active == True) + .order_by(SenderProfile.updated_at.desc()) + ) + profiles = list(result.scalars()) + return templates.TemplateResponse( + request, + "senders.html", + { + "active_page": "senders", + "profiles": profiles, + "sender_type_labels": SENDER_TYPE_LABELS, + "email_type_options": EMAIL_TYPE_OPTIONS, + }, + ) + + +@router.get("/list-partial", response_class=HTMLResponse) +async def senders_list_partial(request: Request, db: AsyncSession = Depends(get_db)): + """Htmx partial: just the profile cards, used to refresh after save.""" + result = await db.execute( + select(SenderProfile) + .where(SenderProfile.is_active == True) + .order_by(SenderProfile.updated_at.desc()) + ) + profiles = list(result.scalars()) + return templates.TemplateResponse( + request, + "senders_list_partial.html", + { + "profiles": profiles, + "sender_type_labels": SENDER_TYPE_LABELS, + "email_type_options": EMAIL_TYPE_OPTIONS, + }, + ) + + +@router.get("/form", response_class=HTMLResponse) +async def sender_form( + request: Request, + email: str = Query(""), + name: str = Query(""), + profile_id: Optional[uuid.UUID] = Query(None), + db: AsyncSession = Depends(get_db), +): + """Return the modal form fragment. Pre-fill from existing profile if profile_id given.""" + profile = None + if profile_id: + profile = await db.get(SenderProfile, profile_id) + elif email: + from sqlalchemy import func + result = await db.execute( + select(SenderProfile) + .where(func.lower(SenderProfile.email_address) == email.lower()) + .where(SenderProfile.is_active == True) + ) + profile = result.scalar_one_or_none() + + return templates.TemplateResponse( + request, + "sender_profile_form.html", + { + "profile": profile, + "prefill_email": email if not profile else "", + "prefill_name": name if not profile else "", + "sender_types": _SENDER_TYPES, + "sender_type_labels": SENDER_TYPE_LABELS, + "email_type_options": EMAIL_TYPE_OPTIONS, + }, + ) + + +@router.post("", response_class=HTMLResponse) +async def sender_create( + request: Request, + email_address: Annotated[str, Form()], + display_name: Annotated[str, Form()] = "", + sender_type: Annotated[str, Form()] = "unknown", + email_types: Annotated[list[str], Form()] = [], + processing_instructions: Annotated[str, Form()] = "", + notes: Annotated[str, Form()] = "", + db: AsyncSession = Depends(get_db), +): + from sqlalchemy import func + + addr = email_address.lower().strip() + # Upsert: if a profile already exists for this address, update it + existing = await db.scalar( + select(SenderProfile) + .where(func.lower(SenderProfile.email_address) == addr) + ) + if existing: + existing.display_name = display_name.strip() or None + existing.sender_type = sender_type if sender_type in _SENDER_TYPES else "unknown" + existing.email_types = email_types + existing.processing_instructions = processing_instructions.strip() or None + existing.notes = notes.strip() or None + existing.is_active = True + profile = existing + else: + profile = SenderProfile( + email_address=addr, + display_name=display_name.strip() or None, + sender_type=sender_type if sender_type in _SENDER_TYPES else "unknown", + email_types=email_types, + processing_instructions=processing_instructions.strip() or None, + notes=notes.strip() or None, + ) + db.add(profile) + await db.flush() + + response = templates.TemplateResponse( + request, + "sender_profile_saved.html", + {"profile": profile, "sender_type_labels": SENDER_TYPE_LABELS}, + ) + response.headers["HX-Trigger"] = "profileSaved" + return response + + +@router.put("/{profile_id}", response_class=HTMLResponse) +async def sender_update( + request: Request, + profile_id: uuid.UUID, + email_address: Annotated[str, Form()], + display_name: Annotated[str, Form()] = "", + sender_type: Annotated[str, Form()] = "unknown", + email_types: Annotated[list[str], Form()] = [], + processing_instructions: Annotated[str, Form()] = "", + notes: Annotated[str, Form()] = "", + db: AsyncSession = Depends(get_db), +): + profile = await db.get(SenderProfile, profile_id) + if not profile: + return HTMLResponse("Not found", status_code=404) + + profile.email_address = email_address.lower().strip() + profile.display_name = display_name.strip() or None + profile.sender_type = sender_type if sender_type in _SENDER_TYPES else "unknown" + profile.email_types = email_types + profile.processing_instructions = processing_instructions.strip() or None + profile.notes = notes.strip() or None + await db.flush() + + response = templates.TemplateResponse( + request, + "sender_profile_saved.html", + {"profile": profile, "sender_type_labels": SENDER_TYPE_LABELS}, + ) + response.headers["HX-Trigger"] = "profileSaved" + return response + + +@router.delete("/{profile_id}", response_class=HTMLResponse) +async def sender_delete( + profile_id: uuid.UUID, + db: AsyncSession = Depends(get_db), +): + profile = await db.get(SenderProfile, profile_id) + if profile: + profile.is_active = False + await db.flush() + response = HTMLResponse("") + response.headers["HX-Trigger"] = "profileSaved" + return response + + +@router.get("/profile-card", response_class=HTMLResponse) +async def sender_profile_card( + request: Request, + email: str = Query(""), + db: AsyncSession = Depends(get_db), +): + """Partial: the profile info card for a given email address (used on log detail page).""" + profile = None + if email: + from sqlalchemy import func + result = await db.execute( + select(SenderProfile) + .where(func.lower(SenderProfile.email_address) == email.lower()) + .where(SenderProfile.is_active == True) + ) + profile = result.scalar_one_or_none() + + return templates.TemplateResponse( + request, + "sender_profile_card.html", + { + "profile": profile, + "email": email, + "sender_type_labels": SENDER_TYPE_LABELS, + "email_type_options": EMAIL_TYPE_OPTIONS, + }, + ) diff --git a/howl/web/templates/analysis.html b/howl/web/templates/analysis.html new file mode 100644 index 0000000..8d3eda0 --- /dev/null +++ b/howl/web/templates/analysis.html @@ -0,0 +1,330 @@ +{% extends "base.html" %} +{% block title %}Analysis{% endblock %} + +{% block head %} +{% if active_tab == 'analysis' %} + +{% endif %} +{% endblock %} + +{% block content %} +
+ +

Mailbox Analysis

+ + {% set period_qs = "days=" ~ lookback_param ~ ("&custom_days=" ~ (custom_days or '')) if lookback_param == 'custom' else "days=" ~ lookback_param %} + + +
+ + Analysis + + + Mailbox Scanner + +
+ + {% if active_tab == 'analysis' %} + + + +
+
+
+
+

Live Sender Analysis

+

Scanned directly from the mailbox via Graph API.

+
+
+
+ +
+ {% for label, val in [('30d','30'),('60d','60'),('90d','90'),('6m','180'),('1y','365')] %} + + {% endfor %} +
+
+ +
+
+
+
+ + + + + + + + + + + + + + + + +
Sender ↕Type ↕Count ↓First Seen ↕Last Seen ↕
+ + Scanning mailbox… +
+
+
+ + +
+ +
+
+

Daily Volume — Processed Emails

+

Emails processed by Howl in this period

+
+ +
+
+
+ +
+
+

Classification Breakdown

+

LLM classifications from processed emails

+
+ +
+
+
+ +
+ + {% elif active_tab == 'scanner' %} + + + {% set scanner_qs = period_qs ~ "&mailbox=" ~ (selected_mailbox | urlencode) %} + +
+
+
+
+

Mailbox Scanner

+

Scan the mailbox and create sender profiles. Use AI to auto-recommend profile settings.

+
+
+ {% if mailboxes | length > 1 %} + + {% endif %} +
+ + +
+ {% for label, val in [('7d','7'),('14d','14'),('30d','30'),('60d','60'),('90d','90')] %} + + {% endfor %} +
+
+ +
+
+
+
+ + + + + + + + + + + + + + +
ReceivedSenderSubject
+ + Scanning mailbox… +
+
+
+ + {% endif %} + +
+ + +{% endblock %} diff --git a/howl/web/templates/analysis_live_error.html b/howl/web/templates/analysis_live_error.html new file mode 100644 index 0000000..fa91b31 --- /dev/null +++ b/howl/web/templates/analysis_live_error.html @@ -0,0 +1,6 @@ + + +
Graph API scan failed
+
{{ error }}
+ + diff --git a/howl/web/templates/analysis_live_rows.html b/howl/web/templates/analysis_live_rows.html new file mode 100644 index 0000000..a648f2a --- /dev/null +++ b/howl/web/templates/analysis_live_rows.html @@ -0,0 +1,51 @@ +{% for s in senders %} + + +
{{ s.address }}
+ {% if s.name %}
{{ s.name }}
{% endif %} + + + + {{ s.sender_type_label }} + + + {{ s.email_count }} + + {{ s.first_seen.strftime('%Y-%m-%d') if s.first_seen else '—' }} + + + {{ s.last_seen.strftime('%Y-%m-%d') if s.last_seen else '—' }} + + +
+ {% if s.has_profile %} + + {% else %} + + {% endif %} + {% if s.has_purge_rule %} + purged + {% else %} + + {% endif %} +
+ + +{% endfor %} diff --git a/howl/web/templates/analysis_scanner_rows.html b/howl/web/templates/analysis_scanner_rows.html new file mode 100644 index 0000000..c042333 --- /dev/null +++ b/howl/web/templates/analysis_scanner_rows.html @@ -0,0 +1,60 @@ +{% for m in messages %} + + + {% if m.received_at %} + + {% else %}—{% endif %} + + +
+
+
{{ m.sender_address }}
+ {% if m.sender_name %}
{{ m.sender_name }}
{% endif %} +
+ + {{ m.sender_type_label }} + +
+ + + {{ m.subject }} + {% if m.has_attachments %} + + + + {% endif %} + + +
+ {% if m.has_profile %} + + {% else %} + + + {% endif %} +
+ + +{% endfor %} diff --git a/howl/web/templates/base.html b/howl/web/templates/base.html new file mode 100644 index 0000000..b0dcedc --- /dev/null +++ b/howl/web/templates/base.html @@ -0,0 +1,128 @@ + + + + + + {% block title %}Howl{% endblock %} — Howl + + + + {% block head %}{% endblock %} + + + + +
+ + + +
+ + +
+ + Howl +
+ + +
+ {% block content %}{% endblock %} +
+ +
+ + +
+ + +
+
+ + + +
+
+ +
+
+
+ + + + diff --git a/howl/web/templates/dashboard.html b/howl/web/templates/dashboard.html new file mode 100644 index 0000000..72597ee --- /dev/null +++ b/howl/web/templates/dashboard.html @@ -0,0 +1,164 @@ +{% extends "base.html" %} +{% block title %}Dashboard{% endblock %} + +{% block content %} +
+ +
+

Dashboard

+
+
+ +
+
+ + +
+ +
+
Emails Today
+
{{ total_today }}
+
{{ total_week }} this week
+
+ +
+
Total Processed
+
{{ total_all }}
+
all time
+
+ +
+
Failures
+
{{ failed_count }}
+
processing errors
+
+ +
+
Avg Confidence
+
+ {% if avg_conf %}{{ avg_conf }}%{% else %}—{% endif %} +
+
LLM classification
+
+ +
+ + +
+ +
+
+

Action Breakdown

+ {% if action_counts %} +
+ + + + + + + + + + {% set max_count = action_counts[0].count if action_counts else 1 %} + {% for item in action_counts %} + + + + + + {% endfor %} + +
ActionCount
{{ item.label }}{{ item.count }} +
+
+
+
+
+ {% else %} +

No data yet.

+ {% endif %} +
+
+ +
+
+

Rule Overrides

+
+ {{ override_count }} +
+

+ Times a business rule changed the LLM's recommended action. +

+
+
+ +
+ + +
+
+
+

Recent Activity

+ View all → +
+ {% if recent %} +
+ + + + + + + + + + + + + {% for e in recent %} + + + + + + + + + {% endfor %} + +
ReceivedSenderSubjectTypeActionStatus
+ {{ e.received_at.strftime('%m/%d %H:%M') if e.received_at else '—' }} + {{ e.sender_address }}{{ e.subject or '—' }} + {{ e.sender_type }} + {{ action_labels.get(e.final_action, e.final_action) }} + {% if e.status == 'completed' %} + done + {% elif e.status == 'failed' %} + failed + {% elif e.status == 'processing' %} + processing + {% else %} + {{ e.status }} + {% endif %} +
+
+ {% else %} +

No emails processed yet.

+ {% endif %} +
+
+ +
+{% endblock %} diff --git a/howl/web/templates/fetch_result.html b/howl/web/templates/fetch_result.html new file mode 100644 index 0000000..fceec66 --- /dev/null +++ b/howl/web/templates/fetch_result.html @@ -0,0 +1,22 @@ +{% if error %} +
+ + + + Fetch failed: {{ error }} +
+{% elif processed == 0 %} +
+ + + + No new messages. +
+{% else %} +
+ + + + Processed {{ processed }} message{{ 's' if processed != 1 else '' }}. +
+{% endif %} diff --git a/howl/web/templates/log.html b/howl/web/templates/log.html new file mode 100644 index 0000000..a5f9301 --- /dev/null +++ b/howl/web/templates/log.html @@ -0,0 +1,145 @@ +{% extends "base.html" %} +{% block title %}Email Log{% endblock %} + +{% block content %} +
+ +

Email Log

+ + +
+
+
+ +
+ + +
+ +
+ + +
+ +
+ + +
+ +
+ + +
+ + + {% if filter_status or filter_sender_type or filter_classification or filter_q %} + Clear + {% endif %} + +
+
+
+ + +
+ + + + + + + + + + + + + + + {% for e in entries %} + + + + + + + + + + + {% else %} + + + + {% endfor %} + +
ReceivedSenderSubjectTypeClassificationConfActionStatus
+ {{ e.received_at.strftime('%Y-%m-%d %H:%M') if e.received_at else '—' }} + +
{{ e.sender_address }}
+ {% if e.sender_name %}
{{ e.sender_name }}
{% endif %} + 		{{ e.subject or '—' }} + + {{ e.sender_type }} + + {{ classification_labels.get(e.llm_classification, e.llm_classification or '—') }} + {% if e.llm_confidence is not none %} + + {{ (e.llm_confidence * 100)|int }}% + + {% else %}—{% endif %} + + {{ action_labels.get(e.final_action, e.final_action) }} + {% if e.action_overridden %}⚡{% endif %} + + {% if e.status == 'completed' %}done + {% elif e.status == 'failed' %}failed + {% elif e.status == 'processing' %}processing + {% elif e.status == 'skipped' %}skipped + {% else %}{{ e.status }}{% endif %} +
No entries match your filters.
+
+ + + {% if total_pages > 1 %} +
+ + Showing {{ ((page - 1) * per_page) + 1 }}–{{ [page * per_page, total] | min }} of {{ total }} + +
+ {% if page > 1 %} + « + {% endif %} + {% for p in range([1, page - 2] | max, [total_pages + 1, page + 3] | min) %} + {{ p }} + {% endfor %} + {% if page < total_pages %} + » + {% endif %} +
+
+ {% else %} +

{{ total }} result{{ 's' if total != 1 else '' }}

+ {% endif %} + +
+{% endblock %} diff --git a/howl/web/templates/log_detail.html b/howl/web/templates/log_detail.html new file mode 100644 index 0000000..378031d --- /dev/null +++ b/howl/web/templates/log_detail.html @@ -0,0 +1,201 @@ +{% extends "base.html" %} +{% block title %}Email Detail{% endblock %} + +{% block content %} +
+ +
+ ← Back +

{{ entry.subject or '(no subject)' }}

+
+ + +
+ {{ status_labels.get(entry.status, entry.status) }} + {% if entry.error_message %} + — {{ entry.error_message }} + {% endif %} +
+ +
+ + +
+
+

Envelope

+
+
+
From
+
+ {{ entry.sender_name or '' }} + {% if entry.sender_name %}<{{ entry.sender_address }}>{% else %}{{ entry.sender_address }}{% endif %} +
+
+
+
Mailbox
+
{{ entry.mailbox }}
+
+
+
Received
+
{{ entry.received_at.strftime('%Y-%m-%d %H:%M:%S UTC') if entry.received_at else '—' }}
+
+
+
Attachments
+
{% if entry.has_attachments %}yes{% else %}no{% endif %}
+
+
+
Sender Type
+
+ + {{ sender_type_labels.get(entry.sender_type, entry.sender_type) }} + +
+
+
+
+
+ + +
+
+

Decision

+
+
+
Final Action
+
{{ action_labels.get(entry.final_action, entry.final_action) }}
+
+ {% if entry.action_overridden %} +
+
LLM Suggested
+
{{ action_labels.get(entry.llm_suggested_action, entry.llm_suggested_action or '—') }}
+
+
+ ⚡ Overridden by business rule +
+ {% endif %} +
+
Executed At
+
{{ entry.action_executed_at.strftime('%H:%M:%S UTC') if entry.action_executed_at else '—' }}
+
+ {% if entry.action_error %} +
+
Error
+
{{ entry.action_error }}
+
+ {% endif %} +
+
Retry Count
+
{{ entry.retry_count }}
+
+
+
+
+ +
+ + +
+
+

LLM Classification

+ {% if entry.llm_classification %} +
+
+
Classification
+
{{ classification_labels.get(entry.llm_classification, entry.llm_classification) }}
+
+
+
Confidence
+
+ {% if entry.llm_confidence is not none %}{{ (entry.llm_confidence * 100)|int }}%{% else %}—{% endif %} +
+
+
+
Model
+
{{ entry.llm_model or '—' }}
+
+
+
Tokens
+
+ {% if entry.llm_input_tokens %}{{ entry.llm_input_tokens + (entry.llm_output_tokens or 0) }}{% else %}—{% endif %} +
+ {% if entry.llm_input_tokens %} +
{{ entry.llm_input_tokens }} in / {{ entry.llm_output_tokens or 0 }} out
+ {% endif %} +
+
+ {% if entry.llm_reasoning %} +
+
Reasoning
+
{{ entry.llm_reasoning }}
+
+ {% endif %} + {% if entry.llm_raw_response %} +
+ Raw LLM response + 
{{ entry.llm_raw_response | tojson(indent=2) }}
+
+ {% endif %} + {% else %} +

No LLM classification data.

+ {% endif %} +
+
+ + +
+
+
+

Sender Profile

+ {% if not profile %} + + {% endif %} +
+
+ {% include "sender_profile_card.html" %} +
+
+
+ + +
+
+

Pipeline Timing

+
+
+
Started
+
{{ entry.processing_started_at.strftime('%H:%M:%S UTC') if entry.processing_started_at else '—' }}
+
+
+
Completed
+
{{ entry.processing_completed_at.strftime('%H:%M:%S UTC') if entry.processing_completed_at else '—' }}
+
+
+
Duration
+
+ {% if entry.processing_started_at and entry.processing_completed_at %} + {{ ((entry.processing_completed_at - entry.processing_started_at).total_seconds() * 1000)|int }}ms + {% else %}—{% endif %} +
+
+
+
Log Created
+
{{ entry.created_at.strftime('%Y-%m-%d %H:%M:%S UTC') if entry.created_at else '—' }}
+
+
+
Graph Message ID
+
{{ entry.graph_message_id[:40] }}…
+
+
+
+
+ +
+{% endblock %} diff --git a/howl/web/templates/purge.html b/howl/web/templates/purge.html new file mode 100644 index 0000000..36cf903 --- /dev/null +++ b/howl/web/templates/purge.html @@ -0,0 +1,55 @@ +{% extends "base.html" %} +{% block title %}Purge Rules{% endblock %} + +{% block content %} +
+ +
+
+

Purge Rules

+

+ Emails from these senders older than the configured threshold will be deleted when howl purge runs. +

+
+ +
+ +
+ + + + + + + + + + + + + + {% for rule in rules %} + {% include "purge_rule_row.html" %} + {% else %} + + + + {% endfor %} + +
Email AddressDisplay NameDelete AfterStatusNotesAdded
+ No purge rules yet. Click + Add Rule to get started. +
+
+ +
+ {{ rules | length }} rule{{ 's' if rules | length != 1 else '' }} total  ·  + {{ rules | selectattr('is_active') | list | length }} active +
+ +
+{% endblock %} diff --git a/howl/web/templates/purge_form_row.html b/howl/web/templates/purge_form_row.html new file mode 100644 index 0000000..2a23223 --- /dev/null +++ b/howl/web/templates/purge_form_row.html @@ -0,0 +1,51 @@ +{% if rule %} + +
+ + + + + + + + + +
+ + +
+ +
+ +{% else %} + +
+ + + + — + + + +
+ + +
+ +
+ +{% endif %} diff --git a/howl/web/templates/purge_rule_row.html b/howl/web/templates/purge_rule_row.html new file mode 100644 index 0000000..8218c76 --- /dev/null +++ b/howl/web/templates/purge_rule_row.html @@ -0,0 +1,29 @@ + + {{ rule.email_address }} + {{ rule.display_name or '—' }} + {{ rule.older_than_days }}d + + {% if rule.is_active %} + active + {% else %} + inactive + {% endif %} + + {{ rule.notes or '—' }} + + {{ rule.created_at.strftime('%Y-%m-%d') if rule.created_at else '—' }} + + +
+ + +
+ + diff --git a/howl/web/templates/runner.html b/howl/web/templates/runner.html new file mode 100644 index 0000000..29da63a --- /dev/null +++ b/howl/web/templates/runner.html @@ -0,0 +1,156 @@ +{% extends "base.html" %} +{% block title %}Run Purge{% endblock %} + +{% block content %} +
+ +
+

Run Purge

+

+ Scan the mailbox and delete matching messages. Always preview first. +

+
+ +
+
+ + +
+ +
+ + +
+
+ + +
+ {% if active_rules %} +
+ {% for r in active_rules %} +
+ {{ r.email_address }} + delete >{{ r.older_than_days }}d old +
+ {% endfor %} +
+ {% else %} +
+ No active purge rules. + Add rules → +
+ {% endif %} +
+ + + + + +
+ + +
+ +
+
+ + +
+ +
+ + + + + +{% endblock %} diff --git a/howl/web/templates/runner_results.html b/howl/web/templates/runner_results.html new file mode 100644 index 0000000..39cd47a --- /dev/null +++ b/howl/web/templates/runner_results.html @@ -0,0 +1,88 @@ +{% if error %} +
+ + + +
+
Error
+
{{ error }}
+
+
+ +{% elif rows %} +
+
+ + +
+
+ {% if dry_run %} +
Preview — no messages deleted
+
+ Found {{ total }} message{{ 's' if total != 1 else '' }} across + {{ rows | length }} sender{{ 's' if rows | length != 1 else '' }} + that match the purge rules. +
+ {% else %} +
Purge complete
+
+ Deleted {{ total }} message{{ 's' if total != 1 else '' }} across + {{ rows | length }} sender{{ 's' if rows | length != 1 else '' }}. +
+ {% endif %} +
+
+ + + + + + + + + + + + {% for row in rows %} + + + + + + {% endfor %} + +
SenderMessagesSample subjects
+
{{ row.address }}
+ {% if row.label != row.address %} +
{{ row.label }}
+ {% endif %} + 		+ {% if dry_run %}~{% endif %}{{ row.count }} + + {% for s in row.subjects %} +
{{ s }}
+ {% else %} + — + {% endfor %} +
+ + {% if dry_run %} +
+ This was a preview. Click Execute Purge to permanently delete these messages. +
+ {% endif %} + +
+
+ +{% else %} +
+ + + +
+
Nothing to purge
+
No messages matched the purge rules in the current mailbox.
+
+
+{% endif %} diff --git a/howl/web/templates/sender_profile_card.html b/howl/web/templates/sender_profile_card.html new file mode 100644 index 0000000..8ce5167 --- /dev/null +++ b/howl/web/templates/sender_profile_card.html @@ -0,0 +1,37 @@ +{% if profile %} +
+
+ + {{ sender_type_labels.get(profile.sender_type, profile.sender_type) }} + + {% for t in (profile.email_types or []) %} + {% set label = dict(email_type_options).get(t, t) %} + {{ label }} + {% endfor %} +
+ {% if profile.processing_instructions %} +
{{ profile.processing_instructions }}
+ {% endif %} + {% if profile.notes %} +

Notes: {{ profile.notes }}

+ {% endif %} +
+ +
+
+{% else %} +
+

No profile for {{ email }} yet.

+ +
+{% endif %} diff --git a/howl/web/templates/sender_profile_form.html b/howl/web/templates/sender_profile_form.html new file mode 100644 index 0000000..40a9ce4 --- /dev/null +++ b/howl/web/templates/sender_profile_form.html @@ -0,0 +1,116 @@ +{% set is_edit = profile is not none %} +{% set form_email = profile.email_address if is_edit else prefill_email %} +{% set form_name = profile.display_name or '' if is_edit else prefill_name %} + +
+

+ {% if is_edit %}Edit Sender Profile{% else %}Profile Sender{% endif %} +

+
+
+ +{% if prefill_error is defined and prefill_error %} +
+ + + + {{ prefill_error }} +
+{% endif %} + +{% if prefill_reasoning is defined and prefill_reasoning %} +
+ + + +
+
AI Recommendation
+

{{ prefill_reasoning }}

+
+
+{% endif %} + +
+ +
+
+ + +
+
+ + +
+
+ +
+ + +
+ +
+ +
+ {% for val, label in email_type_options %} + + {% endfor %} +
+
+ +
+ + +
+ +
+ + +
+ +
+ + +
+ diff --git a/howl/web/templates/sender_profile_saved.html b/howl/web/templates/sender_profile_saved.html new file mode 100644 index 0000000..e63e809 --- /dev/null +++ b/howl/web/templates/sender_profile_saved.html @@ -0,0 +1,15 @@ +
+
✓
+
+

Profile saved

+

{{ profile.email_address }}

+

+ {{ sender_type_labels.get(profile.sender_type, profile.sender_type) }} + {% if profile.email_types %}· {{ profile.email_types | join(', ') }}{% endif %} +

+
+
+
+ View All Profiles +
+
diff --git a/howl/web/templates/senders.html b/howl/web/templates/senders.html new file mode 100644 index 0000000..230af7d --- /dev/null +++ b/howl/web/templates/senders.html @@ -0,0 +1,30 @@ +{% extends "base.html" %} +{% block title %}Sender Profiles{% endblock %} + +{% block content %} +
+ +
+
+

Sender Profiles

+

+ Rich per-sender context fed to the AI on every email — classification type, handling notes, and processing instructions. +

+
+ +
+ +
+ {% include "senders_list_partial.html" %} +
+ +
+{% endblock %} diff --git a/howl/web/templates/senders_list_partial.html b/howl/web/templates/senders_list_partial.html new file mode 100644 index 0000000..7edbdd0 --- /dev/null +++ b/howl/web/templates/senders_list_partial.html @@ -0,0 +1,56 @@ +{% if profiles %} +
+ {% for p in profiles %} +
+
+
+
+
{{ p.email_address }}
+ {% if p.display_name %} +
{{ p.display_name }}
+ {% endif %} +
+ + {{ sender_type_labels.get(p.sender_type, p.sender_type) }} + +
+ + {% if p.email_types %} +
+ {% for t in p.email_types %} + {% set label = dict(email_type_options).get(t, t) %} + {{ label }} + {% endfor %} +
+ {% endif %} + + {% if p.processing_instructions %} +

{{ p.processing_instructions }}

+ {% endif %} + +
+ + +
+
+
+ {% endfor %} +
+
{{ profiles | length }} profile{{ 's' if profiles | length != 1 else '' }}
+{% else %} +
+
+

No sender profiles yet.

+

Click + New Profile or use the Profile Sender button on any email log entry.

+
+
+{% endif %} diff --git a/migrations/versions/0001_initial_schema.py b/migrations/versions/0001_initial_schema.py index 2c76ff4..cdac61d 100644 --- a/migrations/versions/0001_initial_schema.py +++ b/migrations/versions/0001_initial_schema.py @@ -148,7 +148,7 @@ def upgrade() -> None: sa.Column("subject", sa.Text), sa.Column("received_at", sa.DateTime(timezone=True)), sa.Column("has_attachments", sa.Boolean, nullable=False, server_default="FALSE"), - sa.Column("sender_type", sa.Enum("customer", "vendor", "whitelist", "unknown", name="sender_type", create_type=False), nullable=False, server_default="unknown"), + sa.Column("sender_type", postgresql.ENUM("customer", "vendor", "whitelist", "unknown", name="sender_type", create_type=False), nullable=False, server_default="unknown"), sa.Column("matched_entity_id", postgresql.UUID(as_uuid=True)), sa.Column("matched_entity_table", sa.Text), sa.Column("llm_model", sa.Text), @@ -158,12 +158,12 @@ def upgrade() -> None: sa.Column("llm_classification", sa.Text), sa.Column("llm_confidence", sa.Numeric(4, 3)), sa.Column("llm_reasoning", sa.Text), - sa.Column("llm_suggested_action", sa.Enum("inbox_keep", "flag_follow_up", "move_customer", "move_vendor", "move_whitelist", "move_spam", "move_review", "escalate", name="email_action", create_type=False)), - sa.Column("final_action", sa.Enum("inbox_keep", "flag_follow_up", "move_customer", "move_vendor", "move_whitelist", "move_spam", "move_review", "escalate", name="email_action", create_type=False), nullable=False, server_default="inbox_keep"), + sa.Column("llm_suggested_action", postgresql.ENUM("inbox_keep", "flag_follow_up", "move_customer", "move_vendor", "move_whitelist", "move_spam", "move_review", "escalate", name="email_action", create_type=False)), + sa.Column("final_action", postgresql.ENUM("inbox_keep", "flag_follow_up", "move_customer", "move_vendor", "move_whitelist", "move_spam", "move_review", "escalate", name="email_action", create_type=False), nullable=False, server_default="inbox_keep"), sa.Column("action_overridden", sa.Boolean, nullable=False, server_default="FALSE"), sa.Column("action_executed_at", sa.DateTime(timezone=True)), sa.Column("action_error", sa.Text), - sa.Column("status", sa.Enum("pending", "processing", "completed", "failed", "skipped", name="processing_status", create_type=False), nullable=False, server_default="pending"), + sa.Column("status", postgresql.ENUM("pending", "processing", "completed", "failed", "skipped", name="processing_status", create_type=False), nullable=False, server_default="pending"), sa.Column("processing_started_at", sa.DateTime(timezone=True)), sa.Column("processing_completed_at", sa.DateTime(timezone=True)), sa.Column("error_message", sa.Text), diff --git a/migrations/versions/0002_add_infosec_action.py b/migrations/versions/0002_add_infosec_action.py new file mode 100644 index 0000000..70a04f4 --- /dev/null +++ b/migrations/versions/0002_add_infosec_action.py @@ -0,0 +1,25 @@ +"""Add move_infosec to email_action enum + +Revision ID: 0002 +Revises: 0001 +Create Date: 2026-04-01 +""" +from __future__ import annotations + +from typing import Sequence, Union + +from alembic import op + +revision: str = "0002" +down_revision: Union[str, None] = "0001" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.execute("ALTER TYPE email_action ADD VALUE IF NOT EXISTS 'move_infosec'") + + +def downgrade() -> None: + # Postgres does not support removing enum values; a full recreate would be needed. + pass diff --git a/migrations/versions/0003_add_purge_rules.py b/migrations/versions/0003_add_purge_rules.py new file mode 100644 index 0000000..2268bfa --- /dev/null +++ b/migrations/versions/0003_add_purge_rules.py @@ -0,0 +1,37 @@ +"""Add purge_rules table + +Revision ID: 0003 +Revises: 0002 +Create Date: 2026-04-01 +""" +from __future__ import annotations + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0003" +down_revision: Union[str, None] = "0002" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "purge_rules", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")), + sa.Column("email_address", sa.Text, nullable=False, unique=True), + sa.Column("display_name", sa.Text), + sa.Column("older_than_days", sa.Integer, nullable=False, server_default="30"), + sa.Column("is_active", sa.Boolean, nullable=False, server_default="TRUE"), + sa.Column("notes", sa.Text), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")), + ) + op.create_index("idx_purge_rules_active", "purge_rules", ["email_address"], + postgresql_where=sa.text("is_active = TRUE")) + + +def downgrade() -> None: + op.drop_table("purge_rules") diff --git a/migrations/versions/0004_add_sender_profiles.py b/migrations/versions/0004_add_sender_profiles.py new file mode 100644 index 0000000..978cfe8 --- /dev/null +++ b/migrations/versions/0004_add_sender_profiles.py @@ -0,0 +1,54 @@ +"""Add sender_profiles table + +Revision ID: 0004 +Revises: 0003 +Create Date: 2026-04-02 +""" +from __future__ import annotations + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op +from sqlalchemy.dialects import postgresql + +revision: str = "0004" +down_revision: Union[str, None] = "0003" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "sender_profiles", + sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True, server_default=sa.text("gen_random_uuid()")), + sa.Column("email_address", sa.Text, nullable=False, unique=True), + sa.Column("display_name", sa.Text), + sa.Column( + "sender_type", + postgresql.ENUM("customer", "vendor", "whitelist", "unknown", name="sender_type", create_type=False), + nullable=False, + server_default="unknown", + ), + sa.Column("email_types", postgresql.JSONB, nullable=False, server_default=sa.text("'[]'")), + sa.Column("processing_instructions", sa.Text), + sa.Column("notes", sa.Text), + sa.Column("is_active", sa.Boolean, nullable=False, server_default="TRUE"), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.text("NOW()")), + ) + op.create_index( + "idx_sender_profiles_address", + "sender_profiles", + [sa.text("LOWER(email_address)")], + ) + op.execute(""" + CREATE TRIGGER trg_sender_profiles_updated_at + BEFORE UPDATE ON sender_profiles + FOR EACH ROW EXECUTE FUNCTION set_updated_at(); + """) + + +def downgrade() -> None: + op.execute("DROP TRIGGER IF EXISTS trg_sender_profiles_updated_at ON sender_profiles") + op.drop_table("sender_profiles") diff --git a/pyproject.toml b/pyproject.toml index c7c9a6c..42e14aa 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,6 +33,11 @@ dependencies = [ "tenacity>=9.0.0", # HTML stripping for email bodies "beautifulsoup4>=4.12.0", + # Web UI + "fastapi>=0.115.0", + "uvicorn[standard]>=0.32.0", + "jinja2>=3.1.0", + "python-multipart>=0.0.12", ] [project.scripts] diff --git a/seed_from_pulse.py b/seed_from_pulse.py new file mode 100644 index 0000000..dad9295 --- /dev/null +++ b/seed_from_pulse.py @@ -0,0 +1,170 @@ +#!/usr/bin/env python3 +""" +Seed howl database from pulse_autotask (Autotask contacts/companies). + +Sources: + customers + customer_emails <- company_type=1 (Customer) active contacts w/ emails + vendors + vendor_emails <- company_type=7 (Partner) active contacts w/ emails + whitelist <- unique domains from all client company email addresses +""" + +import psycopg2 +import re + +PULSE_PARAMS = dict(host='localhost', port=5432, user='pulse_user', + password='9KuYTjjGEB7NsJc_togj6R9wYLRRrhudZiR4@i@N', + dbname='pulse_autotask') +HOWL_PARAMS = dict(host='localhost', port=5434, user='howl', + password='ztPEGUexRYkfWqUtd15WMdJF98N1jsui', + dbname='howl') + +EMAIL_RE = re.compile(r'^[^@\s]+@[^@\s]+\.[^@\s]+$') + +def valid_email(e): + return e and EMAIL_RE.match(e.strip()) + +def domain_of(e): + return e.strip().lower().split('@')[1] if e and '@' in e else None + +def seed_contacts(pulse_cur, howl_cur, company_type, parent_table, email_table, fk_col): + """Generic: pull contacts of given company_type, insert parent rows then email rows.""" + pulse_cur.execute(""" + SELECT + trim(concat(c.first_name, ' ', c.last_name)) AS name, + co.company_name AS company, + COALESCE(NULLIF(c.phone,''), NULLIF(co.phone,'')) AS phone, + c.email_address, + c.email_address2, + c.email_address3, + c.primary_contact + FROM contacts c + JOIN companies co ON c.company_id = co.id + WHERE co.company_type = %s + AND co.is_active = true AND co.is_deleted = false + AND c.is_active = true AND c.is_deleted = false + AND c.email_address IS NOT NULL AND c.email_address != '' + ORDER BY co.company_name, c.primary_contact DESC, c.last_name, c.first_name + """, (company_type,)) + rows = pulse_cur.fetchall() + + inserted_parents = 0 + inserted_emails = 0 + skipped_emails = 0 + seen_emails = set() # guard against dupes within this batch + + for row in rows: + name, company, phone, e1, e2, e3, is_primary = row + + # Collect valid emails for this contact + emails = [] + for addr in [e1, e2, e3]: + if valid_email(addr): + emails.append(addr.strip().lower()) + if not emails: + continue + + # Insert parent row + howl_cur.execute( + f"INSERT INTO {parent_table} (name, company, phone) VALUES (%s,%s,%s) RETURNING id", + (name or 'Unknown', company, phone) + ) + parent_id = howl_cur.fetchone()[0] + inserted_parents += 1 + + # Insert email rows + for i, addr in enumerate(emails): + if addr in seen_emails: + skipped_emails += 1 + continue + try: + howl_cur.execute( + f"INSERT INTO {email_table} ({fk_col}, email_address, label, is_primary) VALUES (%s,%s,%s,%s)", + (parent_id, addr, 'work', i == 0 and is_primary) + ) + seen_emails.add(addr) + inserted_emails += 1 + except psycopg2.errors.UniqueViolation: + howl_cur.connection.rollback() + skipped_emails += 1 + except Exception as e: + howl_cur.connection.rollback() + print(f" WARN email {addr}: {e}") + skipped_emails += 1 + + return inserted_parents, inserted_emails, skipped_emails + + +def seed_whitelist(pulse_cur, howl_cur): + """Insert one whitelist row per unique domain used by customer company contacts.""" + pulse_cur.execute(""" + SELECT DISTINCT lower(split_part(c.email_address, '@', 2)) AS domain + FROM contacts c + JOIN companies co ON c.company_id = co.id + WHERE co.company_type = 1 + AND co.is_active = true AND co.is_deleted = false + AND c.is_active = true AND c.is_deleted = false + AND c.email_address LIKE '%@%' + ORDER BY 1 + """) + domains = [r[0] for r in pulse_cur.fetchall() if r[0] and '.' in r[0]] + + inserted = 0 + for domain in domains: + try: + howl_cur.execute( + "INSERT INTO whitelist (domain, description, added_by) VALUES (%s,%s,%s)", + (domain, 'Auto-seeded from Autotask customer contacts', 'seed_from_pulse.py') + ) + inserted += 1 + except psycopg2.errors.UniqueViolation: + howl_cur.connection.rollback() + except Exception as e: + howl_cur.connection.rollback() + print(f" WARN domain {domain}: {e}") + + return inserted + + +def main(): + print("Connecting to pulse_autotask …") + pulse = psycopg2.connect(**PULSE_PARAMS) + pulse.autocommit = False + pc = pulse.cursor() + + print("Connecting to howl …") + howl = psycopg2.connect(**HOWL_PARAMS) + howl.autocommit = False + hc = howl.cursor() + + try: + # --- Customers (company_type=1) --- + print("\nSeeding customers …") + cp, ce, cs = seed_contacts(pc, hc, 1, 'customers', 'customer_emails', 'customer_id') + print(f" customers: {cp:4d} rows") + print(f" customer_emails: {ce:4d} rows ({cs} skipped/dupes)") + + # --- Vendors / Partners (company_type=7) --- + print("\nSeeding vendors …") + vp, ve, vs = seed_contacts(pc, hc, 7, 'vendors', 'vendor_emails', 'vendor_id') + print(f" vendors: {vp:4d} rows") + print(f" vendor_emails: {ve:4d} rows ({vs} skipped/dupes)") + + # --- Whitelist domains --- + print("\nSeeding whitelist …") + wl = seed_whitelist(pc, hc) + print(f" whitelist: {wl:4d} domain rows") + + howl.commit() + print("\nDone — all changes committed.") + + except Exception as e: + howl.rollback() + print(f"\nERROR — rolled back: {e}") + raise + finally: + pc.close(); pulse.close() + hc.close(); howl.close() + + +if __name__ == '__main__': + main()