feat: file search, type/tag filters, and pagination

Add tsvector full-text search, type filter (image/document/text/
spreadsheet/archive), tag filter dropdown, and pagination (50/page).
Replace folder button bar with compact dropdown. All filters combine
and carry through sort and pagination links.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-03-03 01:05:10 +00:00
parent 41b974c804
commit 75b055299a
2 changed files with 173 additions and 33 deletions

View File

@@ -144,12 +144,27 @@ SORT_OPTIONS = {
"date_asc": "created_at ASC",
}
# Map type filter values to mime prefixes/patterns
TYPE_FILTERS = {
"image": "image/%",
"document": "application/pdf",
"text": "text/%",
"spreadsheet": "application/vnd.%sheet%",
"archive": "application/%zip%",
}
PER_PAGE = 50
@router.get("/")
async def list_files(
request: Request,
folder: Optional[str] = None,
sort: Optional[str] = None,
q: Optional[str] = None,
file_type: Optional[str] = None,
tag: Optional[str] = None,
page: int = 1,
context_type: Optional[str] = None,
context_id: Optional[str] = None,
db: AsyncSession = Depends(get_db),
@@ -163,38 +178,86 @@ async def list_files(
order_by = SORT_OPTIONS.get(sort, "storage_path ASC")
# Normalize folder param from form: " " (space) = root, "" = all, None = all
if folder is not None:
if folder.strip() == "" and folder != "":
# Space-only value means root folder
folder = ""
elif folder.strip() == "":
# Empty string means "all folders" (no filter)
folder = None
# Build dynamic WHERE clauses
where_clauses = ["f.is_deleted = false"]
params = {}
if context_type and context_id:
# Files attached to a specific entity
result = await db.execute(text(f"""
where_clauses.append("fm.context_type = :ct AND fm.context_id = :cid")
params["ct"] = context_type
params["cid"] = context_id
if folder is not None:
if folder == "":
where_clauses.append("f.storage_path NOT LIKE '%/%'")
else:
where_clauses.append("f.storage_path LIKE :prefix")
params["prefix"] = folder + "/%"
if q and q.strip():
search_terms = q.strip().split()
tsquery = " & ".join(f"{t}:*" for t in search_terms)
where_clauses.append("f.search_vector @@ to_tsquery('english', :tsquery)")
params["tsquery"] = tsquery
if file_type and file_type in TYPE_FILTERS:
where_clauses.append("f.mime_type LIKE :mime_pattern")
params["mime_pattern"] = TYPE_FILTERS[file_type]
if tag and tag.strip():
where_clauses.append(":tag = ANY(f.tags)")
params["tag"] = tag.strip()
where_sql = " AND ".join(where_clauses)
# Count total for pagination
if context_type and context_id:
count_sql = f"""
SELECT COUNT(*) FROM files f
JOIN file_mappings fm ON fm.file_id = f.id
WHERE {where_sql}
"""
else:
count_sql = f"SELECT COUNT(*) FROM files f WHERE {where_sql}"
total = (await db.execute(text(count_sql), params)).scalar()
# Paginate
if page < 1:
page = 1
offset = (page - 1) * PER_PAGE
total_pages = max(1, (total + PER_PAGE - 1) // PER_PAGE)
# Query
if context_type and context_id:
query_sql = f"""
SELECT f.*, fm.context_type, fm.context_id
FROM files f
JOIN file_mappings fm ON fm.file_id = f.id
WHERE f.is_deleted = false AND fm.context_type = :ct AND fm.context_id = :cid
WHERE {where_sql}
ORDER BY {order_by}
"""), {"ct": context_type, "cid": context_id})
elif folder is not None:
if folder == "":
# Root folder: files with no directory separator in storage_path
result = await db.execute(text(f"""
SELECT * FROM files
WHERE is_deleted = false AND storage_path NOT LIKE '%/%'
ORDER BY {order_by}
"""))
else:
# Specific folder: storage_path starts with folder/
result = await db.execute(text(f"""
SELECT * FROM files
WHERE is_deleted = false AND storage_path LIKE :prefix
ORDER BY {order_by}
"""), {"prefix": folder + "/%"})
LIMIT :lim OFFSET :off
"""
else:
# All files
result = await db.execute(text(f"""
SELECT * FROM files
WHERE is_deleted = false
query_sql = f"""
SELECT f.* FROM files f
WHERE {where_sql}
ORDER BY {order_by}
"""))
LIMIT :lim OFFSET :off
"""
params["lim"] = PER_PAGE
params["off"] = offset
result = await db.execute(text(query_sql), params)
items = [dict(r._mapping) for r in result]
# Add derived folder field for display
@@ -202,10 +265,23 @@ async def list_files(
dirname = os.path.dirname(item["storage_path"])
item["folder"] = dirname if dirname else "/"
# Get all unique tags for the tag filter dropdown
tag_result = await db.execute(text(
"SELECT DISTINCT unnest(tags) AS tag FROM files WHERE is_deleted = false AND tags IS NOT NULL ORDER BY tag"
))
all_tags = [r._mapping["tag"] for r in tag_result]
return templates.TemplateResponse("files.html", {
"request": request, "sidebar": sidebar, "items": items,
"folders": folders, "current_folder": folder,
"current_sort": sort or "path",
"current_q": q or "",
"current_type": file_type or "",
"current_tag": tag or "",
"current_page": page,
"total_pages": total_pages,
"total_files": total,
"all_tags": all_tags,
"sync_result": sync_result,
"context_type": context_type or "",
"context_id": context_id or "",