120 lines
4.4 KiB
Python
120 lines
4.4 KiB
Python
from fastapi import FastAPI
|
|
from fastapi.responses import HTMLResponse, JSONResponse
|
|
import httpx, asyncio, re, html as html_lib
|
|
from xml.etree import ElementTree as ET
|
|
|
|
app = FastAPI(redirect_slashes=False)
|
|
|
|
KIWIX_URL = "http://kiwix:8080"
|
|
PDFS_URL = "http://pdfs:8000"
|
|
FREQS_URL = "http://freqs:8000"
|
|
INV_URL = "http://inventory:8000"
|
|
|
|
def text_match(item, q, fields):
|
|
ql = q.lower()
|
|
for f in fields:
|
|
v = item.get(f) or ""
|
|
if ql in str(v).lower():
|
|
return True
|
|
return False
|
|
|
|
async def search_kiwix(client, q, limit=30):
|
|
"""Use OPDS catalog search to find matching ZIM books."""
|
|
try:
|
|
r = await client.get(
|
|
f"{KIWIX_URL}/library/catalog/v2/entries",
|
|
params={"q": q, "count": limit},
|
|
timeout=8.0
|
|
)
|
|
if r.status_code != 200:
|
|
return []
|
|
text = r.text
|
|
results = []
|
|
for entry in re.findall(r"<entry>(.*?)</entry>", text, re.DOTALL)[:limit]:
|
|
t = re.search(r"<title[^>]*>(.*?)</title>", entry, re.DOTALL)
|
|
name = re.search(r"<name>(.*?)</name>", entry, re.DOTALL)
|
|
s = re.search(r"<summary[^>]*>(.*?)</summary>", entry, re.DOTALL)
|
|
# Find the content link
|
|
content = re.search(r"<link[^>]*href=\"(/library/content/[^\"]+)\"", entry)
|
|
if t and content:
|
|
results.append({
|
|
"source": "Library",
|
|
"title": html_lib.unescape(t.group(1).strip()),
|
|
"url": content.group(1),
|
|
"snippet": re.sub(r"<[^>]+>", "", html_lib.unescape(s.group(1)))[:300] if s else "",
|
|
"context": name.group(1).strip() if name else "",
|
|
})
|
|
return results
|
|
except Exception:
|
|
return []
|
|
|
|
async def search_pdfs(client, q, limit=20):
|
|
try:
|
|
r = await client.get(f"{PDFS_URL}/api/list", timeout=5.0)
|
|
if r.status_code != 200: return []
|
|
items = r.json()
|
|
matches = [i for i in items if text_match(i, q, ["title", "tag", "filename"])][:limit]
|
|
return [{
|
|
"source": "PDFs",
|
|
"title": i["title"],
|
|
"url": f"/pdfs/view/{i['id']}",
|
|
"snippet": f"Tag: {i.get('tag') or 'none'}",
|
|
"context": i.get("tag", ""),
|
|
} for i in matches]
|
|
except: return []
|
|
|
|
async def search_freqs(client, q, limit=20):
|
|
try:
|
|
r = await client.get(f"{FREQS_URL}/api/list", timeout=5.0)
|
|
if r.status_code != 200: return []
|
|
items = r.json()
|
|
matches = [i for i in items if text_match(i, q, ["name","freq","band","mode","tone","tag","description"])][:limit]
|
|
return [{
|
|
"source": "Frequencies",
|
|
"title": f"{i.get('freq','')} - {i.get('name','')}",
|
|
"url": "/freqs/",
|
|
"snippet": i.get("description") or f"{i.get('band','')} {i.get('mode','')} {i.get('tone','')}".strip(),
|
|
"context": i.get("band", ""),
|
|
} for i in matches]
|
|
except: return []
|
|
|
|
async def search_inventory(client, q, limit=20):
|
|
try:
|
|
r = await client.get(f"{INV_URL}/api/list", timeout=5.0)
|
|
if r.status_code != 200: return []
|
|
items = r.json()
|
|
matches = [i for i in items if text_match(i, q, ["name","category","location","notes"])][:limit]
|
|
return [{
|
|
"source": "Inventory",
|
|
"title": i["name"],
|
|
"url": "/inventory/",
|
|
"snippet": f"Qty: {i.get('quantity',0)} {i.get('unit','')} @ {i.get('location','?')} | {i.get('notes','')}".strip(),
|
|
"context": i.get("category", ""),
|
|
} for i in matches]
|
|
except: return []
|
|
|
|
@app.get("/api/search")
|
|
async def search(q: str = ""):
|
|
q = q.strip()
|
|
if not q:
|
|
return JSONResponse({"results": {}, "total": 0})
|
|
async with httpx.AsyncClient() as client:
|
|
zims, pdfs, freqs, inv = await asyncio.gather(
|
|
search_kiwix(client, q),
|
|
search_pdfs(client, q),
|
|
search_freqs(client, q),
|
|
search_inventory(client, q),
|
|
)
|
|
grouped = {}
|
|
if zims: grouped["Library"] = zims
|
|
if pdfs: grouped["PDFs"] = pdfs
|
|
if freqs: grouped["Frequencies"] = freqs
|
|
if inv: grouped["Inventory"] = inv
|
|
total = sum(len(v) for v in grouped.values())
|
|
return {"results": grouped, "total": total}
|
|
|
|
@app.get("/", response_class=HTMLResponse)
|
|
def root():
|
|
with open("/static/index.html") as f:
|
|
return f.read()
|