apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-06-28 21:49:40 +02:00
Code Issues Projects Releases Packages Wiki Activity

chore: ran frontend and backend linting

Browse source
This commit is contained in:
Anish Sarkar 2026-02-01 22:54:25 +05:30
parent ff4a574248
commit 085653d3e3
32 changed files with 288 additions and 210 deletions
Download patch file Download diff file Expand all files Collapse all files

8
surfsense_backend/app/connectors/composio_gmail_connector.py
View file

@ -10,10 +10,6 @@ from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from typing import Any
# Heartbeat configuration
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
HEARTBEAT_INTERVAL_SECONDS = 30
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload
@ -32,6 +28,10 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Heartbeat configuration
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
HEARTBEAT_INTERVAL_SECONDS = 30
logger = logging.getLogger(__name__)

8
surfsense_backend/app/connectors/composio_google_calendar_connector.py
View file

@ -10,10 +10,6 @@ from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from typing import Any
# Heartbeat configuration
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
HEARTBEAT_INTERVAL_SECONDS = 30
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy.orm import selectinload
@ -35,6 +31,10 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Heartbeat configuration
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
HEARTBEAT_INTERVAL_SECONDS = 30
logger = logging.getLogger(__name__)

8
surfsense_backend/app/connectors/composio_google_drive_connector.py
View file

@ -15,10 +15,6 @@ from datetime import UTC, datetime
from pathlib import Path
from typing import Any
# Heartbeat configuration
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
HEARTBEAT_INTERVAL_SECONDS = 30
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm.attributes import flag_modified
@ -35,6 +31,10 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Heartbeat configuration
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
HEARTBEAT_INTERVAL_SECONDS = 30
logger = logging.getLogger(__name__)

6
surfsense_backend/app/connectors/discord_connector.py
View file

@ -61,7 +61,9 @@ class DiscordConnector(commands.Bot):
self.token = None
self._bot_task = None # Holds the async bot task
self._is_running = False # Flag to track if the bot is running
self._start_called_event = asyncio.Event() # Event to signal when start() is called
self._start_called_event = (
asyncio.Event()
) # Event to signal when start() is called
# Event to confirm bot is ready
@self.event
@ -293,7 +295,7 @@ class DiscordConnector(commands.Bot):
logger.error("start_bot() did not call start() within 30 seconds")
raise RuntimeError(
"Discord client failed to initialize - start() was never called"
)
) from None
try:
await asyncio.wait_for(self.wait_until_ready(), timeout=60.0)

8
surfsense_backend/app/connectors/google_calendar_connector.py
View file

