一. 前言：可以上傳，不代表可以無限上傳 #
拍拍君做了一個小工具，讓使用者上傳 JSONL，再下載筆數摘要。 一個檔案時很乖，五個分頁一起開，記憶體就開始鬧脾氣。 更尷尬的是：換了輸入，下載按鈕卻還在提供上一批的報告。
問題不是按鈕少一個參數，而是我們沒說清楚資料的生命週期。 這次不做可編輯表格，也不做一般效能分析。 目標只有三件事：有限度地接受輸入、明確地準備產物、避免下載過期結果。
如果你要驗證 CSV 欄位與逐列資料，請接著看 Data Editor 工作流。 如果想量 RSS、cache hit 與延遲，則看 Streamlit 效能診斷。 今天先把檔案進出門的規矩訂好。
二. 先分清四種不同的限制 #
| 邊界 | 範例規則 | 真正限制什麼 |
|---|---|---|
| Widget 單檔限制 | 每檔 8 MiB | 一個上傳檔案的大小 |
| 批次 admission | 最多 4 檔、合計 24 MiB | 本次願意處理的輸入集合 |
| Parser 限制 | 每行 64 KiB、總共 5 萬筆 | 處理時的工作量 |
| 服務容量 | 同時 session 數與進程上限 | 整台服務的資源使用 |
這幾層不能互相冒充。
accept_multiple_files=True 不會把單檔限制變成整批配額。
App 裡的檢查也不是反向代理、認證、速率限制或全站 admission controller。
尤其重要：Python 拿到 UploadedFile 時，檔案已經上傳並進入框架管理的記憶體。 後面逐行解析，可以避免再建立整份文字或 DataFrame，卻不能消除最初的上傳緩衝。 App 拒絕處理，不代表那批位元組從來沒佔過 RAM。
三. 安裝：固定實際驗證過的版本 #
以下範例使用 Python 3.12 與 Streamlit 1.66.0。
舊環境若沒有 max_upload_size，請升級或改由設定檔控制，不要直接刪掉檢查。
uv init pypy-file-boundaries
cd pypy-file-boundaries
uv add 'streamlit==1.66.0'
uv add --dev pytest
mkdir -p .streamlit
建立 .streamlit/config.toml：
[server]
maxUploadSize = 8
這是未個別指定時的單檔預設；本篇也在 Widget 明確傳入 max_upload_size=8。
該參數可覆寫設定值，因此設定檔不是不可突破的全站硬上限。
應用內以 8 * 1024 * 1024 bytes 訂定自己的明確契約。
實際部署還要對齊代理與平台限制，並測試邊界檔案。
專案包含 transfer.py、app.py、test_transfer.py。
下面三段 Python 可以分別存成這三個檔案。
四. Admission：先檢查數量與總量，再碰內容 #
我們只接受未壓縮的 .jsonl，每行一個 JSON object。
副檔名只是一道使用者提示，不是內容證明。
這裡不接受 ZIP，也不把瀏覽器回報的 MIME 當成可信鑑定。
建立 transfer.py，先放入以下函式：
# transfer.py
import hashlib
import json
from pathlib import PurePosixPath
MIB = 1024 * 1024
PER_FILE = 8 * MIB
TOTAL = 24 * MIB
MAX_FILES = 4
MAX_LINE = 64 * 1024
MAX_ROWS = 50_000
MAX_EXPORT = 64 * 1024
def admit(files):
if not 1 <= len(files) <= MAX_FILES:
raise ValueError("請選擇 1 到 4 個檔案")
if any(not 0 < f.size <= PER_FILE for f in files):
raise ValueError("單檔必須非空，且不超過 8 MiB")
if sum(f.size for f in files) > TOTAL:
raise ValueError("這一批合計不能超過 24 MiB")
identity = []
for f in files:
if PurePosixPath(f.name).suffix.lower() != ".jsonl":
raise ValueError("只接受未壓縮的 JSONL")
with f.getbuffer() as view:
if view.nbytes != f.size:
raise ValueError("檔案大小資訊不一致")
digest = hashlib.sha256(view).hexdigest()
identity.append((f.name, f.size, digest))
return tuple(identity)
前面先用數量與 metadata 擋掉不願處理的集合，再核對實際 buffer 長度。
getbuffer() 提供 view，這裡沒有主動複製整份 bytes；用完就釋放 view。
不要把 view 存進 session，否則可能讓 buffer 無法關閉或調整。
身分包含檔名、大小與內容 SHA-256。 同名同大小不代表同內容；相同內容重新上傳，則可視為同一份輸入。 這是內容版本識別，不是使用者授權，也不是防毒掃描。
每輪重新雜湊仍有 O(總 bytes) 成本，所以範例刻意維持小配額。 不要為了避免這個成本，改成只比檔名，然後讓過期下載偷偷混進來。
五. Parser：借用指標，也要還回去 #
同一個 file-like object 被讀到結尾，再讀一次就會得到空 bytes。
把 read() 的結果當成「原檔案不見了」，通常只是忘記處理 cursor。
接續貼到 transfer.py：
def reject_constant(value):
raise ValueError(f"不接受 JSON 非有限常數：{value}")
def count_records(file, remaining, skip_blank):
position = file.tell()
rows = 0
try:
file.seek(0)
while True:
line = file.readline(MAX_LINE + 1)
if not line:
break
if len(line) > MAX_LINE:
raise ValueError("單行超過 64 KiB，包含換行 bytes")
text = line.decode("utf-8")
if not text.strip() and skip_blank:
continue
obj = json.loads(text, parse_constant=reject_constant)
if not isinstance(obj, dict):
raise ValueError("每行必須是一個 JSON object")
rows += 1
if rows > remaining:
raise ValueError("整批不能超過 50,000 筆")
return rows
finally:
file.seek(position)
def prepare(files, skip_blank=False):
identity = admit(files)
result = []
total_rows = 0
for number, f in enumerate(files, start=1):
rows = count_records(f, MAX_ROWS - total_rows, skip_blank)
total_rows += rows
result.append({"file": number, "bytes": f.size, "records": rows})
payload = json.dumps(
{"files": result, "total_records": total_rows},
ensure_ascii=False, indent=2,
).encode("utf-8")
if len(payload) > MAX_EXPORT:
raise ValueError("摘要輸出超過 64 KiB")
return (identity, skip_blank), payload
finally 保證成功或失敗都還回原來的位置。
錯誤會中止整批；不回傳半成功報告，不默默跳過壞 JSON。
預設空白行也是錯誤；是否略過空白行，由明確選項決定。
UTF-8 BOM 不在這份契約內；請輸出標準 UTF-8 JSONL。
這不是完整資料 schema，也不是不可信程式的執行沙箱。 每行大小限制降低單次解析工作量，但不能宣稱它完整解決所有 CPU／巢狀結構攻擊。 面向公開網路的大型服務，還需要獨立 worker、執行期限與進程級資源限制。
六. Rerun：舊產物必須跟著輸入失效 #
下載報告不能只用一個 report 變數永遠存著。
我們把產物和 (內容身分, 解析選項) 綁在一起。
任何輸入變更，就先撤下原本的下載按鈕，再等使用者重新準備。
建立完整 app.py：
# app.py
import streamlit as st
from transfer import admit, prepare
st.title("拍拍君 JSONL 摘要")
st.session_state.setdefault("upload_epoch", 0)
def clear_batch():
st.session_state.pop("artifact", None)
st.session_state["upload_epoch"] += 1
st.button("清除這一批", on_click=clear_batch)
files = st.file_uploader(
"上傳 JSONL：最多 4 檔、合計 24 MiB",
type=["jsonl"], accept_multiple_files=True,
max_upload_size=8,
key=f"uploads_{st.session_state['upload_epoch']}",
)
skip_blank = st.checkbox("略過空白行", value=False)
if not files:
st.session_state.pop("artifact", None)
st.info("請先選擇檔案")
st.stop()
try:
source = (admit(files), skip_blank)
except ValueError as exc:
st.session_state.pop("artifact", None)
st.error(str(exc))
st.stop()
artifact = st.session_state.get("artifact")
if artifact is not None and artifact["source"] != source:
st.session_state.pop("artifact", None)
artifact = None
if st.button("準備本批摘要", key="prepare"):
st.session_state.pop("artifact", None)
artifact = None
try:
source, payload = prepare(files, skip_blank)
except (ValueError, RecursionError):
st.error("內容不符合契約：請檢查 UTF-8、JSON、行長與筆數")
else:
artifact = {"source": source, "data": payload}
st.session_state["artifact"] = artifact
if artifact is not None:
st.download_button(
"下載本批摘要", data=artifact["data"],
file_name="pypy-summary.json", mime="application/json",
key="download", on_click="ignore",
)
執行 uv run streamlit run app.py，上傳例如 {"name":"拍拍醬"} 的單行檔案。
再換同名、同大小但內容不同的檔案，舊下載應該消失。
更改空白行選項也一樣：產物對應的是內容與設定，不只是某個 Widget。
「清除」用新的 uploader key 重建選擇狀態，不直接寫入 uploader 的 session value。 這是清除目前 App 引用與 UI 的方式，不是立刻歸零整個進程 RSS 的保證。 Session State 跟 WebSocket session 綁定，也不是永久保存或跨使用者共享資料庫。
七. 下載：先準備，再點擊，不把成功訊號搞反 #
這個範例刻意先產生不可變 bytes，使用者再下載。
on_click="ignore" 只是不因下載按鈕本身觸發 rerun。
它不阻止其他 Widget 的 rerun，也不保證瀏覽器已把檔案寫到磁碟。
因此不要在點擊後立刻宣告「使用者已成功收檔」。
換了輸入後，我們撤掉後續畫面的舊按鈕；這不是撤銷已發出的 bytes 或舊 URL。 同時發生的瀏覽器事件與 HTTP 請求，也不能靠 UI 失效邏輯變成原子交易。 需要即時撤銷與逐次權限檢查時，應使用另外實作授權的下載端點。
新版 download_button 也支援 callable 延後生成，但延後不等於無緩衝串流。
本篇小摘要不需要它；避免為了一個 1 KiB 報告，額外引入背景執行與狀態競爭。
不要跨 rerun 保存 with open(...) 已關閉的 handle；先確定物件有效期再交給元件。
八. 記憶體帳：24 MiB 配額不是 24 MiB RSS #
最少要盤點這些可能同時存在的資料：
- 框架管理的上傳 bytes 與 UploadedFile 參照。
- Parser 暫時持有的一行 bytes、解碼字串與 JSON object。
- Session 保存的摘要 bytes，以及框架的下載媒體資料。
- 其他 session、Python 執行環境、套件與配置器保留的容量。
範例只把小摘要存入 session，沒有保留完整解析結果或原始 bytes 副本。
但這不是固定倍數公式：BytesIO、字串與 JSON object 的成本不同。
getvalue() 也不能被武斷宣稱「一定零拷貝」或「一定多一份」；別把實作最佳化當契約。
粗估容量可先用「並行 session × 每批輸入預算」算輸入層，再另外留解析、輸出與基礎負載餘裕。 這是規劃，不是量測結果，更不是框架提供的全站硬配額。 真實壓測必須涵蓋多分頁、連續換檔、慢下載與中斷連線，而不只是單人跑一次。
如果需求是數 GB 檔案，拍拍君會改成物件儲存直傳、後端排程處理與到期下載連結。 仍要限制授權、物件大小、存活時間與清理工作；預簽 URL 本身也不能公開亂貼。 Streamlit 留作控制面板，不必硬扛所有檔案流量。
九. 用測試守住指標與過期資料邊界 #
建立 test_transfer.py，先驗證最容易被 rerun 踩壞的契約：
# test_transfer.py
from io import BytesIO
import json
import pytest
from transfer import admit, prepare
def upload(data, name="pypy.jsonl"):
f = BytesIO(data)
f.name, f.size = name, len(data)
return f
def test_cursor_and_summary():
f = upload(b'{"n":1}\n{"n":2}\n')
f.seek(3)
_, payload = prepare([f])
assert f.tell() == 3
assert json.loads(payload)["total_records"] == 2
def test_same_name_size_different_content():
assert admit([upload(b'{"n":1}\n')]) != admit([upload(b'{"n":2}\n')])
def test_failure_restores_cursor():
f = upload(b'not-json\n')
f.seek(2)
with pytest.raises(ValueError):
prepare([f])
assert f.tell() == 2
執行 uv run pytest -q。
再補單檔與總量上界、空檔、過長單行、跨檔總筆數、無效 UTF-8、NaN 和空白行政策。
UI 邏輯則確認換檔／切選項／清除會移除舊 artifact，兩個 session 不共用產物。
本篇實際以 Streamlit 1.66.0 的 UploadedFile 及上述契約做自動化檢查。 AppTest 的注入式 server 邏輯測試不等於瀏覽器真的完成 multipart 上傳或下載。 正式環境仍需另測代理限制、網路失敗、HTTP 授權與瀏覽器收檔。
結語：下載按鈕背後，也需要所有權 #
小工具不一定要很龐大，但它得知道哪批資料由誰持有、何時失效。 單檔限制管不了總量；逐行處理消不掉初始上傳緩衝；按下下載也不代表成功收檔。 把配額、指標與產物身分寫成契約，才不會讓好用的小工具變成記憶體與舊資料的抽獎機。
拍拍君的選擇很簡單：小檔案做得清楚，大檔案交給合適的傳輸架構。