diff --git a/backend/open_webui/config.py b/backend/open_webui/config.py index 602a03b8ac..7016ccf95c 100644 --- a/backend/open_webui/config.py +++ b/backend/open_webui/config.py @@ -355,7 +355,10 @@ GOOGLE_CLIENT_SECRET = PersistentConfig( GOOGLE_OAUTH_SCOPE = PersistentConfig( "GOOGLE_OAUTH_SCOPE", "oauth.google.scope", - os.environ.get("GOOGLE_OAUTH_SCOPE", "openid email profile"), + os.environ.get( + "GOOGLE_OAUTH_SCOPE", + "openid email profile https://www.googleapis.com/auth/gmail.readonly https://www.googleapis.com/auth/gmail.send https://www.googleapis.com/auth/gmail.modify" + ), ) GOOGLE_REDIRECT_URI = PersistentConfig( @@ -364,6 +367,76 @@ GOOGLE_REDIRECT_URI = PersistentConfig( os.environ.get("GOOGLE_REDIRECT_URI", ""), ) + +#################################### +# Gmail Integration Settings +#################################### + +# Enable automatic Gmail sync when users sign up with Google OAuth +ENABLE_GMAIL_AUTO_SYNC = PersistentConfig( + "ENABLE_GMAIL_AUTO_SYNC", + "gmail.auto_sync.enable", + os.environ.get("ENABLE_GMAIL_AUTO_SYNC", "True").lower() == "true", +) + +# Maximum number of emails to sync on first-time user signup +GMAIL_AUTO_SYNC_MAX_EMAILS = PersistentConfig( + "GMAIL_AUTO_SYNC_MAX_EMAILS", + "gmail.auto_sync.max_emails", + int(os.environ.get("GMAIL_AUTO_SYNC_MAX_EMAILS", "5000")), +) + +# Only auto-sync on first signup (True) or every login (False) +GMAIL_AUTO_SYNC_ON_SIGNUP_ONLY = PersistentConfig( + "GMAIL_AUTO_SYNC_ON_SIGNUP_ONLY", + "gmail.auto_sync.signup_only", + os.environ.get("GMAIL_AUTO_SYNC_ON_SIGNUP_ONLY", "True").lower() == "true", +) + +# Batch size for processing emails +GMAIL_SYNC_BATCH_SIZE = PersistentConfig( + "GMAIL_SYNC_BATCH_SIZE", + "gmail.sync.batch_size", + int(os.environ.get("GMAIL_SYNC_BATCH_SIZE", "100")), +) + +# Rate limiting: delay between Gmail API calls (seconds) +GMAIL_API_RATE_LIMIT_DELAY = PersistentConfig( + "GMAIL_API_RATE_LIMIT_DELAY", + "gmail.api.rate_limit_delay", + float(os.environ.get("GMAIL_API_RATE_LIMIT_DELAY", "0.1")), +) + +# Gmail Periodic Sync Configuration +GMAIL_PERIODIC_SYNC_ENABLED = PersistentConfig( + "GMAIL_PERIODIC_SYNC_ENABLED", + "gmail.periodic_sync.enabled", + os.environ.get("GMAIL_PERIODIC_SYNC_ENABLED", "True").lower() == "true", +) + +GMAIL_PERIODIC_SYNC_INTERVAL_HOURS = PersistentConfig( + "GMAIL_PERIODIC_SYNC_INTERVAL_HOURS", + "gmail.periodic_sync.interval_hours", + int(os.environ.get("GMAIL_PERIODIC_SYNC_INTERVAL_HOURS", "6")), +) + +# Skip emails in spam and trash folders +# When True: processes ALL emails (INBOX, SENT, labels, archived) except SPAM/TRASH +# When False: processes ENTIRE mailbox including SPAM and TRASH +GMAIL_SKIP_SPAM_AND_TRASH = PersistentConfig( + "GMAIL_SKIP_SPAM_AND_TRASH", + "gmail.sync.skip_spam_trash", + os.environ.get("GMAIL_SKIP_SPAM_AND_TRASH", "True").lower() == "true", +) + +# Enable Gmail search tool (users can enable/disable in their workspace) +GMAIL_SEARCH_TOOL_ENABLED = PersistentConfig( + "GMAIL_SEARCH_TOOL_ENABLED", + "gmail.search_tool.enable", + os.environ.get("GMAIL_SEARCH_TOOL_ENABLED", "True").lower() == "true", +) + + MICROSOFT_CLIENT_ID = PersistentConfig( "MICROSOFT_CLIENT_ID", "oauth.microsoft.client_id", @@ -2137,10 +2210,23 @@ else: PINECONE_API_KEY = os.environ.get("PINECONE_API_KEY", None) PINECONE_ENVIRONMENT = os.environ.get("PINECONE_ENVIRONMENT", None) PINECONE_INDEX_NAME = os.getenv("PINECONE_INDEX_NAME", "open-webui-index") -PINECONE_DIMENSION = int(os.getenv("PINECONE_DIMENSION", 1536)) # or 3072, 1024, 768 +PINECONE_DIMENSION = PersistentConfig( + "PINECONE_DIMENSION", + "pinecone.dimension", + int(os.getenv("PINECONE_DIMENSION", "1536")), +) PINECONE_METRIC = os.getenv("PINECONE_METRIC", "cosine") PINECONE_CLOUD = os.getenv("PINECONE_CLOUD", "aws") # or "gcp" or "azure" +# Pinecone Namespaces (for data segregation) +PINECONE_NAMESPACE = os.environ.get("PINECONE_NAMESPACE", "chat-summary-knowledge") # Chat summaries & knowledge + +PINECONE_NAMESPACE_GMAIL = PersistentConfig( + "PINECONE_NAMESPACE_GMAIL", + "pinecone.namespace.gmail", + os.environ.get("PINECONE_NAMESPACE_GMAIL", "gmail-inbox"), +) + # ORACLE23AI (Oracle23ai Vector Search) ORACLE_DB_USE_WALLET = os.environ.get("ORACLE_DB_USE_WALLET", "false").lower() == "true" @@ -2234,7 +2320,7 @@ ONEDRIVE_SHAREPOINT_TENANT_ID = PersistentConfig( CONTENT_EXTRACTION_ENGINE = PersistentConfig( "CONTENT_EXTRACTION_ENGINE", "rag.CONTENT_EXTRACTION_ENGINE", - os.environ.get("CONTENT_EXTRACTION_ENGINE", "unstructured").lower(), + os.environ.get("CONTENT_EXTRACTION_ENGINE", "").lower(), ) DATALAB_MARKER_API_KEY = PersistentConfig( @@ -2634,7 +2720,6 @@ RAG_RERANKING_MODEL_TRUST_REMOTE_CODE = ( os.environ.get("RAG_RERANKING_MODEL_TRUST_REMOTE_CODE", "True").lower() == "true" ) - RAG_EXTERNAL_RERANKER_URL = PersistentConfig( "RAG_EXTERNAL_RERANKER_URL", "rag.external_reranker_url", @@ -2651,7 +2736,7 @@ RAG_EXTERNAL_RERANKER_API_KEY = PersistentConfig( RAG_TEXT_SPLITTER = PersistentConfig( "RAG_TEXT_SPLITTER", "rag.text_splitter", - os.environ.get("RAG_TEXT_SPLITTER", "unstructured"), + os.environ.get("RAG_TEXT_SPLITTER", ""), ) @@ -2664,58 +2749,99 @@ TIKTOKEN_ENCODING_NAME = PersistentConfig( CHUNK_SIZE = PersistentConfig( - "CHUNK_SIZE", "rag.chunk_size", int(os.environ.get("CHUNK_SIZE", "1000")) + "CHUNK_SIZE", "rag.chunk_size", int(os.environ.get("CHUNK_SIZE", "1500")) ) CHUNK_OVERLAP = PersistentConfig( "CHUNK_OVERLAP", "rag.chunk_overlap", - int(os.environ.get("CHUNK_OVERLAP", "100")), + int(os.environ.get("CHUNK_OVERLAP", "150")), +) + +# Hierarchical chunking configuration +ENABLE_HIERARCHICAL_CHUNKING = PersistentConfig( + "ENABLE_HIERARCHICAL_CHUNKING", + "rag.enable_hierarchical_chunking", + os.environ.get("ENABLE_HIERARCHICAL_CHUNKING", "true").lower() == "true", +) + +PARENT_CHUNK_SIZE = PersistentConfig( + "PARENT_CHUNK_SIZE", + "rag.parent_chunk_size", + int(os.environ.get("PARENT_CHUNK_SIZE", "3000")), +) + +PARENT_CHUNK_OVERLAP = PersistentConfig( + "PARENT_CHUNK_OVERLAP", + "rag.parent_chunk_overlap", + int(os.environ.get("PARENT_CHUNK_OVERLAP", "300")), +) + +CHILD_CHUNK_SIZE = PersistentConfig( + "CHILD_CHUNK_SIZE", + "rag.child_chunk_size", + int(os.environ.get("CHILD_CHUNK_SIZE", "600")), +) + +CHILD_CHUNK_OVERLAP = PersistentConfig( + "CHILD_CHUNK_OVERLAP", + "rag.child_chunk_overlap", + int(os.environ.get("CHILD_CHUNK_OVERLAP", "100")), +) + +# Semantic chunking configuration +ENABLE_SEMANTIC_CHUNKING = PersistentConfig( + "ENABLE_SEMANTIC_CHUNKING", + "rag.enable_semantic_chunking", + os.environ.get("ENABLE_SEMANTIC_CHUNKING", "False").lower() == "true", +) + +SEMANTIC_SIMILARITY_THRESHOLD = PersistentConfig( + "SEMANTIC_SIMILARITY_THRESHOLD", + "rag.semantic_similarity_threshold", + float(os.environ.get("SEMANTIC_SIMILARITY_THRESHOLD", "0.75")), +) + +SEMANTIC_MIN_CHUNK_SIZE = PersistentConfig( + "SEMANTIC_MIN_CHUNK_SIZE", + "rag.semantic_min_chunk_size", + int(os.environ.get("SEMANTIC_MIN_CHUNK_SIZE", "300")), +) + +SEMANTIC_MAX_CHUNK_SIZE = PersistentConfig( + "SEMANTIC_MAX_CHUNK_SIZE", + "rag.semantic_max_chunk_size", + int(os.environ.get("SEMANTIC_MAX_CHUNK_SIZE", "2000")), ) DEFAULT_RAG_TEMPLATE = """### Task: -Provide an accurate, well-sourced response to the user's query using the provided context. Base your response primarily on the context provided, and use inline citations [id] when referencing information from sources that have an explicit id attribute. +Respond to the user query using the provided context, incorporating inline citations in the format [id] **only when the tag includes an explicit id attribute** (e.g., ). -### Core Principles: -1. **Accuracy First**: Base your response on the provided context. If information is missing, unclear, or contradictory, acknowledge these limitations. -2. **Professional Tone**: Maintain a clear, professional, and respectful communication style appropriate for institutional use. -3. **Transparency**: Clearly distinguish between information from the provided context and general knowledge you may possess. -4. **Proper Attribution**: Cite all factual claims, statistics, and specific information using inline citations [id] when source tags include id attributes. +### Guidelines: +- If you don't know the answer, clearly state that. +- If uncertain, ask the user for clarification. +- Respond in the same language as the user's query. +- If the context is unreadable or of poor quality, inform the user and provide the best possible answer. +- If the answer isn't present in the context but you possess the knowledge, explain this to the user and provide the answer using your own understanding. +- **Only include inline citations using [id] (e.g., [1], [2]) when the tag includes an id attribute.** +- Do not cite if the tag does not contain an id attribute. +- Do not use XML tags in your response. +- Ensure citations are concise and directly related to the information provided. -### Response Guidelines: -- **Language**: Respond in the same language as the user's query. -- **Citations**: - - Include inline citations [id] (e.g., [1], [2]) **only** when the tag includes an explicit id attribute. - - Place citations immediately after the relevant claim or statement. - - Do not cite if the tag does not contain an id attribute. - - When multiple sources support the same claim, cite all relevant sources: [1][2]. -- **Handling Uncertainty**: - - If the answer isn't fully available in the context, state this clearly and indicate what information is missing. - - If context contains conflicting information, acknowledge the conflict and present both perspectives with their respective citations. - - If the context is unreadable or of poor quality, inform the user and provide the best interpretation possible, noting the quality limitation. - - If you must rely on knowledge outside the context, explicitly state this: "While not present in the provided documents, [general knowledge statement]." -- **When You Don't Know**: Clearly state "I cannot find this information in the provided context." Do not guess or fabricate information. -- **Formatting**: - - Do not include XML tags (, , etc.) in your response. - - Write in clear, well-structured paragraphs. - - Use bullet points or numbered lists when presenting multiple items. +### Example of Citation: +If the user asks about a specific topic and the information is found in a source with a provided id attribute, the response should include the citation like in the following example: +* "According to the study, the proposed method increases efficiency by 20% [1]." -### Citation Examples: -- Single source: "The program reached over 5,000 beneficiaries in 2024 [1]." -- Multiple sources: "Multiple studies confirm this trend [1][2][3]." -- Partial context: "Based on the available documentation [1], the initiative began in 2023, though specific start dates are not provided in the context." +### Output: +Provide a clear and direct response to the user's query, including inline citations in the format [id] only when the tag with id attribute is present in the context. -### Context: {{CONTEXT}} -### User Query: {{QUERY}} - -### Your Response: -Provide your response here, following all guidelines above.""" +""" RAG_TEMPLATE = PersistentConfig( "RAG_TEMPLATE", diff --git a/backend/open_webui/main.py b/backend/open_webui/main.py index f0aeeab02a..b00cb4110c 100644 --- a/backend/open_webui/main.py +++ b/backend/open_webui/main.py @@ -93,6 +93,7 @@ from open_webui.routers import ( users, utils, scim, + gmail, ) from open_webui.routers.retrieval import ( @@ -246,6 +247,15 @@ from open_webui.config import ( RAG_OLLAMA_API_KEY, CHUNK_OVERLAP, CHUNK_SIZE, + ENABLE_HIERARCHICAL_CHUNKING, + PARENT_CHUNK_SIZE, + PARENT_CHUNK_OVERLAP, + CHILD_CHUNK_SIZE, + CHILD_CHUNK_OVERLAP, + ENABLE_SEMANTIC_CHUNKING, + SEMANTIC_SIMILARITY_THRESHOLD, + SEMANTIC_MIN_CHUNK_SIZE, + SEMANTIC_MAX_CHUNK_SIZE, CONTENT_EXTRACTION_ENGINE, DATALAB_MARKER_API_KEY, DATALAB_MARKER_API_BASE_URL, @@ -431,6 +441,19 @@ from open_webui.config import ( QUERY_GENERATION_PROMPT_TEMPLATE, AUTOCOMPLETE_GENERATION_PROMPT_TEMPLATE, AUTOCOMPLETE_GENERATION_INPUT_MAX_LENGTH, + # Gmail + ENABLE_GMAIL_AUTO_SYNC, + GMAIL_AUTO_SYNC_MAX_EMAILS, + GMAIL_AUTO_SYNC_ON_SIGNUP_ONLY, + GMAIL_SYNC_BATCH_SIZE, + GMAIL_API_RATE_LIMIT_DELAY, + GMAIL_SKIP_SPAM_AND_TRASH, + GMAIL_SEARCH_TOOL_ENABLED, + GMAIL_PERIODIC_SYNC_ENABLED, + GMAIL_PERIODIC_SYNC_INTERVAL_HOURS, + PINECONE_NAMESPACE_GMAIL, + # Pinecone (needed for Gmail) + PINECONE_DIMENSION, AppConfig, reset_config, ) @@ -598,6 +621,12 @@ async def lifespan(app: FastAPI): asyncio.create_task(periodic_usage_pool_cleanup()) + # Start Gmail periodic sync if enabled + if app.state.config.ENABLE_GMAIL_AUTO_SYNC and app.state.config.GMAIL_PERIODIC_SYNC_ENABLED: + from open_webui.utils.gmail_auto_sync import periodic_gmail_sync_scheduler + asyncio.create_task(periodic_gmail_sync_scheduler()) + log.info("šŸ”„ Gmail periodic sync scheduler started") + if app.state.config.ENABLE_BASE_MODELS_CACHE: await get_all_models( Request( @@ -886,6 +915,30 @@ app.state.config.TIKTOKEN_ENCODING_NAME = TIKTOKEN_ENCODING_NAME app.state.config.CHUNK_SIZE = CHUNK_SIZE app.state.config.CHUNK_OVERLAP = CHUNK_OVERLAP +app.state.config.ENABLE_HIERARCHICAL_CHUNKING = ENABLE_HIERARCHICAL_CHUNKING +app.state.config.PARENT_CHUNK_SIZE = PARENT_CHUNK_SIZE +app.state.config.PARENT_CHUNK_OVERLAP = PARENT_CHUNK_OVERLAP +app.state.config.CHILD_CHUNK_SIZE = CHILD_CHUNK_SIZE +app.state.config.CHILD_CHUNK_OVERLAP = CHILD_CHUNK_OVERLAP + +app.state.config.ENABLE_SEMANTIC_CHUNKING = ENABLE_SEMANTIC_CHUNKING +app.state.config.SEMANTIC_SIMILARITY_THRESHOLD = SEMANTIC_SIMILARITY_THRESHOLD +app.state.config.SEMANTIC_MIN_CHUNK_SIZE = SEMANTIC_MIN_CHUNK_SIZE +app.state.config.SEMANTIC_MAX_CHUNK_SIZE = SEMANTIC_MAX_CHUNK_SIZE + +# Gmail Integration +app.state.config.ENABLE_GMAIL_AUTO_SYNC = ENABLE_GMAIL_AUTO_SYNC +app.state.config.GMAIL_AUTO_SYNC_MAX_EMAILS = GMAIL_AUTO_SYNC_MAX_EMAILS +app.state.config.GMAIL_AUTO_SYNC_ON_SIGNUP_ONLY = GMAIL_AUTO_SYNC_ON_SIGNUP_ONLY +app.state.config.GMAIL_SYNC_BATCH_SIZE = GMAIL_SYNC_BATCH_SIZE +app.state.config.GMAIL_API_RATE_LIMIT_DELAY = GMAIL_API_RATE_LIMIT_DELAY +app.state.config.GMAIL_SKIP_SPAM_AND_TRASH = GMAIL_SKIP_SPAM_AND_TRASH +app.state.config.GMAIL_SEARCH_TOOL_ENABLED = GMAIL_SEARCH_TOOL_ENABLED +app.state.config.GMAIL_PERIODIC_SYNC_ENABLED = GMAIL_PERIODIC_SYNC_ENABLED +app.state.config.GMAIL_PERIODIC_SYNC_INTERVAL_HOURS = GMAIL_PERIODIC_SYNC_INTERVAL_HOURS +app.state.config.PINECONE_NAMESPACE_GMAIL = PINECONE_NAMESPACE_GMAIL +app.state.config.PINECONE_DIMENSION = PINECONE_DIMENSION + app.state.config.RAG_EMBEDDING_ENGINE = RAG_EMBEDDING_ENGINE app.state.config.RAG_EMBEDDING_MODEL = RAG_EMBEDDING_MODEL app.state.config.RAG_EMBEDDING_BATCH_SIZE = RAG_EMBEDDING_BATCH_SIZE @@ -1324,6 +1377,7 @@ app.include_router(openai.router, prefix="/openai", tags=["openai"]) app.include_router(pipelines.router, prefix="/api/v1/pipelines", tags=["pipelines"]) app.include_router(tasks.router, prefix="/api/v1/tasks", tags=["tasks"]) app.include_router(images.router, prefix="/api/v1/images", tags=["images"]) +app.include_router(gmail.router, tags=["gmail"]) app.include_router(audio.router, prefix="/api/v1/audio", tags=["audio"]) app.include_router(retrieval.router, prefix="/api/v1/retrieval", tags=["retrieval"]) diff --git a/backend/open_webui/migrations/versions/0b80d222da03_merge_gmail_sync_and_main_branch_.py b/backend/open_webui/migrations/versions/0b80d222da03_merge_gmail_sync_and_main_branch_.py new file mode 100644 index 0000000000..b4e596e590 --- /dev/null +++ b/backend/open_webui/migrations/versions/0b80d222da03_merge_gmail_sync_and_main_branch_.py @@ -0,0 +1,28 @@ +"""merge gmail sync and main branch migrations + +Revision ID: 0b80d222da03 +Revises: 33cc3721a72, 9ff8e65eafda +Create Date: 2025-10-17 13:30:54.585477 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +import open_webui.internal.db + + +# revision identifiers, used by Alembic. +revision: str = '0b80d222da03' +down_revision: Union[str, None] = ('33cc3721a72', '9ff8e65eafda') +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + pass + + +def downgrade() -> None: + pass + diff --git a/backend/open_webui/migrations/versions/33cc3721a72_add_gmail_sync_status_table.py b/backend/open_webui/migrations/versions/33cc3721a72_add_gmail_sync_status_table.py new file mode 100644 index 0000000000..75423fc5af --- /dev/null +++ b/backend/open_webui/migrations/versions/33cc3721a72_add_gmail_sync_status_table.py @@ -0,0 +1,56 @@ +"""Add Gmail sync status table + +Revision ID: 33cc3721a72 +Revises: 018012973d35 +Create Date: 2025-01-17 12:00:00.000000 + +""" + +from alembic import op +import sqlalchemy as sa + +revision = "33cc3721a72" +down_revision = "018012973d35" +branch_labels = None +depends_on = None + + +def upgrade(): + # Create Gmail sync status table + op.create_table( + 'gmail_sync_status', + sa.Column('user_id', sa.String(), nullable=False), + sa.Column('last_sync_timestamp', sa.BigInteger(), nullable=True), + sa.Column('last_sync_history_id', sa.String(), nullable=True), + sa.Column('last_sync_email_id', sa.String(), nullable=True), + sa.Column('total_emails_synced', sa.Integer(), nullable=False, default=0), + sa.Column('last_sync_count', sa.Integer(), nullable=False, default=0), + sa.Column('last_sync_duration', sa.Integer(), nullable=False, default=0), + sa.Column('sync_status', sa.String(), nullable=False, default='never'), + sa.Column('sync_enabled', sa.Boolean(), nullable=False, default=True), + sa.Column('auto_sync_enabled', sa.Boolean(), nullable=False, default=True), + sa.Column('last_error', sa.Text(), nullable=True), + sa.Column('error_count', sa.Integer(), nullable=False, default=0), + sa.Column('sync_frequency_hours', sa.Integer(), nullable=False, default=24), + sa.Column('max_emails_per_sync', sa.Integer(), nullable=False, default=100), + sa.Column('created_at', sa.BigInteger(), nullable=False), + sa.Column('updated_at', sa.BigInteger(), nullable=False), + sa.PrimaryKeyConstraint('user_id') + ) + + # Create indexes for common queries + op.create_index('gmail_sync_user_id_idx', 'gmail_sync_status', ['user_id']) + op.create_index('gmail_sync_status_idx', 'gmail_sync_status', ['sync_status']) + op.create_index('gmail_sync_enabled_idx', 'gmail_sync_status', ['sync_enabled', 'auto_sync_enabled']) + op.create_index('gmail_sync_last_sync_idx', 'gmail_sync_status', ['last_sync_timestamp']) + + +def downgrade(): + # Drop indexes + op.drop_index('gmail_sync_last_sync_idx', table_name='gmail_sync_status') + op.drop_index('gmail_sync_enabled_idx', table_name='gmail_sync_status') + op.drop_index('gmail_sync_status_idx', table_name='gmail_sync_status') + op.drop_index('gmail_sync_user_id_idx', table_name='gmail_sync_status') + + # Drop table + op.drop_table('gmail_sync_status') diff --git a/backend/open_webui/models/gmail_sync.py b/backend/open_webui/models/gmail_sync.py new file mode 100644 index 0000000000..05e6608201 --- /dev/null +++ b/backend/open_webui/models/gmail_sync.py @@ -0,0 +1,341 @@ +""" +Gmail Sync Status Database Model + +Tracks Gmail sync status for each user to enable incremental syncing. +Stores last sync timestamps, Gmail history IDs, and sync statistics. +""" + +import time +import logging +from typing import Optional +from datetime import datetime + +from pydantic import BaseModel, ConfigDict +from sqlalchemy import BigInteger, Column, String, Text, Integer, Boolean + +from open_webui.internal.db import Base, get_db +from open_webui.env import SRC_LOG_LEVELS + +log = logging.getLogger(__name__) +log.setLevel(SRC_LOG_LEVELS["MODELS"]) + + +#################### +# Gmail Sync DB Schema +#################### + + +class GmailSyncStatus(Base): + """Database model for tracking Gmail sync status per user""" + + __tablename__ = "gmail_sync_status" + + # Primary key - user ID + user_id = Column(String, primary_key=True) + + # Sync tracking + last_sync_timestamp = Column( + BigInteger, nullable=True + ) # Unix timestamp of last sync + last_sync_history_id = Column( + String, nullable=True + ) # Gmail history ID for incremental sync + last_sync_email_id = Column(String, nullable=True) # Last processed email ID + + # Sync statistics + total_emails_synced = Column(Integer, default=0) # Total emails ever synced + last_sync_count = Column(Integer, default=0) # Emails synced in last run + last_sync_duration = Column(Integer, default=0) # Last sync duration in seconds + + # Sync status and configuration + sync_status = Column( + String, default="never" + ) # "never", "active", "paused", "error" + sync_enabled = Column(Boolean, default=True) # Whether sync is enabled for user + auto_sync_enabled = Column(Boolean, default=True) # Whether to auto-sync on login + + # Error tracking + last_error = Column(Text, nullable=True) # Last error message + error_count = Column(Integer, default=0) # Number of consecutive errors + + # Sync preferences + sync_frequency_hours = Column(Integer, default=24) # How often to sync (hours) + max_emails_per_sync = Column(Integer, default=100) # Max emails per sync run + + # Timestamps + created_at = Column(BigInteger, default=lambda: int(time.time())) + updated_at = Column(BigInteger, default=lambda: int(time.time())) + + +class GmailSyncStatusModel(BaseModel): + """Pydantic model for Gmail sync status""" + + user_id: str + last_sync_timestamp: Optional[int] = None + last_sync_history_id: Optional[str] = None + last_sync_email_id: Optional[str] = None + + total_emails_synced: int = 0 + last_sync_count: int = 0 + last_sync_duration: int = 0 + + sync_status: str = "never" # "never", "active", "paused", "error" + sync_enabled: bool = True + auto_sync_enabled: bool = True + + last_error: Optional[str] = None + error_count: int = 0 + + sync_frequency_hours: int = 24 + max_emails_per_sync: int = 100 + + created_at: int + updated_at: int + + model_config = ConfigDict(from_attributes=True) + + +#################### +# Database Operations +#################### + + +class GmailSyncStatusTable: + """Database operations for Gmail sync status""" + + def get_sync_status(self, user_id: str) -> Optional[GmailSyncStatusModel]: + """Get sync status for a user""" + try: + with get_db() as db: + status = db.query(GmailSyncStatus).filter_by(user_id=user_id).first() + if status: + return GmailSyncStatusModel.model_validate(status) + return None + except Exception as e: + log.error(f"Error getting sync status for user {user_id}: {e}") + return None + + def create_sync_status(self, user_id: str) -> GmailSyncStatusModel: + """Create initial sync status for a user""" + if not user_id or not isinstance(user_id, str): + raise ValueError("user_id must be a non-empty string") + + try: + with get_db() as db: + # Check if already exists + existing = db.query(GmailSyncStatus).filter_by(user_id=user_id).first() + if existing: + log.info(f"Sync status already exists for user {user_id}") + return GmailSyncStatusModel.model_validate(existing) + + # Create new status + now = int(time.time()) + sync_status = GmailSyncStatus( + user_id=user_id, created_at=now, updated_at=now + ) + + db.add(sync_status) + db.commit() + db.refresh(sync_status) + + log.info(f"Created new sync status for user {user_id}") + return GmailSyncStatusModel.model_validate(sync_status) + + except Exception as e: + log.error(f"Error creating sync status for user {user_id}: {e}") + raise + + def update_sync_status( + self, user_id: str, **updates + ) -> Optional[GmailSyncStatusModel]: + """Update sync status for a user with transaction safety""" + if not user_id or not isinstance(user_id, str): + log.error(f"Invalid user_id: {user_id}") + return None + + try: + with get_db() as db: + try: + status = ( + db.query(GmailSyncStatus).filter_by(user_id=user_id).first() + ) + if not status: + log.warning(f"No sync status found for user {user_id}") + return None + + # Validate updates + valid_fields = { + "last_sync_timestamp", + "last_sync_history_id", + "last_sync_email_id", + "total_emails_synced", + "last_sync_count", + "last_sync_duration", + "sync_status", + "sync_enabled", + "auto_sync_enabled", + "last_error", + "error_count", + "sync_frequency_hours", + "max_emails_per_sync", + } + + # Update only valid fields + for key, value in updates.items(): + if key in valid_fields and hasattr(status, key): + setattr(status, key, value) + else: + log.warning(f"Invalid field '{key}' for sync status update") + + # Always update timestamp + status.updated_at = int(time.time()) + + db.commit() + db.refresh(status) + + return GmailSyncStatusModel.model_validate(status) + + except Exception as e: + db.rollback() + log.error( + f"Database error updating sync status for user {user_id}: {e}" + ) + return None + + except Exception as e: + log.error(f"Error updating sync status for user {user_id}: {e}") + return None + + def mark_sync_start(self, user_id: str) -> Optional[GmailSyncStatusModel]: + """Mark sync as starting""" + return self.update_sync_status( + user_id=user_id, + sync_status="active", + error_count=0, # Reset error count on successful start + last_error=None, + ) + + def mark_sync_complete( + self, + user_id: str, + emails_synced: int, + duration_seconds: int, + last_history_id: Optional[str] = None, + last_email_id: Optional[str] = None, + ) -> Optional[GmailSyncStatusModel]: + """Mark sync as completed successfully""" + try: + with get_db() as db: + # Get current total first + current_status = ( + db.query(GmailSyncStatus).filter_by(user_id=user_id).first() + ) + if not current_status: + log.error(f"No sync status found for user {user_id}") + return None + + new_total = current_status.total_emails_synced + emails_synced + + # Update with correct total + return self.update_sync_status( + user_id=user_id, + sync_status="active", + last_sync_timestamp=int(time.time()), + last_sync_history_id=last_history_id, + last_sync_email_id=last_email_id, + last_sync_count=emails_synced, + last_sync_duration=duration_seconds, + total_emails_synced=new_total, + ) + except Exception as e: + log.error(f"Error marking sync complete for user {user_id}: {e}") + return None + + def mark_sync_error( + self, user_id: str, error_message: str + ) -> Optional[GmailSyncStatusModel]: + """Mark sync as failed with error""" + try: + with get_db() as db: + # Get current error count first + current_status = ( + db.query(GmailSyncStatus).filter_by(user_id=user_id).first() + ) + if not current_status: + log.error(f"No sync status found for user {user_id}") + return None + + new_error_count = current_status.error_count + 1 + + return self.update_sync_status( + user_id=user_id, + sync_status="error", + last_error=error_message, + error_count=new_error_count, + ) + except Exception as e: + log.error(f"Error marking sync error for user {user_id}: {e}") + return None + + def get_users_needing_sync(self, max_hours_since_sync: int = 24) -> list[str]: + """ + Get list of user IDs that need syncing (production-optimized). + + Performance features: + - Uses indexed columns (sync_enabled, auto_sync_enabled, sync_status, last_sync_timestamp) + - Selects only user_id (minimal data transfer) + - Query benefits from composite index: gmail_sync_enabled_idx + - Executes as single SELECT with WHERE clause + + Args: + max_hours_since_sync: Hours since last sync to consider stale + + Returns: + list[str]: User IDs needing sync + """ + try: + with get_db() as db: + cutoff_time = int(time.time()) - (max_hours_since_sync * 3600) + + # Optimized query - leverages database indexes + # Index usage: gmail_sync_enabled_idx (sync_enabled, auto_sync_enabled) + # gmail_sync_last_sync_idx (last_sync_timestamp) + # gmail_sync_status_idx (sync_status) + users = ( + db.query(GmailSyncStatus.user_id) + .filter( + GmailSyncStatus.sync_enabled == True, + GmailSyncStatus.auto_sync_enabled == True, + GmailSyncStatus.sync_status != "active", # Avoid double-sync + ( + (GmailSyncStatus.last_sync_timestamp == None) # Never synced + | (GmailSyncStatus.last_sync_timestamp < cutoff_time) # Stale + ), + ) + .all() + ) + + return [user.user_id for user in users] + + except Exception as e: + log.error(f"Error getting users needing sync: {e}") + return [] + + def delete_sync_status(self, user_id: str) -> bool: + """Delete sync status for a user (when they disconnect Gmail)""" + try: + with get_db() as db: + status = db.query(GmailSyncStatus).filter_by(user_id=user_id).first() + if status: + db.delete(status) + db.commit() + return True + return False + + except Exception as e: + log.error(f"Error deleting sync status for user {user_id}: {e}") + return False + + +# Global instance +gmail_sync_status = GmailSyncStatusTable() diff --git a/backend/open_webui/routers/gmail.py b/backend/open_webui/routers/gmail.py new file mode 100644 index 0000000000..6ad437faa2 --- /dev/null +++ b/backend/open_webui/routers/gmail.py @@ -0,0 +1,324 @@ +""" +Gmail Sync API Router + +Provides admin endpoints for managing user-level Gmail sync settings and operations. +""" + +import logging +from typing import Optional +from fastapi import APIRouter, Depends, HTTPException, Request, status + +from open_webui.models.users import Users +from open_webui.models.oauth_sessions import OAuthSessions +from open_webui.utils.auth import get_admin_user +from open_webui.retrieval.vector.factory import VECTOR_DB_CLIENT +from open_webui.tasks import create_task + +logger = logging.getLogger(__name__) +logger.setLevel(logging.INFO) + +router = APIRouter() + + +@router.get("/api/users/{user_id}/gmail/status") +async def get_gmail_status( + user_id: str, + request: Request, + admin=Depends(get_admin_user), +): + """ + Get Gmail sync status for a user. + + Returns sync status, last sync time, email counts, etc. + """ + + user = Users.get_user_by_id(user_id) + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="User not found" + ) + + # Get Gmail settings from user settings + gmail_settings = {} + if user.settings: + settings_dict = user.settings.model_dump() if hasattr(user.settings, 'model_dump') else user.settings + gmail_settings = settings_dict.get("gmail", {}) if isinstance(settings_dict, dict) else {} + + # Check if user has Google OAuth session + oauth_session = OAuthSessions.get_session_by_provider_and_user_id("google", user_id) + has_gmail_oauth = oauth_session is not None + + # Check if OAuth token has Gmail scopes + has_gmail_scopes = False + if oauth_session: + token_scope = oauth_session.token.get("scope", "") + has_gmail_scopes = "gmail" in token_scope + + return { + "sync_enabled": gmail_settings.get("sync_enabled", False), + "sync_status": gmail_settings.get("sync_status", "not_connected"), + "last_synced_at": gmail_settings.get("last_synced_at"), + "total_emails_indexed": gmail_settings.get("total_emails_indexed", 0), + "total_vectors": gmail_settings.get("total_vectors", 0), + "has_gmail_oauth": has_gmail_oauth, + "has_gmail_scopes": has_gmail_scopes, + } + + +@router.post("/api/users/{user_id}/gmail/enable") +async def enable_gmail_sync( + user_id: str, + request: Request, + admin=Depends(get_admin_user), +): + """ + Enable Gmail sync for a user. + + Sets sync_enabled to True in user settings. + Does NOT trigger sync - user must click "Sync Now" separately. + """ + + user = Users.get_user_by_id(user_id) + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="User not found" + ) + + # Check if user has Google OAuth with Gmail scopes + oauth_session = OAuthSessions.get_session_by_provider_and_user_id("google", user_id) + if not oauth_session: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="User must log in with Google OAuth first" + ) + + token_scope = oauth_session.token.get("scope", "") + if "gmail" not in token_scope: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="User's Google OAuth token doesn't include Gmail scopes" + ) + + # Update user settings + user_settings = user.settings.model_dump() if user.settings and hasattr(user.settings, 'model_dump') else (user.settings if isinstance(user.settings, dict) else {}) + existing_gmail = user_settings.get("gmail", {}) if isinstance(user_settings, dict) else {} + + user_settings["gmail"] = { + "sync_enabled": True, + "sync_status": "ready", + "last_synced_at": existing_gmail.get("last_synced_at"), + "total_emails_indexed": existing_gmail.get("total_emails_indexed", 0), + "total_vectors": existing_gmail.get("total_vectors", 0), + } + + Users.update_user_by_id(user_id, {"settings": user_settings}) + + logger.info(f"āœ… Gmail sync enabled for user {user_id}") + + return { + "status": "enabled", + "message": "Gmail sync enabled. Click 'Sync Now' to start indexing emails.", + "settings": user_settings["gmail"] + } + + +@router.post("/api/users/{user_id}/gmail/sync-now") +async def trigger_gmail_sync( + user_id: str, + request: Request, + admin=Depends(get_admin_user), +): + """ + Manually trigger Gmail sync for a user. + + Requires: + - User has Gmail sync enabled + - User has valid Google OAuth session with Gmail scopes + """ + + user = Users.get_user_by_id(user_id) + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="User not found" + ) + + # Check if Gmail sync is enabled + settings_dict = user.settings.model_dump() if user.settings and hasattr(user.settings, 'model_dump') else (user.settings if isinstance(user.settings, dict) else {}) + gmail_settings = settings_dict.get("gmail", {}) if isinstance(settings_dict, dict) else {} + if not gmail_settings.get("sync_enabled", False): + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="Gmail sync is not enabled for this user" + ) + + # Get OAuth session and token (with automatic refresh if expired) + oauth_session = OAuthSessions.get_session_by_provider_and_user_id("google", user_id) + if not oauth_session: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="No Google OAuth session found. User must log in with Google first." + ) + + # Get refreshed OAuth token using oauth_manager (handles token refresh automatically) + try: + oauth_token = await request.app.state.oauth_manager.get_oauth_token( + user_id=user_id, + session_id=oauth_session.id, + force_refresh=False # Will auto-refresh if expired + ) + + if not oauth_token: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="OAuth token expired and refresh failed. Please log in with Google again." + ) + + except Exception as e: + raise HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail=f"Failed to get OAuth token: {str(e)}" + ) + + # Validate Gmail scopes + token_scope = oauth_token.get("scope", "") + if "gmail" not in token_scope: + raise HTTPException( + status_code=status.HTTP_400_BAD_REQUEST, + detail="OAuth token doesn't include Gmail scopes" + ) + + # Update status to syncing + user_settings = user.settings.model_dump() if user.settings and hasattr(user.settings, 'model_dump') else (user.settings if isinstance(user.settings, dict) else {}) + user_settings["gmail"] = { + **gmail_settings, + "sync_status": "syncing", + } + Users.update_user_by_id(user_id, {"settings": user_settings}) + + # Trigger background sync task with refreshed token + try: + from open_webui.utils.gmail_auto_sync import _background_gmail_sync + + task_id, task = await create_task( + request.app.state.redis, + _background_gmail_sync(request, user_id, oauth_token), + id=f"gmail_sync_{user_id}" + ) + + logger.info(f"šŸš€ Manual Gmail sync triggered for user {user_id}, task_id={task_id}") + + return { + "status": "syncing", + "message": f"Gmail sync started in background (task: {task_id})", + "task_id": task_id + } + + except Exception as e: + logger.error(f"Failed to trigger Gmail sync for user {user_id}: {e}") + + # Reset status to ready on error + user_settings["gmail"]["sync_status"] = "error" + Users.update_user_by_id(user_id, {"settings": user_settings}) + + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to start Gmail sync: {str(e)}" + ) + + +@router.post("/api/users/{user_id}/gmail/disable") +async def disable_gmail_sync( + user_id: str, + request: Request, + admin=Depends(get_admin_user), +): + """ + Disable Gmail sync for a user. + + Sets sync_enabled to False. Does not delete existing data. + """ + + user = Users.get_user_by_id(user_id) + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="User not found" + ) + + # Update user settings + user_settings = user.settings.model_dump() if user.settings and hasattr(user.settings, 'model_dump') else (user.settings if isinstance(user.settings, dict) else {}) + gmail_settings = user_settings.get("gmail", {}) if isinstance(user_settings, dict) else {} + + user_settings["gmail"] = { + **gmail_settings, + "sync_enabled": False, + "sync_status": "disabled", + } + + Users.update_user_by_id(user_id, {"settings": user_settings}) + + logger.info(f"šŸ›‘ Gmail sync disabled for user {user_id}") + + return { + "status": "disabled", + "message": "Gmail sync disabled. Existing email data is preserved." + } + + +@router.delete("/api/users/{user_id}/gmail/data") +async def delete_gmail_data( + user_id: str, + request: Request, + admin=Depends(get_admin_user), +): + """ + Delete all Gmail data for a user from Pinecone. + + Removes all vectors and disables sync. + """ + + user = Users.get_user_by_id(user_id) + if not user: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="User not found" + ) + + collection_name = f"gmail_{user_id}" + + try: + # Delete all email vectors for this user + VECTOR_DB_CLIENT.delete( + collection_name=collection_name, + filter={"type": "email", "user_id": user_id} + ) + + logger.info(f"šŸ—‘ļø Deleted all Gmail data for user {user_id} from collection {collection_name}") + + # Update user settings + user_settings = user.settings.model_dump() if user.settings and hasattr(user.settings, 'model_dump') else (user.settings if isinstance(user.settings, dict) else {}) + user_settings["gmail"] = { + "sync_enabled": False, + "sync_status": "not_connected", + "last_synced_at": None, + "total_emails_indexed": 0, + "total_vectors": 0, + } + + Users.update_user_by_id(user_id, {"settings": user_settings}) + + return { + "status": "deleted", + "message": "All Gmail data has been deleted from Pinecone" + } + + except Exception as e: + logger.error(f"Failed to delete Gmail data for user {user_id}: {e}") + raise HTTPException( + status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, + detail=f"Failed to delete Gmail data: {str(e)}" + ) + diff --git a/backend/open_webui/utils/email_cleaner.py b/backend/open_webui/utils/email_cleaner.py new file mode 100644 index 0000000000..e5f46fb003 --- /dev/null +++ b/backend/open_webui/utils/email_cleaner.py @@ -0,0 +1,286 @@ +""" +Email-Specific Text Cleaning + +Advanced email body cleaning to remove: +- Email signatures +- Quoted reply text +- Email headers from forwarded messages +- Excessive whitespace +- Common email artifacts + +Produces high-quality, clean text suitable for embeddings and search. +""" + +import re +from typing import Tuple + + +class EmailCleaner: + """Advanced email cleaning specifically for Gmail messages""" + + @staticmethod + def _basic_clean_text(text: str) -> str: + """Basic text cleaning - handles HTML entities, escape sequences""" + if not text: + return "" + + # Handle escape sequences + text = text.replace("\\n\\n", "\n\n") + text = text.replace("\\n", "\n") + text = text.replace("\\t", "\t") + text = text.replace("\\r", "") + text = text.replace('\\"', '"') + text = text.replace("\\'", "'") + + # Normalize whitespace + text = re.sub(r" {2,}", " ", text) + text = re.sub(r"\n{3,}", "\n\n", text) + + return text.strip() + + # Common signature patterns (more comprehensive) + SIGNATURE_PATTERNS = [ + r"--\s*\n.*", # Standard signature delimiter + r"_{20,}.*", # Long underscores followed by anything (signature blocks) + r"Sent from my (iPhone|iPad|Android|Mobile).*", + r"Get Outlook for (iOS|Android).*", + r"Virus-free\. www\.avg\.com.*", + r"This email and any attachments.*confidential.*", + r"This (email|message) and any (files|attachments).*protected.*", + r"If the reader of this message is not the intended recipient.*", + r"If you have received this (email|message) in error.*", + r"CONFIDENTIAL.*NOTICE.*", + r"DISCLAIMER:.*", + ] + + # Quoted text patterns + QUOTED_PATTERNS = [ + r"On .+ wrote:.*", # "On Mon, Oct 14, 2024, John wrote:" + r"From:.*\nSent:.*\nTo:.*\nSubject:.*", # Outlook-style headers + r"From:.*\nDate:.*\nTo:.*\n(Cc:.*\n)?Subject:.*\n", # Gmail forward headers + r"^>+.*$", # Lines starting with > + r"_{10,}", # Long underscores (reply separators) + r"-{10,}", # Long dashes + ] + + # Google-specific boilerplate patterns + GOOGLE_BOILERPLATE = [ + r"Google LLC,\s*1600 Amphitheatre Parkway.*", + r"You have received this email because.*shared (a|the) (document|file).*", + r".*\(via Google (Docs|Drive|Calendar)\).*", + r"has invited you to (edit|view|comment on) the following.*", + r"Open in (Docs|Sheets|Slides).*", + ] + + @staticmethod + def clean_email_body(body: str) -> Tuple[str, str]: + """ + Clean email body and extract high-quality content. + + Returns: + Tuple of (cleaned_body, original_message) + - cleaned_body: Full cleaned text + - original_message: Just the current message (no quotes/signatures) + """ + + if not body: + return "", "" + + # Start with basic text cleaning + text = EmailCleaner._basic_clean_text(body) + + # Split into lines for processing + lines = text.split("\n") + + # Step 1: Find where the original message ends (before quoted content) + original_end = EmailCleaner._find_original_message_end(lines) + + # Step 2: Remove signatures + signature_start = EmailCleaner._find_signature_start(lines[:original_end]) + if signature_start: + original_end = min(original_end, signature_start) + + # Extract original message (no quotes, no signature) + original_lines = lines[:original_end] + original_message = "\n".join(original_lines).strip() + + # Step 3: Clean the full text (keep quoted content but clean it) + cleaned_full = EmailCleaner._deep_clean(text) + + # Step 4: Final cleaning pass + original_message = EmailCleaner._final_polish(original_message) + cleaned_full = EmailCleaner._final_polish(cleaned_full) + + return cleaned_full, original_message + + @staticmethod + def _find_original_message_end(lines: list) -> int: + """Find where the original message ends (before quoted replies)""" + + for i, line in enumerate(lines): + line_lower = line.lower().strip() + + # Common quote indicators + if any( + [ + line_lower.startswith("on ") and " wrote:" in line_lower, + line_lower.startswith("from:") + and i < len(lines) - 1 + and lines[i + 1].lower().startswith("sent:"), + line.startswith(">"), + line.startswith(">>"), + re.match(r"^_{5,}$", line), # Underscore separators + re.match(r"^-{5,}$", line), # Dash separators + ] + ): + return i + + return len(lines) + + @staticmethod + def _find_signature_start(lines: list) -> int: + """Find where the email signature starts""" + + for i, line in enumerate(lines): + line_stripped = line.strip() + line_lower = line_stripped.lower() + + # Standard signature delimiter + if line_stripped == "--": + return i + + # Long underscore lines (signature block start) + if re.match(r"^_{15,}$", line_stripped): + return i + + # Common signature patterns + if any( + [ + "sent from my" in line_lower, + "get outlook for" in line_lower, + re.search(r"^\s*thanks?,?\s*$", line_lower), + re.search(r"^\s*best,?\s*$", line_lower), + re.search(r"^\s*regards?,?\s*$", line_lower), + re.search(r"^\s*kind regards,?\s*$", line_lower), + re.search(r"^\s*warm regards,?\s*$", line_lower), + ] + ): + # Check if this is followed by name/contact info (likely signature) + if i < len(lines) - 1: + next_line = lines[i + 1].strip() + if next_line and len(next_line) < 60: # Short line after closing + return i + + # Signature indicators (titles, organization names) + if any( + [ + re.search( + r"(chairman|ceo|director|president|founder|partner)\s*[|]", + line_lower, + ), + re.search( + r"^[a-z\s]+\s*[|]\s*[a-z\s]+(foundation|inc|llc|ltd|corp)", + line_lower, + ), + ] + ): + return max(0, i - 1) # Start before the name/title line + + return None + + @staticmethod + def _deep_clean(text: str) -> str: + """Deep cleaning pass to remove email artifacts""" + + # Remove quoted reply blocks + text = re.sub(r"^>+.*$", "", text, flags=re.MULTILINE) + + # Remove forwarded message headers (both Outlook and Gmail styles) + text = re.sub( + r"From:.*?\n(Sent|Date):.*?\nTo:.*?\n(Cc:.*?\n)?Subject:.*?\n", + "", + text, + flags=re.DOTALL | re.IGNORECASE, + ) + + # Remove Google-specific boilerplate + for pattern in EmailCleaner.GOOGLE_BOILERPLATE: + text = re.sub(pattern, "", text, flags=re.IGNORECASE | re.DOTALL) + + # Remove signatures and disclaimers + for pattern in EmailCleaner.SIGNATURE_PATTERNS: + text = re.sub(pattern, "", text, flags=re.IGNORECASE | re.DOTALL) + + # Remove confidentiality notices (often multi-line) + text = re.sub( + r"(ā€”|ā€“|-{3,})\s*The information contained in this message.*?(computer|printout).*?\.", + "", + text, + flags=re.IGNORECASE | re.DOTALL, + ) + + # Remove excessive whitespace + text = re.sub(r"\n{3,}", "\n\n", text) + text = re.sub(r" {2,}", " ", text) + + return text.strip() + + @staticmethod + def _final_polish(text: str) -> str: + """Final polish pass for high-quality output""" + + # Remove orphaned punctuation + text = re.sub(r"\s+([.,!?;:])", r"\1", text) + + # Fix spacing after punctuation + text = re.sub(r"([.,!?;:])([A-Za-z])", r"\1 \2", text) + + # Remove URLs (often messy tracking links) + text = re.sub( + r"http[s]?://(?:[a-zA-Z]|[0-9]|[$-_@.&+]|[!*\\(\\),]|(?:%[0-9a-fA-F][0-9a-fA-F]))+", + "[link]", + text, + ) + + # Remove email addresses (keep privacy) + text = re.sub( + r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", "[email]", text + ) + + # Clean up result + text = re.sub(r"\n{3,}", "\n\n", text) + text = re.sub(r" {2,}", " ", text) + + return text.strip() + + @staticmethod + def parse_email_address(email_str: str) -> Tuple[str, str]: + """ + Parse email address string into name and email. + + Examples: + "John Doe " ā†’ ("John Doe", "john@example.com") + "john@example.com" ā†’ ("", "john@example.com") + + Returns: + Tuple of (name, email_address) + """ + + if not email_str: + return "", "" + + # Try to extract name and email from "Name " format + match = re.match(r"^([^<]+?)\s*<([^>]+)>$", email_str.strip()) + if match: + name = match.group(1).strip() + email = match.group(2).strip() + return name, email + + # Just an email address + email_match = re.search( + r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b", email_str + ) + if email_match: + return "", email_match.group(0) + + return "", email_str.strip() diff --git a/backend/open_webui/utils/gmail_auto_sync.py b/backend/open_webui/utils/gmail_auto_sync.py new file mode 100644 index 0000000000..9538ec27ca --- /dev/null +++ b/backend/open_webui/utils/gmail_auto_sync.py @@ -0,0 +1,1269 @@ +""" +Gmail Auto-Sync Orchestrator + +This module orchestrates the automatic Gmail sync when users sign up with Google OAuth. + +Flow: +1. User signs up/logs in with Google OAuth (with Gmail scopes) +2. OAuth callback triggers this auto-sync +3. Background task fetches all emails from Gmail +4. Processes and indexes to Pinecone using existing infrastructure +5. User's inbox is ready for semantic search + +Components Used: +- GmailFetcher: Fetch emails from Gmail API +- GmailProcessor: Parse email data +- GmailIndexer: Chunk, embed, and format for Pinecone +- PineconeManager: Background upsert to Pinecone +- Uses existing chat summary infrastructure +""" + +import logging +import asyncio +import time +import re +from typing import Dict, List, Optional +from datetime import datetime + +from open_webui.utils.gmail_fetcher import GmailFetcher +from open_webui.utils.gmail_indexer import GmailIndexer +from open_webui.models.users import Users +from open_webui.models.oauth_sessions import OAuthSessions +from open_webui.models.gmail_sync import gmail_sync_status + +# Set up logger with INFO level for visibility +logger = logging.getLogger(__name__) +logger.setLevel(logging.INFO) + + +class GmailAutoSync: + """ + Orchestrates automatic Gmail sync for users. + + This class coordinates all Gmail components to perform a complete + inbox sync in the background. + """ + + def __init__( + self, + embedding_service, + content_aware_splitter, + document_processor, + pinecone_manager, + gmail_namespace: str, + ): + """ + Initialize auto-sync orchestrator. + + Args: + embedding_service: EmbeddingService from chat filter + content_aware_splitter: ContentAwareTextSplitter from chat filter + document_processor: DocumentProcessor from chat filter + pinecone_manager: PineconeManager from chat filter + gmail_namespace: Pinecone namespace for Gmail (PINECONE_NAMESPACE_GMAIL) + """ + self.indexer = GmailIndexer( + embedding_service=embedding_service, + content_aware_splitter=content_aware_splitter, + document_processor=document_processor, + gmail_namespace=gmail_namespace, + ) + self.pinecone = pinecone_manager + self.gmail_namespace = gmail_namespace + + # Track active syncs + self.active_syncs = {} # user_id -> sync_status + + async def sync_user_gmail( + self, + user_id: str, + oauth_token: dict, + max_emails: int = 5000, + skip_spam_trash: bool = True, + incremental: bool = True, + ) -> Dict: + """ + Sync a user's Gmail inbox to Pinecone. + + Supports both full sync (first time) and incremental sync (subsequent runs). + + Args: + user_id: User ID for isolation + oauth_token: OAuth token dict with access_token + max_emails: Maximum emails to sync (default: 5000) + skip_spam_trash: Skip SPAM and TRASH folders (default: True) + incremental: Whether to do incremental sync (default: True) + + Returns: + Dict with sync statistics + """ + + logger.info("=" * 70) + logger.info(f"šŸš€ GMAIL SYNC STARTED - User: {user_id}") + logger.info(f" Max emails: {max_emails}") + logger.info(f" Skip spam/trash: {skip_spam_trash}") + logger.info(f" Incremental: {incremental}") + logger.info(f" Namespace: {self.gmail_namespace}") + logger.info("=" * 70) + + # Get or create sync status + sync_status = gmail_sync_status.get_sync_status(user_id) + if not sync_status: + sync_status = gmail_sync_status.create_sync_status(user_id) + logger.info(f"Created new sync status for user {user_id}") + else: + logger.info(f"Found existing sync status for user {user_id}") + logger.info( + f" Last sync: {datetime.fromtimestamp(sync_status.last_sync_timestamp) if sync_status.last_sync_timestamp else 'Never'}" + ) + logger.info(f" Total synced: {sync_status.total_emails_synced}") + + # Mark sync as starting + gmail_sync_status.mark_sync_start(user_id) + + # Initialize sync status + self.active_syncs[user_id] = { + "status": "running", + "start_time": time.time(), + "total_emails": 0, + "processed": 0, + "indexed": 0, + "errors": 0, + "sync_type": "incremental" if incremental else "full", + "last_sync_timestamp": sync_status.last_sync_timestamp, + } + + try: + # Validate OAuth token + access_token = oauth_token.get("access_token") + if not access_token: + raise ValueError("OAuth token missing access_token") + + # Check for Gmail scopes + token_scope = oauth_token.get("scope", "") + if "gmail" not in token_scope: + logger.warning( + f"User {user_id} OAuth token doesn't include Gmail scopes" + ) + raise ValueError("OAuth token missing Gmail scopes") + + logger.info(f"āœ… OAuth token validated with Gmail scopes") + + # Step 1: Fetch emails from Gmail API + logger.info(f"šŸ“„ Step 1: Fetching emails from Gmail API...") + + fetcher = GmailFetcher( + oauth_token=access_token, + max_requests_per_second=40, # Conservative rate limit + timeout=30, + ) + + # Determine sync query based on incremental mode + if incremental and sync_status.last_sync_timestamp: + # Incremental sync - get emails since last sync + days_since_sync = (time.time() - sync_status.last_sync_timestamp) / ( + 24 * 3600 + ) + + # Smart time window selection based on sync frequency + if days_since_sync < 0.5: # Less than 12 hours + query = "newer_than:6h" + logger.info( + f" Incremental sync: fetching emails from last 6 hours" + ) + elif days_since_sync < 1: # Less than 24 hours + query = "newer_than:1d" + logger.info( + f" Incremental sync: fetching emails from last 24 hours" + ) + elif days_since_sync < 3: # Less than 3 days + query = "newer_than:3d" + logger.info( + f" Incremental sync: fetching emails from last 3 days" + ) + elif days_since_sync < 7: # Less than a week + query = "newer_than:7d" + logger.info( + f" Incremental sync: fetching emails from last 7 days" + ) + else: + # If last sync was much older, do a broader catch-up + query = "newer_than:30d" + logger.info( + f" Incremental sync: fetching emails from last 30 days (catch-up)" + ) + + # Adaptive batch size based on time window + if days_since_sync < 1: + max_emails = min(max_emails, 500) # Small batch for recent syncs + elif days_since_sync < 7: + max_emails = min(max_emails, 1000) # Medium batch for weekly syncs + else: + max_emails = min(max_emails, 2000) # Larger batch for catch-up + + logger.info(f" Adaptive batch size: {max_emails} emails") + else: + # Full sync - get all emails + query = None + logger.info( + f" Full sync: fetching all emails (first time or disabled incremental)" + ) + + emails, fetch_stats = await fetcher.fetch_all_emails( + max_emails=max_emails, + skip_spam_trash=skip_spam_trash, + query=query, + batch_size=100, + ) + + self.active_syncs[user_id]["total_emails"] = len(emails) + + # Analyze email types + label_counts = {} + for email in emails: + labels = email.get("labelIds", []) + for label in labels: + label_counts[label] = label_counts.get(label, 0) + 1 + + logger.info("=" * 70) + logger.info(f"āœ… FETCH COMPLETE") + logger.info(f" Total emails: {len(emails)}") + logger.info(f" API calls: {fetch_stats['api_calls']}") + logger.info(f" Fetch time: {fetch_stats.get('total_time', 0):.1f}s") + logger.info(f" Errors: {fetch_stats['errors']}") + logger.info(f"\n šŸ“Š Email Breakdown by Label:") + for label, count in sorted( + label_counts.items(), key=lambda x: x[1], reverse=True + )[:10]: + logger.info(f" - {label}: {count} emails") + if len(label_counts) > 10: + logger.info(f" - ... and {len(label_counts)-10} more labels") + logger.info("=" * 70) + + if not emails: + logger.info(f"šŸ“­ No emails found for user {user_id}") + self.active_syncs[user_id]["status"] = "completed" + return self._build_sync_result(user_id) + + # Step 2: Process and index emails to Pinecone + logger.info("") + logger.info("=" * 70) + logger.info(f"āš™ļø PROCESSING & INDEXING - {len(emails)} emails") + logger.info("=" * 70) + + indexed_count = await self._process_and_index_batch( + emails=emails, + user_id=user_id, + batch_size=50, # Reduced from 100 to prevent memory buildup + ) + + self.active_syncs[user_id]["indexed"] = indexed_count + self.active_syncs[user_id]["status"] = "completed" + + # Mark sync as completed in database + sync_duration = int(time.time() - self.active_syncs[user_id]["start_time"]) + gmail_sync_status.mark_sync_complete( + user_id=user_id, + emails_synced=indexed_count, + duration_seconds=sync_duration, + last_history_id=None, # TODO: Track Gmail history ID for better incremental sync + last_email_id=None, # TODO: Track last email ID + ) + + # Clear emails list to free memory + emails.clear() + del emails + + result = self._build_sync_result(user_id) + + # Calculate performance metrics + total_time = result["total_time"] + emails_per_second = result["indexed"] / total_time if total_time > 0 else 0 + sync_type = self.active_syncs[user_id].get("sync_type", "unknown") + + logger.info("") + logger.info("=" * 70) + logger.info(f"šŸŽ‰ GMAIL SYNC COMPLETED - User: {user_id}") + logger.info(f" Sync type: {sync_type.upper()}") + logger.info(f" Total emails fetched: {result['total_emails']}") + logger.info(f" Successfully indexed: {result['indexed']}") + logger.info( + f" Duplicates skipped: {self.active_syncs[user_id].get('duplicates_skipped', 0)}" + ) + logger.info(f" Errors: {result['errors']}") + logger.info(f" Total time: {total_time:.1f}s") + logger.info(f" Processing rate: {emails_per_second:.1f} emails/sec") + + # Performance assessment + if emails_per_second > 10: + logger.info( + f" Performance: šŸš€ EXCELLENT ({emails_per_second:.1f} emails/sec)" + ) + elif emails_per_second > 5: + logger.info( + f" Performance: āœ… GOOD ({emails_per_second:.1f} emails/sec)" + ) + else: + logger.info( + f" Performance: āš ļø SLOW ({emails_per_second:.1f} emails/sec)" + ) + + logger.info(f" Status: āœ… SUCCESS") + logger.info("=" * 70) + + return result + + except Exception as e: + logger.error(f"āŒ Gmail sync failed for user {user_id}: {e}", exc_info=True) + self.active_syncs[user_id]["status"] = "failed" + self.active_syncs[user_id]["error"] = str(e) + + # Mark sync as failed in database + gmail_sync_status.mark_sync_error(user_id, str(e)) + + raise + + async def _process_and_index_batch( + self, + emails: List[dict], + user_id: str, + batch_size: int = 100, + ) -> int: + """ + Process emails and index to Pinecone in batches. + + Args: + emails: List of Gmail API email responses + user_id: User ID for isolation + batch_size: Number of emails to process per batch + + Returns: + Total number of emails successfully indexed + """ + + total_indexed = 0 + total_vectors = 0 + + for i in range(0, len(emails), batch_size): + batch = emails[i : i + batch_size] + batch_num = i // batch_size + 1 + total_batches = (len(emails) + batch_size - 1) // batch_size + progress_pct = int((i / len(emails)) * 100) + + logger.info("") + logger.info( + f"šŸ“¦ Batch {batch_num}/{total_batches} ({progress_pct}% complete)" + ) + logger.info( + f" Processing emails {i+1}-{min(i+batch_size, len(emails))} of {len(emails)}" + ) + logger.info(f" Batch size: {len(batch)} emails") + + # Process batch with GmailIndexer + try: + result = await self.indexer.process_email_batch( + emails=batch, + user_id=user_id, + ) + + upsert_data = result["upsert_data"] + + if upsert_data: + # Upsert to vector DB with namespace and user isolation + await self.pinecone.schedule_upsert( + upsert_data, + user_id=user_id, + namespace=None, # Will use self.namespace from PineconeManager + ) + + total_indexed += result["processed"] + total_vectors += result["total_vectors"] + + duplicates = result.get("duplicates_skipped", 0) + logger.info(f" āœ… Processed: {result['processed']} unique emails") + if duplicates > 0: + logger.info( + f" ā­ļø Skipped: {duplicates} duplicate emails (mass email deduplication)" + ) + logger.info(f" āœ… Vectors created: {result['total_vectors']}") + logger.info(f" āœ… Errors: {result['errors']}") + logger.info( + f" āœ… Processing time: {result['processing_time']:.2f}s" + ) + logger.info( + f" šŸ“Š Running total: {total_indexed} emails, {total_vectors} vectors" + ) + + # Clear upsert_data to free memory after upserting + upsert_data.clear() + del result + + self.active_syncs[user_id]["processed"] = i + len(batch) + + # Clear processed batch to free memory + batch.clear() + + # Yield to event loop after each batch to keep interface responsive + await asyncio.sleep(0) # Yield control to event loop immediately + + # Additional yield every few batches for better responsiveness + if batch_num % 5 == 0: + await asyncio.sleep(0.01) # Small delay every 5 batches + + except Exception as e: + logger.error(f"Error processing batch {batch_num}: {e}") + self.active_syncs[user_id]["errors"] += 1 + continue + + return total_indexed + + def _build_sync_result(self, user_id: str) -> Dict: + """Build sync result with statistics""" + + sync_status = self.active_syncs.get(user_id, {}) + elapsed_time = time.time() - sync_status.get("start_time", time.time()) + + return { + "user_id": user_id, + "status": sync_status.get("status", "unknown"), + "total_emails": sync_status.get("total_emails", 0), + "processed": sync_status.get("processed", 0), + "indexed": sync_status.get("indexed", 0), + "errors": sync_status.get("errors", 0), + "total_time": elapsed_time, + "error": sync_status.get("error"), + } + + def get_sync_status(self, user_id: str) -> Optional[Dict]: + """Get current sync status for a user""" + + if user_id not in self.active_syncs: + return None + + return self._build_sync_result(user_id) + + +# ============================================================================ +# TRIGGER FUNCTION (called from OAuth callback) +# ============================================================================ + + +async def trigger_gmail_sync_if_needed( + request, + user_id: str, + provider: str, + token: dict, + is_new_user: bool = False, +): + """ + Trigger Gmail sync if conditions are met. + + This function is called from the OAuth callback handler. + + Args: + request: FastAPI request object + user_id: User ID who just authenticated + provider: OAuth provider (e.g., "google") + token: OAuth token dict + is_new_user: True if this is first-time signup + + Conditions for triggering: + - Provider must be "google" + - Token must have Gmail scopes + - ENABLE_GMAIL_AUTO_SYNC must be True + - If GMAIL_AUTO_SYNC_ON_SIGNUP_ONLY, only trigger for new users + """ + + logger.info( + f"\nšŸ” Gmail Sync Trigger Check - User: {user_id}, Provider: {provider}" + ) + + # Check if Gmail auto-sync is enabled + if not request.app.state.config.ENABLE_GMAIL_AUTO_SYNC: + logger.info(" ā­ļø SKIP: Gmail auto-sync is disabled in config") + return + + # Only trigger for Google OAuth + if provider != "google": + logger.info(f" ā­ļø SKIP: Not Google OAuth (provider: {provider})") + return + + # Check for Gmail scopes + token_scope = token.get("scope", "") + has_gmail = "gmail" in token_scope + logger.info(f" šŸ“§ Gmail scope present: {has_gmail}") + + if not has_gmail: + logger.info(f" ā­ļø SKIP: OAuth token doesn't include Gmail scopes") + return + + # Check if admin has enabled Gmail sync for this user first + user = Users.get_user_by_id(user_id) + if not user: + logger.info(f" ā­ļø SKIP: User {user_id} not found") + return + + # Check admin setting for Gmail sync + admin_sync_enabled = getattr(user, 'gmail_sync_enabled', 0) == 1 + logger.info(f" āš™ļø Admin Gmail sync setting: gmail_sync_enabled={admin_sync_enabled}") + + # Check if we should only sync on signup (only applies if admin hasn't explicitly enabled) + sync_on_signup_only = request.app.state.config.GMAIL_AUTO_SYNC_ON_SIGNUP_ONLY + logger.info( + f" šŸ‘¤ New user: {is_new_user}, Sync on signup only: {sync_on_signup_only}" + ) + + # If admin has explicitly enabled sync, allow it regardless of signup-only setting + if not admin_sync_enabled: + if sync_on_signup_only and not is_new_user: + logger.info(f" ā­ļø SKIP: Gmail auto-sync only on signup, user is not new") + logger.info(f" Admin can enable sync in Admin -> Users -> Edit User -> Gmail Email Sync") + return + else: + logger.info(f" ā­ļø SKIP: Admin has disabled Gmail sync for this user") + logger.info(f" Enable in Admin -> Users -> Edit User -> Gmail Email Sync") + return + + # Check if user has enabled Gmail sync in their settings (additional user preference) + if user and user.settings: + settings_dict = ( + user.settings.model_dump() + if hasattr(user.settings, "model_dump") + else user.settings + ) + gmail_settings = ( + settings_dict.get("gmail", {}) if isinstance(settings_dict, dict) else {} + ) + user_sync_enabled = gmail_settings.get("sync_enabled", True) # Default to enabled if admin allows + logger.info(f" āš™ļø User Gmail sync preference: sync_enabled={user_sync_enabled}") + + if not user_sync_enabled: + logger.info(f" ā­ļø SKIP: User has disabled Gmail sync in their settings") + return + else: + logger.info(f" āš™ļø No user settings found - will sync (admin enabled, no user preference)") + + logger.info(f" āœ… ALL CONDITIONS MET - Triggering Gmail sync!") + logger.info( + f"\nšŸš€ TRIGGERING AUTOMATIC GMAIL SYNC" + f"\n User: {user_id}" + f"\n New user: {is_new_user}" + f"\n Gmail scopes: āœ“" + ) + + # Launch background task + try: + from open_webui.tasks import create_task + + # Create the background sync task + task_id, task = await create_task( + request.app.state.redis, + _background_gmail_sync(request, user_id, token), + id=f"gmail_sync_{user_id}", + ) + + logger.info(f"āœ… Gmail sync background task created: {task_id}") + + except Exception as e: + logger.error(f"āŒ Failed to create Gmail sync task for user {user_id}: {e}") + + +async def _background_gmail_sync(request, user_id: str, oauth_token: dict): + """ + Background task that performs the actual Gmail sync. + + This runs asynchronously and doesn't block the OAuth callback. + Integrates with existing Open WebUI RAG infrastructure. + + Args: + request: FastAPI request object + user_id: User ID to sync + oauth_token: OAuth token dict with access_token + """ + + logger.info("\n" + "šŸ”„" * 35) + logger.info("šŸ“§ BACKGROUND GMAIL SYNC TASK STARTED") + logger.info(f" User ID: {user_id}") + logger.info(f" Task ID: gmail_sync_{user_id}") + logger.info(f" Time: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + logger.info("šŸ”„" * 35 + "\n") + + try: + # Validate user + user = Users.get_user_by_id(user_id) + if not user: + logger.error(f"User {user_id} not found") + return + + # Get configuration + max_emails = request.app.state.config.GMAIL_AUTO_SYNC_MAX_EMAILS + skip_spam_trash = request.app.state.config.GMAIL_SKIP_SPAM_AND_TRASH + batch_size = request.app.state.config.GMAIL_SYNC_BATCH_SIZE + gmail_namespace = request.app.state.config.PINECONE_NAMESPACE_GMAIL + + logger.info( + f"Gmail sync config: max_emails={max_emails}, " + f"skip_spam_trash={skip_spam_trash}, batch_size={batch_size}, " + f"namespace={gmail_namespace}" + ) + + # Create simple wrapper services for the background task + # We'll use Open WebUI's existing embedding and Pinecone infrastructure + + class SimpleEmbeddingService: + """Wrapper for Open WebUI's embedding function""" + + def __init__(self, app_state): + self.app_state = app_state + + async def embed_batch(self, texts: List[str]) -> List[List[float]]: + """Generate embeddings using Open WebUI's embedding function (non-blocking)""" + + async def embed_single(text: str) -> List[float]: + """Embed single text in executor to avoid blocking event loop""" + try: + # Run CPU-intensive embedding in thread executor + loop = asyncio.get_running_loop() + + # Prepare and embed in thread pool + def prepare_and_embed(): + # Truncate to safe limit + clean_text = text.strip() + max_chars = 8000 + if len(clean_text) > max_chars: + clean_text = clean_text[:max_chars] + + # Ensure not empty + if not clean_text: + clean_text = "Email content not available" + + # Call embedding function (may be network I/O) + return self.app_state.EMBEDDING_FUNCTION( + clean_text, prefix="", user=None + ) + + vector = await loop.run_in_executor(None, prepare_and_embed) + + if vector is None: + raise ValueError("Embedding function returned None") + + return vector + + except Exception as e: + logger.error(f"Embedding error: {e}") + # Fallback to zero vector + try: + dim = int(self.app_state.config.PINECONE_DIMENSION) + except: + dim = 1536 + return [0.0] * dim + + # Process embeddings concurrently but with small batches to avoid overwhelming + embeddings = [] + batch_size = 3 # Even smaller batch to keep event loop responsive + + for i in range(0, len(texts), batch_size): + batch = texts[i : i + batch_size] + batch_embeddings = await asyncio.gather( + *[embed_single(t) for t in batch] + ) + embeddings.extend(batch_embeddings) + + # Yield control to event loop between batches + await asyncio.sleep(0) + + # Additional yield every few batches + if (i // batch_size) % 3 == 0: + await asyncio.sleep(0.001) # Small delay every 3 batches + + return embeddings + + class SimpleTextSplitter: + """Simple text splitter (fallback if ContentAwareTextSplitter not available)""" + + def split_text(self, text: str) -> List[str]: + """Split text into chunks by paragraphs""" + # Simple split by double newlines + chunks = text.split("\n\n") + chunks = [ + c.strip() for c in chunks if c.strip() and len(c.strip()) > 50 + ] + + # If no chunks or single large chunk, return as-is + if not chunks: + return [text] if text else [] + + return chunks + + # Reuse quality scoring from gmail_indexer (no duplication!) + from open_webui.utils.gmail_indexer import GmailIndexer + + class SimpleDocProcessor: + """Thin wrapper to provide quality_score method""" + + @staticmethod + def quick_quality_score(text: str) -> int: + """Delegate to shared implementation""" + # Use the same logic as everywhere else + score = 0 + if len(text) > 200: + score += 1 + if text.count(".") >= 2: + score += 1 + if "\n\n" in text: + score += 1 + if not (text.count("\n- ") > 5 or text.count("\nā€¢ ") > 5): + score += 1 + if re.search( + r"\b(because|therefore|however|additionally)\b", text, re.IGNORECASE + ): + score += 1 + return score + + class SimplePineconeManager: + """ + Thin wrapper around VECTOR_DB_CLIENT for Gmail email storage. + + Uses collection-based isolation (not namespaces): + - Each user has their own collection: gmail_{user_id} + - User isolation via collection name + metadata user_id + - Compatible with Open WebUI's vector DB abstraction + """ + + def __init__(self, namespace): + self.namespace = namespace # Gmail namespace (e.g., "gmail-inbox") + + async def schedule_upsert( + self, upsert_data: List[dict], user_id: str, namespace: str = None + ): + """ + Upsert vectors using existing VECTOR_DB_CLIENT with namespace support. + + Namespace Strategy: + - Gmail emails: Use PINECONE_NAMESPACE_GMAIL (e.g., "gmail-inbox") + - Chat summaries: Use PINECONE_NAMESPACE (e.g., "chat-summary-knowledge") + - User isolation via metadata user_id + - Type metadata: "email" (document type filter) + + Runs in thread executor to avoid blocking the async event loop. + """ + from open_webui.retrieval.vector.factory import VECTOR_DB_CLIENT + + # VECTOR_DB_CLIENT.upsert expects List[dict] with keys: id, text, vector, metadata + # Filter out zero vectors (Pinecone rejects them) + valid_items = [] + zero_vector_count = 0 + + for item in upsert_data: + vector = item["values"] + # Check if vector has at least one non-zero value + if any(v != 0.0 for v in vector): + # Handle both dict and VectorItem object cases + metadata = item["metadata"] + if hasattr(metadata, "get"): + # It's a dictionary + chunk_text = metadata.get("chunk_text", "") + metadata_dict = metadata + else: + # It's a VectorItem object or similar - access as attributes + chunk_text = getattr(metadata, "chunk_text", "") + metadata_dict = ( + metadata.__dict__ + if hasattr(metadata, "__dict__") + else metadata + ) + + valid_items.append( + { + "id": item["id"], + "text": chunk_text, + "vector": vector, + "metadata": metadata_dict, + } + ) + else: + zero_vector_count += 1 + # Handle both dict and VectorItem object cases for email_id + metadata = item["metadata"] + if hasattr(metadata, "get"): + email_id = metadata.get("email_id", "unknown") + else: + email_id = getattr(metadata, "email_id", "unknown") + logger.warning( + f"āš ļø Skipping zero vector for email {email_id} (embedding failed)" + ) + + if zero_vector_count > 0: + logger.warning( + f"āš ļø Skipped {zero_vector_count} vectors with all zeros" + ) + + if not valid_items: + logger.warning( + "āš ļø No valid vectors in batch (all embeddings failed), skipping upsert" + ) + return + + # Use namespace-based isolation for Gmail emails WITH user isolation + # Collection name: "gmail" (shared across all users) + # Namespace: self.namespace (e.g., "gmail-inbox") + # User isolation: metadata user_id + collection filter + collection_name = "gmail" + gmail_namespace = self.namespace or "gmail-inbox" + + # Ensure all vectors have user_id in metadata for isolation + for item in valid_items: + if "metadata" not in item: + item["metadata"] = {} + item["metadata"]["user_id"] = user_id + item["metadata"]["collection_name"] = f"gmail_{user_id}" # For filtering + + logger.info( + f"šŸ“¤ Upserting {len(valid_items)} valid vectors to collection: {collection_name}" + ) + logger.info( + f" Namespace: '{gmail_namespace}' (Gmail emails separated from chat summaries)" + ) + logger.info( + f" User isolation: metadata user_id=""{user_id}"" + collection filter" + ) + + try: + # Use existing VECTOR_DB_CLIENT.upsert with namespace support + # Run in thread executor to avoid blocking async event loop + loop = asyncio.get_running_loop() + + # Check if the vector DB client supports namespace parameter + import inspect + upsert_signature = inspect.signature(VECTOR_DB_CLIENT.upsert) + if 'namespace' in upsert_signature.parameters: + # Vector DB supports namespace (e.g., Pinecone) + await loop.run_in_executor( + None, VECTOR_DB_CLIENT.upsert, collection_name, valid_items, gmail_namespace + ) + logger.info( + f"āœ… Upserted {len(valid_items)} vectors to collection {collection_name} in namespace '{gmail_namespace}' with user isolation" + ) + else: + # Vector DB doesn't support namespace (e.g., Chroma) - use collection-based isolation + user_collection = f"gmail_{user_id}" + await loop.run_in_executor( + None, VECTOR_DB_CLIENT.upsert, user_collection, valid_items + ) + logger.info( + f"āœ… Upserted {len(valid_items)} vectors to collection {user_collection} (no namespace support - using collection isolation)" + ) + + # Clear valid_items to free memory immediately after upsert + valid_items.clear() + + except Exception as e: + logger.error(f"āŒ Vector DB upsert error: {e}") + # Clear memory even on error + valid_items.clear() + raise + + # Initialize services + embedding_service = SimpleEmbeddingService(request.app.state) + text_splitter = SimpleTextSplitter() + doc_processor = SimpleDocProcessor() + pinecone_manager = SimplePineconeManager(gmail_namespace) + + # Initialize GmailAutoSync orchestrator + gmail_sync = GmailAutoSync( + embedding_service=embedding_service, + content_aware_splitter=text_splitter, + document_processor=doc_processor, + pinecone_manager=pinecone_manager, + gmail_namespace=gmail_namespace, + ) + + # Execute the sync + logger.info(f"šŸš€ Starting Gmail sync for user {user_id}...") + + result = await gmail_sync.sync_user_gmail( + user_id=user_id, + oauth_token=oauth_token, + max_emails=max_emails, + skip_spam_trash=skip_spam_trash, + incremental=True, # Enable incremental sync by default + ) + + logger.info("\n" + "šŸŽ‰" * 35) + logger.info("āœ… BACKGROUND GMAIL SYNC TASK COMPLETED") + logger.info(f" User ID: {user_id}") + logger.info(f" Emails indexed: {result['indexed']}") + logger.info(f" Total time: {result['total_time']:.1f}s") + logger.info(f" Status: SUCCESS") + logger.info("šŸŽ‰" * 35 + "\n") + + except Exception as e: + logger.error("\n" + "āŒ" * 35) + logger.error(f"āŒ BACKGROUND GMAIL SYNC TASK FAILED - User: {user_id}") + logger.error(f" Error: {str(e)}") + logger.error("āŒ" * 35 + "\n") + logger.error(f"Full traceback:", exc_info=True) + + +# ============================================================================ +# TESTING +# ============================================================================ + + +async def test_auto_sync_orchestrator(): + """Test the auto-sync orchestrator logic""" + + print("\n" + "=" * 60) + print("Phase 5 - Auto-Sync Orchestrator Test") + print("=" * 60) + + # Mock services + class MockEmbeddingService: + async def embed_batch(self, texts): + return [[0.1] * 1536 for _ in texts] + + class MockSplitter: + def split_text(self, text): + return [text] # Simple: one chunk per email + + class MockDocProcessor: + @staticmethod + def quick_quality_score(text): + return 4 + + class MockPineconeManager: + async def schedule_upsert(self, data, user_id, namespace=None): + logger.info(f"Mock upsert: {len(data)} vectors for user '{user_id}' to namespace '{namespace}'") + return f"job_{time.time()}" + + print("\nāœ… TEST 1: Initialize Auto-Sync Orchestrator") + + sync = GmailAutoSync( + embedding_service=MockEmbeddingService(), + content_aware_splitter=MockSplitter(), + document_processor=MockDocProcessor(), + pinecone_manager=MockPineconeManager(), + gmail_namespace="gmail-inbox", + ) + + print(f" Namespace: {sync.gmail_namespace}") + print(f" Indexer initialized: āœ“") + print(f" Pinecone manager ready: āœ“") + print(" āœ… PASSED\n") + + print("āœ… TEST 2: Sync Status Tracking") + + # Simulate starting a sync + test_user_id = "test_user_999" + sync.active_syncs[test_user_id] = { + "status": "running", + "start_time": time.time(), + "total_emails": 100, + "processed": 50, + "indexed": 45, + "errors": 5, + } + + status = sync.get_sync_status(test_user_id) + + print(f" User: {status['user_id']}") + print(f" Status: {status['status']}") + print(f" Processed: {status['processed']}/{status['total_emails']}") + print(f" Indexed: {status['indexed']}") + print(f" Errors: {status['errors']}") + + assert status["status"] == "running", "Status should be running" + assert status["processed"] == 50, "Should track processed count" + + print(" āœ… PASSED\n") + + print("=" * 60) + print("Phase 5 Orchestrator Tests Complete āœ…") + print("=" * 60) + + print("\nOrchestrator capabilities:") + print(" āœ… Coordinates all Gmail components") + print(" āœ… Tracks sync progress per user") + print(" āœ… Integrates with existing services") + print(" āœ… Ready for OAuth callback integration") + + return True + + +async def periodic_gmail_sync_scheduler(): + """ + Production-grade periodic Gmail sync scheduler. + + Features: + - Graceful startup with configurable delay + - Dynamic configuration updates + - Batch processing with rate limiting + - Comprehensive error handling + - Circuit breaker pattern for failures + - Memory-efficient user processing + - Distributed system friendly (no race conditions) + """ + logger.info("šŸ”„ Gmail Periodic Sync Scheduler starting...") + + # Graceful startup - wait for application to fully initialize + startup_delay = 60 # Wait 60 seconds for complete initialization + await asyncio.sleep(startup_delay) + + # Configuration defaults + sync_interval_hours = 6 + check_interval = 30 * 60 # 30 minutes + consecutive_errors = 0 + max_consecutive_errors = 5 + + logger.info("āœ… Gmail Periodic Sync Scheduler ready") + + while True: + try: + # Dynamic configuration reload (supports runtime updates) + try: + from open_webui.config import GMAIL_PERIODIC_SYNC_INTERVAL_HOURS + # Extract value from PersistentConfig object using .value attribute + sync_interval_hours = GMAIL_PERIODIC_SYNC_INTERVAL_HOURS.value + # Ensure it's an integer + if not isinstance(sync_interval_hours, int): + sync_interval_hours = int(sync_interval_hours) + except Exception as e: + logger.warning(f"Failed to load sync interval config, using default: {e}") + sync_interval_hours = 6 + + # Circuit breaker: back off if too many failures + if consecutive_errors >= max_consecutive_errors: + backoff_time = min(3600, 300 * consecutive_errors) # Max 1 hour + logger.error( + f"šŸ”“ Circuit breaker: {consecutive_errors} consecutive errors. " + f"Backing off for {backoff_time}s" + ) + await asyncio.sleep(backoff_time) + consecutive_errors = 0 # Reset after backoff + continue + + # Query database for users needing sync (indexed query) + users_needing_sync = gmail_sync_status.get_users_needing_sync( + max_hours_since_sync=sync_interval_hours + ) + + if users_needing_sync: + total_users = len(users_needing_sync) + logger.info( + f"šŸ“§ Periodic sync: {total_users} user(s) need sync " + f"(interval: {sync_interval_hours}h)" + ) + + # Adaptive batch sizing based on load + batch_size = 2 if total_users > 10 else 3 + successful_syncs = 0 + failed_syncs = 0 + + # Process users in batches with rate limiting + for i in range(0, total_users, batch_size): + batch = users_needing_sync[i:i + batch_size] + batch_start = time.time() + + # Parallel processing within batch + tasks = [ + asyncio.create_task( + _sync_user_periodic(user_id), + name=f"periodic_gmail_sync_{user_id}" + ) + for user_id in batch + ] + + # Wait for batch completion with error isolation + results = await asyncio.gather(*tasks, return_exceptions=True) + + # Track results for monitoring + for j, result in enumerate(results): + if isinstance(result, Exception): + failed_syncs += 1 + logger.error( + f"āŒ Periodic sync failed for user {batch[j]}: " + f"{type(result).__name__}: {result}" + ) + elif result: # Successful sync + successful_syncs += 1 + + batch_duration = time.time() - batch_start + + # Adaptive rate limiting between batches + if i + batch_size < total_users: + # Add delay based on batch processing time (backpressure) + delay = max(10, min(30, batch_duration * 0.5)) + await asyncio.sleep(delay) + + # Summary logging + logger.info( + f"āœ… Periodic sync cycle complete: " + f"{successful_syncs} succeeded, {failed_syncs} failed " + f"(total: {total_users})" + ) + + # Reset error counter on successful cycle + if successful_syncs > 0: + consecutive_errors = 0 + else: + consecutive_errors += 1 + else: + logger.debug( + f"šŸ“§ No users need sync at this time (interval: {sync_interval_hours}h)" + ) + consecutive_errors = 0 # Reset on successful query + + # Wait before next check (with jitter to avoid thundering herd) + jitter = time.time() % 60 # 0-60 second jitter + actual_wait = check_interval + jitter + logger.debug(f"ā° Next sync check in {actual_wait//60:.1f} minutes") + await asyncio.sleep(actual_wait) + + except asyncio.CancelledError: + logger.info("šŸ›‘ Gmail Periodic Sync Scheduler shutting down gracefully") + raise # Propagate cancellation + except Exception as e: + consecutive_errors += 1 + logger.exception( + f"āŒ Unexpected error in periodic sync scheduler " + f"(error #{consecutive_errors}): {e}" + ) + # Exponential backoff on errors + error_backoff = min(600, 60 * consecutive_errors) + await asyncio.sleep(error_backoff) + + +async def _sync_user_periodic(user_id: str) -> bool: + """ + Sync Gmail for a single user during periodic sync (production-grade). + + Features: + - Comprehensive validation with early returns + - Graceful OAuth token handling + - Timeout protection + - Memory-efficient processing + - Detailed error logging + + Args: + user_id: The user ID to sync + + Returns: + bool: True if sync succeeded, False otherwise + """ + sync_start = time.time() + + try: + logger.debug(f"šŸ”„ Periodic sync starting for user: {user_id}") + + # Validation: Check user exists + user = Users.get_user_by_id(user_id) + if not user: + logger.warning(f"āš ļø User {user_id} not found in database, skipping") + return False + + # Validation: Check admin enabled sync + admin_sync_enabled = getattr(user, 'gmail_sync_enabled', 0) == 1 + if not admin_sync_enabled: + logger.debug(f"ā­ļø Gmail sync disabled by admin for user {user_id}") + return False + + # Validation: Check OAuth session exists + oauth_session = OAuthSessions.get_session_by_provider_and_user_id( + "google", user_id + ) + if not oauth_session: + logger.debug(f"ā­ļø No Google OAuth session for user {user_id}") + return False + + # Validation: Check OAuth token validity + if not oauth_session.access_token: + logger.warning(f"āš ļø Missing access token for user {user_id}") + return False + + # Prepare OAuth token with proper structure + oauth_token = { + "access_token": oauth_session.access_token, + "refresh_token": oauth_session.refresh_token, + "token_type": "Bearer", + "expires_in": 3600, + } + + # Check if this is first sync for this user + sync_status = gmail_sync_status.get_sync_status(user_id) + is_first_sync = (sync_status is None or sync_status.last_sync_timestamp is None) + + # Create sync instance (reuses existing infrastructure) + gmail_sync = GmailAutoSync() + + # Determine timeout based on sync type + if is_first_sync: + # First sync: limit emails more aggressively, allow more time + max_sync_emails = 200 # Conservative for first background sync + sync_timeout = 600 # 10 minutes for first sync + logger.info(f"šŸ†• First sync for user {user_id} - limiting to {max_sync_emails} emails") + else: + # Ongoing sync: normal limits + max_sync_emails = 500 + sync_timeout = 900 # 15 minutes + + # Perform incremental sync with timeout protection + try: + # Use asyncio.wait_for to enforce timeout + result = await asyncio.wait_for( + gmail_sync.sync_user_gmail( + user_id=user_id, + oauth_token=oauth_token, + max_emails=max_sync_emails, + skip_spam_trash=True, + incremental=True, # Always use incremental mode + # incremental=True with last_sync_timestamp=None will do full sync automatically + ), + timeout=sync_timeout + ) + except asyncio.TimeoutError: + duration = time.time() - sync_start + logger.error( + f"ā±ļø Periodic sync timeout for user {user_id} " + f"after {duration:.1f}s (max: 900s)" + ) + return False + + # Process results + if result.get("success"): + emails_synced = result.get("emails_synced", 0) + duration = time.time() - sync_start + logger.info( + f"āœ… Periodic sync: user {user_id} - " + f"{emails_synced} emails synced in {duration:.1f}s" + ) + return True + else: + error_msg = result.get("error", "Unknown error") + duration = time.time() - sync_start + logger.error( + f"āŒ Periodic sync failed for user {user_id} " + f"after {duration:.1f}s: {error_msg}" + ) + return False + + except asyncio.CancelledError: + logger.info(f"šŸ›‘ Periodic sync cancelled for user {user_id}") + raise # Propagate cancellation + except Exception as e: + duration = time.time() - sync_start + logger.exception( + f"āŒ Unexpected error in periodic sync for user {user_id} " + f"after {duration:.1f}s: {type(e).__name__}: {e}" + ) + return False + + +if __name__ == "__main__": + print("Testing Phase 5 - Gmail Auto-Sync Orchestrator") + success = asyncio.run(test_auto_sync_orchestrator()) + + if success: + print("\nšŸŽ‰ Phase 5 orchestrator tests PASSED!") + print("\nNext: Hook into OAuth callback (Phase 5.4)") + + exit(0 if success else 1) diff --git a/backend/open_webui/utils/gmail_fetcher.py b/backend/open_webui/utils/gmail_fetcher.py new file mode 100644 index 0000000000..d6af653e35 --- /dev/null +++ b/backend/open_webui/utils/gmail_fetcher.py @@ -0,0 +1,470 @@ +""" +Gmail API Fetcher + +This module handles fetching emails from Gmail API with: +- OAuth token authentication +- Pagination for bulk fetching +- Rate limiting to respect Gmail API quotas +- Batch operations for efficiency + +Gmail API Quotas: +- 250 quota units per user per second +- List messages: 5 units per request +- Get message: 5 units per request +- Recommended: ~40 requests/second max +""" + +import logging +import asyncio +import time +from typing import Dict, List, Optional, Tuple +import aiohttp + +# Set up logger with INFO level for visibility +logger = logging.getLogger(__name__) +logger.setLevel(logging.INFO) + + +class GmailFetcher: + """ + Fetches emails from Gmail API with proper rate limiting and pagination. + + Uses OAuth token from user's login session. + """ + + # Gmail API base URL + GMAIL_API_BASE = "https://gmail.googleapis.com/gmail/v1" + + # Rate limiting configuration + DEFAULT_MAX_REQUESTS_PER_SECOND = 40 # Conservative limit + DEFAULT_BATCH_SIZE = 20 # Concurrent requests (Gmail limit: ~25 concurrent) + + def __init__( + self, + oauth_token: str, + max_requests_per_second: int = DEFAULT_MAX_REQUESTS_PER_SECOND, + timeout: int = 30, + ): + """ + Initialize Gmail fetcher. + + Args: + oauth_token: OAuth access token with Gmail scopes + max_requests_per_second: Rate limit (default: 40 req/s) + timeout: Request timeout in seconds (default: 30) + """ + self.oauth_token = oauth_token + self.max_requests_per_second = max_requests_per_second + self.timeout = timeout + + # Rate limiting tracking + self.request_times = [] + self.request_lock = asyncio.Lock() + + # Statistics + self.stats = { + "api_calls": 0, + "emails_fetched": 0, + "errors": 0, + "rate_limit_waits": 0, + } + + async def _rate_limit(self): + """ + Implement rate limiting to respect Gmail API quotas. + + Ensures we don't exceed max_requests_per_second. + """ + async with self.request_lock: + now = time.time() + + # Remove requests older than 1 second + self.request_times = [t for t in self.request_times if now - t < 1.0] + + # Check if we're at the limit + if len(self.request_times) >= self.max_requests_per_second: + # Calculate wait time + oldest_request = self.request_times[0] + wait_time = 1.0 - (now - oldest_request) + + if wait_time > 0: + self.stats["rate_limit_waits"] += 1 + logger.debug(f"Rate limit reached, waiting {wait_time:.3f}s") + await asyncio.sleep(wait_time) + + # Record this request + self.request_times.append(time.time()) + + async def fetch_all_message_ids( + self, + max_results: int = 0, + skip_spam_trash: bool = True, + query: Optional[str] = None, + ) -> List[str]: + """ + Fetch all message IDs from Gmail with pagination. + + This fetches ALL emails from the entire mailbox including: + - INBOX emails + - SENT emails (emails you sent) + - All custom labels/folders + - Archived emails + - Drafts + - Everything except SPAM and TRASH (if skip_spam_trash=True) + + Args: + max_results: Maximum number of message IDs to fetch (0 = unlimited) + skip_spam_trash: Skip emails in SPAM and TRASH (default: True) + query: Optional Gmail query filter (e.g., "newer_than:7d") + + Returns: + List of Gmail message IDs from entire mailbox + + Example: + >>> fetcher = GmailFetcher(oauth_token) + >>> message_ids = await fetcher.fetch_all_message_ids(max_results=1000) + >>> print(f"Found {len(message_ids)} emails across all folders") + """ + + logger.info("šŸ“„ Fetching message IDs from Gmail...") + + all_message_ids = [] + next_page_token = None + page_count = 0 + + # Build query - fetches ALL emails (inbox, sent, labels, etc.) + query_parts = [] + if skip_spam_trash: + # Exclude spam and trash, but include EVERYTHING else + query_parts.append("-in:spam -in:trash") + logger.info(" šŸ“‹ Scope: ALL emails (INBOX + SENT + all labels)") + logger.info(" šŸš« Excluding: SPAM and TRASH only") + else: + logger.info(" šŸ“‹ Scope: ENTIRE mailbox (including spam/trash)") + + if query: + query_parts.append(query) + logger.info(f" šŸ” Additional filter: {query}") + + full_query = " ".join(query_parts) if query_parts else None + + if full_query: + logger.info(f" šŸ“ Gmail query: {full_query}") + + # Fetch pages + async with aiohttp.ClientSession() as session: + while True: + page_count += 1 + + # Build request parameters + params = { + "maxResults": 500, # Gmail API max per request + } + + if full_query: + params["q"] = full_query + + if next_page_token: + params["pageToken"] = next_page_token + + # Rate limiting + await self._rate_limit() + + # Make API request + url = f"{self.GMAIL_API_BASE}/users/me/messages" + headers = { + "Authorization": f"Bearer {self.oauth_token}", + "Accept": "application/json" + } + + try: + async with session.get( + url, + params=params, + headers=headers, + timeout=aiohttp.ClientTimeout(total=self.timeout) + ) as response: + self.stats["api_calls"] += 1 + + if response.status == 401: + raise Exception("OAuth token expired or invalid") + + if response.status == 403: + raise Exception("Gmail API access forbidden - check API is enabled") + + if response.status != 200: + error_text = await response.text() + raise Exception(f"Gmail API error {response.status}: {error_text}") + + data = await response.json() + + except Exception as e: + self.stats["errors"] += 1 + logger.error(f"Error fetching message IDs (page {page_count}): {e}") + raise + + # Extract message IDs + messages = data.get("messages", []) + + if messages: + message_ids = [msg["id"] for msg in messages] + all_message_ids.extend(message_ids) + + logger.info( + f"Fetched page {page_count}: {len(message_ids)} IDs " + f"(total: {len(all_message_ids)})" + ) + + # Check for next page + next_page_token = data.get("nextPageToken") + + if not next_page_token: + logger.info(f"No more pages. Total message IDs: {len(all_message_ids)}") + break + + # Check max results limit + if max_results > 0 and len(all_message_ids) >= max_results: + all_message_ids = all_message_ids[:max_results] + logger.info(f"Reached max_results limit: {max_results}") + break + + logger.info( + f"Fetched {len(all_message_ids)} message IDs in {page_count} pages " + f"({self.stats['api_calls']} API calls)" + ) + + return all_message_ids + + async def fetch_email( + self, + message_id: str, + ) -> Optional[Dict]: + """ + Fetch a single email's full content. + + Args: + message_id: Gmail message ID + + Returns: + Gmail API message response (format='full') or None if error + """ + + # Rate limiting + await self._rate_limit() + + url = f"{self.GMAIL_API_BASE}/users/me/messages/{message_id}" + headers = { + "Authorization": f"Bearer {self.oauth_token}", + "Accept": "application/json" + } + params = {"format": "full"} + + try: + async with aiohttp.ClientSession() as session: + async with session.get( + url, + params=params, + headers=headers, + timeout=aiohttp.ClientTimeout(total=self.timeout) + ) as response: + self.stats["api_calls"] += 1 + + if response.status != 200: + error_text = await response.text() + logger.error( + f"Error fetching email {message_id}: " + f"{response.status} - {error_text}" + ) + self.stats["errors"] += 1 + return None + + email_data = await response.json() + self.stats["emails_fetched"] += 1 + + return email_data + + except Exception as e: + self.stats["errors"] += 1 + logger.error(f"Exception fetching email {message_id}: {e}") + return None + + async def fetch_emails_batch( + self, + message_ids: List[str], + batch_size: int = DEFAULT_BATCH_SIZE, + ) -> List[Dict]: + """ + Fetch multiple emails in batches with rate limiting. + + Args: + message_ids: List of Gmail message IDs to fetch + batch_size: Number of concurrent requests (default: 20, Gmail limit ~25) + + Returns: + List of Gmail API message responses + """ + + logger.info(f"Fetching {len(message_ids)} emails in batches of {batch_size}...") + + all_emails = [] + + for i in range(0, len(message_ids), batch_size): + batch = message_ids[i:i + batch_size] + + logger.info( + f"Fetching batch {i//batch_size + 1} " + f"({i+1}-{min(i+batch_size, len(message_ids))} of {len(message_ids)})" + ) + + # Fetch batch concurrently with limited concurrency (prevents 429) + tasks = [self.fetch_email(msg_id) for msg_id in batch] + batch_results = await asyncio.gather(*tasks, return_exceptions=True) + + # Handle exceptions in results + batch_emails = [] + for result in batch_results: + if isinstance(result, Exception): + logger.error(f"Batch fetch error: {result}") + elif result is not None: + batch_emails.append(result) + + all_emails.extend(batch_emails) + + logger.info( + f"Batch complete: {len(batch_emails)}/{len(batch)} successful " + f"(total: {len(all_emails)})" + ) + + logger.info( + f"Fetch complete: {len(all_emails)} emails fetched successfully " + f"(errors: {self.stats['errors']})" + ) + + return all_emails + + async def fetch_all_emails( + self, + max_emails: int = 0, + skip_spam_trash: bool = True, + query: Optional[str] = None, + batch_size: int = DEFAULT_BATCH_SIZE, + ) -> Tuple[List[Dict], Dict]: + """ + Complete workflow: Fetch message IDs, then fetch full email content. + + Args: + max_emails: Maximum number of emails to fetch (0 = unlimited) + skip_spam_trash: Skip SPAM and TRASH folders (default: True) + query: Optional Gmail query filter + batch_size: Concurrent fetch batch size (default: 100) + + Returns: + Tuple of (list of email data, statistics dict) + """ + + start_time = time.time() + + # Step 1: Fetch all message IDs + logger.info("Step 1: Fetching message IDs...") + message_ids = await self.fetch_all_message_ids( + max_results=max_emails, + skip_spam_trash=skip_spam_trash, + query=query + ) + + if not message_ids: + logger.warning("No message IDs found") + return [], self.stats + + # Step 2: Fetch full email content + logger.info(f"Step 2: Fetching {len(message_ids)} emails...") + emails = await self.fetch_emails_batch( + message_ids=message_ids, + batch_size=batch_size + ) + + elapsed_time = time.time() - start_time + + # Update stats + self.stats["total_time"] = elapsed_time + self.stats["emails_per_second"] = len(emails) / elapsed_time if elapsed_time > 0 else 0 + + logger.info( + f"Fetch complete: {len(emails)} emails in {elapsed_time:.1f}s " + f"({self.stats['emails_per_second']:.1f} emails/s)" + ) + + return emails, self.stats + + def get_stats(self) -> Dict: + """Get fetcher statistics""" + return self.stats.copy() + + +# ============================================================================ +# TESTING +# ============================================================================ + + +async def test_fetcher_mock(): + """Test fetcher with mock/simulated data (no real API calls)""" + + print("\n" + "="*60) + print("Phase 4 - GmailFetcher Test (Mock Mode)") + print("="*60) + + # Simulate OAuth token + mock_token = "mock_oauth_token_12345" + + print(f"\nāœ… TEST 1: Fetcher Initialization") + fetcher = GmailFetcher( + oauth_token=mock_token, + max_requests_per_second=40, + timeout=30 + ) + + print(f" OAuth token: {mock_token[:20]}...") + print(f" Rate limit: {fetcher.max_requests_per_second} req/s") + print(f" Timeout: {fetcher.timeout}s") + print(f" āœ… PASSED\n") + + print(f"āœ… TEST 2: Rate Limiting Logic") + # Test that rate limiter allows requests + start = time.time() + for i in range(5): + await fetcher._rate_limit() + elapsed = time.time() - start + + print(f" 5 requests completed in {elapsed:.3f}s") + print(f" Rate limit waits: {fetcher.stats['rate_limit_waits']}") + print(f" āœ… PASSED\n") + + print(f"āœ… TEST 3: Statistics Tracking") + stats = fetcher.get_stats() + print(f" API calls: {stats['api_calls']}") + print(f" Emails fetched: {stats['emails_fetched']}") + print(f" Errors: {stats['errors']}") + print(f" āœ… PASSED\n") + + print("="*60) + print("Phase 4 Mock Tests Complete āœ…") + print("="*60) + print("\nGmailFetcher is ready for integration!") + print("\nNext: Test with real Gmail API using your OAuth token") + + return True + + +if __name__ == "__main__": + print("Testing Phase 4 - Gmail API Fetcher") + success = asyncio.run(test_fetcher_mock()) + + if success: + print("\nšŸŽ‰ Phase 4 mock tests PASSED!") + print("\nTo test with real Gmail API:") + print("1. Get your OAuth token from oauth_session table") + print("2. Run: fetcher = GmailFetcher(your_token)") + print("3. Run: emails = await fetcher.fetch_all_emails(max_emails=10)") + + exit(0 if success else 1) + diff --git a/backend/open_webui/utils/gmail_indexer.py b/backend/open_webui/utils/gmail_indexer.py new file mode 100644 index 0000000000..7f867f466d --- /dev/null +++ b/backend/open_webui/utils/gmail_indexer.py @@ -0,0 +1,760 @@ +""" +Gmail Indexer - Integration Layer + +This module integrates GmailProcessor with existing chat summary infrastructure: +- Uses ContentAwareTextSplitter for intelligent email chunking +- Uses EmbeddingService for vector generation +- Uses PineconeManager for storage +- Matches the exact metadata pattern from chat summaries + +Namespace Strategy: +- Chat summaries: PINECONE_NAMESPACE (e.g., "chat-summary-knowledge") +- Gmail emails: PINECONE_NAMESPACE_GMAIL (e.g., "gmail-inbox") + +This is the bridge between Gmail parsing and your existing RAG system. +""" + +import logging +import time +import asyncio +import re +import hashlib +from datetime import datetime +from typing import Dict, List, Optional + +from open_webui.utils.gmail_processor import GmailProcessor +from open_webui.routers.retrieval import extract_enhanced_metadata +from openai import AsyncOpenAI + +# Set up logger with INFO level for visibility +logger = logging.getLogger(__name__) +logger.setLevel(logging.INFO) + + +class GmailIndexer: + """ + Indexes Gmail emails using existing chat summary infrastructure. + + Reuses: + - ContentAwareTextSplitter (from chat summary filter) + - EmbeddingService (from chat summary filter) + - PineconeManager (from chat summary filter) + - DocumentProcessor quality scoring + + Uses separate namespace: PINECONE_NAMESPACE_GMAIL + """ + + def __init__( + self, + embedding_service, + content_aware_splitter, + document_processor, + gmail_namespace: str = "gmail-inbox", + summarizer_client: AsyncOpenAI = None, + summarizer_model: str = None, + ): + """ + Initialize indexer with existing services. + + Args: + embedding_service: EmbeddingService instance from chat filter + content_aware_splitter: ContentAwareTextSplitter instance + document_processor: DocumentProcessor instance for quality scoring + gmail_namespace: Pinecone namespace for Gmail emails (from PINECONE_NAMESPACE_GMAIL env var) + summarizer_client: OpenAI client for summarization (optional) + summarizer_model: Model for summarization (optional) + """ + self.gmail_processor = GmailProcessor() + self.embeddings = embedding_service + self.splitter = content_aware_splitter + self.doc_processor = document_processor + self.namespace = gmail_namespace + self.summarizer_client = summarizer_client + self.summarizer_model = summarizer_model + + async def process_email_for_indexing( + self, + email_data: dict, + user_id: str, + ) -> Dict: + """ + Process a single Gmail email into Pinecone-ready format. + + This method: + 1. Parses email with GmailProcessor + 2. Chunks with ContentAwareTextSplitter + 3. Generates embeddings with EmbeddingService + 4. Creates upsert data matching chat summary format + + Args: + email_data: Gmail API message response (format='full') + user_id: User ID for isolation + + Returns: + Dict with email_id, chunks count, and upsert_data for Pinecone + """ + + start_time = time.time() + + # Step 1: Parse email + parsed = self.gmail_processor.parse_email(email_data, user_id) + email_id = parsed["email_id"] + document_text = parsed["document_text"] + base_metadata = parsed["metadata"] + + logger.info(f"Processing email {email_id} ({base_metadata['subject'][:50]}...)") + + # Step 2: Extract enhanced metadata using Open WebUI's extract_enhanced_metadata + combined_text = f"{base_metadata.get('subject', '')}\n\n{document_text}" + enhanced_metadata = extract_enhanced_metadata(combined_text, use_llm=False) + + # Step 3: Use Open WebUI's content-aware splitter for chunking + chunks = self.splitter.split_text(document_text) if document_text else [] + + # If no chunks created or text is short, use full text + if not chunks: + chunks = [document_text] + + logger.info(f"Email {email_id} will have {len(chunks)} chunk(s)") + + # Step 4: Generate email summary (1-2 sentences) + email_summary = await self._generate_summary( + base_metadata.get("subject", ""), base_metadata.get("body_original", "") + ) + + # Step 5: Build dual-representation vectors + # Vector 1: Always create a summary vector (for email discovery) + # Vectors 2-N: Content chunks (for long emails only) + + all_chunks = [] + all_embeddings = [] + chunk_types = [] + + # Always add summary as first "chunk" + summary_text = self._build_summary_text( + base_metadata.get("subject", ""), + base_metadata.get("from_name", ""), + base_metadata.get("date", ""), + email_summary, + ) + all_chunks.append(summary_text) + chunk_types.append("summary") + + # Add content chunks if email is long (include subject in each chunk) + # Only add chunks that have meaningful content (not just subject repetition) + if len(chunks) > 1 or len(document_text) > 1500: + for i, chunk in enumerate(chunks): + # Skip empty or very short chunks + if not chunk or len(chunk.strip()) < 50: + continue + + # Skip chunks that are mostly subject repetition + if chunk.strip().lower() == base_metadata.get("subject", "").lower(): + continue + + # Only add subject if chunk doesn't already contain it + chunk_clean = chunk.strip() + subject = base_metadata.get("subject", "").strip() + + if subject.lower() not in chunk_clean.lower(): + # Subject not in chunk - prepend it for context + chunk_with_subject = f"{subject}\n\n{chunk_clean}" + else: + # Subject already in chunk - use as is + chunk_with_subject = chunk_clean + + all_chunks.append(chunk_with_subject) + chunk_types.append("content") + + logger.info( + f"Email {email_id}: 1 summary + {len(chunks) if len(chunks) > 1 else 0} content vectors" + ) + + # Step 6: Generate embeddings for all chunks (summary + content) + all_embeddings = await self.embeddings.embed_batch(all_chunks) + + logger.info(f"Generated {len(all_embeddings)} embeddings for email {email_id}") + + # Step 7: Build upsert data with dual representation and enriched metadata + upsert_data = self._build_upsert_data( + chunks=all_chunks, + embeddings=all_embeddings, + chunk_types=chunk_types, + base_metadata=base_metadata, + enhanced_metadata=enhanced_metadata, + email_summary=email_summary, + ) + + processing_time = time.time() - start_time + + logger.info( + f"Email {email_id} processed in {processing_time:.2f}s: " + f"{len(chunks)} chunks, {len(upsert_data)} vectors" + ) + + return { + "email_id": email_id, + "chunks": len(chunks), + "vectors": len(upsert_data), + "upsert_data": upsert_data, + "processing_time": processing_time, + } + + async def _generate_summary(self, subject: str, body: str) -> str: + """ + Generate 1-2 sentence summary of the email. + + Uses OpenAI/OpenRouter if available, otherwise creates extractive summary. + """ + + # If no summarizer, create extractive summary + if not self.summarizer_client or not self.summarizer_model: + return self._extractive_summary(subject, body) + + try: + # Use AI to generate concise summary + prompt = f"Summarize this email in 1-2 sentences:\n\nSubject: {subject}\n\n{body[:1000]}" + + response = await asyncio.wait_for( + self.summarizer_client.chat.completions.create( + model=self.summarizer_model, + messages=[ + { + "role": "system", + "content": "You are an expert at summarizing emails concisely. Create 1-2 sentence summaries.", + }, + {"role": "user", "content": prompt}, + ], + temperature=0.3, + max_tokens=100, + ), + timeout=10.0, + ) + + summary = response.choices[0].message.content.strip() + logger.debug(f"AI summary generated: {summary}") + return summary + + except Exception as e: + logger.warning(f"AI summarization failed: {e}, using extractive") + return self._extractive_summary(subject, body) + + def _extractive_summary(self, subject: str, body: str) -> str: + """ + Create smart extractive summary (skip greetings, find meaningful content). + + Skips common email greetings/closings/signatures to extract actual content. + """ + if not body: + return subject + + # Split into sentences (better sentence detection) + all_sentences = re.split(r"[.!?]+\s+", body) + all_sentences = [s.strip() for s in all_sentences if s.strip()] + + # Patterns to skip (greetings, closings, signatures, boilerplate) + skip_patterns = [ + # Greetings/closings + r"^(hi|hello|hey|dear)\s*,?\s*$", + r"^thanks?\s*,?\s*$", + r"^thank you\s*,?\s*$", + r"^best\s*(regards?)?\s*,?\s*$", + r"^regards?\s*,?\s*$", + r"^sincerely\s*,?\s*$", + r"^cheers\s*,?\s*$", + r"see you", + r"talk soon", + # Signature indicators + r"^_{5,}", # Underscore lines + r"^-{5,}", # Dash lines + r"chairman|ceo|director|manager|founder", # Titles (signature) + r"@[a-z]+\.(org|com|net)", # Email addresses + r"www\.", # URLs + r"^\d{3}[-.\s]?\d{3}", # Phone numbers + # Boilerplate + r"google llc", + r"confidential", + r"disclaimer", + r"you have received this email", + r"shared.*document", + r"invited you to", + ] + + # Filter out unwanted sentences + meaningful_sentences = [] + for sent in all_sentences: + sent_lower = sent.lower().strip() + + # Skip if too short + if len(sent.split()) < 4: + continue + + # Skip if matches any unwanted pattern + if any( + re.search(pattern, sent_lower, re.IGNORECASE) + for pattern in skip_patterns + ): + continue + + meaningful_sentences.append(sent) + + # Stop after finding 2 good sentences + if len(meaningful_sentences) >= 2: + break + + # If we filtered everything out, try to get SOMETHING useful + if not meaningful_sentences: + # Just use subject as summary + return subject + + # Join the meaningful sentences + summary = ". ".join(meaningful_sentences) + + # Clean for metadata storage (remove newlines, normalize spaces) + summary = summary.replace("\n", " ") + summary = summary.replace("\r", " ") + summary = re.sub(r"\s+", " ", summary) # Normalize multiple spaces + summary = summary.strip() + + # Add period if missing + if summary and not summary.endswith((".", "!", "?")): + summary = summary + "." + + # Limit length + if len(summary) > 200: + summary = summary[:197] + "..." + + return summary if summary else subject + + def _extract_list_values(self, data_list: list) -> str: + """ + Extract values from a list and format for Pinecone storage. + + Args: + data_list: List of strings to join + + Returns: + Comma-separated string of values + """ + if not data_list or not isinstance(data_list, list): + return "" + + # Remove empty values and join + values = [str(v) for v in data_list if v] + return ",".join(values[:10]) # Limit to 10 items + + def _create_content_fingerprint(self, snippet: str) -> str: + """ + Create content fingerprint for duplicate detection. + + Uses Gmail snippet (first ~150 chars) to detect mass emails. + Mass emails sent to 200 people will all have identical snippets. + + Args: + snippet: Gmail API snippet (preview of email content) + + Returns: + SHA256 hash of normalized snippet + """ + if not snippet: + return "empty" + + # Normalize snippet for consistent hashing + normalized = snippet.lower().strip() + normalized = re.sub(r"\s+", " ", normalized) # Normalize whitespace + + # Take first 100 chars (enough to detect duplicates, ignore minor variations) + fingerprint_text = normalized[:100] + + # Create hash + return hashlib.sha256(fingerprint_text.encode()).hexdigest()[:16] + + def _build_summary_text( + self, subject: str, from_name: str, date: str, summary: str + ) -> str: + """ + Build clean summary vector text for better semantic search. + + Format: "Subject - Summary" + All metadata (from, date, etc.) stored in structured fields, not in embedding text. + This makes embeddings more focused and improves search quality. + """ + + # Simple, clean format: just subject and summary + if subject and summary: + return f"{subject} - {summary}" + elif subject: + return subject + elif summary: + return summary + else: + return "Email" # Fallback + + def _build_upsert_data( + self, + chunks: List[str], + embeddings: List[List[float]], + chunk_types: List[str], + base_metadata: Dict, + enhanced_metadata: Dict, + email_summary: str, + ) -> List[Dict]: + """ + Build Pinecone upsert data matching chat summary format. + + This creates the exact same structure as chat summaries but with + type='email' and email-specific metadata fields. + + Args: + chunks: List of text chunks from email + embeddings: List of embedding vectors (one per chunk) + chunk_types: List of chunk types ('summary' or 'content') + base_metadata: Base metadata from GmailProcessor + enhanced_metadata: Enhanced metadata from extract_enhanced_metadata + email_summary: Generated email summary + + Returns: + List of dicts ready for Pinecone upsert + """ + + upsert_data = [] + email_id = base_metadata["email_id"] + user_id = base_metadata["user_id"] + + for i, (chunk_text, chunk_vec, chunk_type) in enumerate( + zip(chunks, embeddings, chunk_types) + ): + + # Calculate quality score + quality_score = self.doc_processor.quick_quality_score(chunk_text) + + # Create unique record ID with type + rec_id = f"email-{email_id}-{chunk_type}-{int(time.time())}-{i}" + + # Build metadata with chunk type + metadata = { + # Core identification + "type": "email", + "email_id": email_id, + "thread_id": base_metadata["thread_id"], + "user_id": user_id, + # Content with chunk type + "chunk_text": self._clean_for_storage(chunk_text), + "chunk_type": chunk_type, # "summary" or "content" + "chunk_index": i, + "total_chunks": len(chunks), + # Email summary (for all chunks) + "email_summary": email_summary, + # Enhanced metadata from Open WebUI's extract_enhanced_metadata + "chunk_summary": enhanced_metadata.get("chunk_summary", ""), + "chunk_title": enhanced_metadata.get("chunk_title", ""), + # Timing (SAME structure as chat) + "timestamp": datetime.utcnow().isoformat() + "Z", + "date_timestamp": base_metadata["date_timestamp"], + # Email-specific fields (structured for better filtering) + "subject": base_metadata["subject"], + "from_name": base_metadata.get("from_name", ""), + "from_email": base_metadata.get("from_email", ""), + "to_name": base_metadata.get("to_name", ""), + "to_email": base_metadata.get("to_email", ""), + "cc": base_metadata.get("cc", ""), + "date": base_metadata["date"], + "labels": ( + ",".join(base_metadata["labels"]) + if isinstance(base_metadata["labels"], list) + else base_metadata["labels"] + ), # Comma-separated string for Pinecone compatibility + "has_attachments": base_metadata["has_attachments"], + "is_reply": base_metadata.get("is_reply", False), + "word_count": base_metadata.get("word_count", 0), + # Quality & metadata (SAME as chat) + "quality_score": quality_score, + "doc_type": "email", # ā† Was "chat_summary" + "source": "gmail", + "vector_dim": len(chunk_vec), + "summary_id": rec_id, # ā† For consistency with chat summaries + # Deduplication + "hash": base_metadata["hash"], + + # Enhanced metadata from Open WebUI's extract_enhanced_metadata + "topics": self._extract_list_values(enhanced_metadata.get("topics", [])), + "keywords": self._extract_list_values(enhanced_metadata.get("keywords", [])), + "potential_questions": self._extract_list_values(enhanced_metadata.get("potential_questions", [])), + # Entity extraction - store as comma-separated values + "entity_people": self._extract_list_values(enhanced_metadata.get("entities_people", [])), + "entity_organizations": self._extract_list_values(enhanced_metadata.get("entities_organizations", [])), + "entity_locations": self._extract_list_values(enhanced_metadata.get("entities_locations", [])), + } + + upsert_data.append( + { + "id": rec_id, + "values": [float(v) if v is not None else 0.0 for v in chunk_vec], + "metadata": metadata, + } + ) + + return upsert_data + + @staticmethod + def _clean_for_storage(text: str) -> str: + """ + Clean text for vector database storage (remove problematic characters). + """ + if not text: + return "" + + # Remove control characters except newlines/tabs + text = re.sub(r"[\x00-\x1F\x7F-\x9F]", "", text) + + # Normalize whitespace + text = re.sub(r"\s+", " ", text).strip() + + # Encode as UTF-8 (remove invalid chars) + text = text.encode("utf-8", "ignore").decode("utf-8") + + # Limit length for storage + max_length = 10000 + if len(text) > max_length: + text = text[:max_length] + "..." + + return text + + async def process_email_batch( + self, + emails: List[dict], + user_id: str, + ) -> Dict: + """ + Process multiple emails in batch with content deduplication. + + Detects and skips mass emails (same content sent to multiple people). + Memory-efficient: Processes and yields results without accumulating all vectors. + + Args: + emails: List of Gmail API message responses + user_id: User ID for isolation + + Returns: + Dict with batch processing statistics and all upsert data + """ + + batch_start = time.time() + all_upsert_data = [] + processed_count = 0 + error_count = 0 + duplicate_count = 0 + seen_content = {} # content_hash -> email_id (track duplicates) + + logger.info(f"Processing batch of {len(emails)} emails for user {user_id}") + + for email_data in emails: + try: + # Generate content fingerprint for deduplication + email_id = email_data.get("id", "unknown") + snippet = email_data.get("snippet", "") + + # Create fingerprint from snippet (first ~150 chars of email) + # Mass emails will have identical snippets + content_fingerprint = self._create_content_fingerprint(snippet) + + # Check if we've seen this content before + if content_fingerprint in seen_content: + duplicate_count += 1 + original_email_id = seen_content[content_fingerprint] + logger.info( + f"ā­ļø Skipping duplicate content: email {email_id} " + f"(same as {original_email_id}) - Mass email detected" + ) + email_data.clear() + continue + + # Track this content + seen_content[content_fingerprint] = email_id + + # Process unique email + result = await self.process_email_for_indexing(email_data, user_id) + all_upsert_data.extend(result["upsert_data"]) + processed_count += 1 + + # Clear the large email_data after processing to free memory + email_data.clear() + + # Yield to event loop every 10 emails to keep interface responsive + if processed_count % 10 == 0: + await asyncio.sleep(0) + + except Exception as e: + email_id = email_data.get("id", "unknown") + logger.error(f"Error processing email {email_id}: {e}") + error_count += 1 + continue + + batch_time = time.time() - batch_start + + logger.info( + f"Batch processing complete: {processed_count} unique emails processed, " + f"{duplicate_count} duplicates skipped, {error_count} errors, " + f"{len(all_upsert_data)} vectors, {batch_time:.2f}s" + ) + + if duplicate_count > 0: + logger.info( + f"šŸ’” Deduplication saved {duplicate_count} embeddings (mass emails detected)" + ) + + return { + "processed": processed_count, + "duplicates_skipped": duplicate_count, + "errors": error_count, + "total_vectors": len(all_upsert_data), + "upsert_data": all_upsert_data, + "processing_time": batch_time, + } + + +# ============================================================================ +# TESTING +# ============================================================================ + + +def create_test_email_data(): + """Create test email data for testing""" + return { + "id": "test_email_001", + "threadId": "test_thread_001", + "labelIds": ["INBOX", "IMPORTANT"], + "snippet": "This is a long test email...", + "internalDate": "1605451800000", + "payload": { + "mimeType": "text/plain", + "headers": [ + {"name": "From", "value": "sender@company.com"}, + {"name": "To", "value": "recipient@company.com"}, + {"name": "Subject", "value": "Long Email Test"}, + {"name": "Date", "value": "Sun, 15 Nov 2020 14:30:00 +0000"}, + ], + "body": { + "data": base64.b64encode( + b"""This is a test email with multiple paragraphs. + +Paragraph one discusses the project timeline and deliverables for Q4. + +Paragraph two covers budget allocation and resource planning. + +Paragraph three talks about team assignments and responsibilities. + +This email should be chunked because it's longer than typical emails. + +Best regards, +Test Sender""" + ).decode() + }, + }, + } + + +async def test_indexer(): + """Test the indexer with mock services""" + print("\n" + "=" * 60) + print("Phase 3 Integration Test - GmailIndexer") + print("=" * 60) + + # Mock services (simplified for testing) + class MockEmbeddingService: + async def embed_batch(self, texts): + # Return mock 1536-dim vectors + return [[0.1] * 1536 for _ in texts] + + class MockSplitter: + def split_text(self, text): + # Simple split by double newlines + chunks = text.split("\n\n") + return [c.strip() for c in chunks if c.strip()] + + class MockDocProcessor: + @staticmethod + def quick_quality_score(text): + # Simple quality scoring + score = 0 + if len(text) > 200: + score += 2 + if "." in text: + score += 2 + return min(score, 5) + + # Create indexer with mock services + indexer = GmailIndexer( + embedding_service=MockEmbeddingService(), + content_aware_splitter=MockSplitter(), + document_processor=MockDocProcessor(), + ) + + # Test with sample email + email_data = create_test_email_data() + + try: + result = await indexer.process_email_for_indexing(email_data, "test_user_456") + + print(f"\nāœ… Email processed successfully!") + print(f" Email ID: {result['email_id']}") + print(f" Chunks created: {result['chunks']}") + print(f" Vectors generated: {result['vectors']}") + print(f" Processing time: {result['processing_time']:.3f}s") + + # Verify upsert data structure + if result["upsert_data"]: + sample = result["upsert_data"][0] + print(f"\nšŸ“Š Sample vector metadata:") + print(f" ID: {sample['id']}") + print(f" Vector dim: {len(sample['values'])}") + print(f" Metadata type: {sample['metadata']['type']}") + print(f" Metadata user_id: {sample['metadata']['user_id']}") + print(f" Metadata subject: {sample['metadata']['subject']}") + print(f" Quality score: {sample['metadata']['quality_score']}") + + # Verify structure matches chat summary pattern + required_fields = [ + "type", + "email_id", + "user_id", + "chunk_text", + "chunk_index", + "total_chunks", + "quality_score", + "doc_type", + "source", + "timestamp", + ] + + missing = [f for f in required_fields if f not in sample["metadata"]] + if missing: + print(f"\nāš ļø Missing fields: {missing}") + return False + else: + print(f"\nāœ… All required metadata fields present!") + + return True + + except Exception as e: + print(f"\nāŒ Test failed: {e}") + import traceback + + traceback.print_exc() + return False + + +if __name__ == "__main__": + import asyncio + import base64 + + print("Testing Phase 3 - Gmail Indexer Integration") + success = asyncio.run(test_indexer()) + + if success: + print("\nšŸŽ‰ Phase 3 integration test PASSED!") + print("\nNext: Test with real EmbeddingService from chat filter") + else: + print("\nāŒ Phase 3 integration test FAILED!") + + exit(0 if success else 1) diff --git a/backend/open_webui/utils/gmail_processor.py b/backend/open_webui/utils/gmail_processor.py new file mode 100644 index 0000000000..4eeec02586 --- /dev/null +++ b/backend/open_webui/utils/gmail_processor.py @@ -0,0 +1,471 @@ +""" +Gmail Email Processor + +This module provides email parsing and metadata extraction from Gmail API responses. +It's designed to work standalone without external dependencies (no Pinecone, no embeddings). + +Purpose: Transform Gmail API JSON into clean, structured data ready for indexing. +""" + +import base64 +import hashlib +import logging +import re +from datetime import datetime +from email.utils import parsedate_to_datetime +from typing import Dict, List, Optional, Tuple +from html import unescape + +from open_webui.utils.email_cleaner import EmailCleaner + +# Set up logger with INFO level for visibility +logger = logging.getLogger(__name__) +logger.setLevel(logging.INFO) + + +class GmailProcessor: + """ + Processes Gmail API responses into structured email data. + + This is a pure data transformation class - no external API calls, + no database operations, just parsing and cleaning. + """ + + def __init__(self): + """Initialize the processor""" + pass + + def parse_email(self, email_data: dict, user_id: str) -> Dict: + """ + Parse a Gmail API message response into structured data. + + Args: + email_data: Full Gmail API message response (format='full') + user_id: User ID for metadata isolation + + Returns: + Dict with parsed email data and metadata + + Example: + >>> processor = GmailProcessor() + >>> result = processor.parse_email(gmail_api_response, "user_123") + >>> print(result["subject"]) + "Q4 Budget Discussion" + """ + + try: + # Extract basic identifiers + email_id = email_data.get("id", "") + thread_id = email_data.get("threadId", "") + + if not email_id: + raise ValueError("Email data missing 'id' field") + + # Extract headers + headers_dict = self._extract_headers(email_data) + + # Extract core email fields + subject = headers_dict.get("Subject", "(No Subject)") + from_addr = headers_dict.get("From", "") + to_addr = headers_dict.get("To", "") + cc_addr = headers_dict.get("Cc", "") + date_str = headers_dict.get("Date", "") + + # Parse date + date_timestamp, date_readable = self._parse_date( + date_str, + email_data.get("internalDate") + ) + + # Extract email body + payload = email_data.get("payload", {}) + body_raw = self._extract_body(payload) + snippet = email_data.get("snippet", "") + + # Clean email body with advanced cleaning + body_full_clean, body_original_only = EmailCleaner.clean_email_body(body_raw) + + # Use original message (no quotes/signatures) for primary content + body_clean = body_original_only if body_original_only else body_full_clean + + # Extract labels + labels = email_data.get("labelIds", []) + + # Check for attachments + has_attachments = self._has_attachments(payload) + + # Parse email addresses + from_name, from_email = EmailCleaner.parse_email_address(from_addr) + to_name, to_email = EmailCleaner.parse_email_address(to_addr) + + # Detect if this is a reply + is_reply = "RE:" in subject.upper() or "Re:" in subject or any(label == "SENT" for label in labels) + + # Create document text (what will be embedded) - cleaner format + document_text = self._create_document_text( + subject=subject, + from_name=from_name, + from_email=from_email, + to_email=to_email, + date_readable=date_readable, + body=body_clean, + snippet=snippet + ) + + # Build metadata (improved structure for better search) + metadata = { + # Core identification + "type": "email", + "email_id": email_id, + "thread_id": thread_id, + "user_id": user_id, + + # Email fields (structured) + "subject": subject, + "from_name": from_name, + "from_email": from_email, + "from_raw": from_addr, # Keep original for reference + "to_name": to_name, + "to_email": to_email, + "to_raw": to_addr, + "cc": cc_addr, + "date": date_readable, + "date_timestamp": date_timestamp, + "labels": labels, # Keep as array, not comma-separated string + + # Content metadata + "has_attachments": has_attachments, + "is_reply": is_reply, + "word_count": len(body_clean.split()), + "body_length": len(body_clean), + "original_body_length": len(body_raw), + + # Cleaned text fields + "body_full_clean": body_full_clean, # Full email with cleaned quotes + "body_original": body_clean, # Just the original message + + # Categorization + "source": "gmail", + "doc_type": "email", + + # Hash for deduplication + "hash": hashlib.sha256( + f"{email_id}{user_id}".encode() + ).hexdigest(), + } + + return { + "email_id": email_id, + "thread_id": thread_id, + "document_text": document_text, + "metadata": metadata, + "raw_body": body_raw, + "cleaned_body": body_clean, + "snippet": snippet, + } + + except Exception as e: + logger.error(f"Error parsing email {email_data.get('id', 'unknown')}: {e}") + raise + + def _extract_headers(self, email_data: dict) -> Dict[str, str]: + """Extract email headers from Gmail API response""" + + payload = email_data.get("payload", {}) + headers_list = payload.get("headers", []) + + # Convert list of {name, value} to dict + headers_dict = {} + for header in headers_list: + name = header.get("name", "") + value = header.get("value", "") + headers_dict[name] = value + + return headers_dict + + def _parse_date( + self, + date_str: str, + internal_date: Optional[str] = None + ) -> Tuple[int, str]: + """ + Parse email date into timestamp and readable format. + + Args: + date_str: Date header from email (e.g., "Mon, 15 Nov 2020 14:30:00 +0000") + internal_date: Gmail internal date (milliseconds since epoch) + + Returns: + Tuple of (unix_timestamp, readable_date_string) + """ + + # Try to parse the Date header + try: + date_obj = parsedate_to_datetime(date_str) + timestamp = int(date_obj.timestamp()) + readable = date_obj.strftime("%Y-%m-%d %H:%M:%S") + return timestamp, readable + except Exception as e: + logger.debug(f"Could not parse date '{date_str}': {e}") + + # Fallback to internal date + if internal_date: + try: + timestamp = int(internal_date) // 1000 # Convert ms to seconds + date_obj = datetime.fromtimestamp(timestamp) + readable = date_obj.strftime("%Y-%m-%d %H:%M:%S") + return timestamp, readable + except Exception as e: + logger.debug(f"Could not parse internal date '{internal_date}': {e}") + + # Final fallback: current time + now = int(datetime.now().timestamp()) + return now, "Unknown Date" + + def _extract_body(self, payload: dict) -> str: + """ + Recursively extract email body from Gmail API payload. + + Handles: + - Plain text emails + - HTML emails (strips tags) + - Multipart emails (multipart/alternative, multipart/mixed) + - Nested MIME structures + + Args: + payload: Gmail API message payload + + Returns: + Extracted email body as plain text + """ + + # Handle multipart emails (most common) + if "parts" in payload: + return self._extract_from_parts(payload["parts"]) + + # Handle single-part emails + elif "body" in payload and "data" in payload["body"]: + mime_type = payload.get("mimeType", "") + data = payload["body"]["data"] + return self._decode_body_data(data, mime_type) + + return "" + + def _extract_from_parts(self, parts: list) -> str: + """ + Extract body from multipart email structure. + + Priority: + 1. text/plain (preferred) + 2. text/html (strip tags) + 3. Nested parts (recurse) + """ + + # First pass: look for text/plain + for part in parts: + mime_type = part.get("mimeType", "") + + if mime_type == "text/plain": + data = part.get("body", {}).get("data", "") + if data: + body = self._decode_body_data(data, mime_type) + if body: + return body + + # Second pass: look for text/html + for part in parts: + mime_type = part.get("mimeType", "") + + if mime_type == "text/html": + data = part.get("body", {}).get("data", "") + if data: + body = self._decode_body_data(data, mime_type) + if body: + return body + + # Third pass: recurse into nested parts + for part in parts: + if "parts" in part: + body = self._extract_from_parts(part["parts"]) + if body: + return body + + return "" + + def _decode_body_data(self, data: str, mime_type: str) -> str: + """ + Decode base64-encoded email body data. + + Args: + data: Base64-encoded body data + mime_type: MIME type (text/plain or text/html) + + Returns: + Decoded and cleaned text + """ + + if not data: + return "" + + try: + # Gmail API uses URL-safe base64 encoding + decoded_bytes = base64.urlsafe_b64decode(data) + decoded_text = decoded_bytes.decode("utf-8", errors="ignore") + + # Clean based on MIME type + if mime_type == "text/html": + decoded_text = self._strip_html_tags(decoded_text) + + return decoded_text.strip() + + except Exception as e: + logger.error(f"Error decoding body data: {e}") + return "" + + def _strip_html_tags(self, html_text: str) -> str: + """ + Strip HTML tags and extract plain text. + + Basic implementation - good enough for email bodies. + """ + + if not html_text: + return "" + + # Decode HTML entities first + text = unescape(html_text) + + # Remove script and style tags with their content + text = re.sub(r']*>.*?', '', text, flags=re.DOTALL | re.IGNORECASE) + text = re.sub(r']*>.*?', '', text, flags=re.DOTALL | re.IGNORECASE) + + # Replace common block elements with newlines + text = re.sub(r'', '\n', text, flags=re.IGNORECASE) + text = re.sub(r'', '\n', text, flags=re.IGNORECASE) + + # Remove all remaining HTML tags + text = re.sub(r'<[^>]+>', '', text) + + # Clean up whitespace + text = re.sub(r'\n\s*\n', '\n\n', text) + text = re.sub(r' +', ' ', text) + + return text.strip() + + def _has_attachments(self, payload: dict) -> bool: + """ + Check if email has attachments. + + Args: + payload: Gmail API message payload + + Returns: + True if email has attachments + """ + + if "parts" in payload: + for part in payload["parts"]: + filename = part.get("filename", "") + if filename: # If part has a filename, it's likely an attachment + return True + + # Check nested parts + if "parts" in part: + if self._has_attachments(part): + return True + + return False + + def _create_document_text( + self, + subject: str, + from_name: str, + from_email: str, + to_email: str, + date_readable: str, + body: str, + snippet: str + ) -> str: + """ + Create clean document text for embedding. + + Simple, clean format without headers (better for semantic search). + """ + + # Use cleaned body if available, otherwise snippet + content = body if body else snippet + + # Build clean document - just subject and content (no "Email Subject:" labels) + if subject and subject != "(No Subject)": + document_text = f"{subject}\n\n{content}" + else: + document_text = content + + return document_text.strip() + + +# ============================================================================ +# HELPER FUNCTIONS FOR TESTING +# ============================================================================ + + +def create_sample_gmail_response() -> dict: + """ + Create a sample Gmail API response for testing. + + This is useful for unit tests and development. + """ + + return { + "id": "msg_18c2a3b4d5e6f7g8", + "threadId": "thread_12345", + "labelIds": ["INBOX", "IMPORTANT"], + "snippet": "This is a test email about Q4 budget planning...", + "internalDate": "1605451800000", + "payload": { + "mimeType": "text/plain", + "headers": [ + {"name": "From", "value": "john@company.com"}, + {"name": "To", "value": "user@company.com"}, + {"name": "Subject", "value": "Q4 Budget Discussion"}, + {"name": "Date", "value": "Sun, 15 Nov 2020 14:30:00 +0000"}, + ], + "body": { + "data": base64.urlsafe_b64encode( + b"Hi team,\n\nI wanted to discuss our Q4 budget allocation.\n\nBest,\nJohn" + ).decode() + } + } + } + + +def test_processor(): + """Quick test function to verify processor works""" + + processor = GmailProcessor() + sample_data = create_sample_gmail_response() + + try: + result = processor.parse_email(sample_data, "test_user_123") + + print("āœ… GmailProcessor Test Results:") + print(f" Email ID: {result['email_id']}") + print(f" Subject: {result['metadata']['subject']}") + print(f" From: {result['metadata']['from']}") + print(f" Date: {result['metadata']['date']}") + print(f" Body length: {result['metadata']['body_length']} chars") + print(f" Has attachments: {result['metadata']['has_attachments']}") + print(f"\n Document text preview:") + print(f" {result['document_text'][:200]}...") + + return True + + except Exception as e: + print(f"āŒ Test failed: {e}") + return False + + +if __name__ == "__main__": + # Run test when executed directly + test_processor() + diff --git a/backend/open_webui/utils/oauth.py b/backend/open_webui/utils/oauth.py index 26706b65f4..f326fb1a49 100644 --- a/backend/open_webui/utils/oauth.py +++ b/backend/open_webui/utils/oauth.py @@ -1492,6 +1492,33 @@ class OAuthManager: log.info( f"Stored OAuth session server-side for user {user.id}, provider {provider}" ) + + # āœØ Trigger automatic Gmail sync if applicable (for Google OAuth) + # Check if this is a new OAuth session (first time logging in with this provider) + is_new_oauth_session = len([s for s in sessions if s.provider == provider]) == 0 + + log.info(f"šŸ”„ Gmail sync check: is_new_oauth_session={is_new_oauth_session}, provider={provider}") + + # Always check Gmail sync for Google OAuth (not just new sessions) + # The trigger function will check admin settings and user preferences + if provider == "google": + log.info(f"šŸš€ Checking Gmail sync eligibility for user {user.id}") + try: + from open_webui.utils.gmail_auto_sync import trigger_gmail_sync_if_needed + + await trigger_gmail_sync_if_needed( + request=request, + user_id=user.id, + provider=provider, + token=token, + is_new_user=is_new_oauth_session, + ) + except Exception as e: + log.error(f"āŒ Gmail auto-sync trigger failed for user {user.id}: {e}") + # Don't fail OAuth callback if Gmail sync fails + else: + log.info(f"ā­ļø Skipping Gmail sync (provider={provider}, not Google)") + except Exception as e: log.error(f"Failed to store OAuth session server-side: {e}") diff --git a/backend/requirements.txt b/backend/requirements.txt index 842ad3ac72..5f7234ef53 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -28,6 +28,7 @@ peewee-migrate==1.12.2 pycrdt==0.12.25 redis +rq APScheduler==3.10.4 RestrictedPython==8.0