api_dm_since. Chat/group feature helper or route handler. This docstring was expanded to make future maintenance easier. Args: peer: Parameter. Returns: Varies.
(peer: str)
| 13728 | with _safe_urlopen(req, timeout=12) as resp: |
| 13729 | ctype = (resp.headers.get("Content-Type") or "").split(";")[0].strip().lower() |
| 13730 | if ctype: |
| 13731 | mime = ctype |
| 13732 | # Allow gif and mp4/webm (many "GIFs" are delivered as mp4) |
| 13733 | if not (mime == "image/gif" or mime.startswith("video/")): |
| 13734 | # Some CDNs mislabel; fall back to extension check |
| 13735 | path = (u.path or "").lower() |
| 13736 | if not (path.endswith(".gif") or path.endswith(".mp4") or path.endswith(".webm")): |
| 13737 | raise ValueError(f"Unsupported GIF mime: {mime}") |
| 13738 | if path.endswith(".mp4"): |
| 13739 | mime = "video/mp4" |
| 13740 | elif path.endswith(".webm"): |
| 13741 | mime = "video/webm" |
| 13742 | else: |
| 13743 | mime = "image/gif" |
| 13744 | |
| 13745 | if mime == "image/gif": |
| 13746 | filename = f"gif_{fid}.gif" |
| 13747 | elif mime == "video/mp4": |
| 13748 | filename = f"gif_{fid}.mp4" |
| 13749 | elif mime == "video/webm": |
| 13750 | filename = f"gif_{fid}.webm" |
| 13751 | else: |
| 13752 | filename = f"file_{fid}" |
| 13753 | |
| 13754 | with open(stored_path, "wb") as f: |
| 13755 | while True: |
| 13756 | chunk = resp.read(64 * 1024) |
| 13757 | if not chunk: |
| 13758 | break |
| 13759 | size += len(chunk) |
| 13760 | if size > max_bytes: |
| 13761 | raise ValueError("GIF too large") |
| 13762 | f.write(chunk) |
| 13763 | except urllib.error.HTTPError as e: |
| 13764 | # Clean up DB row on failure. |
| 13765 | try: |
| 13766 | c = db_connect() |
| 13767 | c.execute("DELETE FROM dm_files WHERE id=?", (fid,)) |
| 13768 | c.execute("UPDATE dm_messages SET has_file=0 WHERE id=?", (dm_id,)) |
| 13769 | c.commit() |
| 13770 | c.close() |
| 13771 | except Exception: |
| 13772 | pass |
| 13773 | raise ValueError(f"GIF download failed ({e.code})") from e |
| 13774 | except Exception: |
| 13775 | try: |
| 13776 | if os.path.exists(stored_path): |
| 13777 | os.remove(stored_path) |
| 13778 | except Exception: |
| 13779 | pass |
| 13780 | try: |
| 13781 | c = db_connect() |
| 13782 | c.execute("DELETE FROM dm_files WHERE id=?", (fid,)) |
| 13783 | c.execute("UPDATE dm_messages SET has_file=0 WHERE id=?", (dm_id,)) |
| 13784 | c.commit() |
| 13785 | c.close() |
| 13786 | except Exception: |
| 13787 | pass |
nothing calls this directly
no test coverage detected