1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
|
"""Session management API routes."""
from __future__ import annotations
import logging
from typing import Any
from starlette.requests import Request
from starlette.responses import JSONResponse
from ..sessions import SessionStore
from ..tools.pdf import extract_pdf_text
logger = logging.getLogger("ollama_web.sessions")
def _get_store(request: Request) -> SessionStore:
return request.app.state.session_store
async def list_sessions(request: Request) -> JSONResponse:
store = _get_store(request)
return JSONResponse({"sessions": store.list_sessions()})
async def create_session(request: Request) -> JSONResponse:
store = _get_store(request)
body = await request.json() if await request.body() else {}
title = body.get("title")
session = store.create(title=title)
return JSONResponse(session)
async def get_session(request: Request) -> JSONResponse:
store = _get_store(request)
session_id = request.path_params["session_id"]
session = store.get(session_id)
if session is None:
return JSONResponse({"error": "Session not found"}, status_code=404)
return JSONResponse(session)
async def delete_session(request: Request) -> JSONResponse:
store = _get_store(request)
session_id = request.path_params["session_id"]
ok = store.delete(session_id)
if not ok:
return JSONResponse({"error": "Session not found"}, status_code=404)
return JSONResponse({"deleted": True})
async def upload_file(request: Request) -> JSONResponse:
store = _get_store(request)
session_id = request.path_params["session_id"]
session = store.get(session_id)
if session is None:
return JSONResponse({"error": "Session not found"}, status_code=404)
max_size = request.app.state.settings.max_upload_mb * 1024 * 1024
try:
form = await request.form()
except Exception as exc: # noqa: BLE001
logger.error("failed to parse multipart form: %s", exc)
return JSONResponse({"error": f"Invalid form data: {exc}"}, status_code=400)
files = form.getlist("files") if hasattr(form, "getlist") else [form.get("files")]
results: list[dict[str, Any]] = []
for upload in files:
if upload is None:
continue
try:
data = await upload.read()
except Exception as exc: # noqa: BLE001
results.append({"name": getattr(upload, "filename", "unknown"), "error": str(exc)})
continue
if len(data) > max_size:
results.append(
{
"name": getattr(upload, "filename", "unknown"),
"error": f"File exceeds {request.app.state.settings.max_upload_mb} MB limit",
}
)
continue
name = getattr(upload, "filename", "uploaded-file") or "uploaded-file"
content_type = getattr(upload, "content_type", None) or "application/octet-stream"
text = ""
if content_type == "application/pdf" or name.lower().endswith(".pdf"):
text = extract_pdf_text(data)
else:
# Try decode as text for text-like uploads.
try:
text = data.decode("utf-8")
except UnicodeDecodeError:
try:
text = data.decode("cp932")
except UnicodeDecodeError:
text = data.decode("utf-8", errors="replace")
chat_file = store.add_file(
session_id=session_id,
name=name,
data=data,
mime=content_type,
text=text,
source="upload",
)
results.append(chat_file.to_dict())
return JSONResponse({"files": results})
async def delete_file(request: Request) -> JSONResponse:
store = _get_store(request)
session_id = request.path_params["session_id"]
file_id = request.path_params["file_id"]
ok = store.remove_file(session_id, file_id)
if not ok:
return JSONResponse({"error": "File not found"}, status_code=404)
return JSONResponse({"deleted": True})
|
