sabo
46a2f2272b
Accept ZIP/single help files, process with help_processor into OUTPUT_DIR+Postgres, pollable jobs, optional RESCAN_TOKEN. Co-authored-by: Cursor <cursoragent@cursor.com>
352 lines
11 KiB
Python
352 lines
11 KiB
Python
"""
|
||
FastAPI webapp за RIP Help System.
|
||
|
||
Endpoint-и:
|
||
GET / — viewer HTML (?prefix=&home=)
|
||
GET /images/{filename} — картинки от OUTPUT_DIR/images
|
||
GET /home-image — Bairaci / HOME_IMAGE
|
||
GET /api/sections — JSON секции (?prefix=)
|
||
GET /api/search — търсене (?q=&prefix=)
|
||
GET /api/section/{code} — една секция с пълен текст
|
||
POST /api/keywords/{code} — обновяване на keywords
|
||
GET /healthz — health + OUTPUT_DIR статус
|
||
|
||
Env:
|
||
HELP_DB_CONN — libpq Postgres
|
||
OUTPUT_DIR — локален/mounted път към секциите (default: share)
|
||
HOME_IMAGE — път към home картинка
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
import psycopg2
|
||
from fastapi import FastAPI, HTTPException, Query, Request
|
||
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
|
||
from fastapi.templating import Jinja2Templates
|
||
from pydantic import BaseModel
|
||
|
||
# ──────────────────────────────────────────────
|
||
# Configuration
|
||
# ──────────────────────────────────────────────
|
||
|
||
DEFAULT_SHARE = "/mnt/mssql/share/RIP/RIP_Help_Source/Output"
|
||
|
||
CONN_STR = os.getenv(
|
||
"HELP_DB_CONN",
|
||
"host=192.168.88.18 port=5432 dbname=rip_help_system user=sa password=Parola~12345!!!",
|
||
)
|
||
OUTPUT_DIR = Path(os.getenv("OUTPUT_DIR", DEFAULT_SHARE))
|
||
HOME_IMAGE = os.getenv("HOME_IMAGE") # absolute path or None
|
||
|
||
_IMG_PLACEHOLDER_RE = re.compile(r"\[IMG:\s*([^\]]+?)\s*\]")
|
||
|
||
BASE_DIR = Path(__file__).parent
|
||
from webapp import rescan as rescan_mod
|
||
|
||
app = FastAPI(title="RIP Help System", version="0.3.0")
|
||
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
|
||
app.include_router(rescan_mod.router)
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# Helpers
|
||
# ──────────────────────────────────────────────
|
||
|
||
def db_conn():
|
||
return psycopg2.connect(CONN_STR)
|
||
|
||
|
||
def _esc(s: str) -> str:
|
||
return (
|
||
str(s or "")
|
||
.replace("&", "&")
|
||
.replace("<", "<")
|
||
.replace(">", ">")
|
||
.replace('"', """)
|
||
)
|
||
|
||
|
||
def _basename(path_str: str) -> str:
|
||
return Path(str(path_str).replace("\\", "/")).name
|
||
|
||
|
||
def _resolve_section_txt(code: str, output_path: Optional[str]) -> Optional[Path]:
|
||
"""Намира .txt в OUTPUT_DIR по code / basename (игнорира q:\\ от БД)."""
|
||
candidates = []
|
||
if code:
|
||
candidates.append(OUTPUT_DIR / f"{code}.txt")
|
||
if output_path:
|
||
candidates.append(OUTPUT_DIR / _basename(output_path))
|
||
# ако output_path вече е абсолютен linux път и съществува
|
||
p = Path(output_path)
|
||
if p.is_file():
|
||
candidates.insert(0, p)
|
||
for c in candidates:
|
||
try:
|
||
if c.is_file():
|
||
return c
|
||
except OSError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _read_section_body(code: str, output_path: Optional[str]) -> str:
|
||
path = _resolve_section_txt(code, output_path)
|
||
if not path:
|
||
return ""
|
||
try:
|
||
raw = path.read_text(encoding="utf-8")
|
||
except Exception:
|
||
try:
|
||
raw = path.read_text(encoding="cp1251")
|
||
except Exception:
|
||
return ""
|
||
parts = raw.split("─" * 60, 1)
|
||
return parts[1].strip() if len(parts) > 1 else raw
|
||
|
||
|
||
def _text_to_html(text: str) -> str:
|
||
parts = []
|
||
last = 0
|
||
for m in _IMG_PLACEHOLDER_RE.finditer(text):
|
||
parts.append(_esc(text[last:m.start()]))
|
||
rel = m.group(1).strip().replace("\\", "/")
|
||
fname = rel.split("/", 1)[1] if rel.startswith("images/") else rel
|
||
parts.append(
|
||
f'<img src="/images/{_esc(fname)}" alt="" '
|
||
f'style="max-width:100%;max-height:240px;display:block;margin:8px 0;'
|
||
f'border:1px solid #d8dce3;border-radius:6px">'
|
||
)
|
||
last = m.end()
|
||
parts.append(_esc(text[last:]))
|
||
return "".join(parts).replace("\n", "<br>")
|
||
|
||
|
||
def _rich_html_with_images(html: str) -> str:
|
||
def sub(m):
|
||
rel = m.group(1).strip().replace("\\", "/")
|
||
fname = rel.split("/", 1)[1] if rel.startswith("images/") else rel
|
||
return (
|
||
f'<img src="/images/{_esc(fname)}" alt="" '
|
||
f'style="max-width:100%;max-height:240px;display:block;margin:8px 0;'
|
||
f'border:1px solid #d8dce3;border-radius:6px">'
|
||
)
|
||
|
||
return _IMG_PLACEHOLDER_RE.sub(sub, html)
|
||
|
||
|
||
def _row_to_dict(cols, r, *, full_text: bool = False) -> dict:
|
||
d = dict(zip(cols, r))
|
||
d["updated_at"] = str(d["updated_at"])[:16] if d["updated_at"] else ""
|
||
try:
|
||
d["images"] = json.loads(d["images"]) if d.get("images") else []
|
||
except Exception:
|
||
d["images"] = []
|
||
|
||
body = _read_section_body(d.get("code") or "", d.get("output_path"))
|
||
# нормализиран път за клиента / дебъг
|
||
resolved = _resolve_section_txt(d.get("code") or "", d.get("output_path"))
|
||
d["resolved_path"] = str(resolved) if resolved else None
|
||
|
||
if d.get("html_text"):
|
||
d["text_html"] = _rich_html_with_images(d["html_text"])
|
||
d["text"] = body if full_text else (body[:800] if body else "")
|
||
else:
|
||
if full_text:
|
||
d["text"] = body
|
||
d["text_html"] = _text_to_html(body) if body else ""
|
||
else:
|
||
d["text"] = body[:800] if body else ""
|
||
d["text_html"] = _text_to_html(body[:1200]) if body else ""
|
||
return d
|
||
|
||
|
||
def fetch_sections(
|
||
prefix: Optional[str] = None,
|
||
q: Optional[str] = None,
|
||
*,
|
||
full_text: bool = False,
|
||
) -> list[dict]:
|
||
conn = db_conn()
|
||
cur = conn.cursor()
|
||
sql = """
|
||
SELECT s.prefix, s.code, s.title, s.keywords, s.char_count,
|
||
s.source_file, s.output_path, s.updated_at,
|
||
s.images, s.html_text, f.section_count
|
||
FROM rip_help_sections s
|
||
LEFT JOIN rip_help_files f
|
||
ON f.file_path = s.source_file AND f.prefix = s.prefix
|
||
WHERE 1=1
|
||
"""
|
||
params: list = []
|
||
if prefix:
|
||
sql += " AND s.prefix = %s"
|
||
params.append(prefix)
|
||
if q:
|
||
like = f"%{q}%"
|
||
sql += """ AND (
|
||
s.code ILIKE %s OR s.title ILIKE %s OR COALESCE(s.keywords,'') ILIKE %s
|
||
OR COALESCE(s.html_text,'') ILIKE %s
|
||
)"""
|
||
params.extend([like, like, like, like])
|
||
sql += " ORDER BY s.prefix, s.code"
|
||
|
||
cur.execute(sql, params)
|
||
cols = [c[0] for c in cur.description]
|
||
rows = [_row_to_dict(cols, r, full_text=full_text) for r in cur.fetchall()]
|
||
conn.close()
|
||
return rows
|
||
|
||
|
||
# ──────────────────────────────────────────────
|
||
# Routes
|
||
# ──────────────────────────────────────────────
|
||
|
||
@app.get("/", response_class=HTMLResponse)
|
||
def viewer(
|
||
request: Request,
|
||
prefix: Optional[str] = Query(None),
|
||
home: Optional[str] = Query(None),
|
||
):
|
||
sections = fetch_sections(prefix)
|
||
home_url = "/home-image" if (home or HOME_IMAGE or (BASE_DIR.parent / "Bairaci.png").exists()) else None
|
||
return templates.TemplateResponse(
|
||
request,
|
||
"viewer.html",
|
||
{
|
||
"sections_json": json.dumps(sections, ensure_ascii=False, default=str),
|
||
"section_count": len(sections),
|
||
"prefix": prefix or "",
|
||
"home_url": home_url,
|
||
"generated": datetime.now().strftime("%d.%m.%Y %H:%M"),
|
||
},
|
||
)
|
||
|
||
|
||
@app.get("/home-image")
|
||
def home_image():
|
||
img_path = HOME_IMAGE
|
||
if not img_path:
|
||
for candidate in (
|
||
OUTPUT_DIR / "Bairaci.png",
|
||
OUTPUT_DIR.parent / "Bairaci.png",
|
||
BASE_DIR.parent / "Bairaci.png",
|
||
):
|
||
if candidate.is_file():
|
||
img_path = str(candidate)
|
||
break
|
||
if not img_path or not Path(img_path).is_file():
|
||
raise HTTPException(404, "home image not configured")
|
||
return FileResponse(img_path)
|
||
|
||
|
||
@app.get("/images/{filename:path}")
|
||
def serve_image(filename: str):
|
||
# само basename — без path traversal
|
||
safe = Path(filename).name
|
||
path = OUTPUT_DIR / "images" / safe
|
||
if not path.is_file():
|
||
raise HTTPException(404, f"image not found: {safe}")
|
||
return FileResponse(path)
|
||
|
||
|
||
@app.get("/api/sections")
|
||
def api_sections(prefix: Optional[str] = Query(None)):
|
||
return JSONResponse(fetch_sections(prefix))
|
||
|
||
|
||
@app.get("/api/search")
|
||
def api_search(
|
||
q: str = Query(..., min_length=1),
|
||
prefix: Optional[str] = Query(None),
|
||
):
|
||
rows = fetch_sections(prefix, q=q.strip())
|
||
return JSONResponse({"q": q, "prefix": prefix, "count": len(rows), "results": rows})
|
||
|
||
|
||
@app.get("/api/section/{code}")
|
||
def api_section(code: str):
|
||
conn = db_conn()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"""
|
||
SELECT s.prefix, s.code, s.title, s.keywords, s.char_count,
|
||
s.source_file, s.output_path, s.updated_at,
|
||
s.images, s.html_text, f.section_count
|
||
FROM rip_help_sections s
|
||
LEFT JOIN rip_help_files f
|
||
ON f.file_path = s.source_file AND f.prefix = s.prefix
|
||
WHERE s.code = %s
|
||
""",
|
||
(code,),
|
||
)
|
||
row = cur.fetchone()
|
||
cols = [c[0] for c in cur.description] if cur.description else []
|
||
conn.close()
|
||
if not row:
|
||
raise HTTPException(404, f"section {code} not found")
|
||
return JSONResponse(_row_to_dict(cols, row, full_text=True))
|
||
|
||
|
||
class KeywordsUpdate(BaseModel):
|
||
keywords: str
|
||
|
||
|
||
@app.post("/api/keywords/{code}")
|
||
def update_keywords(code: str, body: KeywordsUpdate):
|
||
conn = db_conn()
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"UPDATE rip_help_sections SET keywords=%s, updated_at=NOW() WHERE code=%s",
|
||
(body.keywords, code),
|
||
)
|
||
if cur.rowcount == 0:
|
||
conn.close()
|
||
raise HTTPException(404, f"section {code} not found")
|
||
conn.commit()
|
||
conn.close()
|
||
return {"ok": True, "code": code}
|
||
|
||
|
||
@app.get("/healthz")
|
||
def healthz():
|
||
db_status = "ok"
|
||
try:
|
||
conn = db_conn()
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT 1")
|
||
cur.fetchone()
|
||
conn.close()
|
||
except Exception as e:
|
||
return JSONResponse(
|
||
{"status": "error", "db": str(e), "output_dir": str(OUTPUT_DIR)},
|
||
status_code=503,
|
||
)
|
||
|
||
out_ok = OUTPUT_DIR.is_dir()
|
||
sample = None
|
||
txt_count = 0
|
||
if out_ok:
|
||
try:
|
||
txts = list(OUTPUT_DIR.glob("*.txt"))
|
||
txt_count = len(txts)
|
||
sample = txts[0].name if txts else None
|
||
except Exception:
|
||
pass
|
||
|
||
return {
|
||
"status": "ok",
|
||
"db": db_status,
|
||
"output_dir": str(OUTPUT_DIR),
|
||
"output_dir_ok": out_ok,
|
||
"txt_count": txt_count,
|
||
"sample": sample,
|
||
}
|