@ -252,12 +252,16 @@ class GoogleCalendarConnector:
if dt_start.tzinfo is None:
dt_start = dt_start.replace(hour=0, minute=0, second=0, tzinfo=pytz.UTC)
else:
dt_start = dt_start.astimezone(pytz.UTC).replace(hour=0, minute=0, second=0)
dt_start = dt_start.astimezone(pytz.UTC).replace(
hour=0, minute=0, second=0
)
if dt_end.tzinfo is None:
dt_end = dt_end.replace(hour=23, minute=59, second=59, tzinfo=pytz.UTC)
else:
dt_end = dt_end.astimezone(pytz.UTC).replace(hour=23, minute=59, second=59)
dt_end = dt_end.astimezone(pytz.UTC).replace(
hour=23, minute=59, second=59
)
if dt_start >= dt_end:
return [], (

99
surfsense_backend/app/routes/discord_add_connector_route.py
View file

@ -46,6 +46,11 @@ SCOPES = [
"guilds.members.read", # Read member information
]
# Discord permission bits
VIEW_CHANNEL = 1 << 10 # 1024
READ_MESSAGE_HISTORY = 1 << 16 # 65536
ADMINISTRATOR = 1 << 3 # 8
# Initialize security utilities
_state_manager = None
_token_encryption = None
@ -542,25 +547,25 @@ def _compute_channel_permissions(
) -> int:
"""
Compute effective permissions for a channel based on role permissions and overwrites.
Discord permission computation follows this order (per official docs):
1. Start with base permissions from roles
2. Apply @everyone role overwrites (deny, then allow)
3. Apply role-specific overwrites (deny, then allow)
4. Apply member-specific overwrites (deny, then allow)
Args:
base_permissions: Combined permissions from all bot roles
bot_role_ids: Set of role IDs the bot has
bot_user_id: The bot's user ID for member-specific overwrites
channel_overwrites: List of permission overwrites for the channel
guild_id: Guild ID (same as @everyone role ID)
Returns:
Computed permission integer
"""
permissions = base_permissions
# Permission overwrites are applied in order: @everyone, roles, member
everyone_allow = 0
everyone_deny = 0
@ -568,13 +573,13 @@ def _compute_channel_permissions(
role_deny = 0
member_allow = 0
member_deny = 0
for overwrite in channel_overwrites:
overwrite_id = overwrite.get("id")
overwrite_type = overwrite.get("type") # 0 = role, 1 = member
allow = int(overwrite.get("allow", 0))
deny = int(overwrite.get("deny", 0))
if overwrite_type == 0: # Role overwrite
if overwrite_id == guild_id: # @everyone role
everyone_allow = allow
@ -582,11 +587,11 @@ def _compute_channel_permissions(
elif overwrite_id in bot_role_ids:
role_allow |= allow
role_deny |= deny
elif overwrite_type == 1: # Member overwrite
if bot_user_id and overwrite_id == bot_user_id:
member_allow = allow
member_deny = deny
elif overwrite_type == 1 and bot_user_id and overwrite_id == bot_user_id:
# Member-specific overwrite for the bot
member_allow = allow
member_deny = deny
# Apply in order per Discord docs:
# 1. @everyone deny, then allow
permissions &= ~everyone_deny
@ -597,7 +602,7 @@ def _compute_channel_permissions(
# 3. Member deny, then allow (applied LAST, highest priority)
permissions &= ~member_deny
permissions |= member_allow
return permissions
@ -609,7 +614,7 @@ async def get_discord_channels(
):
"""
Get list of Discord text channels for a connector with permission info.
Uses Discord's HTTP REST API directly instead of WebSocket bot connection.
Computes effective permissions to determine if bot can read message history.
@ -623,18 +628,14 @@ async def get_discord_channels(
"""
from sqlalchemy import select
# Discord permission bits
VIEW_CHANNEL = 1 << 10 # 1024
READ_MESSAGE_HISTORY = 1 << 16 # 65536
ADMINISTRATOR = 1 << 3 # 8
try:
# Get connector and verify ownership
result = await session.execute(
select(SearchSourceConnector).where(
SearchSourceConnector.id == connector_id,
SearchSourceConnector.user_id == user.id,
SearchSourceConnector.connector_type == SearchSourceConnectorType.DISCORD_CONNECTOR,
SearchSourceConnector.connector_type
== SearchSourceConnectorType.DISCORD_CONNECTOR,
)
)
connector = result.scalar_one_or_none()
@ -675,7 +676,7 @@ async def get_discord_channels(
)
headers = {"Authorization": f"Bot {bot_token}"}
async with httpx.AsyncClient() as client:
# Fetch bot's user info to get bot user ID
bot_user_response = await client.get(
@ -683,55 +684,61 @@ async def get_discord_channels(
headers=headers,
timeout=30.0,
)
if bot_user_response.status_code != 200:
logger.warning(f"Failed to fetch bot user info: {bot_user_response.text}")
logger.warning(
f"Failed to fetch bot user info: {bot_user_response.text}"
)
bot_user_id = None
else:
bot_user_id = bot_user_response.json().get("id")
# Fetch guild info to get roles
guild_response = await client.get(
f"https://discord.com/api/v10/guilds/{guild_id}",
headers=headers,
timeout=30.0,
)
if guild_response.status_code != 200:
raise HTTPException(
status_code=guild_response.status_code,
detail="Failed to fetch guild information",
)
guild_data = guild_response.json()
guild_roles = {role["id"]: role for role in guild_data.get("roles", [])}
# Fetch bot's member info to get its roles
bot_member_response = await client.get(
f"https://discord.com/api/v10/guilds/{guild_id}/members/{bot_user_id}",
headers=headers,
timeout=30.0,
)
if bot_member_response.status_code != 200:
logger.warning(f"Failed to fetch bot member info: {bot_member_response.text}")
logger.warning(
f"Failed to fetch bot member info: {bot_member_response.text}"
)
bot_role_ids = {guild_id} # At minimum, bot has @everyone role
base_permissions = int(guild_roles.get(guild_id, {}).get("permissions", 0))
base_permissions = int(
guild_roles.get(guild_id, {}).get("permissions", 0)
)
else:
bot_member_data = bot_member_response.json()
bot_role_ids = set(bot_member_data.get("roles", []))
bot_role_ids.add(guild_id) # @everyone role is always included
# Compute base permissions from all bot roles
base_permissions = 0
for role_id in bot_role_ids:
if role_id in guild_roles:
role_perms = int(guild_roles[role_id].get("permissions", 0))
base_permissions |= role_perms
# Check if bot has administrator permission (bypasses all checks)
is_admin = (base_permissions & ADMINISTRATOR) == ADMINISTRATOR
# Fetch channels
channels_response = await client.get(
f"https://discord.com/api/v10/guilds/{guild_id}/channels",
@ -767,7 +774,7 @@ async def get_discord_channels(
# 0 = GUILD_TEXT, 2 = GUILD_VOICE, 4 = GUILD_CATEGORY, 5 = GUILD_ANNOUNCEMENT
# We want text channels (type 0) and announcement channels (type 5)
text_channel_types = {0, 5}
text_channels = []
for ch in channels_data:
if ch.get("type") in text_channel_types:
@ -784,20 +791,24 @@ async def get_discord_channels(
channel_overwrites,
guild_id,
)
# Bot can index if it has both VIEW_CHANNEL and READ_MESSAGE_HISTORY
has_view = (effective_perms & VIEW_CHANNEL) == VIEW_CHANNEL
has_read_history = (effective_perms & READ_MESSAGE_HISTORY) == READ_MESSAGE_HISTORY
has_read_history = (
effective_perms & READ_MESSAGE_HISTORY
) == READ_MESSAGE_HISTORY
can_index = has_view and has_read_history
text_channels.append({
"id": ch["id"],
"name": ch["name"],
"type": "text" if ch["type"] == 0 else "announcement",
"position": ch.get("position", 0),
"category_id": ch.get("parent_id"),
"can_index": can_index,
})
text_channels.append(
{
"id": ch["id"],
"name": ch["name"],
"type": "text" if ch["type"] == 0 else "announcement",
"position": ch.get("position", 0),
"category_id": ch.get("parent_id"),
"can_index": can_index,
}
)
# Sort by position
text_channels.sort(key=lambda x: x["position"])

19
surfsense_backend/app/routes/search_source_connectors_routes.py
View file

@ -1168,9 +1168,10 @@ async def _run_indexing_with_notifications(
supports_retry_callback: Whether the indexing function supports on_retry_callback
supports_heartbeat_callback: Whether the indexing function supports on_heartbeat_callback
"""
from celery.exceptions import SoftTimeLimitExceeded
from uuid import UUID
from celery.exceptions import SoftTimeLimitExceeded
notification = None
# Track indexed count for retry notifications and heartbeat
current_indexed_count = 0
@ -1241,11 +1242,13 @@ async def _run_indexing_with_notifications(
if notification:
try:
await session.refresh(notification)
await NotificationService.connector_indexing.notify_indexing_progress(
session=session,
notification=notification,
indexed_count=indexed_count,
stage="processing",
await (
NotificationService.connector_indexing.notify_indexing_progress(
session=session,
notification=notification,
indexed_count=indexed_count,
stage="processing",
)
)
await session.commit()
except Exception as e:
@ -1447,7 +1450,9 @@ async def _run_indexing_with_notifications(
)
await session.commit()
except Exception as notif_error:
logger.error(f"Failed to update notification on soft timeout: {notif_error!s}")
logger.error(
f"Failed to update notification on soft timeout: {notif_error!s}"
)
# Re-raise so Celery knows the task was terminated
raise

3
surfsense_backend/app/routes/slack_add_connector_route.py
View file

@ -547,7 +547,8 @@ async def get_slack_channels(
select(SearchSourceConnector).where(
SearchSourceConnector.id == connector_id,
SearchSourceConnector.user_id == user.id,
SearchSourceConnector.connector_type == SearchSourceConnectorType.SLACK_CONNECTOR,
SearchSourceConnector.connector_type
== SearchSourceConnectorType.SLACK_CONNECTOR,
)
)
connector = result.scalar_one_or_none()

27
surfsense_backend/app/tasks/celery_tasks/stale_notification_cleanup_task.py
View file

@ -9,8 +9,7 @@ frontend from showing a perpetual "syncing" state.
import logging
from datetime import UTC, datetime, timedelta
from sqlalchemy import and_, update
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy import and_
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
from sqlalchemy.future import select
from sqlalchemy.orm.attributes import flag_modified
@ -42,12 +41,12 @@ def get_celery_session_maker():
def cleanup_stale_indexing_notifications_task():
"""
Check for stale connector indexing notifications and mark them as failed.
This task finds notifications that:
- Have type = 'connector_indexing'
- Have metadata.status = 'in_progress'
- Have updated_at older than STALE_NOTIFICATION_TIMEOUT_MINUTES
And marks them as failed with an appropriate error message.
"""
import asyncio
@ -78,7 +77,8 @@ async def _cleanup_stale_notifications():
select(Notification).filter(
and_(
Notification.type == "connector_indexing",
Notification.notification_metadata["status"].astext == "in_progress",
Notification.notification_metadata["status"].astext
== "in_progress",
Notification.updated_at < cutoff_time,
)
)
@ -98,22 +98,28 @@ async def _cleanup_stale_notifications():
for notification in stale_notifications:
try:
# Get current indexed count from metadata if available
indexed_count = notification.notification_metadata.get("indexed_count", 0)
connector_name = notification.notification_metadata.get("connector_name", "Unknown")
indexed_count = notification.notification_metadata.get(
"indexed_count", 0
)
connector_name = notification.notification_metadata.get(
"connector_name", "Unknown"
)
# Calculate how long it's been stale
stale_duration = datetime.now(UTC) - notification.updated_at
stale_minutes = int(stale_duration.total_seconds() / 60)
# Update notification metadata
notification.notification_metadata["status"] = "failed"
notification.notification_metadata["completed_at"] = datetime.now(UTC).isoformat()
notification.notification_metadata["completed_at"] = datetime.now(
UTC
).isoformat()
notification.notification_metadata["error_message"] = (
f"Indexing task appears to have crashed or timed out. "
f"No activity detected for {stale_minutes} minutes. "
f"Please try syncing again."
)
# Flag the JSONB column as modified for SQLAlchemy to detect the change
flag_modified(notification, "notification_metadata")
@ -138,4 +144,3 @@ async def _cleanup_stale_notifications():
except Exception as e:
logger.error(f"Error cleaning up stale notifications: {e!s}", exc_info=True)
await session.rollback()

6
surfsense_backend/app/tasks/composio_indexer.py
View file

@ -12,9 +12,6 @@ import logging
from collections.abc import Awaitable, Callable
from importlib import import_module
# Type alias for heartbeat callback function
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
@ -26,6 +23,9 @@ from app.db import (
from app.services.composio_service import INDEXABLE_TOOLKITS, TOOLKIT_TO_INDEXER
from app.services.task_logging_service import TaskLoggingService
# Type alias for heartbeat callback function
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Set up logging
logger = logging.getLogger(__name__)

23
surfsense_backend/app/tasks/connector_indexers/airtable_indexer.py
View file

@ -20,12 +20,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
calculate_date_range,
check_document_by_unique_identifier,
@ -36,6 +30,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_airtable_records(
session: AsyncSession,
@ -145,7 +144,11 @@ async def index_airtable_records(
# Process each base
for base in bases:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time)
>= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(total_documents_indexed)
last_heartbeat_time = time.time()
base_id = base.get("id")
@ -224,7 +227,11 @@ async def index_airtable_records(
# Process each record
for record in records:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time)
>= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(total_documents_indexed)
last_heartbeat_time = time.time()

16
surfsense_backend/app/tasks/connector_indexers/bookstack_indexer.py
View file

@ -21,12 +21,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
calculate_date_range,
check_document_by_unique_identifier,
@ -37,6 +31,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_bookstack_pages(
session: AsyncSession,
@ -194,7 +193,10 @@ async def index_bookstack_pages(
for page in pages:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
try:

17
surfsense_backend/app/tasks/connector_indexers/clickup_indexer.py
View file

@ -22,12 +22,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
check_document_by_unique_identifier,
check_duplicate_document_by_hash,
@ -37,6 +31,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_clickup_tasks(
session: AsyncSession,
@ -184,7 +183,11 @@ async def index_clickup_tasks(
for task in tasks:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time)
>= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()

16
surfsense_backend/app/tasks/connector_indexers/confluence_indexer.py
View file

@ -22,12 +22,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
calculate_date_range,
check_document_by_unique_identifier,
@ -38,6 +32,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_confluence_pages(
session: AsyncSession,
@ -190,7 +189,10 @@ async def index_confluence_pages(
for page in pages:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
try:

6
surfsense_backend/app/tasks/connector_indexers/discord_indexer.py
View file

@ -305,7 +305,11 @@ async def index_discord_messages(
try:
for guild in guilds:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time)
>= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
guild_id = guild["id"]

18
surfsense_backend/app/tasks/connector_indexers/elasticsearch_indexer.py
View file

@ -21,18 +21,18 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
check_document_by_unique_identifier,
check_duplicate_document_by_hash,
get_current_timestamp,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
logger = logging.getLogger(__name__)
@ -186,7 +186,11 @@ async def index_elasticsearch_documents(
fields=config.get("ELASTICSEARCH_FIELDS"),
):
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time)
>= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_processed)
last_heartbeat_time = time.time()

17
surfsense_backend/app/tasks/connector_indexers/github_indexer.py
View file

@ -24,12 +24,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
check_document_by_unique_identifier,
check_duplicate_document_by_hash,
@ -38,6 +32,12 @@ from .base import (
logger,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
# Maximum tokens for a single digest before splitting
# Most LLMs can handle 128k+ tokens now, but we'll be conservative
MAX_DIGEST_CHARS = 500_000 # ~125k tokens
@ -184,7 +184,10 @@ async def index_github_repos(
for repo_full_name in repo_full_names_to_index:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
if not repo_full_name or not isinstance(repo_full_name, str):

18
surfsense_backend/app/tasks/connector_indexers/google_calendar_indexer.py
View file

@ -6,8 +6,6 @@ import time
from collections.abc import Awaitable, Callable
from datetime import datetime, timedelta
import pytz
from dateutil.parser import isoparse
from google.oauth2.credentials import Credentials
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession
@ -23,12 +21,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
check_document_by_unique_identifier,
check_duplicate_document_by_hash,
@ -38,6 +30,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_google_calendar_events(
session: AsyncSession,
@ -296,7 +293,10 @@ async def index_google_calendar_events(
for event in events:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
try:

10
surfsense_backend/app/tasks/connector_indexers/google_drive_indexer.py
View file

@ -420,7 +420,10 @@ async def _index_full_scan(
while folders_to_process and files_processed < max_files:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
current_folder_id, current_folder_name = folders_to_process.pop(0)
@ -541,7 +544,10 @@ async def _index_with_delta_sync(
for change in changes:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
if files_processed >= max_files:

16
surfsense_backend/app/tasks/connector_indexers/google_gmail_indexer.py
View file

@ -25,12 +25,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
calculate_date_range,
check_document_by_unique_identifier,
@ -41,6 +35,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_google_gmail_messages(
session: AsyncSession,
@ -228,7 +227,10 @@ async def index_google_gmail_messages(
for message in messages:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
try:

16
surfsense_backend/app/tasks/connector_indexers/jira_indexer.py
View file

@ -22,12 +22,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
calculate_date_range,
check_document_by_unique_identifier,
@ -38,6 +32,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_jira_issues(
session: AsyncSession,
@ -184,7 +183,10 @@ async def index_jira_issues(
for issue in issues:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
try:

16
surfsense_backend/app/tasks/connector_indexers/linear_indexer.py
View file

@ -21,12 +21,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
calculate_date_range,
check_document_by_unique_identifier,
@ -37,6 +31,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_linear_issues(
session: AsyncSession,
@ -210,7 +209,10 @@ async def index_linear_issues(
# Process each issue
for issue in issues:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()

16
surfsense_backend/app/tasks/connector_indexers/luma_indexer.py
View file

@ -21,12 +21,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
check_document_by_unique_identifier,
check_duplicate_document_by_hash,
@ -36,6 +30,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_luma_events(
session: AsyncSession,
@ -236,7 +235,10 @@ async def index_luma_events(
for event in events:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
try:

5
surfsense_backend/app/tasks/connector_indexers/notion_indexer.py
View file

@ -234,7 +234,10 @@ async def index_notion_pages(
# Process each page
for page in pages:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()

16
surfsense_backend/app/tasks/connector_indexers/obsidian_indexer.py
View file

@ -27,12 +27,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
build_document_metadata_string,
check_document_by_unique_identifier,
@ -43,6 +37,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
def parse_frontmatter(content: str) -> tuple[dict | None, str]:
"""
@ -320,7 +319,10 @@ async def index_obsidian_vault(
for file_info in files:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(indexed_count)
last_heartbeat_time = time.time()
try:

16
surfsense_backend/app/tasks/connector_indexers/slack_indexer.py
View file

@ -20,12 +20,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
build_document_metadata_markdown,
calculate_date_range,
@ -37,6 +31,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_slack_messages(
session: AsyncSession,
@ -187,7 +186,10 @@ async def index_slack_messages(
# Process each channel
for channel_obj in channels:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
channel_id = channel_obj["id"]

16
surfsense_backend/app/tasks/connector_indexers/teams_indexer.py
View file

@ -19,12 +19,6 @@ from app.utils.document_converters import (
generate_unique_identifier_hash,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
build_document_metadata_markdown,
calculate_date_range,
@ -36,6 +30,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds - update notification every 30 seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_teams_messages(
session: AsyncSession,
@ -200,7 +199,10 @@ async def index_teams_messages(
# Process each team
for team in teams:
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()

16
surfsense_backend/app/tasks/connector_indexers/webcrawler_indexer.py
View file

@ -22,12 +22,6 @@ from app.utils.document_converters import (
)
from app.utils.webcrawler_utils import parse_webcrawler_urls
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
from .base import (
check_document_by_unique_identifier,
check_duplicate_document_by_hash,
@ -37,6 +31,11 @@ from .base import (
update_connector_last_indexed,
)
# Type hint for heartbeat callback
HeartbeatCallbackType = Callable[[int], Awaitable[None]]
# Heartbeat interval in seconds
HEARTBEAT_INTERVAL_SECONDS = 30
async def index_crawled_urls(
session: AsyncSession,
@ -155,7 +154,10 @@ async def index_crawled_urls(
for idx, url in enumerate(urls, 1):
# Check if it's time for a heartbeat update
if on_heartbeat_callback and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS:
if (
on_heartbeat_callback
and (time.time() - last_heartbeat_time) >= HEARTBEAT_INTERVAL_SECONDS
):
await on_heartbeat_callback(documents_indexed)
last_heartbeat_time = time.time()
try: