Update features.py
This commit is contained in:
parent
513ea3e577
commit
842f97428d
145
features.py
145
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}$")
|
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)
|
PAGE_LINK_RE = re.compile(r"\]\((?:/teams/[^/]+/pages/)?([a-z0-9-]{1,48})\)", re.IGNORECASE)
|
||||||
|
|
||||||
|
|
||||||
def valid_hex_color(value: object) -> bool:
|
def valid_hex_color(value: object) -> bool:
|
||||||
return bool(isinstance(value, str) and HEX_COLOR_RE.fullmatch(value.strip()))
|
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": brand, "label": label}
|
||||||
|
|
||||||
return {"host": host, "brand": "web", "label": "website"}
|
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(
|
def attachment_kind_from_name_and_mime(
|
||||||
original_name: Optional[str],
|
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]:
|
def load_page_children(get_db, page_id: int) -> list[dict]:
|
||||||
with get_db() as db:
|
if not sqlite_column_exists(get_db, "pages", "parent_page_id"):
|
||||||
has_parent_page = bool(
|
return []
|
||||||
db.execute(
|
|
||||||
"""
|
|
||||||
SELECT 1
|
|
||||||
FROM pragma_table_info('pages')
|
|
||||||
WHERE name = 'parent_page_id'
|
|
||||||
"""
|
|
||||||
).fetchone()
|
|
||||||
)
|
|
||||||
if not has_parent_page:
|
|
||||||
return []
|
|
||||||
|
|
||||||
|
with get_db() as db:
|
||||||
rows = db.execute(
|
rows = db.execute(
|
||||||
"""
|
"""
|
||||||
SELECT
|
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]:
|
def load_page_comments(get_db, page_id: int) -> list[dict]:
|
||||||
with get_db() as db:
|
if not sqlite_table_exists(get_db, "page_comments"):
|
||||||
table_exists = bool(
|
return []
|
||||||
db.execute(
|
|
||||||
"""
|
|
||||||
SELECT 1
|
|
||||||
FROM sqlite_master
|
|
||||||
WHERE type = 'table'
|
|
||||||
AND name = 'page_comments'
|
|
||||||
"""
|
|
||||||
).fetchone()
|
|
||||||
)
|
|
||||||
if not table_exists:
|
|
||||||
return []
|
|
||||||
|
|
||||||
|
with get_db() as db:
|
||||||
rows = db.execute(
|
rows = db.execute(
|
||||||
"""
|
"""
|
||||||
SELECT
|
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:
|
def next_page_version_number(get_db, page_id: int) -> int:
|
||||||
with get_db() as db:
|
if not sqlite_table_exists(get_db, "page_versions"):
|
||||||
table_exists = bool(
|
return 1
|
||||||
db.execute(
|
|
||||||
"""
|
|
||||||
SELECT 1
|
|
||||||
FROM sqlite_master
|
|
||||||
WHERE type = 'table'
|
|
||||||
AND name = 'page_versions'
|
|
||||||
"""
|
|
||||||
).fetchone()
|
|
||||||
)
|
|
||||||
if not table_exists:
|
|
||||||
return 1
|
|
||||||
|
|
||||||
|
with get_db() as db:
|
||||||
row = db.execute(
|
row = db.execute(
|
||||||
"""
|
"""
|
||||||
SELECT COALESCE(MAX(version_number), 0) AS value
|
SELECT COALESCE(MAX(version_number), 0) AS value
|
||||||
@ -275,19 +267,8 @@ def snapshot_page_version(
|
|||||||
created_by_user_id: int,
|
created_by_user_id: int,
|
||||||
created_at: str,
|
created_at: str,
|
||||||
) -> int:
|
) -> int:
|
||||||
with get_db() as db:
|
if not sqlite_table_exists(get_db, "page_versions"):
|
||||||
table_exists = bool(
|
return 0
|
||||||
db.execute(
|
|
||||||
"""
|
|
||||||
SELECT 1
|
|
||||||
FROM sqlite_master
|
|
||||||
WHERE type = 'table'
|
|
||||||
AND name = 'page_versions'
|
|
||||||
"""
|
|
||||||
).fetchone()
|
|
||||||
)
|
|
||||||
if not table_exists:
|
|
||||||
return 0
|
|
||||||
|
|
||||||
version_number = next_page_version_number(get_db, page_id)
|
version_number = next_page_version_number(get_db, page_id)
|
||||||
with get_db() as db:
|
with get_db() as db:
|
||||||
@ -325,20 +306,10 @@ def snapshot_page_version(
|
|||||||
|
|
||||||
|
|
||||||
def load_page_versions(get_db, page_id: int) -> list[dict]:
|
def load_page_versions(get_db, page_id: int) -> list[dict]:
|
||||||
with get_db() as db:
|
if not sqlite_table_exists(get_db, "page_versions"):
|
||||||
table_exists = bool(
|
return []
|
||||||
db.execute(
|
|
||||||
"""
|
|
||||||
SELECT 1
|
|
||||||
FROM sqlite_master
|
|
||||||
WHERE type = 'table'
|
|
||||||
AND name = 'page_versions'
|
|
||||||
"""
|
|
||||||
).fetchone()
|
|
||||||
)
|
|
||||||
if not table_exists:
|
|
||||||
return []
|
|
||||||
|
|
||||||
|
with get_db() as db:
|
||||||
rows = db.execute(
|
rows = db.execute(
|
||||||
"""
|
"""
|
||||||
SELECT
|
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):
|
def load_page_version_detail(get_db, version_id: int):
|
||||||
with get_db() as db:
|
if not sqlite_table_exists(get_db, "page_versions"):
|
||||||
table_exists = bool(
|
return None
|
||||||
db.execute(
|
|
||||||
"""
|
|
||||||
SELECT 1
|
|
||||||
FROM sqlite_master
|
|
||||||
WHERE type = 'table'
|
|
||||||
AND name = 'page_versions'
|
|
||||||
"""
|
|
||||||
).fetchone()
|
|
||||||
)
|
|
||||||
if not table_exists:
|
|
||||||
return None
|
|
||||||
|
|
||||||
|
with get_db() as db:
|
||||||
row = db.execute(
|
row = db.execute(
|
||||||
"""
|
"""
|
||||||
SELECT
|
SELECT
|
||||||
@ -423,8 +384,6 @@ def user_can_see_last_active(settings_row, viewer: Optional[dict], target_user_i
|
|||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
def page_payload(app_module, page, user: Optional[dict]) -> dict:
|
def page_payload(app_module, page, user: Optional[dict]) -> dict:
|
||||||
can_read = bool(
|
can_read = bool(
|
||||||
page["visibility"] == "public"
|
page["visibility"] == "public"
|
||||||
@ -991,19 +950,19 @@ def profile_payload(
|
|||||||
explicit_name_color = settings_row["name_color"] if settings_row else None
|
explicit_name_color = settings_row["name_color"] if settings_row else None
|
||||||
resolved = resolve_name_color(user_row["role"], explicit_name_color)
|
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:
|
if profile_visibility not in PROFILE_VISIBILITIES:
|
||||||
profile_visibility = "public"
|
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:
|
if last_active_visibility not in LAST_ACTIVE_VISIBILITIES:
|
||||||
last_active_visibility = "everyone"
|
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:
|
if presence_mode not in PRESENCE_MODES:
|
||||||
presence_mode = "online"
|
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:
|
if dm_layout_mode not in DM_LAYOUT_MODES:
|
||||||
dm_layout_mode = "cards"
|
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:
|
def dispatch_scheduled_messages(get_db, now_iso_func, app_logger) -> None:
|
||||||
with get_db() as db:
|
if not sqlite_table_exists(get_db, "scheduled_messages"):
|
||||||
table_exists = bool(
|
return
|
||||||
db.execute(
|
|
||||||
"""
|
|
||||||
SELECT 1
|
|
||||||
FROM sqlite_master
|
|
||||||
WHERE type = 'table'
|
|
||||||
AND name = 'scheduled_messages'
|
|
||||||
"""
|
|
||||||
).fetchone()
|
|
||||||
)
|
|
||||||
if not table_exists:
|
|
||||||
return
|
|
||||||
|
|
||||||
now_value = now_iso_func()
|
now_value = now_iso_func()
|
||||||
|
with get_db() as db:
|
||||||
rows = db.execute(
|
rows = db.execute(
|
||||||
"""
|
"""
|
||||||
SELECT
|
SELECT
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user