mirror of
https://github.com/open-webui/open-webui.git
synced 2026-08-29 08:07:34 -05:00
Add Gmail sync core files from feat/gmail-sync branch
Cherry-picked essential Gmail sync functionality: Core Files (7): - utils/gmail_fetcher.py - Fetches emails from Gmail API - utils/gmail_processor.py - Parses Gmail API responses - utils/email_cleaner.py - Email-specific text cleaning - utils/gmail_indexer.py - Creates vector embeddings using extract_enhanced_metadata - utils/gmail_auto_sync.py - Orchestrates sync process - routers/gmail.py - Gmail sync API endpoints - models/gmail_sync.py - Gmail sync status tracking Integration Files (3): - main.py - Register Gmail router and background tasks - oauth.py - Trigger Gmail sync after Google OAuth - config.py - Gmail sync configuration settings Database (2): - migrations/33cc3721a72_add_gmail_sync_status_table.py - migrations/0b80d222da03_merge_gmail_sync_and_main_branch_.py Dependencies: - requirements.txt - Updated unstructured[all-docs]==0.18.15 for full document support
This commit is contained in:
PVBLIC Foundation
+165
-39
@@ -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 <source> tag includes an explicit id attribute** (e.g., <source id="1">).
|
||||
|
||||
### 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 <source> tag includes an id attribute.**
|
||||
- Do not cite if the <source> 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 <source> tag includes an explicit id attribute.
|
||||
- Place citations immediately after the relevant claim or statement.
|
||||
- Do not cite if the <source> 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 (<source>, <context>, 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 <source> tag with id attribute is present in the context.
|
||||
|
||||
### Context:
|
||||
<context>
|
||||
{{CONTEXT}}
|
||||
</context>
|
||||
|
||||
### User Query:
|
||||
<user_query>
|
||||
{{QUERY}}
|
||||
</user_query>
|
||||
|
||||
### Your Response:
|
||||
Provide your response here, following all guidelines above."""
|
||||
"""
|
||||
|
||||
RAG_TEMPLATE = PersistentConfig(
|
||||
"RAG_TEMPLATE",
|
||||
|
||||
@@ -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"])
|
||||
|
||||
+28
@@ -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
|
||||
|
||||
@@ -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')
|
||||
@@ -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()
|
||||
@@ -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)}"
|
||||
)
|
||||
|
||||
@@ -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@example.com>" → ("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 <email>" 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()
|
||||
File diff suppressed because it is too large
Load Diff
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
@@ -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'<script[^>]*>.*?</script>', '', text, flags=re.DOTALL | re.IGNORECASE)
|
||||
text = re.sub(r'<style[^>]*>.*?</style>', '', text, flags=re.DOTALL | re.IGNORECASE)
|
||||
|
||||
# Replace common block elements with newlines
|
||||
text = re.sub(r'</(p|div|h[1-6]|li|tr)>', '\n', text, flags=re.IGNORECASE)
|
||||
text = re.sub(r'<br\s*/?>', '\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()
|
||||
|
||||
@@ -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}")
|
||||
|
||||
|
||||
@@ -28,6 +28,7 @@ peewee-migrate==1.12.2
|
||||
|
||||
pycrdt==0.12.25
|
||||
redis
|
||||
rq
|
||||
|
||||
APScheduler==3.10.4
|
||||
RestrictedPython==8.0
|
||||
|
||||
Reference in New Issue
Block a user