104 lines
5.4 KiB
Python
104 lines
5.4 KiB
Python
"""Tools für abgeglichene Ordner: Dateien auflisten, im Volltext suchen, lesen (nur lesend)."""
|
||
|
||
from pydantic import Field
|
||
from sqlalchemy import func, literal_column, select
|
||
|
||
from app.documents import service as documents
|
||
from app.documents.models import OK, DocFile, DocFolder
|
||
from app.tools.context import ToolContext
|
||
from app.tools.registry import Args, Tool, ToolError, register
|
||
|
||
MAX_LIST = 300
|
||
|
||
|
||
class ListArgs(Args):
|
||
folder: str | None = Field(None, description="Name des Ordners; leer = alle.")
|
||
path_prefix: str = Field("", description="Nur Dateien, deren Pfad so beginnt, z. B. 'Verträge/'.")
|
||
|
||
|
||
class SearchArgs(Args):
|
||
query: str = Field(min_length=2, max_length=300,
|
||
description="Suchbegriffe (Volltext, deutsche Wortformen; \"genaue Phrase\", -ausschließen, OR).")
|
||
folder: str | None = Field(None, description="Name des Ordners; leer = alle.")
|
||
max_results: int = Field(10, ge=1, le=30)
|
||
|
||
|
||
class ReadArgs(Args):
|
||
folder: str = Field(description="Name des Ordners.")
|
||
path: str = Field(description="Pfad der Datei im Ordner, wie in docs_list.")
|
||
start_line: int = Field(1, ge=1)
|
||
max_lines: int = Field(400, ge=1, le=2000)
|
||
|
||
|
||
def folders_of(context: ToolContext, session, name: str | None) -> list[DocFolder]:
|
||
found = documents.for_run(session, context.workspace_id, context.user_id, context.chat_session_id)
|
||
if not found:
|
||
raise ToolError("In diesem Chat sind keine abgeglichenen Ordner verfügbar.")
|
||
if name:
|
||
found = [f for f in found if f.name == name]
|
||
if not found:
|
||
raise ToolError(f"Ordner {name!r} gibt es hier nicht.")
|
||
return found
|
||
|
||
|
||
def list_files(context: ToolContext, args: ListArgs) -> str:
|
||
with context.db.session() as session:
|
||
found = folders_of(context, session, args.folder)
|
||
lines = []
|
||
for f in found:
|
||
rows = session.execute(select(DocFile.path, DocFile.size, DocFile.status, DocFile.error)
|
||
.where(DocFile.folder_id == f.id, DocFile.path.startswith(args.path_prefix))
|
||
.order_by(DocFile.path).limit(MAX_LIST + 1)).all()
|
||
synced = f.synced.astimezone().strftime("%d.%m.%Y %H:%M") if f.synced else "nie"
|
||
lines.append(f"Ordner {f.name} (abgeglichen: {synced}):")
|
||
for path, size, status, error in rows[:MAX_LIST]:
|
||
state = "wird ausgelesen" if status == "pending" else "nicht lesbar"
|
||
note = "" if status == OK else f" [{state}{': ' + error[:80] if error else ''}]"
|
||
lines.append(f"- {path} ({size // 1024 + 1} KB){note}")
|
||
if len(rows) > MAX_LIST:
|
||
lines.append("… weitere Dateien; mit path_prefix eingrenzen.")
|
||
if not rows:
|
||
lines.append("- (keine Dateien)")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def search(context: ToolContext, args: SearchArgs) -> str:
|
||
with context.db.session() as session:
|
||
found = folders_of(context, session, args.folder)
|
||
names = {f.id: f.name for f in found}
|
||
query = func.websearch_to_tsquery(literal_column("'german'"), args.query)
|
||
headline = func.ts_headline(literal_column("'german'"), DocFile.text, query,
|
||
"MaxFragments=2, MinWords=8, MaxWords=30, StartSel=«, StopSel=»")
|
||
rows = session.execute(
|
||
select(DocFile.folder_id, DocFile.path, func.ts_rank(DocFile.search, query).label("rank"), headline)
|
||
.where(DocFile.folder_id.in_(names), DocFile.status == OK, DocFile.search.op("@@")(query))
|
||
.order_by(literal_column("rank").desc()).limit(args.max_results)).all()
|
||
if not rows:
|
||
return "Keine Treffer. Andere Begriffe oder Wortstämme versuchen; docs_list zeigt die Dateien."
|
||
return "\n\n".join(f"{names[folder_id]} / {path}\n{snippet.strip()}" for folder_id, path, _, snippet in rows)
|
||
|
||
|
||
def read(context: ToolContext, args: ReadArgs) -> str:
|
||
with context.db.session() as session:
|
||
[f] = folders_of(context, session, args.folder)
|
||
file = session.scalar(select(DocFile).where(DocFile.folder_id == f.id,
|
||
DocFile.path == documents.check_path(args.path)))
|
||
if file is None:
|
||
raise ToolError(f"Datei {args.path!r} gibt es im Ordner {f.name} nicht; docs_list zeigt die Dateien.")
|
||
if file.status != OK:
|
||
raise ToolError(f"{file.path} ist (noch) nicht lesbar: {file.error or 'wird ausgelesen'}")
|
||
lines = file.text.splitlines()
|
||
start = args.start_line - 1
|
||
part = lines[start:start + args.max_lines]
|
||
more = (f"\n[… {len(lines) - start - len(part)} weitere Zeilen; mit start_line={start + len(part) + 1} "
|
||
"weiterlesen]") if start + len(part) < len(lines) else ""
|
||
return f"{f.name} / {file.path}, Zeilen {args.start_line}–{start + len(part)} von {len(lines)}\n" \
|
||
+ "\n".join(part) + more
|
||
|
||
|
||
register(Tool("docs_list", "docs", "Dateien der abgeglichenen Ordner auflisten (Pfad, Größe, Lesbarkeit).",
|
||
ListArgs, "read", run=list_files))
|
||
register(Tool("docs_search", "docs", "In den abgeglichenen Ordnern im Volltext suchen; liefert Dateien mit "
|
||
"Textausschnitten. Danach die passenden Dateien mit docs_read lesen.", SearchArgs, "read", run=search))
|
||
register(Tool("docs_read", "docs", "Ausgelesenen Text einer Datei aus einem abgeglichenen Ordner lesen (Office, "
|
||
"PDF und Bilder als Text).", ReadArgs, "read", run=read))
|