algonotes_rag/scripts/delete.py

88 lines
2.5 KiB
Python

# scripts/delete.py
# CLI backend: delete a note from three-layer storage.
#
# Order: Vector → File → SQL
import argparse
from dataclasses import dataclass
import json
from pathlib import Path
from src.store.file_store import get_file_store
from src.store.sql_store import get_sql_store
from src.store.vector_store import get_vector_store
from src.logger import setup_logger
_file_store = get_file_store()
_sql_store = get_sql_store()
_vector_store = get_vector_store()
logger = setup_logger("algonotes.cli.delete")
@dataclass
class DeleteResult:
"""Result of a single note deletion."""
file_name: str
success: bool
def delete_note(file_name: str) -> DeleteResult:
"""Delete a note from all three storage layers.
Deletes in order: VectorStore → FileStore → SQLStore.
If the note does not exist in SQLStore, returns success=False.
Args:
file_name: The note filename (e.g. ``"fenwick.md"``).
Returns:
A ``DeleteResult`` indicating whether deletion succeeded.
"""
sql_query = _sql_store.get_by_filename(file_name)
if not sql_query:
logger.warning(f"Note not found: {file_name}")
return DeleteResult(file_name, False)
vec_success: bool = _vector_store.delete(file_name) > 0
file_success: bool = _file_store.delete(file_name)
sql_success: bool = _sql_store.delete(sql_query.get("id"))
success = vec_success and file_success and sql_success
logger.info(f"Delete {'' if success else ''} {file_name}")
return DeleteResult(file_name, success)
if __name__ == "__main__":
parser = argparse.ArgumentParser(
prog="algonotes delete",
description="给定文件名,从三层存储删除笔记",
)
parser.add_argument('filename', help='笔记文件名')
parser.add_argument("--force", action="store_true",
help="跳过确认提示")
parser.add_argument("--json", type=Path, metavar="FILE",
help="输出结果到 JSON 文件")
args = parser.parse_args()
if not args.force:
ipt = input(f"请确认删除 {args.filename}(键入 stop 以撤销操作): ")
if ipt.lower() == "stop":
exit(0)
result: DeleteResult = delete_note(args.filename)
if result.success:
print(f"成功删除 {result.file_name}")
else:
print(f"{result.file_name} 文件不存在,无法进行删除")
if args.json:
args.json.write_text(
json.dumps(result.__dict__, ensure_ascii=False, indent=2),
encoding="utf-8",
)