sabo/rip-help-system
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Align Coolify webapp with WinSCP share and remote output paths.

Browse Source 
Resolve section text from OUTPUT_DIR by basename, add /api/search and richer /healthz, default mount to /mnt/.../Output, support --remote-root in processor.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
sabo
2026-08-25 19:47:09 +03:00
parent 9613420d1d
commit a81d5794d4
9 changed files with 342 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

347
webapp/main.py
View File

@@ -2,63 +2,58 @@
FastAPI webapp за RIP Help System.
Endpoint-и:
GET / — viewer HTML (с опционален ?prefix=&home=)
GET /images/{path} — статичен сервинг на картинки от OUTPUT_DIR
GET /home-image — home image ако HOME_IMAGE е зададен
GET /api/sections — JSON списък със секции (?prefix=)
POST /api/keywords/{code} — обновяване на keywords в БД
GET /healthz — health check
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 vars):
HELP_DB_CONN — libpq формат за Postgres
OUTPUT_DIR — директорията със секциите и картинките (default: ./data)
HOME_IMAGE — път към home картинка (опционален)
Env:
HELP_DB_CONN — libpq Postgres
OUTPUT_DIR — локален/mounted път към секциите (default: share)
HOME_IMAGE — път към home картинка
"""
import os, sys, json, re
from pathlib import Path
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, Request, Query
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from fastapi import FastAPI, HTTPException, Query, Request
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
from fastapi.templating import Jinja2Templates
from pydantic import BaseModel
# ──────────────────────────────────────────────
# Configuration
# ──────────────────────────────────────────────
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", Path(__file__).parent.parent / "data"))
HOME_IMAGE = os.getenv("HOME_IMAGE") # абсолютен път или None
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*\]")
# ──────────────────────────────────────────────
# App
# ──────────────────────────────────────────────
BASE_DIR = Path(__file__).parent
app = FastAPI(title="RIP Help System")
app = FastAPI(title="RIP Help System", version="0.2.0")
templates = Jinja2Templates(directory=str(BASE_DIR / "templates"))
# Static mount за картинките — само ако директорията съществува
_images_dir = OUTPUT_DIR / "images"
if _images_dir.exists():
app.mount("/images", StaticFiles(directory=str(_images_dir)), name="images")
# ──────────────────────────────────────────────
# DB
# Helpers
# ──────────────────────────────────────────────
def db_conn():
@@ -66,19 +61,60 @@ def db_conn():
def _esc(s: str) -> str:
return (str(s or "")
.replace("&", "&amp;").replace("<", "&lt;")
.replace(">", "&gt;").replace('"', "&quot;"))
return (
str(s or "")
.replace("&", "&amp;")
.replace("<", "&lt;")
.replace(">", "&gt;")
.replace('"', "&quot;")
)
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:
"""Конвертира [IMG: images/foo.png] към <img src="/images/foo.png">."""
parts = []
last = 0
for m in _IMG_PLACEHOLDER_RE.finditer(text):
parts.append(_esc(text[last:m.start()]))
rel = m.group(1).strip().replace("\\", "/")
# Очакваме path като "images/<filename>" — превръщаме в URL /images/<filename>
fname = rel.split("/", 1)[1] if rel.startswith("images/") else rel
parts.append(
f'<img src="/images/{_esc(fname)}" alt="" '
@@ -91,70 +127,77 @@ def _text_to_html(text: str) -> str:
def _rich_html_with_images(html: str) -> str:
"""Подменя [IMG: ...] placeholder-и с <img src="/images/...">; не escape-ва HTML."""
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 (
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 fetch_sections(prefix: Optional[str] = None) -> list[dict]:
conn = db_conn()
cur = conn.cursor()
if prefix:
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.prefix = %s
ORDER BY s.code
""", (prefix,))
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:
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
ORDER BY s.prefix, s.code
""")
cols = [c[0] for c in cur.description]
rows = []
for r in cur.fetchall():
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"] = []
# Render rich HTML или fallback към plain text + image placeholders
if d.get("html_text"):
d["text_html"] = _rich_html_with_images(d["html_text"])
if full_text:
d["text"] = body
d["text_html"] = _text_to_html(body) if body else ""
else:
# Plain text fallback — четем от output_path ако може
body = ""
op = d.get("output_path")
if op:
# Опитваме се да намерим файла в OUTPUT_DIR (по basename)
txt_name = Path(op).name
txt_path = OUTPUT_DIR / txt_name
if txt_path.exists():
try:
raw = txt_path.read_text(encoding="utf-8")
parts = raw.split("─" * 60, 1)
body = parts[1].strip() if len(parts) > 1 else raw
except Exception:
pass
d["text"] = body[:800]
d["text"] = body[:800] if body else ""
d["text_html"] = _text_to_html(body[:1200]) if body else ""
rows.append(d)
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
@@ -164,38 +207,91 @@ def fetch_sections(prefix: Optional[str] = None) -> list[dict]:
# ──────────────────────────────────────────────
@app.get("/", response_class=HTMLResponse)
def viewer(request: Request,
prefix: Optional[str] = Query(None),
home: Optional[str] = Query(None)):
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) 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"),
})
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
# Ако HOME_IMAGE не е зададен, пробваме Bairaci.png в проекта
if not img_path:
candidate = BASE_DIR.parent / "Bairaci.png"
if candidate.exists():
img_path = str(candidate)
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
@@ -203,10 +299,10 @@ class KeywordsUpdate(BaseModel):
@app.post("/api/keywords/{code}")
def update_keywords(code: str, body: KeywordsUpdate):
conn = db_conn()
cur = conn.cursor()
cur = conn.cursor()
cur.execute(
"UPDATE rip_help_sections SET keywords=%s, updated_at=NOW() WHERE code=%s",
(body.keywords, code)
(body.keywords, code),
)
if cur.rowcount == 0:
conn.close()
@@ -218,12 +314,35 @@ def update_keywords(code: str, body: KeywordsUpdate):
@app.get("/healthz")
def healthz():
db_status = "ok"
try:
conn = db_conn()
cur = conn.cursor()
cur = conn.cursor()
cur.execute("SELECT 1")
cur.fetchone()
conn.close()
return {"status": "ok", "db": "ok"}
except Exception as e:
return JSONResponse({"status": "error", "db": str(e)}, status_code=503)
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,
}