"""Week 2 · M11 · 文件与序列化进阶 + SQLite——参考答案
对应练习：exercises/week2/w2_08_files_sqlite.py
"""
import csv
import hashlib
import json
import os
import re
import sqlite3
from collections.abc import Iterator, Sequence
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path
from typing import Any
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_created_at ON notes (created_at);
"""
def iter_markdown_files(root: Path) -> Iterator[Path]:
"""递归找出 root 下所有 .md 文件，按路径排序后逐个 yield。
例：iter_markdown_files(Path("research")) -> 逐个 Path
"""
if not root.exists():
raise FileNotFoundError(f"目录不存在：{root}")
for path in sorted(root.rglob("*")):
if not path.is_file() or path.suffix.lower() != ".md":
continue
if any(part.startswith(".") for part in path.relative_to(root).parts[:-1]):
continue
yield path
def file_sha256(path: Path, chunk_size: int = 65536) -> str:
"""算文件的 SHA-256 十六进制摘要，分块读。
例：内容为 b"hello" 的文件 -> "2cf24dba...9824"
"""
digest = hashlib.sha256()
with path.open("rb") as fp:
while chunk := fp.read(chunk_size):
digest.update(chunk)
return digest.hexdigest()
def json_default(obj: Any) -> Any:
"""给 json.dump 的 default= 用：把它不认识的类型转成能序列化的东西。
例：json_default(Decimal("1.5")) -> "1.5"
"""
if isinstance(obj, datetime):
return obj.isoformat()
if isinstance(obj, Decimal):
return str(obj)
if isinstance(obj, set | frozenset):
return sorted(obj)
raise TypeError(f"不支持的类型：{type(obj).__name__}")
def export_json(path: Path, notes: list[dict[str, Any]]) -> None:
"""把 notes 原子地写成 JSON 文件。
例：export_json(p, [{"title": "中文"}]) 后 p 里是 UTF-8 的 JSON 数组
"""
path.parent.mkdir(parents=True, exist_ok=True)
tmp = path.with_suffix(path.suffix + ".tmp")
try:
with tmp.open("w", encoding="utf-8") as fp:
json.dump(notes, fp, ensure_ascii=False, indent=2, default=json_default)
os.replace(tmp, path)
except BaseException:
tmp.unlink(missing_ok=True)
raise
def read_csv_with_bom(path: Path) -> list[dict[str, str]]:
"""读 Excel 导出的 CSV（带 BOM），返回每行一个 dict。
例：表头 name,score 的文件 -> [{"name": "小明", "score": "90"}]
"""
with path.open("r", encoding="utf-8-sig", newline="") as fp:
return [dict(row) for row in csv.DictReader(fp)]
def utc_now_iso() -> str:
"""当前时间的 UTC ISO 8601 字符串（带时区）。
例：utc_now_iso() -> "2026-09-14T02:03:04.567890+00:00"
"""
return datetime.now(UTC).isoformat()
def migrate(conn: sqlite3.Connection) -> None:
"""建表建索引，幂等。
例：migrate(conn); migrate(conn) 不报错
"""
with conn:
conn.executescript(SCHEMA)
conn.execute("PRAGMA journal_mode=WAL")
class SqliteNoteRepository:
"""用 SQLite 存笔记：add/get/list/search/delete。
例：repo = SqliteNoteRepository(tmp_path / "kb.db"); repo.add("标题") -> 1
"""
def __init__(self, db_path: Path) -> None:
"""连上数据库，设置 row_factory，并跑一次 migrate。"""
db_path.parent.mkdir(parents=True, exist_ok=True)
self.conn = sqlite3.connect(db_path)
self.conn.row_factory = sqlite3.Row
migrate(self.conn)
def add(self, title: str, body: str = "", tags: Sequence[str] = ()) -> int:
"""插入一条笔记，返回新的 id。
例：repo.add("标题", tags=["python"]) -> 1
"""
with self.conn:
cursor = self.conn.execute(
"INSERT INTO notes (title, body, tags, created_at) VALUES (?, ?, ?, ?)",
(title, body, ",".join(tags), utc_now_iso()),
)
return int(cursor.lastrowid or 0)
def get(self, note_id: int) -> dict[str, Any] | None:
"""按 id 取一条，不存在返回 None。
例：repo.get(999) -> None
"""
row = self.conn.execute(
"SELECT * FROM notes WHERE id = ?", (note_id,)
).fetchone()
return None if row is None else self._row_to_dict(row)
def list(self) -> list[dict[str, Any]]:
"""按 id 升序返回全部笔记。
例：repo.list() -> [{"id": 1, ...}]
"""
rows = self.conn.execute("SELECT * FROM notes ORDER BY id").fetchall()
return [self._row_to_dict(row) for row in rows]
def search(self, pattern: str, *, regex: bool = False) -> list[dict[str, Any]]:
"""在 title 和 body 里搜，返回命中的笔记。
例：repo.search("python") -> 命中的那些
"""
if regex:
compiled = re.compile(pattern, re.IGNORECASE)
return [
note
for note in self.list()
if compiled.search(f"{note['title']}\n{note['body']}")
]
rows = self.conn.execute(
"SELECT * FROM notes WHERE title LIKE ? OR body LIKE ? ORDER BY id",
(f"%{pattern}%", f"%{pattern}%"),
).fetchall()
return [self._row_to_dict(row) for row in rows]
def delete(self, note_id: int) -> bool:
"""删一条，返回是否真的删掉了。
例：repo.delete(1) -> True；再删一次 -> False
"""
with self.conn:
cursor = self.conn.execute("DELETE FROM notes WHERE id = ?", (note_id,))
return cursor.rowcount > 0
def close(self) -> None:
"""关闭连接。"""
self.conn.close()
def _row_to_dict(self, row: sqlite3.Row) -> dict[str, Any]:
"""把一行 sqlite3.Row 变成 dict，并把 tags 还原成 list[str]。
例：tags 字段是 "a,b" -> {"tags": ["a", "b"], ...}
"""
note = dict(row)
raw_tags = str(note.get("tags") or "")
note["tags"] = [tag for tag in raw_tags.split(",") if tag]
return note