627 lines
20 KiB
Python
627 lines
20 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 статус + version/git
|
||
|
||
Env:
|
||
HELP_DB_CONN — libpq Postgres
|
||
OUTPUT_DIR — локален/mounted път към секциите (default: share)
|
||
HOME_IMAGE — път към home картинка
|
||
GIT_SHA / SOURCE_COMMIT / COMMIT_SHA — git SHA (Coolify / Docker build-arg)
|
||
|
||
Версия: префиксът е файлът VERSION в корена на репото (същият като show-version.bat).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Optional
|
||
|
||
import psycopg2
|
||
from fastapi import FastAPI, Header, HTTPException, Query, Request
|
||
from fastapi.responses import FileResponse, HTMLResponse, JSONResponse
|
||
from fastapi.templating import Jinja2Templates
|
||
from pydantic import BaseModel
|
||
|
||
from help_codes import (
|
||
WIPED_HASH,
|
||
numbering_report,
|
||
parse_code,
|
||
remove_section_outputs,
|
||
source_basename,
|
||
)
|
||
|
||
# ──────────────────────────────────────────────
|
||
# 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
|
||
_NO_CACHE = {"Cache-Control": "no-store, no-cache, max-age=0"}
|
||
|
||
|
||
def _short_id(value: str, n: int = 7) -> str:
|
||
value = value.strip()
|
||
if not value:
|
||
return ""
|
||
return value[:n] if len(value) > n else value
|
||
|
||
|
||
def _resolve_app_version() -> str:
|
||
"""Prefix from repo-root VERSION (same file as show-version.bat)."""
|
||
for candidate in (BASE_DIR.parent / "VERSION", Path("/app/VERSION")):
|
||
try:
|
||
if candidate.is_file():
|
||
text = candidate.read_text(encoding="utf-8").strip().splitlines()[0].strip()
|
||
if text:
|
||
return text
|
||
except OSError:
|
||
pass
|
||
return "0.3.0"
|
||
|
||
|
||
APP_VERSION = _resolve_app_version()
|
||
|
||
|
||
def _resolve_git() -> str:
|
||
"""Git SHA / build id that changes on each image rebuild.
|
||
|
||
Prefers Coolify/runtime env, then /app/BUILD_ID (baked in Docker),
|
||
then local ``git rev-parse``.
|
||
"""
|
||
for key in ("GIT_SHA", "APP_GIT_SHA", "SOURCE_COMMIT", "COMMIT_SHA"):
|
||
raw = (os.getenv(key) or "").strip()
|
||
if raw and raw.lower() not in {"unknown", "none", "null"}:
|
||
return _short_id(raw)
|
||
build_id = Path("/app/BUILD_ID")
|
||
if build_id.is_file():
|
||
try:
|
||
text = build_id.read_text(encoding="utf-8").strip()
|
||
if text:
|
||
return _short_id(text, 7 if len(text) >= 40 else 12)
|
||
except OSError:
|
||
pass
|
||
try:
|
||
repo = BASE_DIR.parent
|
||
r = subprocess.run(
|
||
["git", "rev-parse", "--short=7", "HEAD"],
|
||
cwd=str(repo),
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=1,
|
||
)
|
||
if r.returncode == 0 and r.stdout.strip():
|
||
return r.stdout.strip()
|
||
except (OSError, subprocess.SubprocessError):
|
||
pass
|
||
return "dev"
|
||
|
||
|
||
GIT_SHA = _resolve_git()
|
||
|
||
from webapp import rescan as rescan_mod
|
||
|
||
app = FastAPI(title="RIP Help System", version=APP_VERSION)
|
||
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 _fmt_ts(value, n: int = 19) -> str:
|
||
return str(value)[:n] if value else ""
|
||
|
||
|
||
def _row_to_dict(cols, r, *, full_text: bool = False) -> dict:
|
||
d = dict(zip(cols, r))
|
||
d["updated_at"] = _fmt_ts(d.get("updated_at"), 16)
|
||
if "created_at" in d:
|
||
d["created_at"] = _fmt_ts(d.get("created_at"))
|
||
if "processed_at" in d:
|
||
d["processed_at"] = _fmt_ts(d.get("processed_at"))
|
||
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.id, s.prefix, s.code, s.title, s.keywords, s.char_count,
|
||
s.source_file, s.output_path, s.updated_at, s.created_at,
|
||
f.processed_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)
|
||
# Escape < so </script> inside text_html cannot break the page script tag
|
||
sections_json = json.dumps(sections, ensure_ascii=False, default=str).replace("<", "\\u003c")
|
||
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": sections_json,
|
||
"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.id, s.prefix, s.code, s.title, s.keywords, s.char_count,
|
||
s.source_file, s.output_path, s.updated_at, s.created_at,
|
||
f.processed_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}
|
||
|
||
|
||
def _safe_help_filename(file: str) -> str:
|
||
name = Path(str(file or "").replace("\\", "/")).name
|
||
if not name or name in (".", ".."):
|
||
raise HTTPException(400, "file is required")
|
||
return name
|
||
|
||
|
||
def _ensure_file_index_col(cur):
|
||
cur.execute(
|
||
"ALTER TABLE rip_help_files ADD COLUMN IF NOT EXISTS file_index INTEGER"
|
||
)
|
||
|
||
|
||
def _matching_source_paths(cur, prefix: str, identity: str) -> list[str]:
|
||
name = source_basename(identity).lower()
|
||
if not name:
|
||
return []
|
||
cur.execute(
|
||
"""
|
||
SELECT file_path FROM rip_help_files WHERE prefix=%s
|
||
UNION
|
||
SELECT source_file FROM rip_help_sections WHERE prefix=%s
|
||
""",
|
||
(prefix, prefix),
|
||
)
|
||
found: list[str] = []
|
||
seen: set[str] = set()
|
||
for (p,) in cur.fetchall():
|
||
if p and source_basename(p).lower() == name and p not in seen:
|
||
seen.add(p)
|
||
found.append(p)
|
||
return found
|
||
|
||
|
||
def _file_index_for(cur, prefix: str, paths: list[str]) -> Optional[int]:
|
||
if not paths:
|
||
return None
|
||
cur.execute(
|
||
"SELECT file_index FROM rip_help_files "
|
||
"WHERE prefix=%s AND file_path = ANY(%s) AND file_index IS NOT NULL",
|
||
(prefix, list(paths)),
|
||
)
|
||
from_col = [r[0] for r in cur.fetchall() if r[0]]
|
||
if from_col:
|
||
return min(from_col)
|
||
cur.execute(
|
||
"SELECT code FROM rip_help_sections WHERE prefix=%s AND source_file = ANY(%s)",
|
||
(prefix, list(paths)),
|
||
)
|
||
found: list[int] = []
|
||
for (code,) in cur.fetchall():
|
||
parsed = parse_code(code)
|
||
if parsed:
|
||
found.append(parsed[1])
|
||
if not found:
|
||
return None
|
||
tally: dict[int, int] = {}
|
||
for idx in found:
|
||
tally[idx] = tally.get(idx, 0) + 1
|
||
return max(tally, key=lambda k: (tally[k], -k))
|
||
|
||
|
||
@app.get("/api/file-extractions")
|
||
def api_file_extractions(
|
||
file: str = Query(..., min_length=1),
|
||
prefix: str = Query("RIP"),
|
||
):
|
||
"""Брой и номерация на секциите за даден help файл."""
|
||
name = _safe_help_filename(file)
|
||
conn = db_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
_ensure_file_index_col(cur)
|
||
conn.commit()
|
||
paths = _matching_source_paths(cur, prefix, name)
|
||
rows = []
|
||
if paths:
|
||
cur.execute(
|
||
"SELECT code, source_file FROM rip_help_sections "
|
||
"WHERE prefix=%s AND source_file = ANY(%s) ORDER BY code",
|
||
(prefix, list(paths)),
|
||
)
|
||
rows = cur.fetchall()
|
||
codes = [r[0] for r in rows]
|
||
report = numbering_report(codes)
|
||
fi = _file_index_for(cur, prefix, paths)
|
||
if fi is not None:
|
||
report["file_index"] = fi
|
||
return {
|
||
"file": name,
|
||
"prefix": prefix,
|
||
"sources": sorted({source_basename(r[1]) for r in rows if r[1]}),
|
||
**report,
|
||
}
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
@app.delete("/api/file-extractions")
|
||
def api_delete_file_extractions(
|
||
file: str = Query(..., min_length=1),
|
||
prefix: str = Query("RIP"),
|
||
x_rescan_token: Optional[str] = Header(None, alias="X-Rescan-Token"),
|
||
token: Optional[str] = Query(None),
|
||
):
|
||
"""Изтрива всички секции за файла. Самият help файл не се пипа."""
|
||
rescan_mod._check_token(x_rescan_token or token)
|
||
name = _safe_help_filename(file)
|
||
conn = db_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
_ensure_file_index_col(cur)
|
||
paths = _matching_source_paths(cur, prefix, name)
|
||
rows = []
|
||
if paths:
|
||
cur.execute(
|
||
"SELECT code, output_path FROM rip_help_sections "
|
||
"WHERE prefix=%s AND source_file = ANY(%s)",
|
||
(prefix, list(paths)),
|
||
)
|
||
rows = cur.fetchall()
|
||
codes = [r[0] for r in rows]
|
||
file_index = _file_index_for(cur, prefix, paths)
|
||
remove_section_outputs(OUTPUT_DIR, codes, [r[1] for r in rows if r[1]])
|
||
if file_index:
|
||
cur.execute(
|
||
"SELECT code, source_file FROM rip_help_sections WHERE prefix=%s",
|
||
(prefix,),
|
||
)
|
||
name_l = name.lower()
|
||
shared = False
|
||
for code, src in cur.fetchall():
|
||
parsed = parse_code(code)
|
||
if (
|
||
parsed
|
||
and parsed[1] == file_index
|
||
and source_basename(src).lower() != name_l
|
||
):
|
||
shared = True
|
||
break
|
||
if shared:
|
||
file_index = None
|
||
if paths:
|
||
cur.execute(
|
||
"DELETE FROM rip_help_sections WHERE prefix=%s AND source_file = ANY(%s)",
|
||
(prefix, list(paths)),
|
||
)
|
||
cur.execute(
|
||
"DELETE FROM rip_help_files WHERE prefix=%s AND file_path = ANY(%s)",
|
||
(prefix, list(paths)),
|
||
)
|
||
if file_index:
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO rip_help_files (prefix, file_path, file_hash, section_count, file_index)
|
||
VALUES (%s, %s, %s, 0, %s)
|
||
ON CONFLICT (prefix, file_path) DO UPDATE SET
|
||
file_hash = EXCLUDED.file_hash,
|
||
section_count = 0,
|
||
processed_at = NOW(),
|
||
file_index = EXCLUDED.file_index
|
||
""",
|
||
(prefix, name, WIPED_HASH, file_index),
|
||
)
|
||
conn.commit()
|
||
return {
|
||
"ok": True,
|
||
"file": name,
|
||
"prefix": prefix,
|
||
"deleted": len(codes),
|
||
"codes": codes,
|
||
"file_index": file_index,
|
||
}
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
@app.get("/healthz", summary="Статус на услугата и БД; version/git, output_dir_ok, txt_count.")
|
||
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(
|
||
{
|
||
"ok": False,
|
||
"status": "error",
|
||
"version": f"{APP_VERSION}+{GIT_SHA}",
|
||
"git": GIT_SHA,
|
||
"db": str(e),
|
||
"output_dir": str(OUTPUT_DIR),
|
||
},
|
||
status_code=503,
|
||
headers=_NO_CACHE,
|
||
)
|
||
|
||
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 JSONResponse(
|
||
{
|
||
"ok": True,
|
||
"status": "ok",
|
||
"version": f"{APP_VERSION}+{GIT_SHA}",
|
||
"git": GIT_SHA,
|
||
"db": db_status,
|
||
"output_dir": str(OUTPUT_DIR),
|
||
"output_dir_ok": out_ok,
|
||
"txt_count": txt_count,
|
||
"sample": sample,
|
||
},
|
||
headers=_NO_CACHE,
|
||
)
|