diff --git a/features.py b/features.py index 9846634..35999e4 100644 --- a/features.py +++ b/features.py @@ -38,6 +38,7 @@ REACTION_EMOJI_RE = re.compile(r"^.{1,16}$") HEX_COLOR_RE = re.compile(r"^#[0-9a-fA-F]{6}$") PAGE_LINK_RE = re.compile(r"\]\((?:/teams/[^/]+/pages/)?([a-z0-9-]{1,48})\)", re.IGNORECASE) + def valid_hex_color(value: object) -> bool: return bool(isinstance(value, str) and HEX_COLOR_RE.fullmatch(value.strip())) @@ -111,7 +112,27 @@ def detect_website_brand(website: str) -> dict: return {"host": host, "brand": brand, "label": label} return {"host": host, "brand": "web", "label": "website"} - + + +def sqlite_table_exists(get_db, table_name: str) -> bool: + with get_db() as db: + row = db.execute( + """ + SELECT 1 + FROM sqlite_master + WHERE type = 'table' + AND name = ? + """, + (table_name,), + ).fetchone() + return bool(row) + + +def sqlite_column_exists(get_db, table_name: str, column_name: str) -> bool: + with get_db() as db: + rows = db.execute(f"PRAGMA table_info({table_name})").fetchall() + return any(row["name"] == column_name for row in rows) + def attachment_kind_from_name_and_mime( original_name: Optional[str], @@ -130,19 +151,10 @@ def attachment_kind_from_name_and_mime( def load_page_children(get_db, page_id: int) -> list[dict]: - with get_db() as db: - has_parent_page = bool( - db.execute( - """ - SELECT 1 - FROM pragma_table_info('pages') - WHERE name = 'parent_page_id' - """ - ).fetchone() - ) - if not has_parent_page: - return [] + if not sqlite_column_exists(get_db, "pages", "parent_page_id"): + return [] + with get_db() as db: rows = db.execute( """ SELECT @@ -184,20 +196,10 @@ def load_page_backlinks(get_db, page_id: int, page_slug: str, team_id: int) -> l def load_page_comments(get_db, page_id: int) -> list[dict]: - with get_db() as db: - table_exists = bool( - db.execute( - """ - SELECT 1 - FROM sqlite_master - WHERE type = 'table' - AND name = 'page_comments' - """ - ).fetchone() - ) - if not table_exists: - return [] + if not sqlite_table_exists(get_db, "page_comments"): + return [] + with get_db() as db: rows = db.execute( """ SELECT @@ -238,20 +240,10 @@ def load_page_comments(get_db, page_id: int) -> list[dict]: def next_page_version_number(get_db, page_id: int) -> int: - with get_db() as db: - table_exists = bool( - db.execute( - """ - SELECT 1 - FROM sqlite_master - WHERE type = 'table' - AND name = 'page_versions' - """ - ).fetchone() - ) - if not table_exists: - return 1 + if not sqlite_table_exists(get_db, "page_versions"): + return 1 + with get_db() as db: row = db.execute( """ SELECT COALESCE(MAX(version_number), 0) AS value @@ -275,19 +267,8 @@ def snapshot_page_version( created_by_user_id: int, created_at: str, ) -> int: - with get_db() as db: - table_exists = bool( - db.execute( - """ - SELECT 1 - FROM sqlite_master - WHERE type = 'table' - AND name = 'page_versions' - """ - ).fetchone() - ) - if not table_exists: - return 0 + if not sqlite_table_exists(get_db, "page_versions"): + return 0 version_number = next_page_version_number(get_db, page_id) with get_db() as db: @@ -325,20 +306,10 @@ def snapshot_page_version( def load_page_versions(get_db, page_id: int) -> list[dict]: - with get_db() as db: - table_exists = bool( - db.execute( - """ - SELECT 1 - FROM sqlite_master - WHERE type = 'table' - AND name = 'page_versions' - """ - ).fetchone() - ) - if not table_exists: - return [] + if not sqlite_table_exists(get_db, "page_versions"): + return [] + with get_db() as db: rows = db.execute( """ SELECT @@ -365,20 +336,10 @@ def load_page_versions(get_db, page_id: int) -> list[dict]: def load_page_version_detail(get_db, version_id: int): - with get_db() as db: - table_exists = bool( - db.execute( - """ - SELECT 1 - FROM sqlite_master - WHERE type = 'table' - AND name = 'page_versions' - """ - ).fetchone() - ) - if not table_exists: - return None + if not sqlite_table_exists(get_db, "page_versions"): + return None + with get_db() as db: row = db.execute( """ SELECT @@ -423,8 +384,6 @@ def user_can_see_last_active(settings_row, viewer: Optional[dict], target_user_i return False - - def page_payload(app_module, page, user: Optional[dict]) -> dict: can_read = bool( page["visibility"] == "public" @@ -991,19 +950,19 @@ def profile_payload( explicit_name_color = settings_row["name_color"] if settings_row else None resolved = resolve_name_color(user_row["role"], explicit_name_color) - profile_visibility = settings_row["profile_visibility"] if settings_row else "public" + profile_visibility = settings_row["profile_visibility"] if settings_row and "profile_visibility" in settings_row.keys() else "public" if profile_visibility not in PROFILE_VISIBILITIES: profile_visibility = "public" - last_active_visibility = settings_row["last_active_visibility"] if settings_row else "everyone" + last_active_visibility = settings_row["last_active_visibility"] if settings_row and "last_active_visibility" in settings_row.keys() else "everyone" if last_active_visibility not in LAST_ACTIVE_VISIBILITIES: last_active_visibility = "everyone" - presence_mode = settings_row["presence_mode"] if settings_row else "online" + presence_mode = settings_row["presence_mode"] if settings_row and "presence_mode" in settings_row.keys() else "online" if presence_mode not in PRESENCE_MODES: presence_mode = "online" - dm_layout_mode = settings_row["dm_layout_mode"] if settings_row else "cards" + dm_layout_mode = settings_row["dm_layout_mode"] if settings_row and "dm_layout_mode" in settings_row.keys() else "cards" if dm_layout_mode not in DM_LAYOUT_MODES: dm_layout_mode = "cards" @@ -1490,21 +1449,11 @@ def call_payload(get_db, call_id: int) -> Optional[dict]: def dispatch_scheduled_messages(get_db, now_iso_func, app_logger) -> None: - with get_db() as db: - table_exists = bool( - db.execute( - """ - SELECT 1 - FROM sqlite_master - WHERE type = 'table' - AND name = 'scheduled_messages' - """ - ).fetchone() - ) - if not table_exists: - return + if not sqlite_table_exists(get_db, "scheduled_messages"): + return - now_value = now_iso_func() + now_value = now_iso_func() + with get_db() as db: rows = db.execute( """ SELECT