"""仓储层：NoteRepository 协议 + SQLite 实现 + 内存实现（测试/范例用）。"""
import logging
import re
import sqlite3
from pathlib import Path
from typing import Protocol
from kb.models import Note
logger = logging.getLogger(__name__)
SCHEMA = """
CREATE TABLE IF NOT EXISTS notes (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
body TEXT NOT NULL DEFAULT '',
tags TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_notes_title ON notes(title);
"""
class NoteRepository(Protocol):
"""存储层接口：服务层只依赖它，不关心背后是 SQLite 还是内存。"""
def add(self, note: Note) -> Note:
"""保存并返回带 id 的副本。"""
...
def get(self, note_id: int) -> Note | None:
"""按 id 取一条，不存在返回 None。"""
...
def list(self, tag: str | None = None) -> list[Note]:
"""按 id 升序列出（可按标签过滤）。"""
...
def search(self, pattern: str, *, regex: bool = False) -> list[Note]:
"""在标题与正文里按关键词或正则搜索。"""
...
def delete(self, note_id: int) -> bool:
"""按 id 删除，返回是否删掉。"""
...
def _matches(note: Note, pattern: str, regex: bool) -> bool:
"""关键词（不分大小写）或正则是否命中标题/正文。"""
haystack = f"{note.title}\n{note.body}"
if regex:
return re.search(pattern, haystack, re.IGNORECASE) is not None
return pattern.lower() in haystack.lower()
class InMemoryNoteRepository:
"""字典实现：给测试与范例用（已完整实现，请对照着写 SQLite 版）。"""
def __init__(self) -> None:
"""初始化。"""
self._notes: dict[int, Note] = {}
self._next_id = 1
def add(self, note: Note) -> Note:
"""新增一条笔记，返回带 id 的副本。"""
stored = note.model_copy(update={"id": self._next_id})
self._notes[self._next_id] = stored
self._next_id += 1
return stored
def get(self, note_id: int) -> Note | None:
"""按 id 取一条，不存在返回 None。"""
return self._notes.get(note_id)
def list(self, tag: str | None = None) -> list[Note]:
"""按 id 升序返回全部（可按标签过滤）。"""
notes = sorted(self._notes.values(), key=lambda n: n.id or 0)
if tag is not None:
notes = [n for n in notes if tag.lower() in n.tags]
return notes
def search(self, pattern: str, *, regex: bool = False) -> list[Note]:
"""在标题与正文里按关键词（或正则）搜索。"""
return [n for n in self.list() if _matches(n, pattern, regex)]
def delete(self, note_id: int) -> bool:
"""按 id 删除，返回是否真的删掉了。"""
return self._notes.pop(note_id, None) is not None
class SqliteNoteRepository:
"""SQLite 实现：参数化查询、`with conn:` 事务、Row 工厂、幂等建表。"""
def __init__(self, db_path: Path) -> None:
"""初始化。"""
db_path.parent.mkdir(parents=True, exist_ok=True)
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
with self.conn:
self.conn.executescript(SCHEMA)
logger.debug("已连接数据库 %s", db_path)
def close(self) -> None:
"""关闭数据库连接。"""
self.conn.close()
def add(self, note: Note) -> Note:
"""新增一条笔记，返回带 id 的副本。"""
# TODO: 在这里实现
raise NotImplementedError
def get(self, note_id: int) -> Note | None:
"""按 id 取一条，不存在返回 None。"""
# TODO: 在这里实现
raise NotImplementedError
def list(self, tag: str | None = None) -> list[Note]:
"""按 id 升序返回全部（可按标签过滤）。"""
# TODO: 在这里实现
raise NotImplementedError
def search(self, pattern: str, *, regex: bool = False) -> list[Note]:
"""在标题与正文里按关键词（或正则）搜索。"""
# TODO: 在这里实现
raise NotImplementedError
def delete(self, note_id: int) -> bool:
"""按 id 删除，返回是否真的删掉了。"""
# TODO: 在这里实现
raise NotImplementedError
@staticmethod
def _to_note(row: sqlite3.Row) -> Note:
"""_to_note。"""
# TODO: 在这里实现
raise NotImplementedError