api_users. Route handler or application helper. This docstring was expanded to make future maintenance easier. Returns: Varies.
()
| 13657 | |
| 13658 | if mime == "image/gif": |
| 13659 | filename = f"gif_{fid}.gif" |
| 13660 | elif mime == "video/mp4": |
| 13661 | filename = f"gif_{fid}.mp4" |
| 13662 | elif mime == "video/webm": |
| 13663 | filename = f"gif_{fid}.webm" |
| 13664 | else: |
| 13665 | filename = f"file_{fid}" |
| 13666 | |
| 13667 | with open(stored_path, "wb") as f: |
| 13668 | while True: |
| 13669 | chunk = resp.read(64 * 1024) |
| 13670 | if not chunk: |
| 13671 | break |
| 13672 | size += len(chunk) |
| 13673 | if size > max_bytes: |
| 13674 | raise ValueError("GIF too large") |
| 13675 | f.write(chunk) |
| 13676 | |
| 13677 | conn.execute("UPDATE dm_files SET stored_path=?, size=?, mime=?, filename=? WHERE id=?", |
| 13678 | (stored_path, int(size), mime, filename, fid)) |
| 13679 | conn.execute("UPDATE dm_messages SET has_file=1 WHERE id=?", (dm_id,)) |
| 13680 | return {"id": fid, "mime": mime, "filename": filename, "size": int(size), "stored_path": stored_path} |
| 13681 | |
| 13682 | def _dm_store_remote_gif(dm_id: int, gif_url: str) -> Dict[str, str]: |
| 13683 | """Download and store a remote GIF (or short MP4) as a DM attachment. |
| 13684 | |
| 13685 | This enables an Instagram-like GIF picker that sends a URL rather than an uploaded file. |
| 13686 | We keep this intentionally strict for safety: |
| 13687 | - only http(s) |
| 13688 | - size cap |
| 13689 | """ |
| 13690 | gif_url = (gif_url or "").strip() |
| 13691 | if not gif_url: |
| 13692 | raise ValueError("Empty gif url") |
| 13693 | |
| 13694 | try: |
| 13695 | u = urllib.parse.urlparse(gif_url) |
| 13696 | except Exception as e: |
| 13697 | raise ValueError("Bad gif url") from e |
| 13698 | if u.scheme not in ("http", "https"): |
| 13699 | raise ValueError("GIF url must be http(s)") |
| 13700 | |
| 13701 | # Insert DB row first so we have a deterministic stored_path. |
| 13702 | conn = db_connect() |
| 13703 | cur = conn.cursor() |
| 13704 | cur.execute( |
| 13705 | "INSERT INTO dm_files(dm_id, filename, mime, stored_path, size, created_at) VALUES(?,?,?,?,?,?)", |
| 13706 | (dm_id, "gif.gif", "image/gif", "PENDING", 0, now_z()) |
| 13707 | ) |
| 13708 | fid = cur.lastrowid |
| 13709 | stored_path = os.path.join(DM_FILES_DIR, f"dm_file_{fid}.bin") |
| 13710 | conn.commit() |
| 13711 | conn.close() |
| 13712 | |
| 13713 | # Download with a cap to avoid abuse. |
| 13714 | max_bytes = CHAT_MAX_BYTES # 33 MB |
nothing calls this directly
no test coverage detected