forked from fangtianchen/algonotes_rag
692 lines
29 KiB
Python
692 lines
29 KiB
Python
# scripts/cli.py
|
||
# Unified CLI entry point for `algonotes` command.
|
||
#
|
||
# Architecture:
|
||
# 1. Parser builders (_add_*_parser) — define arguments only
|
||
# 2. Handler functions (_handle_*) — validate args → delegate → output
|
||
# 3. Dispatch tables — map command strings to handlers
|
||
# 4. main() — build parser, parse, dispatch
|
||
|
||
import argparse
|
||
import json
|
||
import sys
|
||
from pathlib import Path
|
||
from typing import Any, Callable
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Setup
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _setup_cli_logging() -> None:
|
||
"""Configure logging before any module with loggers is imported.
|
||
|
||
Checks sys.argv for ``-v`` / ``--verbose`` before argparse runs,
|
||
so the flag only needs to be *present* — not fully parsed.
|
||
When present, enables human-readable stderr output at INFO level
|
||
alongside the default JSON-lines file logger.
|
||
"""
|
||
console = "-v" in sys.argv or "--verbose" in sys.argv
|
||
from src.logger import setup_logger
|
||
setup_logger("algonotes", console=console)
|
||
|
||
|
||
def _ensure_utf8_stdout() -> None:
|
||
"""Reconfigure stdout to UTF-8 on Windows (e.g. GBK/cp936 → UTF-8).
|
||
|
||
Agent responses may contain emoji or characters outside the system
|
||
codepage. When ``reconfigure`` fails (e.g. stdout is piped to a
|
||
non-reconfigurable stream), a warning is printed to stderr instead
|
||
of silently swallowing the error.
|
||
"""
|
||
try:
|
||
if sys.stdout.encoding and sys.stdout.encoding.lower() not in (
|
||
"utf-8", "utf8",
|
||
):
|
||
sys.stdout.reconfigure(encoding="utf-8")
|
||
except (OSError, AttributeError):
|
||
print("⚠ 无法将 stdout 重配置为 UTF-8（可能被管道占用）",
|
||
file=sys.stderr)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Output helper
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _output(data: Any, json_path: Path | None, *,
|
||
display: Callable[[Any], None] | None = None) -> None:
|
||
"""Unified output: write JSON and/or call ``display(data)``.
|
||
|
||
Eliminates the repeated ``if args.json: … else: print(…)`` pattern
|
||
that was duplicated across every handler.
|
||
|
||
When *json_path* is given the data is written as UTF-8 JSON (parent
|
||
directories are created automatically). When *display* is given it
|
||
is called with *data* for console output. Both can fire together —
|
||
they are independent; ``--json`` does not suppress console output.
|
||
|
||
Args:
|
||
data: Result to output. For JSON mode this is serialised with
|
||
``json.dumps(ensure_ascii=False, indent=2)``. For console
|
||
mode it is passed directly to *display*.
|
||
json_path: Path from the ``--json`` flag, or ``None`` to skip.
|
||
display: Callable that renders *data* to stdout, or ``None``.
|
||
If both *json_path* and *display* are ``None`` the call is
|
||
a silent no-op.
|
||
"""
|
||
if json_path:
|
||
json_path.parent.mkdir(parents=True, exist_ok=True)
|
||
json_path.write_text(
|
||
json.dumps(data, ensure_ascii=False, indent=2),
|
||
encoding="utf-8",
|
||
)
|
||
if display:
|
||
display(data)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Parser builders
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _add_ingest_parser(sub: argparse._SubParsersAction) -> None:
|
||
"""Register the ``ingest`` subcommand and its arguments.
|
||
|
||
Adds ``algonotes ingest`` with mutually exclusive ``-i/--path``
|
||
(local file or directory) and ``-u/--url`` (web page) sources,
|
||
plus optional ``--no-tag``, ``--type``, ``--author``, ``-v``,
|
||
and ``--json`` flags.
|
||
"""
|
||
p = sub.add_parser("ingest", help="导入笔记")
|
||
src = p.add_mutually_exclusive_group(required=True)
|
||
src.add_argument("-i", "--path", type=Path, help="本地文件或目录路径")
|
||
src.add_argument("-u", "--url", help="网页 URL 地址")
|
||
p.add_argument("--no-tag", action="store_true", help="跳过打 tag")
|
||
p.add_argument("--type", choices=["note", "solution", "template"],
|
||
default="note", help="笔记类型")
|
||
p.add_argument("--author", help="笔记作者")
|
||
p.add_argument("-n", "--filename", help="自定义文件名（默认使用源文件名或URL派生名）")
|
||
p.add_argument("-v", "--verbose", action="store_true", help="详细输出")
|
||
p.add_argument("--json", type=Path, metavar="FILE", help="输出结果到 JSON 文件")
|
||
|
||
|
||
def _add_update_parser(sub: argparse._SubParsersAction) -> None:
|
||
"""Register the ``update`` subcommand with ``content`` and ``metadata`` sub-subcommands.
|
||
|
||
``algonotes update content`` re-processes a note's file content
|
||
through the full ingestion pipeline (split → tag → re-index).
|
||
``algonotes update metadata`` edits tags/title/author/type in-place
|
||
without re-embedding.
|
||
"""
|
||
p = sub.add_parser("update", help="更新笔记")
|
||
u_sub = p.add_subparsers(dest="update_command", required=True)
|
||
|
||
p_content = u_sub.add_parser("content", help="更新笔记内容（重新处理文件）")
|
||
p_content.add_argument("filename", help="笔记文件名")
|
||
p_content.add_argument("--file", type=Path, help="新的本地文件路径")
|
||
p_content.add_argument("--no-tag", action="store_true", help="跳过打 tag")
|
||
p_content.add_argument("-v", "--verbose", action="store_true", help="详细输出")
|
||
p_content.add_argument("--json", type=Path, metavar="FILE", help="输出结果到 JSON 文件")
|
||
|
||
p_meta = u_sub.add_parser("metadata", help="更新笔记元数据（不修改内容）")
|
||
p_meta.add_argument("filename", help="笔记文件名")
|
||
p_meta.add_argument("--title", help="新标题")
|
||
p_meta.add_argument("--tags", help="新标签（逗号分隔）")
|
||
p_meta.add_argument("--author", help="新作者")
|
||
p_meta.add_argument("--type", choices=["note", "solution", "template"],
|
||
help="笔记类型")
|
||
p_meta.add_argument("--json", type=Path, metavar="FILE", help="输出结果到 JSON 文件")
|
||
|
||
|
||
def _add_delete_parser(sub: argparse._SubParsersAction) -> None:
|
||
"""Register the ``delete`` subcommand and its arguments.
|
||
|
||
Adds ``algonotes delete <filename>`` with optional ``--force``
|
||
(skip confirmation prompt) and ``--json`` flags.
|
||
"""
|
||
p = sub.add_parser("delete", help="删除笔记")
|
||
p.add_argument("filename", help="笔记文件名")
|
||
p.add_argument("--force", action="store_true", help="跳过确认提示")
|
||
p.add_argument("--json", type=Path, metavar="FILE", help="输出结果到 JSON 文件")
|
||
|
||
|
||
def _add_query_parser(sub: argparse._SubParsersAction) -> None:
|
||
"""Register the ``query`` subcommand and its six sub-subcommands.
|
||
|
||
Adds ``algonotes query {show, list, info, search, export, ask}``
|
||
with their respective arguments. This is the most complex parser
|
||
group because it covers all read-oriented operations.
|
||
"""
|
||
p = sub.add_parser("query", help="查询笔记")
|
||
q_sub = p.add_subparsers(dest="query_command", required=True)
|
||
|
||
p_show = q_sub.add_parser("show", help="查看笔记原文")
|
||
p_show.add_argument("filename", help="笔记文件名")
|
||
p_show.add_argument("--lines", type=int, help="只显示前 N 行")
|
||
p_show.add_argument("--json", type=Path, metavar="FILE", help="输出到 JSON 文件")
|
||
|
||
p_list = q_sub.add_parser("list", help="列出笔记")
|
||
p_list.add_argument("--tag", help="按标签筛选")
|
||
p_list.add_argument("-v", "--verbose", action="store_true", help="显示详细信息")
|
||
p_list.add_argument("--json", type=Path, metavar="FILE", help="输出到 JSON 文件")
|
||
|
||
p_info = q_sub.add_parser("info", help="查看笔记详情")
|
||
p_info.add_argument("filename", nargs="?", help="笔记文件名")
|
||
p_info.add_argument("--id", type=int, help="笔记 ID")
|
||
p_info.add_argument("--json", type=Path, metavar="FILE", help="输出到 JSON 文件")
|
||
|
||
p_search = q_sub.add_parser("search", help="语义搜索")
|
||
p_search.add_argument("query", help="搜索关键词")
|
||
p_search.add_argument("--top-k", type=int, default=5, help="返回结果数量")
|
||
p_search.add_argument("--json", type=Path, metavar="FILE", help="输出到 JSON 文件")
|
||
|
||
p_export = q_sub.add_parser("export",
|
||
help="导出笔记（YAML frontmatter，兼容 Obsidian）")
|
||
p_export.add_argument("filename", nargs="?", help="笔记文件名")
|
||
p_export.add_argument("--all", action="store_true", help="导出所有笔记")
|
||
p_export.add_argument("-o", "--output", type=Path, default=Path("./export"),
|
||
help="输出目录")
|
||
p_export.add_argument("--json", type=Path, metavar="FILE", help="输出到 JSON 文件")
|
||
|
||
p_ask = q_sub.add_parser("ask", help="RAG 问答（单次，直接输出答案）")
|
||
p_ask.add_argument("question", help="问题")
|
||
p_ask.add_argument("--top-k", type=int, default=5, help="检索结果数量（默认 5）")
|
||
p_ask.add_argument("--stream", action="store_true", help="流式输出答案")
|
||
p_ask.add_argument("--json", type=Path, metavar="FILE", help="输出到 JSON 文件")
|
||
|
||
|
||
def _add_chat_parser(sub: argparse._SubParsersAction) -> None:
|
||
"""Register the ``chat`` subcommand for interactive RAG sessions.
|
||
|
||
Adds ``algonotes chat`` with an optional ``--thread-id`` argument
|
||
that isolates conversation history (default: ``"default"``).
|
||
"""
|
||
p = sub.add_parser("chat", help="交互式问答")
|
||
p.add_argument("--thread-id", default="default",
|
||
help="会话标识符，用于区分不同对话")
|
||
|
||
|
||
def _add_mcp_parser(sub: argparse._SubParsersAction) -> None:
|
||
"""Register the ``mcp`` subcommand to start the MCP server.
|
||
|
||
Adds ``algonotes mcp`` with optional ``--host`` (default ``127.0.0.1``)
|
||
and ``--port`` (default ``8000``) to control the SSE transport binding.
|
||
"""
|
||
p = sub.add_parser("mcp", help="启动 MCP 服务器")
|
||
p.add_argument("--host", default="127.0.0.1", help="监听地址（默认 127.0.0.1）")
|
||
p.add_argument("--port", type=int, default=8000, help="监听端口（默认 8000）")
|
||
|
||
|
||
def _add_perf_parser(sub: argparse._SubParsersAction) -> None:
|
||
"""Register the ``perf`` subcommand for performance reporting.
|
||
|
||
Adds ``algonotes perf`` with optional ``--detail`` (show per-entry
|
||
breakdown) and ``--json`` flags.
|
||
"""
|
||
p = sub.add_parser("perf", help="性能报告")
|
||
p.add_argument("--detail", action="store_true", help="显示每条记录的详细信息")
|
||
p.add_argument("--json", type=Path, metavar="FILE", help="输出结果到 JSON 文件")
|
||
|
||
|
||
def _build_parser() -> argparse.ArgumentParser:
|
||
"""Construct the full argument parser with all subcommands.
|
||
|
||
Returns an ``ArgumentParser`` that is ready for ``parse_args()``.
|
||
Each subcommand group is delegated to its own ``_add_*_parser``
|
||
helper so parser construction stays modular.
|
||
"""
|
||
parser = argparse.ArgumentParser(
|
||
prog="algonotes",
|
||
description="AlgoNotes RAG CLI — 算法竞赛笔记管理工具",
|
||
)
|
||
sub = parser.add_subparsers(dest="command", required=True)
|
||
_add_ingest_parser(sub)
|
||
_add_update_parser(sub)
|
||
_add_delete_parser(sub)
|
||
_add_query_parser(sub)
|
||
_add_chat_parser(sub)
|
||
_add_mcp_parser(sub)
|
||
_add_perf_parser(sub)
|
||
return parser
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Handler: ingest
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _handle_ingest(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes ingest`` — import notes from local files or web URLs.
|
||
|
||
Dispatches to:
|
||
- ``ingest_local`` for ``-i <file>`` (single .md file)
|
||
- ``ingest_locals`` for ``-i <dir>`` (batch, all *.md in directory)
|
||
- ``ingest_web`` for ``-u <url>`` (web page with LLM cleaning)
|
||
|
||
Outputs a summary line for single imports; batch imports print
|
||
per-file progress from within ``ingest_locals`` and stay silent here.
|
||
Supports ``--json`` for structured output.
|
||
"""
|
||
from scripts.ingest import ingest_local, ingest_locals, ingest_web
|
||
|
||
tag = not args.no_tag
|
||
is_dir = args.path and args.path.is_dir()
|
||
|
||
if args.path:
|
||
if args.path.is_file():
|
||
result = ingest_local(args.path, tag=tag, verbose=args.verbose,
|
||
type=args.type, author=args.author,
|
||
filename=args.filename)
|
||
elif is_dir:
|
||
results = ingest_locals(args.path, tag=tag, verbose=args.verbose,
|
||
type=args.type, author=args.author)
|
||
else:
|
||
print(f"❌ 路径不存在: {args.path}")
|
||
sys.exit(1)
|
||
elif args.url:
|
||
result = ingest_web(args.url, tag=tag, verbose=args.verbose,
|
||
type=args.type, author=args.author,
|
||
filename=args.filename)
|
||
|
||
if is_dir:
|
||
output_data = [r.__dict__ for r in results]
|
||
else:
|
||
output_data = result.__dict__
|
||
|
||
def _display(data):
|
||
# Directory ingest is silent here — ingest_locals already
|
||
# prints per-file progress lines during iteration.
|
||
if not is_dir:
|
||
print(f" ✅ {data['file_name']} ({data['chunk_count']} chunks, tags: {data['tags']})")
|
||
|
||
_output(output_data, args.json, display=_display)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Handlers: update
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _handle_update_content(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes update content <filename>`` — reprocess a note's file.
|
||
|
||
Validates the note exists in SQLStore and that a content source
|
||
is available (``--file`` or stored ``source_url``), then runs the
|
||
full ingestion pipeline: load → split → tag → re-index in both
|
||
SQLStore and VectorStore (old chunks are deleted first).
|
||
|
||
Exits with code 1 if the note is not found or no source is available.
|
||
"""
|
||
from scripts.update import update_note
|
||
from src.store.sql_store import get_sql_store
|
||
|
||
sql_store = get_sql_store()
|
||
tag = not args.no_tag
|
||
|
||
sql_record = sql_store.get_by_filename(args.filename)
|
||
if not sql_record:
|
||
print(f"❌ 笔记不存在: {args.filename}")
|
||
sys.exit(1)
|
||
|
||
if not args.file and not sql_record.get("source_url"):
|
||
print(f"❌ 笔记 {args.filename} 无 source_url，请使用 --file 指定新文件")
|
||
sys.exit(1)
|
||
|
||
result = update_note(args.filename, file_path=args.file, tag=tag,
|
||
verbose=args.verbose)
|
||
|
||
def _display(data):
|
||
if data["success"]:
|
||
print(f"Update success: {data['file_name']} ({data['chunk_count']} chunks)")
|
||
else:
|
||
print(f"Update failed: {data['file_name']}")
|
||
|
||
_output(result.__dict__, args.json, display=_display)
|
||
|
||
|
||
def _handle_update_metadata(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes update metadata <filename>`` — edit metadata in-place.
|
||
|
||
Updates title, tags, author, and/or type in SQLStore without
|
||
re-processing file content. If tags are changed the VectorStore
|
||
chunk metadata is updated to match (no re-embedding).
|
||
|
||
At least one of ``--title``, ``--tags``, ``--author``, ``--type``
|
||
should be provided; the handler itself does not enforce this —
|
||
``update_metadata`` simply returns with ``fields_updated=[]``.
|
||
"""
|
||
from scripts.update import update_metadata
|
||
|
||
result = update_metadata(args.filename, title=args.title, tags=args.tags,
|
||
author=args.author, type=args.type)
|
||
|
||
def _display(data):
|
||
if data["success"]:
|
||
print(f"Metadata updated: {data['filename']} "
|
||
f"({', '.join(data['fields_updated'])})")
|
||
else:
|
||
print(f"Metadata update failed: {data['filename']}")
|
||
|
||
_output({"filename": result.filename, "success": result.success,
|
||
"fields_updated": result.fields_updated}, args.json, display=_display)
|
||
|
||
|
||
def _handle_update(args: argparse.Namespace) -> None:
|
||
"""Dispatch ``algonotes update {content, metadata}`` to the correct sub-handler.
|
||
|
||
``args.update_command`` is guaranteed to exist because the sub-subparser
|
||
was created with ``required=True``.
|
||
"""
|
||
_UPDATE_HANDLERS[args.update_command](args)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Handler: delete
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _handle_delete(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes delete <filename>`` — remove a note from all three stores.
|
||
|
||
Unless ``--force`` is passed, prompts the user to type the filename
|
||
for confirmation (type ``"stop"`` to abort). Deletion order is
|
||
VectorStore → FileStore → SQLStore (see ``scripts/delete.py``).
|
||
|
||
Does NOT exit with error if the file doesn't exist; prints a
|
||
warning and returns normally.
|
||
"""
|
||
from scripts.delete import delete_note
|
||
|
||
if not args.force:
|
||
ipt = input(f"请确认删除 {args.filename}（键入 stop 以撤销操作）: ")
|
||
if ipt.lower() == "stop":
|
||
sys.exit(0)
|
||
|
||
result = delete_note(args.filename)
|
||
|
||
def _display(data):
|
||
if data["success"]:
|
||
print(f"成功删除 {data['file_name']}")
|
||
else:
|
||
print(f"{data['file_name']} 文件不存在，无法进行删除")
|
||
|
||
_output(result.__dict__, args.json, display=_display)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Handlers: query
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _handle_query_show(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes query show <filename>`` — print raw note content.
|
||
|
||
Reads the file from FileStore and prints it to the terminal via
|
||
``rich.console.Console``. Use ``--lines N`` to limit output to
|
||
the first N lines. Supports ``--json`` for structured output.
|
||
"""
|
||
from scripts.query import show_raw, console
|
||
|
||
content = show_raw(args.filename, args.lines)
|
||
if content is None:
|
||
console.print(f"[red]笔记不存在: {args.filename}[/red]")
|
||
return
|
||
|
||
def _display(data):
|
||
console.print(data["content"], markup=False)
|
||
|
||
_output({"filename": args.filename, "content": content,
|
||
"line_count": len(content.splitlines())},
|
||
args.json, display=_display)
|
||
|
||
|
||
def _handle_query_list(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes query list`` — list all notes as a table.
|
||
|
||
Optionally filter by tag with ``--tag <keyword>`` (SQL LIKE on
|
||
the comma-separated tags field). Renders a ``rich`` table with
|
||
columns: ID, filename, tags, ingestion date. Supports ``--json``.
|
||
"""
|
||
from scripts.query import list_notes, print_list, NoteInfo
|
||
|
||
notes = list_notes(args.tag)
|
||
|
||
def _display(data):
|
||
print_list([NoteInfo.from_dict(d) for d in data])
|
||
|
||
_output([n.__dict__ for n in notes], args.json, display=_display)
|
||
|
||
|
||
def _handle_query_info(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes query info`` — show detailed metadata for one note.
|
||
|
||
Looks up the note by ``<filename>`` or ``--id <n>`` (at least one
|
||
required). Prints each metadata field as a key-value pair via
|
||
``rich`` console. Supports ``--json``.
|
||
"""
|
||
from scripts.query import show_info, console
|
||
from src.store.sql_store import get_sql_store
|
||
|
||
if not args.filename and not args.id:
|
||
console.print("[red]须指定 filename 或 --id[/red]")
|
||
return
|
||
|
||
if args.id:
|
||
record = get_sql_store().get(args.id)
|
||
note = show_info(record["filename"]) if record and record.get("filename") else None
|
||
else:
|
||
note = show_info(args.filename)
|
||
|
||
if note is None:
|
||
console.print(f"[red]笔记不存在: {args.filename or f'id={args.id}'}[/red]")
|
||
return
|
||
|
||
def _display(data):
|
||
for k, v in data.items():
|
||
console.print(f" [cyan]{k}:[/cyan] {v}")
|
||
|
||
_output(note.__dict__, args.json, display=_display)
|
||
|
||
|
||
def _handle_query_search(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes query search <query>`` — semantic vector search.
|
||
|
||
Searches the Chroma vector store for chunks semantically similar
|
||
to *query*. Prints top ``--top-k`` (default 5) results with source
|
||
filename and a 120-character preview. Supports ``--json``.
|
||
"""
|
||
from scripts.query import semantic_search, console
|
||
|
||
docs = semantic_search(args.query, args.top_k)
|
||
if not docs:
|
||
console.print("[yellow]无匹配结果[/yellow]")
|
||
return
|
||
|
||
def _display(data):
|
||
results = data["results"]
|
||
console.print(f"找到 {len(results)} 条相关结果：\n")
|
||
for i, r in enumerate(results, 1):
|
||
console.print(
|
||
f"[cyan][{i}][/cyan] 来自: {r.get('source', '?')}")
|
||
console.print(f" {r['content'][:120]}...\n", markup=False)
|
||
|
||
_output({"query": args.query,
|
||
"results": [{"source": d.metadata.get("source"),
|
||
"content": d.page_content,
|
||
"metadata": {k: v for k, v in d.metadata.items()
|
||
if k != "source"}}
|
||
for d in docs]},
|
||
args.json, display=_display)
|
||
|
||
|
||
def _handle_query_export(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes query export`` — export notes with YAML frontmatter.
|
||
|
||
Supports two modes:
|
||
- ``export <filename>`` — single note
|
||
- ``export --all`` — every note in the store
|
||
|
||
Output files are written to ``-o/--output`` (default ``./export``)
|
||
and are compatible with Obsidian. Catches ``FileNotFoundError``
|
||
for missing notes gracefully. Supports ``--json``.
|
||
"""
|
||
from scripts.query import export_note, export_all
|
||
|
||
try:
|
||
if args.all:
|
||
results = export_all(args.output)
|
||
|
||
def _display(data):
|
||
print(f"Exported {len(data)} notes to {args.output}")
|
||
|
||
_output(results, args.json, display=_display)
|
||
elif args.filename:
|
||
path = export_note(args.filename, args.output)
|
||
|
||
def _display(data):
|
||
print(f"Exported {data['filename']} to {data['path']}")
|
||
|
||
_output({"filename": args.filename, "path": str(path)},
|
||
args.json, display=_display)
|
||
else:
|
||
print("请指定 filename 或 --all")
|
||
except FileNotFoundError as e:
|
||
print(f"❌ 笔记不存在: {e}")
|
||
|
||
|
||
def _handle_query_ask(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes query ask <question>`` — single-shot RAG question.
|
||
|
||
Runs the full RAG pipeline (retrieve → rerank → generate) via
|
||
``scripts.rag.ask_question`` and prints the answer. Use ``--stream``
|
||
for token-by-token output (the handler itself is silent in stream
|
||
mode because ``ask_question`` already prints). Supports ``--json``.
|
||
|
||
Exits with code 1 if the RAG pipeline returns ``success=False``.
|
||
"""
|
||
from scripts.rag import ask_question
|
||
|
||
result = ask_question(args.question, stream=args.stream)
|
||
|
||
if not result.success:
|
||
print(f"❌ 问答失败: {result.error}")
|
||
sys.exit(1)
|
||
|
||
def _display(data):
|
||
if not args.stream:
|
||
print(data["answer"])
|
||
|
||
_output(result.dict, args.json, display=_display)
|
||
|
||
|
||
def _handle_query(args: argparse.Namespace) -> None:
|
||
"""Dispatch ``algonotes query {show, list, info, search, export, ask}``.
|
||
|
||
``args.query_command`` is guaranteed to exist because the sub-subparser
|
||
was created with ``required=True``.
|
||
"""
|
||
_QUERY_HANDLERS[args.query_command](args)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Handlers: chat & mcp
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _handle_chat(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes chat`` — start an interactive RAG session.
|
||
|
||
Launches a REPL loop that preserves conversation context across
|
||
turns via LangGraph's ``InMemorySaver`` checkpointer. The
|
||
``--thread-id`` argument isolates different conversations.
|
||
|
||
Type ``exit`` or ``quit`` to end the session.
|
||
"""
|
||
from scripts.chat import chat_loop
|
||
chat_loop(args.thread_id)
|
||
|
||
|
||
def _handle_mcp(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes mcp`` — start the MCP server over SSE transport.
|
||
|
||
Binds to ``--host``:``--port`` (default ``127.0.0.1:8000``) and
|
||
exposes nine tools (ingest, update, delete, show, list, search,
|
||
metadata, export, ask) that MCP clients such as Claude Desktop
|
||
can call.
|
||
"""
|
||
from src.mcp.server import main as mcp_main
|
||
mcp_main(host=args.host, port=args.port)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Handler: perf
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def _handle_perf(args: argparse.Namespace) -> None:
|
||
"""Handle ``algonotes perf`` — display performance statistics.
|
||
|
||
Parses ``logs/perf.log``, aggregates latency metrics by call_type,
|
||
and prints a summary table. Use ``--detail`` for a per-entry
|
||
breakdown. Supports ``--json`` for structured export.
|
||
"""
|
||
from scripts.perf_report import load_entries, print_report, report_to_dict
|
||
|
||
entries = load_entries()
|
||
|
||
if entries:
|
||
print_report(entries, detail=args.detail)
|
||
|
||
_output(report_to_dict(entries), args.json)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Dispatch tables
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
#: Map ``update_command`` strings to their handler functions.
|
||
_UPDATE_HANDLERS = {
|
||
"content": _handle_update_content,
|
||
"metadata": _handle_update_metadata,
|
||
}
|
||
|
||
#: Map ``query_command`` strings to their handler functions.
|
||
_QUERY_HANDLERS = {
|
||
"show": _handle_query_show,
|
||
"list": _handle_query_list,
|
||
"info": _handle_query_info,
|
||
"search": _handle_query_search,
|
||
"export": _handle_query_export,
|
||
"ask": _handle_query_ask,
|
||
}
|
||
|
||
#: Map top-level ``command`` strings to their handler functions.
|
||
_HANDLERS = {
|
||
"ingest": _handle_ingest,
|
||
"update": _handle_update,
|
||
"delete": _handle_delete,
|
||
"query": _handle_query,
|
||
"chat": _handle_chat,
|
||
"mcp": _handle_mcp,
|
||
"perf": _handle_perf,
|
||
}
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════
|
||
# Entry point
|
||
# ═══════════════════════════════════════════════════════════
|
||
|
||
def main() -> None:
|
||
"""Entry point for the ``algonotes`` CLI.
|
||
|
||
Order of operations:
|
||
1. Configure logging (checks for ``-v/--verbose`` in raw argv).
|
||
2. Ensure stdout uses UTF-8 encoding.
|
||
3. Build the full argparse parser via ``_build_parser()``.
|
||
4. Parse arguments and dispatch to the matching handler.
|
||
"""
|
||
_setup_cli_logging()
|
||
_ensure_utf8_stdout()
|
||
|
||
parser = _build_parser()
|
||
args = parser.parse_args()
|
||
|
||
_HANDLERS[args.command](args)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|