'update'
This commit is contained in:
douboer
@@ -6,18 +6,20 @@ annotationdata.py (OOP版)
|
||||
- 解析iBooks的AEAnnotation.sqlite数据库，提取所有或指定书籍（assetid/bookid）的笔记。
|
||||
- 提供parse_location辅助函数，解析笔记定位信息。
|
||||
- 返回结构化的annotations数据，便于后续章节定位与导出。
|
||||
- 使用 EPUB CFI 解析器实现正确的位置排序
|
||||
|
||||
依赖：config.py 统一管理路径和配置项。
|
||||
主要接口：AnnotationManager
|
||||
- get_annotations(bookid=None)：返回所有或指定assetid的笔记，结构为{assetid: {uuid: {...}}}
|
||||
- get_annotations(bookid=None)：返回所有或指定assetid的笔记，结构为{assetid: {uuid: {...}}}，按CFI位置排序
|
||||
- parse_location(location)：解析ZANNOTATIONLOCATION，返回(idref, filepos)
|
||||
依赖：sqlite3, collections, re, os, datetime
|
||||
依赖：sqlite3, collections, re, os, datetime, epub_cfi_parser
|
||||
"""
|
||||
import config
|
||||
import sqlite3
|
||||
import re
|
||||
import os
|
||||
from collections import defaultdict
|
||||
from epub_cfi_parser import EpubCFIParser
|
||||
|
||||
class AnnotationManager:
|
||||
"""
|
||||
@@ -68,10 +70,10 @@ class AnnotationManager:
|
||||
|
||||
def get_annotations(self, bookid=None):
|
||||
"""
|
||||
从数据库获取笔记数据
|
||||
从数据库获取笔记数据，按 CFI 位置排序
|
||||
|
||||
从iBooks的AEAnnotation.sqlite数据库中提取所有或指定书籍的笔记和高亮内容。
|
||||
自动处理时间戳转换和位置信息解析。
|
||||
自动处理时间戳转换和位置信息解析。现在按照 EPUB CFI 位置进行正确排序。
|
||||
|
||||
Args:
|
||||
bookid (str, optional): 书籍资产ID，如果为None则获取所有书籍的笔记
|
||||
@@ -79,52 +81,66 @@ class AnnotationManager:
|
||||
Returns:
|
||||
dict: 笔记数据字典，结构为：
|
||||
{
|
||||
assetid: {
|
||||
uuid: {
|
||||
assetid: [
|
||||
{
|
||||
'uuid': '笔记唯一标识',
|
||||
'creationdate': '创建日期',
|
||||
'filepos': '文件位置',
|
||||
'idref': '章节标识',
|
||||
'note': '笔记内容',
|
||||
'selectedtext': '选中文本'
|
||||
'selectedtext': '选中文本',
|
||||
'location': 'CFI位置字符串',
|
||||
'chapter_info': '章节信息'
|
||||
}
|
||||
}
|
||||
] # 现在返回按CFI位置排序的列表
|
||||
}
|
||||
|
||||
Note:
|
||||
- 会检查WAL模式相关文件(-wal, -shm)的存在性
|
||||
- 自动转换苹果时间戳格式(以2001-01-01为基准)
|
||||
- 过滤掉既没有笔记也没有选中文本的空记录
|
||||
- 按照 EPUB CFI 位置进行排序，确保笔记按阅读顺序排列
|
||||
"""
|
||||
# 检查WAL模式相关文件
|
||||
# 检查WAL模式相关文件（只显示一次警告）
|
||||
base = self.db_path.rsplit('.', 1)[0]
|
||||
wal_path = base + '.sqlite-wal'
|
||||
shm_path = base + '.sqlite-shm'
|
||||
for f in [self.db_path, wal_path, shm_path]:
|
||||
missing_files = []
|
||||
for f in [wal_path, shm_path]:
|
||||
if not os.path.exists(f):
|
||||
print(f'警告: 缺少 {f}，可能无法获取全部最新笔记')
|
||||
missing_files.append(os.path.basename(f))
|
||||
|
||||
if missing_files and not hasattr(self, '_wal_warning_shown'):
|
||||
print(f'提示: 缺少WAL文件 {", ".join(missing_files)}，这是正常的（数据库未被其他进程打开时）')
|
||||
self._wal_warning_shown = True
|
||||
|
||||
# 连接数据库并执行查询
|
||||
conn = sqlite3.connect(self.db_path)
|
||||
cursor = conn.cursor()
|
||||
|
||||
# 根据是否指定bookid选择不同的查询语句
|
||||
# 根据是否指定bookid选择不同的查询语句，使用已有的列
|
||||
if bookid is not None:
|
||||
cursor.execute('''
|
||||
SELECT ZANNOTATIONASSETID, ZANNOTATIONCREATIONDATE, ZANNOTATIONLOCATION, ZANNOTATIONNOTE, ZANNOTATIONSELECTEDTEXT, ZANNOTATIONUUID
|
||||
SELECT ZANNOTATIONASSETID, ZANNOTATIONCREATIONDATE, ZANNOTATIONLOCATION,
|
||||
ZANNOTATIONNOTE, ZANNOTATIONSELECTEDTEXT, ZANNOTATIONUUID,
|
||||
ZPLABSOLUTEPHYSICALLOCATION
|
||||
FROM ZAEANNOTATION WHERE ZANNOTATIONASSETID=?
|
||||
''', (bookid,))
|
||||
else:
|
||||
cursor.execute('''
|
||||
SELECT ZANNOTATIONASSETID, ZANNOTATIONCREATIONDATE, ZANNOTATIONLOCATION, ZANNOTATIONNOTE, ZANNOTATIONSELECTEDTEXT, ZANNOTATIONUUID
|
||||
SELECT ZANNOTATIONASSETID, ZANNOTATIONCREATIONDATE, ZANNOTATIONLOCATION,
|
||||
ZANNOTATIONNOTE, ZANNOTATIONSELECTEDTEXT, ZANNOTATIONUUID,
|
||||
ZPLABSOLUTEPHYSICALLOCATION
|
||||
FROM ZAEANNOTATION
|
||||
''')
|
||||
|
||||
rows = cursor.fetchall()
|
||||
annotations = defaultdict(dict)
|
||||
annotations = defaultdict(list)
|
||||
import datetime
|
||||
|
||||
# 处理每一行数据
|
||||
for row in rows:
|
||||
assetid, creationdate, location, note, selectedtext, uuid = row
|
||||
assetid, creationdate, location, note, selectedtext, uuid, physical_location = row
|
||||
|
||||
# 转换 creationdate格式为'YYYY-MM-DD HH:MM:SS'，支持苹果时间戳（以2001-01-01为基准）
|
||||
date_str = creationdate
|
||||
@@ -148,15 +164,61 @@ class AnnotationManager:
|
||||
# 过滤空记录（既没有笔记也没有选中文本）
|
||||
if note is None and selectedtext is None:
|
||||
continue
|
||||
|
||||
# 提取章节信息
|
||||
chapter_info = EpubCFIParser.extract_chapter_info(location or "")
|
||||
|
||||
# 构建笔记数据结构
|
||||
annotations[str(assetid)][uuid] = {
|
||||
annotation = {
|
||||
'uuid': uuid,
|
||||
'creationdate': date_str,
|
||||
'filepos': filepos,
|
||||
'idref': idref,
|
||||
'note': note,
|
||||
'selectedtext': selectedtext
|
||||
'selectedtext': selectedtext,
|
||||
'location': location or "", # CFI 字符串
|
||||
'chapter_info': chapter_info,
|
||||
'physical_location': physical_location or 0 # 物理位置作为后备排序
|
||||
}
|
||||
|
||||
annotations[str(assetid)].append(annotation)
|
||||
|
||||
conn.close()
|
||||
|
||||
# 对每本书的标注按 CFI 位置排序
|
||||
for assetid in annotations:
|
||||
annotations[assetid].sort(key=self._create_annotation_sort_key)
|
||||
|
||||
# 根据查询类型返回相应结果
|
||||
if bookid is not None:
|
||||
return {str(bookid): annotations.get(str(bookid), [])}
|
||||
return annotations
|
||||
|
||||
def _create_annotation_sort_key(self, annotation: dict) -> tuple:
|
||||
"""
|
||||
为标注创建排序键，优先使用 CFI，失败时回退到物理位置或创建时间
|
||||
|
||||
Args:
|
||||
annotation: 标注数据字典
|
||||
|
||||
Returns:
|
||||
排序元组
|
||||
"""
|
||||
cfi = annotation.get('location', '')
|
||||
|
||||
if cfi:
|
||||
# 尝试 CFI 排序
|
||||
try:
|
||||
cfi_key = EpubCFIParser.create_sort_key(cfi)
|
||||
# CFI 排序成功，返回 CFI 键（优先级 0）
|
||||
return (0, cfi_key)
|
||||
except Exception as e:
|
||||
print(f"CFI 排序失败: {cfi} -> {e}")
|
||||
|
||||
# CFI 排序失败，使用物理位置或创建时间（优先级 1）
|
||||
physical_location = annotation.get('physical_location', 0)
|
||||
creation_date = annotation.get('creationdate', '')
|
||||
return (1, physical_location, creation_date)
|
||||
|
||||
conn.close()
|
||||
|
||||
|
||||
Reference in New Issue
Block a user