Spaces:
Runtime error
Runtime error
| """Launchpad: the calendar, connected social accounts (OAuth / app password) and tracked short links.""" | |
| import json | |
| import re | |
| import secrets | |
| import uuid | |
| from datetime import UTC, datetime, timedelta | |
| from typing import Literal | |
| from urllib.parse import urlencode | |
| from zoneinfo import ZoneInfo, ZoneInfoNotFoundError | |
| from fastapi import APIRouter, Depends, HTTPException, Request | |
| from fastapi.concurrency import run_in_threadpool | |
| from fastapi.responses import RedirectResponse | |
| from pydantic import BaseModel, Field | |
| from sqlalchemy import select | |
| from sqlalchemy.orm import Session | |
| from app.auth import Ctx, decode_token_full, encode_signed, get_ctx | |
| from app.config import settings | |
| from app.db import get_db | |
| from app.models import Artifact, CalendarItem, Document, EngagementEvent, SocialAccount, TrackedLink, Upload | |
| from app.services import launchpad as lp | |
| from app.services.artifacts import TYPE_LABELS, download_name | |
| from app.services.crypto import decrypt, encrypt | |
| from app.services.social import AUTO_POST, LIMITS, PLATFORM_LABELS, SocialError, bluesky, linkedin, x | |
| from app.services.social.health import pause_account_posts, resume_account_posts | |
| from app.services.social.media import ( | |
| MB, | |
| SIZE_LIMITS, | |
| UPLOAD_TYPES, | |
| carries_media, | |
| check_postable, | |
| duration, | |
| is_blog2video, | |
| media_count, | |
| upload_kind, | |
| ) | |
| from app.services.storage import storage | |
| router = APIRouter(prefix="/api", tags=["launchpad"]) | |
| Platform = Literal["x", "linkedin", "bluesky", "substack_notes"] | |
| # Calendar | |
| class When(BaseModel): | |
| """When a post goes out: the wall time the user picked in their own time zone (local_time "2026-10-07T11:00" + | |
| timezone "Asia/Karachi", on a half hour), converted to UTC here; or an exact scheduled_at (Post now).""" | |
| scheduled_at: datetime | None = None | |
| local_time: str | None = Field(None, max_length=32) | |
| timezone: str | None = Field(None, max_length=64) | |
| def utc(self) -> datetime | None: | |
| if self.local_time: | |
| try: | |
| zone = ZoneInfo(self.timezone or "") | |
| except (ZoneInfoNotFoundError, ValueError) as exc: | |
| raise HTTPException(400, {"code": "bad_timezone", "message": "Unknown time zone."}) from exc | |
| try: | |
| wall = datetime.fromisoformat(self.local_time) | |
| except ValueError as exc: | |
| raise HTTPException(400, {"code": "bad_time", "message": "Pick a date and time."}) from exc | |
| if wall.tzinfo or wall.minute not in (0, 30) or wall.second or wall.microsecond: | |
| raise HTTPException(400, {"code": "half_hour_only", "message": "Pick a time on the hour or half hour."}) | |
| return wall.replace(tzinfo=zone).astimezone(UTC) | |
| return _aware(self.scheduled_at) if self.scheduled_at else None | |
| class ItemIn(When): | |
| platform: Platform | |
| content: str = Field("", max_length=10000) # may be empty when the post carries images or a video | |
| thread: list[str] = Field(default_factory=list, max_length=25) | |
| artifact_id: uuid.UUID | None = None # the attachment (a video, quote card, carousel...), never a Launch Kit | |
| kit_id: uuid.UUID | None = None # the Launch Kit the post was written from (tracked, not attached) | |
| document_id: uuid.UUID | None = None | |
| social_account_id: uuid.UUID | None = None | |
| remind_by_email: bool = False | |
| draft: bool = False | |
| # Post now: published (or, by email, reminded) in this request, never left scheduled; no time needed | |
| publish_now: bool = False | |
| class ItemPatch(When): | |
| artifact_id: uuid.UUID | None = None # set or (null) clear what the post carries | |
| content: str | None = Field(None, max_length=10000) | |
| thread: list[str] | None = None | |
| social_account_id: uuid.UUID | None = None | |
| remind_by_email: bool | None = None | |
| status: Literal["scheduled", "draft", "paused"] | None = None | |
| def _aware(dt: datetime) -> datetime: | |
| return dt if dt.tzinfo else dt.replace(tzinfo=UTC) | |
| # audio can't be posted; "upload" is a file the user brought from their computer | |
| POSTABLE = ("video", "quote_card", "carousel", "summary", "launch_kit", "mind_map", "upload", "slide_deck") | |
| # Listed on the Launchpad too, but only to download: X and LinkedIn take no audio files (check_postable refuses it) | |
| LISTED = (*POSTABLE, "audio_overview") | |
| def _cite_free(text: str) -> str: | |
| return re.sub(r"\s*\[\d+\]", "", text or "").strip() | |
| def _prefill(a: Artifact) -> dict: | |
| """A starting text per platform for a post of this artifact: X as a list of posts (a thread), LinkedIn as one.""" | |
| c = a.content_json or {} | |
| title = c.get("title") or TYPE_LABELS.get(a.type, a.type) | |
| if a.type == "upload": # their own file: the words are theirs to write | |
| return {"x": [""], "linkedin": ""} | |
| if a.type == "launch_kit": | |
| posts = c.get("posts") or {} | |
| x_posts = posts.get("x_thread") or posts.get("x") or [title] | |
| return {"x": x_posts, "linkedin": "\n\n".join(posts.get("linkedin") or x_posts)} | |
| text = { | |
| "summary": _cite_free(c.get("summary") or title), | |
| "quote_card": f'"{c.get("quote")}"' + (f" ({c['source']})" if c.get("source") else "") if c.get("quote") | |
| else title, | |
| "carousel": ((c.get("slides") or [{}])[0].get("heading") or title), | |
| "slide_deck": _deck_prefill(c) or title, | |
| }.get(a.type, title) | |
| return {"x": [text], "linkedin": text} | |
| def _deck_prefill(c: dict) -> str: | |
| """A slide deck's post: its title, its subtitle, and the takeaways of its closing slide.""" | |
| deck = c.get("deck") or {} | |
| slides = deck.get("slides") or [] | |
| closing = next((s for s in reversed(slides) if s.get("layout") == "closing"), {}) | |
| head = "\n\n".join(x for x in (deck.get("title"), deck.get("subtitle")) if x) | |
| takeaways = "\n".join(f"- {t}" for t in closing.get("takeaways") or []) | |
| return "\n\n".join(x for x in (head, takeaways) if x) | |
| def artifact_brief(a: Artifact) -> dict: | |
| """What the composer and the calendar show of the attached artifact.""" | |
| c = a.content_json or {} | |
| thumb = (a.storage_key if a.type == "quote_card" or upload_kind(a) == "image" | |
| else (c.get("slide_keys") or [a.storage_key])[0] if a.type == "carousel" else None) | |
| return {"id": str(a.id), "type": a.type, "type_label": TYPE_LABELS.get(a.type, a.type), | |
| # A deck's title starts "Slide deck: ", which its tag already says | |
| "title": (c.get("title") or TYPE_LABELS.get(a.type, a.type)).removeprefix("Slide deck: " if a.type == "slide_deck" else ""), | |
| "media": upload_kind(a) or ("video" if a.type == "video" else "audio" if a.type == "audio_overview" | |
| else "image" if media_count(a) else "text"), | |
| "media_count": media_count(a), "duration_s": duration(a), | |
| "thumb_url": storage.presign_get(thumb) if thumb else None, | |
| # The user's own file, to preview before posting it. | |
| "view_url": storage.presign_get(a.storage_key) if a.type == "upload" and a.storage_key else None, | |
| # An audio overview can't be posted: it is offered to download instead. | |
| "download_url": (storage.presign_get(a.storage_key, download_name=download_name(a)) | |
| if a.type == "audio_overview" and a.storage_key else None), | |
| # A blog2video video with scenes but no MP4 yet: listed, but it has to be rendered before it can go out. | |
| # Made by the video service (blog2video): it opens in the video editor. | |
| "editable_video": is_blog2video(a), | |
| "needs_render": is_blog2video(a) and not (a.content_json or {}).get("video_url"), | |
| # Its MP4 is being made right now: it can't be scheduled until the render finishes. | |
| "rendering": is_blog2video(a) and a.status == "rendering", | |
| "created_at": a.created_at.isoformat() if a.created_at else None, "prefill": _prefill(a)} | |
| def serialize_item(item: CalendarItem, account: SocialAccount | None = None, clicks: int = 0, | |
| artifact: Artifact | None = None, kit: Artifact | None = None) -> dict: | |
| return { | |
| "id": str(item.id), | |
| "platform": item.platform, | |
| "platform_label": PLATFORM_LABELS.get(item.platform, item.platform), | |
| "scheduled_at": _aware(item.scheduled_at).isoformat(), | |
| "status": item.status, | |
| "content": item.content, | |
| "thread": item.thread or [], | |
| "artifact_id": str(item.artifact_id) if item.artifact_id else None, | |
| "artifact": artifact_brief(artifact) if artifact else None, | |
| "kit_id": str(item.kit_id) if item.kit_id else None, | |
| "kit_title": ((kit.content_json or {}).get("post_title") or (kit.content_json or {}).get("title")) if kit else None, | |
| "document_id": str(item.document_id) if item.document_id else None, | |
| "social_account_id": str(item.social_account_id) if item.social_account_id else None, | |
| "account_handle": account.handle if account else None, | |
| "remind_by_email": item.remind_by_email, | |
| "auto_post": item.platform in AUTO_POST and bool(item.social_account_id) and not item.remind_by_email, | |
| "external_url": item.external_url, | |
| "posted_at": _aware(item.posted_at).isoformat() if item.posted_at else None, | |
| "error": item.error, | |
| "metrics": item.metrics or {}, | |
| "clicks": clicks, | |
| } | |
| def _item(ctx: Ctx, item_id: uuid.UUID) -> CalendarItem: | |
| item = ctx.db.scalar(select(CalendarItem).where(CalendarItem.id == item_id, | |
| CalendarItem.workspace_id == ctx.workspace.id)) | |
| if not item: | |
| raise HTTPException(404, "Calendar item not found") | |
| return item | |
| def _account(ctx: Ctx, account_id: uuid.UUID | None, platform: str) -> SocialAccount | None: | |
| if not account_id: | |
| return None | |
| account = ctx.db.scalar(select(SocialAccount).where(SocialAccount.id == account_id, | |
| SocialAccount.workspace_id == ctx.workspace.id)) | |
| if not account or account.platform != platform: | |
| raise HTTPException(400, "That account does not match the platform") | |
| return account | |
| CONNECTED_ONLY = ("x", "linkedin") # these post only through a live connection (no email fallback) | |
| def can_post_media(a: SocialAccount) -> bool: | |
| return {"x": x.can_post_media, "linkedin": linkedin.can_post_media}.get(a.platform, lambda _: False)(a) | |
| def _require_connected(platform: str, account: SocialAccount | None, artifact: Artifact | None) -> None: | |
| """X and LinkedIn: scheduling needs a live connection (and one allowed to post media, when media is attached).""" | |
| if platform not in CONNECTED_ONLY: | |
| return | |
| label = PLATFORM_LABELS[platform] | |
| if not account or account.status != "active": | |
| raise HTTPException(409, {"code": "not_connected", | |
| "message": f"Connect {label} to schedule posts there." | |
| if not account else f"Your {label} connection is down. Reconnect {label} first."}) | |
| if artifact and media_count(artifact) and not can_post_media(account): | |
| raise HTTPException(409, {"code": "reconnect", "message": f"Reconnect {label} to post images and videos."}) | |
| def _require_words(posts: list[str], artifact: Artifact | None) -> None: | |
| """A post needs words, unless it carries images or a video (those can go out without a caption).""" | |
| if not any(p.strip() for p in posts) and not carries_media(artifact): | |
| raise HTTPException(422, {"code": "no_text", "message": "Write something to post, or attach an image or video."}) | |
| def _validate_lengths(platform: str, posts: list[str]) -> None: | |
| limit = LIMITS.get(platform) | |
| if platform in ("x", "bluesky") and limit: | |
| for i, p in enumerate(posts, start=1): | |
| if len(p) > limit: | |
| raise HTTPException(422, {"code": "too_long", | |
| "message": f"Post {i} is {len(p)} characters; {PLATFORM_LABELS[platform]} " | |
| f"allows {limit}."}) | |
| def _enrich(ctx: Ctx, items: list[CalendarItem]) -> list[dict]: | |
| accounts = {a.id: a for a in ctx.db.scalars( | |
| select(SocialAccount).where(SocialAccount.workspace_id == ctx.workspace.id))} | |
| clicks = lp.clicks_by_item(ctx.db, [i.id for i in items]) | |
| ids = {i.artifact_id for i in items if i.artifact_id} | {i.kit_id for i in items if i.kit_id} | |
| artifacts = {a.id: a for a in ctx.db.scalars(select(Artifact).where(Artifact.id.in_(ids)))} if ids else {} | |
| return [serialize_item(i, accounts.get(i.social_account_id), clicks.get(i.id, 0), artifacts.get(i.artifact_id), | |
| artifacts.get(i.kit_id)) for i in items] | |
| def _kit(ctx: Ctx, kit_id: uuid.UUID) -> Artifact: | |
| """This workspace's Launch Kit (what a post was written from).""" | |
| kit = ctx.db.scalar(select(Artifact).where(Artifact.id == kit_id, Artifact.workspace_id == ctx.workspace.id, | |
| Artifact.type == "launch_kit")) | |
| if not kit: | |
| raise HTTPException(404, "Launch Kit not found") | |
| return kit | |
| def _postable(ctx: Ctx, artifact_id: uuid.UUID, platform: str) -> Artifact: | |
| """This workspace's artifact, checked for the platform (audio, unfinished, too long for X: 400).""" | |
| a = ctx.db.scalar(select(Artifact).where(Artifact.id == artifact_id, Artifact.workspace_id == ctx.workspace.id)) | |
| if not a: | |
| raise HTTPException(404, "Artifact not found") | |
| check_postable(a, platform) | |
| return a | |
| def list_items(ctx: Ctx = Depends(get_ctx), start: datetime | None = None, end: datetime | None = None, | |
| status: str | None = None, limit: int = 500): | |
| q = select(CalendarItem).where(CalendarItem.workspace_id == ctx.workspace.id) | |
| if start: | |
| q = q.where(CalendarItem.scheduled_at >= start) | |
| if end: | |
| q = q.where(CalendarItem.scheduled_at < end) | |
| if status: | |
| q = q.where(CalendarItem.status.in_(status.split(","))) | |
| items = list(ctx.db.scalars(q.order_by(CalendarItem.scheduled_at).limit(min(limit, 1000))).all()) | |
| return _enrich(ctx, items) | |
| def create_item(body: ItemIn, ctx: Ctx = Depends(get_ctx)): | |
| when = datetime.now(UTC) if body.publish_now else body.utc() | |
| if when is None: | |
| raise HTTPException(422, {"code": "no_time", "message": "Pick when it goes out."}) | |
| account = _account(ctx, body.social_account_id, body.platform) | |
| _validate_lengths(body.platform, [body.content, *body.thread]) | |
| artifact = _postable(ctx, body.artifact_id, body.platform) if body.artifact_id else None | |
| _require_words([body.content, *body.thread], artifact) | |
| kit = _kit(ctx, body.kit_id) if body.kit_id else None | |
| _require_connected(body.platform, account, artifact) | |
| if body.document_id and not ctx.db.scalar(select(Document.id).where(Document.id == body.document_id, | |
| Document.workspace_id == ctx.workspace.id)): | |
| raise HTTPException(404, "Document not found") | |
| item = CalendarItem( | |
| workspace_id=ctx.workspace.id, platform=body.platform, scheduled_at=when, | |
| content=body.content, thread=[t for t in body.thread if t.strip()], artifact_id=body.artifact_id, | |
| kit_id=kit.id if kit else None, | |
| document_id=body.document_id, social_account_id=account.id if account else None, | |
| remind_by_email=body.remind_by_email or body.platform not in AUTO_POST or not account, | |
| status="draft" if body.draft else "publishing" if body.publish_now else "scheduled", | |
| ) | |
| ctx.db.add(item) | |
| ctx.db.commit() | |
| if body.publish_now: | |
| _go_now(ctx, item) | |
| return _enrich(ctx, [item])[0] | |
| def _go_now(ctx: Ctx, item: CalendarItem) -> None: | |
| """Publish an item claimed for it ("publishing") now: with images or a video through a job (uploading takes a | |
| while; the page follows its status), anything else right here.""" | |
| if lp.has_media(ctx.db, item): | |
| lp.queue_publish(ctx.db, item) | |
| else: | |
| lp.publish_item(ctx.db, item) | |
| def update_item(item_id: uuid.UUID, body: ItemPatch, ctx: Ctx = Depends(get_ctx)): | |
| item = _item(ctx, item_id) | |
| if item.status in ("posted", "publishing"): | |
| raise HTTPException(409, "This post already went out.") | |
| when = body.utc() | |
| if when is not None: | |
| item.scheduled_at = when | |
| if body.content is not None: | |
| item.content = body.content | |
| if body.thread is not None: | |
| item.thread = [t for t in body.thread if t.strip()] | |
| if "artifact_id" in body.model_fields_set: | |
| item.artifact_id = _postable(ctx, body.artifact_id, item.platform).id if body.artifact_id else None | |
| if "social_account_id" in body.model_fields_set: | |
| item.social_account_id = (_account(ctx, body.social_account_id, item.platform).id | |
| if body.social_account_id else None) | |
| if body.remind_by_email is not None: | |
| item.remind_by_email = body.remind_by_email | |
| if not item.social_account_id or item.platform not in AUTO_POST: | |
| item.remind_by_email = True | |
| account = ctx.db.get(SocialAccount, item.social_account_id) if item.social_account_id else None | |
| artifact = ctx.db.get(Artifact, item.artifact_id) if item.artifact_id else None | |
| if body.status == "paused": | |
| item.status = "paused" # held by hand: never attempted until resumed | |
| elif body.status is not None or item.status in ("failed", "reminded", "paused"): | |
| # Going out (again), rescheduled: only through a live connection, and only at a time still ahead (a failed or | |
| # reminded post's old time has passed, and it would otherwise go out at the very next run). | |
| _require_connected(item.platform, account, artifact) | |
| if (body.status or "scheduled") == "scheduled" and _aware(item.scheduled_at) <= datetime.now(UTC): | |
| raise HTTPException(409, {"code": "missed", "message": "Pick a time that is still ahead."}) | |
| item.status = body.status or "scheduled" | |
| elif item.platform in CONNECTED_ONLY and body.model_fields_set & {"social_account_id", "artifact_id"}: | |
| _require_connected(item.platform, account, artifact) | |
| item.error = None | |
| item.attempts = 0 | |
| _require_words([item.content or "", *(item.thread or [])], artifact) # e.g. the caption emptied, or the video removed | |
| _validate_lengths(item.platform, [item.content, *(item.thread or [])]) | |
| ctx.db.commit() | |
| return _enrich(ctx, [item])[0] | |
| def delete_item(item_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)): | |
| item = _item(ctx, item_id) | |
| ctx.db.delete(item) | |
| ctx.db.commit() | |
| return {"ok": True} | |
| def publish_now(item_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)): | |
| """Publish a scheduled post now. A plain (threadpool) endpoint: its database calls and an inline publish must | |
| never hold up the event loop, and with it every other request.""" | |
| item = _item(ctx, item_id) | |
| if item.status in ("posted", "publishing"): | |
| raise HTTPException(409, "This post already went out.") | |
| item.status = "publishing" | |
| item.attempts = 0 | |
| ctx.db.commit() | |
| _go_now(ctx, item) | |
| return _enrich(ctx, [item])[0] | |
| class UploadIn(BaseModel): | |
| upload_id: uuid.UUID | |
| duration_s: float | None = Field(None, ge=0, le=24 * 3600) # a video's length, read by the browser | |
| def register_upload(body: UploadIn, ctx: Ctx = Depends(get_ctx)): | |
| """A file the user uploaded from their computer (through /api/storage/uploads), made postable: it becomes an | |
| `upload` artifact, attached and posted like anything Notestack made, and offered again in the picker later.""" | |
| upload = ctx.db.scalar(select(Upload).where(Upload.id == body.upload_id, Upload.workspace_id == ctx.workspace.id)) | |
| if not upload: | |
| raise HTTPException(404, "Upload not found") | |
| if upload.status != "complete": | |
| raise HTTPException(409, "This file hasn't finished uploading.") | |
| kind = UPLOAD_TYPES.get(upload.content_type) | |
| if not kind: | |
| raise HTTPException(415, "LinkedIn takes JPG, PNG or GIF images and MP4 videos.") | |
| low, high = SIZE_LIMITS[("linkedin", kind)] | |
| size = upload.size_bytes or 0 | |
| if size > high: | |
| raise HTTPException(413, f"This {kind} is {size / MB:.0f} MB; LinkedIn takes up to {high // MB} MB.") | |
| if size < low: | |
| raise HTTPException(400, f"This {kind} is too small for LinkedIn (at least {low // 1024} KB).") | |
| title = upload.filename.rsplit(".", 1)[0] if "." in upload.filename else upload.filename | |
| a = Artifact(workspace_id=ctx.workspace.id, type="upload", status="ready", storage_key=upload.key, | |
| content_json={"title": title[:200] or "Upload", "filename": upload.filename, | |
| "content_type": upload.content_type, "media": kind, "size_bytes": size, | |
| **({"duration_s": body.duration_s} if body.duration_s else {})}) | |
| ctx.db.add(a) | |
| ctx.db.commit() | |
| return artifact_brief(a) | |
| def postable(ctx: Ctx = Depends(get_ctx), q: str = "", limit: int = 60): | |
| """What can be attached to a post: finished videos, quote cards, carousels, summaries, launch kits, mind maps, | |
| slide decks, newest first, and audio overviews (listed to download: they can't be posted). Videos made but not | |
| rendered yet are listed too, flagged needs_render (they can only be posted once rendered), and so are videos being | |
| rendered, flagged rendering.""" | |
| rows = ctx.db.scalars(select(Artifact).where(Artifact.workspace_id == ctx.workspace.id, | |
| Artifact.type.in_(LISTED)) | |
| .order_by(Artifact.created_at.desc()).limit(400)) | |
| needle = q.strip().lower() | |
| out = [] | |
| for a in rows: | |
| # A blog2video video counts once blog2video made it (ready), rendered or not; anything else once it is ready. | |
| # A blog2video video mid-render stays listed (flagged rendering). | |
| ready = a.status == "ready" or (is_blog2video(a) and (bool((a.content_json or {}).get("video_url")) | |
| or a.status == "rendering")) | |
| title = ((a.content_json or {}).get("title") or "").lower() | |
| if ready and (not needle or needle in title): | |
| out.append(artifact_brief(a)) | |
| if len(out) >= min(limit, 200): | |
| break | |
| # What each one already has coming up (scheduled, still ahead), soonest first: the list offers to reschedule it | |
| upcoming = ctx.db.scalars(select(CalendarItem).where( | |
| CalendarItem.workspace_id == ctx.workspace.id, CalendarItem.status == "scheduled", | |
| CalendarItem.scheduled_at > datetime.now(UTC), | |
| CalendarItem.artifact_id.in_([uuid.UUID(b["id"]) for b in out])).order_by(CalendarItem.scheduled_at)) | |
| by_artifact: dict[str, list[dict]] = {} | |
| for it in upcoming: | |
| by_artifact.setdefault(str(it.artifact_id), []).append( | |
| {"id": str(it.id), "scheduled_at": _aware(it.scheduled_at).isoformat(), | |
| "platform_label": PLATFORM_LABELS.get(it.platform, it.platform)}) | |
| for b in out: | |
| b["scheduled"] = by_artifact.get(b["id"], []) | |
| return out | |
| # Social accounts | |
| def serialize_account(a: SocialAccount) -> dict: | |
| return {"id": str(a.id), "platform": a.platform, "platform_label": PLATFORM_LABELS[a.platform], | |
| "handle": a.handle, "status": a.status, "avatar_url": a.avatar_url, | |
| "can_post_media": can_post_media(a), | |
| # No refresh token and the connection ends within a week (LinkedIn's 60 day tokens). | |
| "expires_soon": bool(a.expires_at and not a.refresh_token | |
| and _aware(a.expires_at) - datetime.now(UTC) < timedelta(days=7)), | |
| "expires_at": _aware(a.expires_at).isoformat() if a.expires_at else None, | |
| # Down (expired, revoked, disconnected), or unable to post images and videos. | |
| "needs_reconnect": a.status != "active" or (a.platform in CONNECTED_ONLY and not can_post_media(a)), | |
| "connected_at": a.created_at.isoformat() if a.created_at else None} | |
| def list_accounts(ctx: Ctx = Depends(get_ctx)): | |
| accounts = ctx.db.scalars(select(SocialAccount).where(SocialAccount.workspace_id == ctx.workspace.id) | |
| .order_by(SocialAccount.created_at)).all() | |
| return { | |
| "accounts": [serialize_account(a) for a in accounts], | |
| "available": {"x": x.configured(), "linkedin": linkedin.configured(), "bluesky": True}, | |
| "redirect_uris": {"x": x.redirect_uri(), "linkedin": linkedin.redirect_uri()}, | |
| } | |
| def disconnect(account_id: uuid.UUID, ctx: Ctx = Depends(get_ctx)): | |
| account = ctx.db.scalar(select(SocialAccount).where(SocialAccount.id == account_id, | |
| SocialAccount.workspace_id == ctx.workspace.id)) | |
| if not account: | |
| raise HTTPException(404, "Account not found") | |
| # Kept (its posts stay linked to it), tokens gone, posts paused: reconnecting the same account resumes them. | |
| account.status = "revoked" | |
| account.access_token = "" # no usable token kept | |
| account.refresh_token = None | |
| label = PLATFORM_LABELS[account.platform] | |
| pause_account_posts(ctx.db, account, f"{label} disconnected. Reconnect {label} to resume.") | |
| ctx.db.commit() | |
| return {"ok": True} | |
| def _upsert_account(db: Session, workspace_id: uuid.UUID, platform: str, info: dict, secret: str, | |
| refresh: str | None = None, expires_in: int | None = None, scopes: str | None = None) -> None: | |
| account = db.scalar(select(SocialAccount).where(SocialAccount.workspace_id == workspace_id, | |
| SocialAccount.platform == platform, | |
| SocialAccount.external_id == info["external_id"])) | |
| if not account: | |
| account = SocialAccount(workspace_id=workspace_id, platform=platform, external_id=info["external_id"], | |
| handle=info["handle"], access_token="") | |
| db.add(account) | |
| account.handle = info["handle"] | |
| account.avatar_url = info.get("avatar_url") | |
| account.access_token = encrypt(secret) | |
| account.refresh_token = encrypt(refresh) if refresh else account.refresh_token | |
| account.expires_at = datetime.now(UTC) + timedelta(seconds=int(expires_in)) if expires_in else None | |
| account.scopes = scopes | |
| account.status = "active" | |
| resume_account_posts(db, account) # its paused posts still ahead go back to scheduled | |
| db.commit() | |
| class BlueskyIn(BaseModel): | |
| handle: str = Field(min_length=3, max_length=200) | |
| app_password: str = Field(min_length=8, max_length=100) | |
| async def connect_bluesky(body: BlueskyIn, ctx: Ctx = Depends(get_ctx)): | |
| try: | |
| session = await run_in_threadpool(bluesky.create_session, body.handle, body.app_password) | |
| except SocialError as exc: | |
| raise HTTPException(400, {"code": "bluesky_auth", "message": str(exc)}) from exc | |
| _upsert_account(ctx.db, ctx.workspace.id, "bluesky", | |
| {"external_id": session["did"], "handle": session.get("handle", body.handle)}, body.app_password) | |
| return list_accounts(ctx) | |
| # OAuth (X, LinkedIn). The signed state carries the workspace and PKCE verifier; the callback hands a | |
| # short lived encrypted ticket back to the SPA, which completes the link as the signed in user. | |
| STATE_TTL = timedelta(minutes=10) | |
| def _launchpad_redirect(**params: str) -> RedirectResponse: | |
| return RedirectResponse(f"{settings.frontend_url.rstrip('/')}/app/launchpad?{urlencode(params)}", 302) | |
| def oauth_start(platform: Literal["x", "linkedin"], ctx: Ctx = Depends(get_ctx)): | |
| """Returns the provider URL. The SPA calls this with its bearer token, then navigates to `url`.""" | |
| mod = x if platform == "x" else linkedin | |
| if not mod.configured(): | |
| raise HTTPException(503, {"code": "not_configured", | |
| "message": f"Set {platform.upper()}_CLIENT_ID and {platform.upper()}_CLIENT_SECRET."}) | |
| nonce = secrets.token_urlsafe(24) | |
| payload = {"typ": "social_state", "nonce": nonce, "ws": str(ctx.workspace.id), "platform": platform} | |
| if platform == "x": | |
| verifier, challenge = x.pkce_pair() | |
| payload["verifier"] = verifier | |
| state = encode_signed(payload, STATE_TTL) | |
| url = x.authorize_url(state, challenge) | |
| else: | |
| state = encode_signed(payload, STATE_TTL) | |
| url = linkedin.authorize_url(state) | |
| return {"url": url, "nonce": nonce} | |
| def oauth_callback(platform: Literal["x", "linkedin"], code: str | None = None, state: str | None = None, | |
| error: str | None = None, db: Session = Depends(get_db)): | |
| label = PLATFORM_LABELS[platform] | |
| if error or not code or not state: | |
| return _launchpad_redirect(error=f"{label} connection was cancelled.") | |
| try: | |
| saved = decode_token_full(state, expected_type="social_state") | |
| except HTTPException: | |
| return _launchpad_redirect(error=f"{label} connection expired. Try again.") | |
| if saved.get("platform") != platform: | |
| return _launchpad_redirect(error=f"{label} connection failed.") | |
| try: | |
| if platform == "x": | |
| info = x.exchange_code(code, saved["verifier"]) | |
| else: | |
| info = linkedin.exchange_code(code) | |
| except Exception: | |
| return _launchpad_redirect(error=f"{label} connection failed. Try again.") | |
| # Do not link here: this request carries no session. The signed in SPA finishes the link, and only | |
| # for the workspace that started it, so a forwarded authorize link cannot attach someone's account | |
| # to another workspace. | |
| ticket = encode_signed({"typ": "social_link", "ws": saved["ws"], "platform": platform, | |
| "data": encrypt(json.dumps(info))}, timedelta(minutes=10)) | |
| return _launchpad_redirect(link=ticket, platform=platform) | |
| class LinkIn(BaseModel): | |
| ticket: str | |
| def oauth_complete(body: LinkIn, ctx: Ctx = Depends(get_ctx)): | |
| try: | |
| saved = decode_token_full(body.ticket, expected_type="social_link") | |
| except HTTPException as exc: | |
| raise HTTPException(400, {"code": "expired", "message": "That connection expired. Try again."}) from exc | |
| if saved.get("ws") != str(ctx.workspace.id): | |
| raise HTTPException(403, {"code": "wrong_workspace", "message": "This connection was started elsewhere."}) | |
| info = json.loads(decrypt(saved["data"]) or "{}") | |
| if not info.get("access_token"): | |
| raise HTTPException(400, {"code": "expired", "message": "That connection expired. Try again."}) | |
| _upsert_account(ctx.db, ctx.workspace.id, saved["platform"], info, info["access_token"], | |
| info.get("refresh_token"), info.get("expires_in"), info.get("scope")) | |
| return list_accounts(ctx) | |
| # Tracked links (public) | |
| links_router = APIRouter(tags=["links"]) | |
| def follow_link(slug: str, request: Request, db: Session = Depends(get_db)): | |
| link = db.scalar(select(TrackedLink).where(TrackedLink.slug == slug)) | |
| if not link: | |
| raise HTTPException(404, "Link not found") | |
| db.add(EngagementEvent(workspace_id=link.workspace_id, tracked_link_id=link.id, kind="click", | |
| metadata_json={"referer": request.headers.get("referer", "")[:300], | |
| "ua": request.headers.get("user-agent", "")[:200]})) | |
| db.commit() | |
| return RedirectResponse(link.target_url, status_code=302) | |