Spaces:
Sleeping
Sleeping
| """Story engine — CRUD + visibility logic for 24-hour stories.""" | |
| from datetime import datetime, timezone | |
| from db import get_service_client | |
| import engine_user | |
| def _now_iso(): | |
| return datetime.now(timezone.utc).isoformat() | |
| def create_story(username: str, media_path: str, media_type: str, station_name: str = "", caption: str = "") -> dict: | |
| from db import reset_service_client | |
| for attempt in range(2): | |
| try: | |
| sb = get_service_client() | |
| row = sb.table("stories").insert({ | |
| "username": username, | |
| "station_name": station_name, | |
| "media_path": media_path, | |
| "media_type": media_type, | |
| "caption": caption, | |
| }).execute().data[0] | |
| return row | |
| except Exception as e: | |
| if attempt == 0 and "ConnectError" in type(e).__name__ or "10054" in str(e) or "forcibly closed" in str(e): | |
| reset_service_client() | |
| continue | |
| raise | |
| def delete_story(story_id: int, username: str) -> bool: | |
| sb = get_service_client() | |
| # Allow author OR station admin | |
| rows = sb.table("stories").select("id, username, station_name").eq("id", story_id).execute() | |
| if not rows.data: | |
| return False | |
| story = rows.data[0] | |
| is_author = story["username"] == username | |
| is_admin = False | |
| if story.get("station_name"): | |
| admin_rows = sb.table("stations").select("admin").eq("station_name", story["station_name"]).execute() | |
| is_admin = bool(admin_rows.data) and admin_rows.data[0]["admin"] == username | |
| if not is_author and not is_admin: | |
| return False | |
| sb.table("stories").delete().eq("id", story_id).execute() | |
| return True | |
| def get_story_views(story_id: int, requester: str) -> list[dict]: | |
| """Return viewers for a story. Only author or station admin can see.""" | |
| sb = get_service_client() | |
| rows = sb.table("stories").select("username, station_name").eq("id", story_id).execute() | |
| if not rows.data: | |
| return [] | |
| story = rows.data[0] | |
| is_author = story["username"] == requester | |
| is_admin = False | |
| if story.get("station_name"): | |
| admin_rows = sb.table("stations").select("admin").eq("station_name", story["station_name"]).execute() | |
| is_admin = bool(admin_rows.data) and admin_rows.data[0]["admin"] == requester | |
| if not is_author and not is_admin: | |
| return [] | |
| view_rows = sb.table("story_views").select("viewer, viewed_at").eq("story_id", story_id).execute() | |
| # Fetch avatars | |
| viewers = [r["viewer"] for r in view_rows.data] | |
| avatar_map = {} | |
| if viewers: | |
| u_rows = sb.table("users").select("username, avatar").in_("username", viewers).execute() | |
| avatar_map = {r["username"]: r.get("avatar", "") for r in u_rows.data} | |
| return [{"username": r["viewer"], "avatar": avatar_map.get(r["viewer"], ""), "viewed_at": r["viewed_at"]} for r in view_rows.data if r["viewer"] != requester] | |
| def _is_follower(viewer: str, owner: str) -> bool: | |
| if not viewer or not owner: | |
| return False | |
| sb = get_service_client() | |
| rows = sb.table("follows").select("status").eq("follower", viewer).eq("following", owner).execute() | |
| return bool(rows.data) and rows.data[0]["status"] == 1 | |
| def _user_visibility(username: str) -> int: | |
| """0 = public, 1 = private""" | |
| sb = get_service_client() | |
| rows = sb.table("users").select("visibility").eq("username", username).execute() | |
| if not rows.data: | |
| return 0 | |
| return int(rows.data[0].get("visibility") or 0) | |
| def _with_retry(fn): | |
| """Call fn(sb), reset client and retry once on connection errors.""" | |
| from db import reset_service_client | |
| for attempt in range(2): | |
| try: | |
| return fn(get_service_client()) | |
| except Exception as e: | |
| if attempt == 0 and ("ConnectError" in type(e).__name__ or "10054" in str(e) or "forcibly closed" in str(e)): | |
| reset_service_client() | |
| continue | |
| raise | |
| def _fetch_stories(sb): | |
| return sb.table("stories").select( | |
| "id, username, station_name, media_path, media_type, caption, created_at, expires_at" | |
| ).gt("expires_at", _now_iso()).order("created_at", desc=False).execute().data | |
| def get_user_stories(viewer: str) -> list[dict]: | |
| """Return active user stories (station_name='') visible to viewer.""" | |
| sb = get_service_client() | |
| all_rows = _with_retry(_fetch_stories) | |
| rows = [r for r in all_rows if r.get("station_name", "") == ""] | |
| # Get viewer's following list for private-account filtering | |
| following_set: set[str] = set() | |
| if viewer: | |
| f_rows = sb.table("follows").select("following").eq("follower", viewer).eq("status", 1).execute() | |
| following_set = {r["following"] for r in f_rows.data} | |
| # Get visibility map for all unique usernames | |
| usernames = list({r["username"] for r in rows}) | |
| vis_map: dict[str, int] = {} | |
| if usernames: | |
| u_rows = sb.table("users").select("username, visibility").in_("username", usernames).execute() | |
| vis_map = {r["username"]: int(r.get("visibility") if r.get("visibility") is not None else 1) for r in u_rows.data} | |
| # Get viewed story ids for viewer | |
| viewed_ids: set[int] = set() | |
| if viewer: | |
| v_rows = sb.table("story_views").select("story_id").eq("viewer", viewer).execute() | |
| viewed_ids = {r["story_id"] for r in v_rows.data} | |
| # Get avatars | |
| avatar_map: dict[str, str] = {} | |
| if usernames: | |
| a_rows = sb.table("users").select("username, avatar").in_("username", usernames).execute() | |
| avatar_map = {r["username"]: (r.get("avatar") or "") for r in a_rows.data} | |
| result = [] | |
| for r in rows: | |
| owner = r["username"] | |
| if owner == viewer: | |
| r["viewed"] = r["id"] in viewed_ids | |
| r["avatar"] = avatar_map.get(owner, "") | |
| result.append(r) | |
| continue | |
| # Visibility check: 0=private, 1=public | |
| vis = vis_map.get(owner, 1) | |
| if vis == 0 and owner not in following_set: | |
| continue | |
| # Block check | |
| if viewer and engine_user.is_blocked_either_way(viewer, owner): | |
| continue | |
| r["viewed"] = r["id"] in viewed_ids | |
| r["avatar"] = avatar_map.get(owner, "") | |
| result.append(r) | |
| return result | |
| def get_station_stories(viewer: str, station_names: list[str]) -> list[dict]: | |
| """Return active station stories for given stations visible to viewer.""" | |
| if not station_names: | |
| return [] | |
| sb = get_service_client() | |
| all_rows = _with_retry(_fetch_stories) | |
| rows = [r for r in all_rows if r.get("station_name", "") in station_names and r.get("station_name", "") != ""] | |
| # Get viewed story ids | |
| viewed_ids: set[int] = set() | |
| if viewer: | |
| v_rows = sb.table("story_views").select("story_id").eq("viewer", viewer).execute() | |
| viewed_ids = {r["story_id"] for r in v_rows.data} | |
| # Get avatars for story authors | |
| usernames = list({r["username"] for r in rows}) | |
| avatar_map: dict[str, str] = {} | |
| if usernames: | |
| a_rows = sb.table("users").select("username, avatar").in_("username", usernames).execute() | |
| avatar_map = {r["username"]: (r.get("avatar") or "") for r in a_rows.data} | |
| # Get station profile pics | |
| station_pic_map: dict[str, str] = {} | |
| if station_names: | |
| s_rows = sb.table("stations").select("station_name, profile_pic").in_("station_name", station_names).execute() | |
| station_pic_map = {r["station_name"]: (r.get("profile_pic") or "") for r in s_rows.data} | |
| result = [] | |
| for r in rows: | |
| r["viewed"] = r["id"] in viewed_ids | |
| r["avatar"] = avatar_map.get(r["username"], "") | |
| r["station_pic"] = station_pic_map.get(r["station_name"], "") | |
| result.append(r) | |
| return result | |
| def mark_viewed(story_id: int, viewer: str) -> None: | |
| if not viewer: | |
| return | |
| sb = get_service_client() | |
| sb.table("story_views").upsert({"story_id": story_id, "viewer": viewer}).execute() | |