Build a DM message view dict (same logic as chat_with) for one dm_messages row.
(conn, r, me: str)
| 13333 | c.execute("UPDATE dm_voice SET stored_path=? WHERE id=?", (stored_path, vid)) |
| 13334 | c.commit() |
| 13335 | c.close() |
| 13336 | |
| 13337 | _bg_encrypt_file(tmp_plain, stored_path, finalize) |
| 13338 | |
| 13339 | conn.execute("UPDATE dm_messages SET has_voice=1 WHERE id=?", (dm_id,)) |
| 13340 | conn.commit() |
| 13341 | conn.close() |
| 13342 | return vid, mime |
| 13343 | |
| 13344 | |
| 13345 | def _dm_store_file(dm_id: int, file_storage) -> Dict[str, str]: |
| 13346 | """Store a regular DM attachment (plaintext on disk).""" |
| 13347 | filename, mime = _validate_chat_file_upload(file_storage) |
| 13348 | |
| 13349 | conn = db_connect() |
| 13350 | cur = conn.cursor() |
| 13351 | cur.execute( |
| 13352 | "INSERT INTO dm_files(dm_id, filename, mime, stored_path, size, created_at) VALUES(?,?,?,?,?,?)", |
| 13353 | (dm_id, filename, mime, "PENDING", 0, now_z()) |
| 13354 | ) |
| 13355 | fid = cur.lastrowid |
| 13356 | |
| 13357 | stored_path = os.path.join(DM_FILES_DIR, f"dm_file_{fid}.bin") |
| 13358 | |
| 13359 | # Save upload bytes directly (no background encryption). |
| 13360 | size = _save_plain_upload(file_storage, stored_path) |
| 13361 | if int(size) > CHAT_MAX_BYTES: |
| 13362 | try: |
| 13363 | os.remove(stored_path) |
| 13364 | except Exception: |
| 13365 | pass |
| 13366 | conn.execute("DELETE FROM dm_files WHERE id=?", (fid,)) |
| 13367 | conn.execute("UPDATE dm_messages SET has_file=0 WHERE id=?", (dm_id,)) |
| 13368 | conn.commit() |
| 13369 | conn.close() |
| 13370 | raise ValueError("too_large") |
| 13371 | |
| 13372 | conn.execute("UPDATE dm_files SET stored_path=?, size=? WHERE id=?", (stored_path, int(size), fid)) |
| 13373 | conn.execute("UPDATE dm_messages SET has_file=1 WHERE id=?", (dm_id,)) |
no test coverage detected