profiler. Route handler or application helper. This docstring was expanded to make future maintenance easier. Returns: Varies.
()
| 15896 | LIMIT 80 |
| 15897 | """, |
| 15898 | (gid, after_id), |
| 15899 | ).fetchall() |
| 15900 | |
| 15901 | items = [] |
| 15902 | for r in rows: |
| 15903 | v = conn.execute("SELECT id, mime FROM group_voice WHERE gm_id=? ORDER BY id DESC LIMIT 1", (r["id"],)).fetchone() |
| 15904 | has_voice = bool(v) |
| 15905 | try: |
| 15906 | body_txt = aesgcm_decrypt_text(r["body_enc"]) |
| 15907 | except Exception: |
| 15908 | body_txt = "[decrypt failed]" |
| 15909 | items.append({ |
| 15910 | "id": int(r["id"]), |
| 15911 | "sender": r["sender"], |
| 15912 | "kind": "voice" if has_voice else "text", |
| 15913 | "body": body_txt if not has_voice else "", |
| 15914 | "created_at_local": _local_time_str(r["created_at"]), |
| 15915 | "created_at_utc": r["created_at"], |
| 15916 | "voice_id": v["id"] if v else None, |
| 15917 | "voice_mime": v["mime"] if v else "audio/webm", |
| 15918 | "voice_url": (url_for("group_voice_stream", vid=v["id"]) if v else None), |
| 15919 | }) |
| 15920 | return jsonify(ok=True, items=items) |
| 15921 | finally: |
| 15922 | try: |
| 15923 | conn.close() |
| 15924 | except Exception: |
| 15925 | pass |
| 15926 | |
| 15927 | @app.route("/group/voice/<int:vid>") |
| 15928 | @login_required |
| 15929 | def group_voice_stream(vid: int): |
| 15930 | """Stream a group voice message.""" |
| 15931 | me = current_user() |
| 15932 | conn = db_connect() |
| 15933 | row = conn.execute(""" |
| 15934 | SELECT v.id, v.mime, v.stored_path, m.group_id |
| 15935 | FROM group_voice v |
| 15936 | JOIN group_messages m ON m.id=v.gm_id |
| 15937 | WHERE v.id=? |
| 15938 | """, (vid,)).fetchone() |
| 15939 | conn.close() |
| 15940 | if not row: |
| 15941 | abort(404) |
| 15942 | if not group_role(int(row["group_id"]), me): |
| 15943 | abort(403) |
| 15944 | if not _chat_lock_is_unlocked(me, "group", int(row["group_id"])): |
| 15945 | return redirect(url_for("group_chat", gid=int(row["group_id"]))) |
| 15946 | |
| 15947 | sp = row["stored_path"] |
| 15948 | if not sp or not os.path.exists(sp): |
| 15949 | abort(404) |
| 15950 | |
| 15951 | mime = row["mime"] or "audio/webm" |
nothing calls this directly
no test coverage detected