Codex Release Bot
120 lines
3.9 KiB
Python
120 lines
3.9 KiB
Python
"""Nexus — Global Search API Route
|
|
Cross-entity search across servers, scripts, credentials, and schedules.
|
|
All operations require JWT authentication.
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import APIRouter, Depends, Query
|
|
|
|
from server.api.dependencies import get_db
|
|
from server.api.auth_jwt import get_current_admin
|
|
from server.domain.models import Admin, Server, Script, DbCredential, PushSchedule
|
|
|
|
from sqlalchemy import or_
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
router = APIRouter(prefix="/api/search", tags=["search"])
|
|
|
|
logger = logging.getLogger("nexus.search")
|
|
|
|
|
|
def _escape_like(s: str) -> str:
|
|
"""Escape LIKE wildcard characters in user input.
|
|
|
|
Without this, a user searching for ``%`` matches all rows and ``_``
|
|
matches any single character — a form of information disclosure.
|
|
"""
|
|
return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
|
|
|
|
|
|
@router.get("/", response_model=dict)
|
|
async def global_search(
|
|
q: str = Query(..., min_length=1, max_length=100, description="Search query"),
|
|
admin: Admin = Depends(get_current_admin),
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Global search across servers, scripts, credentials, and schedules.
|
|
|
|
Returns top 10 matches per entity type.
|
|
"""
|
|
from sqlalchemy import select
|
|
|
|
pattern = f"%{_escape_like(q)}%"
|
|
results = {"servers": [], "scripts": [], "credentials": [], "schedules": []}
|
|
|
|
# Servers
|
|
try:
|
|
stmt = select(Server).where(
|
|
or_(
|
|
Server.name.ilike(pattern, escape="\\"),
|
|
Server.domain.ilike(pattern, escape="\\"),
|
|
Server.description.ilike(pattern, escape="\\"),
|
|
Server.category.ilike(pattern, escape="\\"),
|
|
)
|
|
).limit(10)
|
|
res = await db.execute(stmt)
|
|
for s in res.scalars().all():
|
|
results["servers"].append({
|
|
"id": s.id, "name": s.name, "domain": s.domain,
|
|
"category": s.category, "is_online": s.is_online,
|
|
})
|
|
except Exception:
|
|
logger.warning("Search failed for 'servers' entity", exc_info=True)
|
|
|
|
# Scripts
|
|
try:
|
|
stmt = select(Script).where(
|
|
or_(
|
|
Script.name.ilike(pattern, escape="\\"),
|
|
Script.category.ilike(pattern, escape="\\"),
|
|
Script.description.ilike(pattern, escape="\\"),
|
|
)
|
|
).limit(10)
|
|
res = await db.execute(stmt)
|
|
for s in res.scalars().all():
|
|
results["scripts"].append({
|
|
"id": s.id, "name": s.name, "category": s.category,
|
|
})
|
|
except Exception:
|
|
logger.warning("Search failed for 'scripts' entity", exc_info=True)
|
|
|
|
# Credentials
|
|
try:
|
|
stmt = select(DbCredential).where(
|
|
or_(
|
|
DbCredential.name.ilike(pattern, escape="\\"),
|
|
DbCredential.host.ilike(pattern, escape="\\"),
|
|
)
|
|
).limit(10)
|
|
res = await db.execute(stmt)
|
|
for c in res.scalars().all():
|
|
results["credentials"].append({
|
|
"id": c.id, "name": c.name, "db_type": c.db_type, "host": c.host,
|
|
})
|
|
except Exception:
|
|
logger.warning("Search failed for 'credentials' entity", exc_info=True)
|
|
|
|
# Schedules
|
|
try:
|
|
stmt = select(PushSchedule).where(
|
|
or_(
|
|
PushSchedule.name.ilike(pattern, escape="\\"),
|
|
PushSchedule.source_path.ilike(pattern, escape="\\"),
|
|
)
|
|
).limit(10)
|
|
res = await db.execute(stmt)
|
|
for s in res.scalars().all():
|
|
results["schedules"].append({
|
|
"id": s.id, "name": s.name, "cron_expr": s.cron_expr, "enabled": s.enabled,
|
|
})
|
|
except Exception:
|
|
logger.warning("Search failed for 'schedules' entity", exc_info=True)
|
|
|
|
# Total count
|
|
total = sum(len(v) for v in results.values())
|
|
results["total"] = total
|
|
results["query"] = q
|
|
|
|
return results
|