Spaces:
Runtime error
Runtime error
| """What a scheduled post carries from the artifact it was made from: images (quote cards, carousel slides, a slide | |
| deck's slides, drawn when the post goes out) or one | |
| video (an audiogram, or a blog2video video), or a file the user uploaded from their computer (an `upload`: one image or | |
| one MP4). Summaries, launch kits and mind maps post as text; audio overviews can't | |
| be posted (X and LinkedIn take no audio files). | |
| Videos are streamed to a temporary file (never held whole in memory) and read in slices by the uploaders; the files are | |
| removed when the post is done (media_for is a context manager). Sizes are checked before anything is sent.""" | |
| import contextlib | |
| import os | |
| import tempfile | |
| from collections.abc import Iterator | |
| from dataclasses import dataclass | |
| import httpx | |
| from fastapi import HTTPException | |
| from app.models import Artifact | |
| from app.services.social import SocialError | |
| MAX_IMAGES = {"x": 4, "linkedin": 20} | |
| X_VIDEO_SECONDS = 140 # X's limit for videos from standard accounts (2:20) | |
| TEXT_ONLY = {"summary", "launch_kit", "mind_map"} | |
| MEDIA_PLATFORMS = {"x", "linkedin"} | |
| MB = 1024 * 1024 | |
| # What the user can upload from their computer to post: LinkedIn takes JPG, PNG and GIF images and MP4 videos. | |
| UPLOAD_TYPES = {"image/png": "image", "image/jpeg": "image", "image/gif": "image", "video/mp4": "video"} | |
| # (min, max) bytes per platform and kind, checked before any upload starts | |
| SIZE_LIMITS = { | |
| ("x", "image"): (1, 5 * MB), | |
| ("x", "video"): (1, 512 * MB), | |
| ("linkedin", "image"): (1, 36 * MB), | |
| ("linkedin", "video"): (75 * 1024, 500 * MB), | |
| } | |
| class Media: | |
| kind: str # image | video | |
| filename: str | |
| content_type: str | |
| data: bytes | None = None # images: the bytes | |
| path: str | None = None # videos: a temporary file | |
| size: int = 0 | |
| def __post_init__(self): | |
| if self.data is not None: | |
| self.size = len(self.data) | |
| elif self.path: | |
| self.size = os.path.getsize(self.path) | |
| def read(self, start: int = 0, end: int | None = None) -> bytes: | |
| """Bytes [start, end) (all of them by default).""" | |
| end = self.size if end is None else min(end, self.size) | |
| if self.data is not None: | |
| return self.data[start:end] | |
| with open(self.path, "rb") as f: | |
| f.seek(start) | |
| return f.read(end - start) | |
| def is_blog2video(a: Artifact) -> bool: | |
| return (a.content_json or {}).get("provider") == "blog2video" | |
| def media_count(a: Artifact) -> int: | |
| """How many files a post of this artifact carries (before a platform's cap).""" | |
| if a.type == "quote_card": | |
| return 1 | |
| if a.type == "carousel": | |
| return len((a.content_json or {}).get("slide_keys") or ([a.storage_key] if a.storage_key else [])) | |
| if a.type == "slide_deck": | |
| return len(((a.content_json or {}).get("deck") or {}).get("slides") or []) | |
| if a.type in ("video", "upload"): | |
| return 1 | |
| return 0 | |
| def carries_media(a: Artifact | None) -> bool: | |
| """Whether a post of this artifact carries images or a video: such a post may go out without a caption.""" | |
| return a is not None and a.type not in TEXT_ONLY and media_count(a) > 0 | |
| def upload_kind(a: Artifact) -> str | None: | |
| """An uploaded file's kind: image | video.""" | |
| return (a.content_json or {}).get("media") if a.type == "upload" else None | |
| def duration(a: Artifact) -> float | None: | |
| """A video's length when we know it (audiograms record it; blog2video's is only known to blog2video).""" | |
| value = (a.content_json or {}).get("duration_s") | |
| return float(value) if value else None | |
| def check_postable(a: Artifact, platform: str) -> None: | |
| """400 when this artifact can't go out on this platform, so it is refused when scheduled, not at post time.""" | |
| if a.type == "audio_overview": | |
| raise HTTPException(400, "Audio can't be posted: X and LinkedIn don't take audio files.") | |
| if a.type == "launch_kit": # a post is written from a kit (kit_id); the kit itself is never an attachment | |
| raise HTTPException(400, "A launch kit can't be attached. Attach a video, quote card or carousel instead.") | |
| if a.type not in TEXT_ONLY and a.type not in ("quote_card", "carousel", "video", "upload", "slide_deck"): | |
| raise HTTPException(400, "This can't be posted.") | |
| if a.type in TEXT_ONLY: | |
| return | |
| if platform not in MEDIA_PLATFORMS: | |
| raise HTTPException(400, "Images and videos can only be posted to X and LinkedIn.") | |
| if a.type == "slide_deck": # drawn from the deck when the post goes out: nothing stored to wait for | |
| if a.status != "ready" or not media_count(a): | |
| raise HTTPException(400, "This slide deck isn't ready yet. Post it once it has finished.") | |
| elif a.type == "video" and is_blog2video(a): | |
| if not (a.content_json or {}).get("video_url"): | |
| raise HTTPException(400, "This video isn't finished yet. Post it once it has rendered.") | |
| elif a.status != "ready" or not a.storage_key: | |
| raise HTTPException(400, "This isn't ready yet. Post it once it has finished.") | |
| seconds = duration(a) if a.type == "video" else None | |
| if platform == "x" and seconds and seconds > X_VIDEO_SECONDS: | |
| raise HTTPException(400, f"This video is {int(seconds // 60)}:{int(seconds % 60):02d} long; X takes videos " | |
| "up to 2:20. Post it to LinkedIn, or make a shorter one.") | |
| def check_size(m: Media, platform: str) -> None: | |
| """Too big (or, on LinkedIn, too small a video): a clear permanent error before any network call.""" | |
| low, high = SIZE_LIMITS.get((platform, m.kind), (1, 1 << 40)) | |
| label = "X" if platform == "x" else "LinkedIn" | |
| if m.size > high: | |
| raise SocialError(f"This {m.kind} is {m.size / MB:.0f} MB; {label} takes up to {high // MB} MB.", permanent=True) | |
| if m.size < low: | |
| raise SocialError(f"This {m.kind} is too small for {label} (at least {low // 1024} KB).", permanent=True) | |
| def media_for(a: Artifact | None, platform: str) -> Iterator[list[Media]]: | |
| """The files to upload with a post of this artifact (none for text-only ones), size checked. Temporary video | |
| files are deleted when the block ends, also on errors.""" | |
| from app.services.storage import storage | |
| temps: list[str] = [] | |
| try: | |
| out: list[Media] = [] | |
| if a is not None and a.type not in TEXT_ONLY and platform in MEDIA_PLATFORMS: | |
| if a.type == "quote_card": | |
| out = [Media("image", "quote.png", "image/png", data=storage.get_bytes(a.storage_key))] | |
| elif a.type == "carousel": | |
| keys = (a.content_json or {}).get("slide_keys") or [a.storage_key] | |
| out = [Media("image", f"slide-{i + 1}.png", "image/png", data=storage.get_bytes(k)) | |
| for i, k in enumerate(keys[:MAX_IMAGES[platform]])] | |
| elif a.type == "slide_deck": | |
| out = [Media("image", f"slide-{i + 1}.png", "image/png", data=png) | |
| for i, png in enumerate(_slide_images(a, MAX_IMAGES[platform]))] | |
| elif a.type == "upload" and upload_kind(a) == "image": | |
| c = a.content_json or {} | |
| out = [Media("image", c.get("filename") or "image", c.get("content_type") or "image/png", | |
| data=storage.get_bytes(a.storage_key))] | |
| elif a.type == "upload": | |
| path = _temp_file(temps) | |
| with open(path, "wb") as f: | |
| storage.download_to(a.storage_key, f) | |
| out = [Media("video", (a.content_json or {}).get("filename") or "video.mp4", "video/mp4", path=path)] | |
| elif a.type == "video": | |
| path = _temp_file(temps) | |
| with open(path, "wb") as f: | |
| if is_blog2video(a): | |
| _download(a.content_json["video_url"], f) | |
| else: | |
| storage.download_to(a.storage_key, f) | |
| out = [Media("video", "video.mp4", "video/mp4", path=path)] | |
| for m in out: | |
| check_size(m, platform) | |
| yield out | |
| finally: | |
| for path in temps: | |
| with contextlib.suppress(OSError): | |
| os.unlink(path) | |
| def _slide_images(a: Artifact, limit: int) -> list[bytes]: | |
| """A slide deck's first `limit` slides as PNGs, exactly as the deck shows them.""" | |
| from app.infographics.image import ImageUnavailable | |
| from app.slides import export | |
| from app.slides.build import stored | |
| got = stored(a.content_json or {}) | |
| if not got: | |
| raise SocialError("This slide deck has no slides to post.", permanent=True) | |
| try: | |
| return export.render_slide_pngs(*got, limit=limit) | |
| except ImageUnavailable as exc: # no Chrome on this worker right now: worth another try | |
| raise SocialError(f"The slides could not be drawn: {exc}") from exc | |
| def _temp_file(temps: list[str]) -> str: | |
| fd, path = tempfile.mkstemp(prefix="notestack-post-", suffix=".mp4") | |
| os.close(fd) | |
| temps.append(path) | |
| return path | |
| def _download(url: str, f) -> None: | |
| with httpx.stream("GET", url, timeout=120, follow_redirects=True) as resp: | |
| resp.raise_for_status() | |
| for chunk in resp.iter_bytes(1024 * 1024): | |
| f.write(chunk) | |