代码拉取完成，页面将自动刷新
#########################################################
## @file : kman.py
## @desc : kindle note managerment tool
## @create : 2020/05/26
## @author : Chengan
## @email : douboer@gmail.com
#########################################################
import re
import sys
import os
import io
import json
import time
import logging
import platform
import subprocess
from collections import defaultdict
# data structure - use dict
'''
books_data =
{
"bookname_xxx": {
"author": "李",
"section1636": {
"content": "张",
"day": "2020年4月3日",
"meridiem": "下午",
"position": "311-311",
"time": "3:00:53",
"type": "HL",
"week": "星期五"
},
"section1651": {
"content": "治",
"day": "2020年4月3日",
"meridiem": "下午",
"position": "514",
"time": "3:43:50",
"type": "NT",
"week": "星期五"
},
"section1814": {
"content": null,
"day": "2020年4月12日",
"meridiem": "下午",
"position": "5186",
"time": "2:20:12",
"type": "BM",
"week": "星期日"
},
...
},
...
}
'''
# modi clippath for different os
SYS = 'WIN' if platform.system()=='Windows' else \
('LINUX' if platform.system()=='LINUX' else 'MAC')
frozen = 'not'
#CURRPATH = os.path.dirname(os.path.realpath(sys.argv[0]))
CURRPATH = ''
if getattr(sys, 'frozen', False):
# we are running in a bundle
frozen = 'ever so'
CURRPATH = sys._MEIPASS
os.chdir(CURRPATH)
else:
# we are running in a normal Python environment
CURRPATH = os.path.dirname(os.path.abspath(__file__))
# some constants
LASTLINE = '=========='
NTPREF = '--CG注:'
CLIPFN = 'My Clippings.txt'
WORDFN = 'vocab.db'
OUTPREF = os.path.join(CURRPATH,'CLIP')
DEBUG = 1 # 0 - INFO; 1 - DEBUG
LOG2FILE = 1 # 0 - to stdio; 1 - to file
DELIMITER= '|'
LOGFILE = os.path.join(CURRPATH,'log')
BACKUPFOLDER = os.path.join(CURRPATH,'backup')
BACKUPNOTEFN = os.path.join(BACKUPFOLDER,'bk.note.data') # kindle notes
BACKUPWORDFN = os.path.join(BACKUPFOLDER,'bk.word.data') # kindle words
BACKUPINFOFN = os.path.join(BACKUPFOLDER,'bk.info.data') # book information from douban/amazon
#HEADER = {0:'type',1:'bookname',2:'author',3:'position',4:'date',5:'content'}
# log info
logger = logging.getLogger()
#formatter = logging.Formatter
# ('%(asctime)s - %(pathname)s[line:%(lineno)d] - %(levelname)s: %(message)s')
formatter = logging.Formatter('')
if LOG2FILE:
handler = logging.FileHandler(LOGFILE)
handler.setFormatter(formatter)
logger.addHandler(handler)
else:
logging.basicConfig(format='%(levelname)s: %(message)s', level=logging.INFO)
if DEBUG:
logger.setLevel(logging.DEBUG)
#author & bookname info
au = re.compile(
r'''
^\ufeff*
(.+) \( #bookname
(.+)\) #author
''', flags=re.X )
# page & date info
#\(\d\+-\{0,1}\d\+\).\+\(\d\{4}年\d\{1,2}月\d\{1,2}日\)\(星期.\) \(..\)\(\d\{1,2}:\d\{1,2}:\d\{1,2}\)
da = re.compile(
r'''
\#
(\d+-{0,1}\d+) #group1 - page
.+
(笔记|标注|书签) #group2 - type
.+
(\d{4}年\d{1,2}月\d{1,2}日) #group3 - xxxx年xx月xx日
(星期.) #group4 - week
\s
(..) #group5 - pm/am
(\d{1,2}:\d{1,2}:\d{1,2}) #group6 - time
''', flags=re.X )
class kMan:
def __init__(self, parent=None):
self.hlnum = 0
self.ntnum = 0
self.refleshtime = '2020/10/10 10:00:00'
self.status = self.status_info()
def status_info(self):
s1 = u'Hightlight: {} Note: {} RefleshTime: {}'. \
format(self.hlnum,self.ntnum,self.refleshtime)
kp = self.get_kindle_path()
if not kp:
s2 = u'Disconnected ({})'.format(os.path.join(CURRPATH,CLIPFN))
else:
with open(os.path.join(kp,'system','version.txt'), 'r', encoding='utf8', errors='ignore') as f:
s2 = u'Connected ({}) version {}'.format(kp,f.read().strip())
return [s1,s2]
def parse_section(self,s,i):
"""parse section
Args:
s: section line list
i: section index
Returns:
dict like this:
d = { 'bookname':bookname,
bookname: {
'author':author
'0':{
'type':'HL',
'position':'123',
'day':'2020年5月26日',
'week':'星期二',
'meridiem':'PM',
'time':'10:26:31'
'content':content }}}
"""
# 1. highlight over the picture, the content(#3 line) is empty, only two lines
# 2. bookmark section only two lines
# 3. other not correct format < 2
if len(s)<=2:
return False
# parse #2 line
section = defaultdict(dict)
"""
authinfo = sec[0]
dateinfo = sec[1]
content = sec[2] if len(sec)==3 else None
"""
(authinfo, dateinfo, content) = \
(s[0], s[1], s[2] if len(s)==3 else None)
das = da.search(dateinfo)
# type of section
'''
STAT :
START - start line of section
BM - section is a bookmark
HL - section is a highlight
NT - section is a note
'''
tpy = ('HL' if das.group(2)=='标注' else \
('NT' if das.group(2)=='笔记' else 'BM'))
"""
pos = das.group(1)
day = das.group(3)
week = das.group(4)
pmam = das.group(5)
time = das.group(6)
"""
(pos, x, day, week, pmam, time) = das.groups()[0:6]
# parse #1 line
aus = au.search(authinfo)
bookname = aus.group(1).strip().replace(' ','')
author = aus.group(2).strip().replace(' ','')
section[bookname]['author'] = author
section['bookname'] = bookname
section[bookname][str(i)] = {
'type':tpy,
'position':pos,
'day':day,
'week':week,
'meridiem':pmam,
'time':time,
'content':content }
return section
def format_time(self,ds):
""" format date
Args:
ds: 2020年1月13日 星期一 上午 8:11:05
Return:
2020/1/13 20:11:05
"""
d = ds.split(' ')
res = re.search(r'(\d{4}).(\d{1,2}).(\d{1,2})',d[0])
ymd = '/'.join(res.groups())
res = re.search(r'(\d{1,2})(:\d{1,2}:\d{1,2})',d[3])
tm = ' '+str(int(res.group(1)) + (0 if d[2]=='上午' else 12))+res.group(2)
return ymd+tm
def format_note_data(self, bks, ft='MD'):
""" format notes data for MD & CSV
Args:
bks: books dict
f: can be 'MD'/'CSV'
Return:
list [header, sections]
header and sections are lists
"""
hd =[] # header
secs =[] # content
DELIMITER = '|' if ft=='MD' else ','
hd.append(DELIMITER.join(['TYPE','BOOKNAME','AUTHOR','MARKTIME','CONTENT']))
if ft=='MD':
hd.append(DELIMITER.join(['--' for i in range(5)]))
for kb,vb in bks.items():
author = vb['author']
for ks, vs in vb.items():
if ks == 'author': continue
secs.append(DELIMITER.join([vs['type'],kb,author, \
self.format_time(' '.join([vs['day'],vs['week'],\
vs['meridiem'],vs['time']])),vs['content']]))
return hd+secs
def format_word_data(self, wdata, flist, ft='MD'):
""" format words data for MD & CSV
Args:
wdata: words data [bookinfo, words, lookups]
flist: list to be formatted
ft: can be 'MD'/'CSV'
Return:
list [header, content]
header and content are lists
"""
[bookinfo, words, lookups] = wdata
hd =[] # header
secs =[] # content
DELIMITER = '|' if ft=='MD' else ','
hd.append(DELIMITER.join(['WORD','BOOKNAME','AUTHOR','CATEGORY','USAGE','TIMESTAMP']))
if ft=='MD':
hd.append(DELIMITER.join(['--' for i in range(6)]))
for frow in flist:
[fword,fbookname,fauthor,fcategory]=frow
for lrow in lookups:
lbookname = bookinfo[lrow[2]]['bookname']
lauthor = bookinfo[lrow[2]]['author']
[lusage,ltimestamp] = lrow[4:6]
if words[lrow[1]]['word']==fword:
secs.append(DELIMITER.join(
[fword,lbookname,lauthor,fcategory,lusage,ltimestamp]))
return hd+secs
def export_notes(self, bks, fnpref, ft='MD'):
"""format output and write to file
markdown format:
TYPE | bookname | author | marktime | content
--|--|--|--|--
xx|xx|xx|xx|xx
CSV format:
TYPE,bookname,author,marktime,content
xx,xx,xx,xx,xx
marktime: 20200403 PM 3:0:3 星期五
Args:
bks: books dict
f: can be 'MD'/'JSON'/'CSV'
Returns: special format of 'bks' dict
"""
suff = {'MD':'.md','CSV':'.csv','JSON':'.json'}
op = os.path.join(CURRPATH, fnpref+suff[ft])
with open(op, 'w', encoding='utf8', errors='ignore') as fw:
if ft=='JSON':
fw.write(json.dumps(bks, indent=4, sort_keys=True, ensure_ascii=False))
elif ft in ['MD','CSV']:
for s in self.format_note_data(bks, ft):
fw.write(s)
fw.write('\n')
else:
fw.write(json.dumps(bks)) # only for load back
# no good! books & words operation need to be abstracted
def export_words(self, wdata, flist, fnpref, ft='MD'):
"""format output and write to file
markdown format:
TYPE | bookname | author | marktime | content
--|--|--|--|--
xx|xx|xx|xx|xx
CSV format:
TYPE,bookname,author,marktime,content
xx,xx,xx,xx,xx
marktime: 20200403 PM 3:0:3 星期五
Args:
bks: books dict
f: can be 'MD'/'JSON'/'CSV'
Returns: special format of 'bks' dict
"""
suff = {'MD':'.md','CSV':'.csv','JSON':'.json'}
op = os.path.join(CURRPATH,fnpref+suff[ft])
with open(op, 'w', encoding='utf8', errors='ignore') as fw:
if ft=='JSON':
fw.write(json.dumps(flist, indent=4, sort_keys=True, ensure_ascii=False))
elif ft in ['MD','CSV']:
for s in self.format_word_data(wdata, flist, ft):
fw.write(s)
fw.write('\n')
else:
fw.write(json.dumps(flist)) # only for load back
def drop_duplicate(self,bks):
""" drop duplicated section
If I mark second time in same place, kindle will create two note,
so I need to remove the duplication record
Args:
bks: books dict
Return:
books remove duplicate sections
"""
[preks,prevs] = ['',{'content':'!#$%^&$%','type':'xx'}]
for kb,vb in bks.items():
# add copy() or throw RuntimeError: dictionary changed size during iteration
# reference - http://www.cocoachina.com/articles/89748
for ks, vs in vb.copy().items():
if ks == 'author': continue
if (vs['content'] in prevs['content'] or \
prevs['content'] in vs['content']) and \
prevs['type'] == vs['type']:
bks[kb].pop(preks)
#if vs['content'] != prevs['content']:
# print('prevs',prevs['type'],prevs['content'])
# print(' vs', vs['type'], vs['content'])
preks = ks
prevs = vs
return bks
def get_totalnum_nt(self,bks):
""" get total number of note
Args:
bks: books dict
Return: total number of note
"""
nu = 0
for kb,vb in bks.items():
for ks, vs in vb.copy().items():
if ks == 'author': continue
nu += 1
return nu
def get_bookname_num(self,bks):
""" get note number of booknames
Args:
bks: books dict
Return: [total books, note number of books]
"""
bksnum = defaultdict(dict)
nu = 0 # number of books
for kb,vb in bks.items():
bksnum.setdefault(kb, 0)
nu += 1
for ks, vs in vb.copy().items():
if ks == 'author': continue
bksnum[kb] += 1
return [nu, bksnum]
def get_author_num(self,bks):
""" get note number of author
Args:
bks: books dict
Return: [total authors, note number of authors]
"""
bksnum = defaultdict(dict)
nu = 0 # number of authors
for kb,vb in bks.items():
for ks, vs in vb.copy().items():
if ks == 'author':
nu += 1
continue
au = vb['author']
bksnum.setdefault(au, 0)
bksnum[au] += 1
return [nu, bksnum]
def filter_clips(self, bks, info=None, tp=0):
""" filter clips to show the clips in table view
Args:
bks: books dict
info: filter by bookname or author information
tp: type to be filter
0: root item clicked
1: bookname item clicked
2: author item clicked
Return: list: [filter books, filter seclist]
filter_seclist:
[[Type,Bookname,Author,Position,Date,content],
[Type,Bookname,Author,Position,Date,content]
....]
"""
nbks = defaultdict(dict)
# do not filter
if tp==0: nbks = bks
# do filter
for kb, vb in bks.items():
if [info, tp] in ([kb, 1], [vb['author'], 2]):
nbks[kb] = vb
seclist = []
idx = 0
for kb, vb in nbks.items():
for ks,vs in vb.items():
if ks == 'author': continue
tm = self.format_time(' '.join([vs['day'],vs['week'], \
vs['meridiem'],vs['time']]))
nttype = '标注' if vs['type']=='HL' else '笔记'
seclist.append([nttype,kb,vb['author'],vs['position'],tm,vs['content']])
idx += 1
return [nbks, seclist]
def add_note_to_highlight(self,bks):
""" append note content to corresponding highlight
and remove NT sections
Args:
bks: books dict
Return:
changed books
"""
[preks,prevs] = ['',{'content':'!#$%^&$%','type':'xx'}]
for kb,vb in bks.items():
for ks,vs in vb.copy().items():
if ks == 'author': continue
if [prevs['type'], vs['type']] == ['HL','NT']:
bks[kb][preks]['content'] += str(NTPREF+vs['content'])
bks[kb].pop(ks)
preks = ks
prevs = vs
return bks
def search_clip(self, bks, s, t='ALL', p='ALL'):
"""search clip, searching scope may be title/author/content
Args:
input: bks: books dict
s: key word
t: 'ALL'
'HL'
'BM'
'NT'
p: 'ALL'
'TITLE'
'AUTHOR'
'CONTENT'
Return:
[number of result , result dict]
"""
nbks = defaultdict(dict)
nu = 0
for kb,vb in bks.items():
num_nt_book = 0
for ks,vs in vb.copy().items():
if ks == 'author':
nbks[kb][ks] = vs
continue
if t in ['ALL', vs['type']]:
scopestr = {'ALL':''.join([kb,vb['author'],vs['content']]), \
'TITLE':kb, 'AUTHOR':vb['author'], 'CONTENT':vs['content']}
found = re.search(s, scopestr[p])
if found:
nbks[kb][ks] = vs
num_nt_book += 1
nu += 1
if num_nt_book==0:
nbks.pop(kb)
return [nu,nbks]
# to be implement
def statistic(self,bks):
pass
def dict2json(self,d):
"""convert dict to json
Args: d is the dict
Return: json string
"""
jstr = json.dumps(d)
return jstr
def json2dict(self,jf):
"""convert dict to json
Args: jf is the file saved json string
Return: dict
"""
d = {}
with open(jf, 'r', encoding='utf8', errors='ignore') as f:
d=json.load(f)
return d
def get_kindle_path(self):
"""check and return kindle device path
Args:
Return:
if kindle connected, return path string of kindle device
else return false
"""
cmd = "wmic logicaldisk get name,volumename" if os.name=='nt'\
else ("ls /Volumes/Kindle" if os.name=='posix' else '')
# not test for windows & linux
with subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, \
stderr=subprocess.PIPE, bufsize=-1) as s:
#Windows:
# UnicodeDecodeError: 'utf-8' codec can't decode byte 0xd0 in position 48:
# invalid continuation byte
#stream_stdout = io.TextIOWrapper(s.stdout, encoding='utf-8')
#stream_stderr = io.TextIOWrapper(s.stderr, encoding='utf-8')
stream_stdout = io.TextIOWrapper(s.stdout, encoding='GBK')
stream_stderr = io.TextIOWrapper(s.stderr, encoding='GBK')
sout = str(stream_stdout.read())
#serr = str(stream_stderr.read())
#if sout: print('stdout {}'.format(sout))
#if serr: print('stderr {}'.format(serr))
#print('os.name {} sout {}'.format(os.name,sout))
if os.name == 'nt': # windows
for d in sout.split('\n'):
if 'Kindle' in d:
p = re.split(r'\s+',d)[0]
return p
elif os.name == 'posix': # mac os
if sout: return('/Volumes/Kindle/')
else:
pass
"""
# will print error information on stdout
with os.popen(cmd) as s:
sout = s.read()
if os.name == 'nt': # windows
for d in sout.split('\n'):
if 'Kindle' in d: return d.split('\s+')[0]
elif os.name == 'posix': # mac os
if sout: return('/Volumes/Kindle')
else:
pass
"""
return False
def import_clips(self, fp=os.path.join(CURRPATH, CLIPFN)):
"""import clips from local file or kindle
4 lines for each section seperated with '======='
so read 4 lines before '======='
Args: fp - file path
Return: 0 - want to import kindle but kindle is not connected
books_data dict
"""
# check kindle by user just call get_kindle_path()
"""
if tp=='kindle':
kp = get_kindle_path()
if not kp: return 0
else: path = kp
else:
path = fn
"""
if not os.path.exists(fp): return {}
# loop to fill books dict
with open(fp, 'r', encoding='utf8', errors='ignore') as f:
bks = defaultdict(dict)
secd = defaultdict(dict)
sidx = 0
idx = 0
sec = []
for line in f.readlines():
line = line.strip()
if re.match(r'^\s*$',line): continue
idx += 1
if not re.search(LASTLINE,line):
# content more than 1 line
if idx>3:
sec[2] += str(' '+line)
#logger.debug('idx {} {}'.format(idx, sec[2]))
else:
sec.append(line)
#logger.debug('idx {} {}'.format(idx, sec[idx-1]))
else:
idx = 0
sidx += 1
# parsing section & fill data structure
secd = self.parse_section(sec,sidx)
if secd:
bn = secd['bookname']
tpy = secd[bn][str(sidx)]['type']
bks[bn]['author'] = secd[bn]['author']
bks[bn][str(sidx)] = secd[bn][str(sidx)]
# not add note to highlight content here,
# because NT maybe duplicated, we need remove duplication record before
"""
if tpy=='NT' and bks[bn][str(sidx-1)]['type']=='HL':
bks[bn][str(sidx-1)]['content'] += str(NTPREF+sec[2])
"""
if tpy=='HL': self.hlnum += 1
elif tpy=='NT': self.ntnum += 1
else: # BM or not correct format section
sidx -= 1
# initial section for next section loop
sec = []
self.refleshtime = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
return self.drop_duplicate(bks)
# words operations
def import_words(self, fp=(os.path.join(CURRPATH,WORDFN))):
"""import words from local file or kindle
vocab.db -> tables: BOOK_INFO, LOOKUPS, WORDS
Args: fp - file path
Return: words_data = [bookinfo, words, lookups]
bookinfo = { 'book_key': {
'bookname':'bookname_xxx',
'author':'author_xxx'},}
words = { 'word_id': {
'word':'梅花鹿',
'category':0},
}
lookups = [['id','word_key','book_key','pos','usage','timestamp'],
['id','word_key','book_key','pos','usage','timestamp'],]
"""
import sqlite3
conn = sqlite3.connect(fp)
#[bookinfo, words, lookups] = [defaultdict(dict),defaultdict(dict),defaultdict(dict)]
[bookinfo, words, lookups] = [defaultdict(dict),defaultdict(dict),[]]
# select records from vocab.db -> tables: BOOK_INFO, LOOKUPS, WORDS
try:
cursor = conn.cursor()
cursor.execute('''select * from BOOK_INFO;''')
for row in cursor.fetchall():
bookinfo[row[0]]['bookname'] = row[4].replace(' ','')
bookinfo[row[0]]['author'] = row[5]
bookinfo[row[0]]['lang'] = row[5]
cursor.execute('''select * from WORDS;''')
for row in cursor.fetchall():
words[row[0]]['word'] = row[1]
#words[row[0]]['lang'] = row[3]
words[row[0]]['category'] = row[4]
cursor.execute('''select * from LOOKUPS;''')
for row in cursor.fetchall():
timearr = time.localtime(row[6]/1000.) # NOTE: length of timestring is 13
mtime = time.strftime("%Y/%m/%d %H:%M:%S", timearr)
"""
lookups[row[0]]['word_key'] = row[1]
lookups[row[0]]['book_key'] = row[2]
lookups[row[0]]['pos'] = row[4]
lookups[row[0]]['usage'] = row[5]
lookups[row[0]]['timestamp'] = mtime
"""
lookups.append([row[0],row[1],row[2],row[4],row[5],mtime])
except Exception as e:
print('SQL Failure: {}'.format(e))
finally:
conn.close()
words_data = [bookinfo, words, lookups]
return words_data
def filter_words(self, wdata, info=None, tp=0):
""" filter words to show the clips in table view
Args:
wdata = [bookinfo, words, lookups]
info: filter by bookname or word information
tp: type to be filter
0: bookname item clicked
1: word item clicked
Return: [filtted words_data, word_list]
word_list:
[[Word,Bookname,Author,category],
[word,Bookname,Author,category],
....]
"""
[bookinfo, words, lookups] = wdata
word_list = []
# bookname tree clicked
if tp == 0:
for row in lookups:
word = words[row[1]]['word']
category = 'learning' if words[row[1]]['category']==0 else 'mastered'
bookname = bookinfo[row[2]]['bookname']
author = bookinfo[row[2]]['author']
if bookname==info or info == None:
# maybe have duplication records
word_list.append((word,bookname,author,category))
# word tree clicked
elif tp == 1:
pass
else:
return
# remove duplication
return list(set(word_list))
def get_lookups_word_num(self, wdata):
pass
def get_word_num(self, wdata):
pass
# show in tree view
def get_book_word_num(self,wdata):
[bookinfo, words, lookups] = wdata
bk_wd_num = defaultdict(dict)
nu = 0
for row in lookups:
nu += 1
word = words[row[1]]['word']
bookname = bookinfo[row[2]]['bookname']
bk_wd_num.setdefault(bookname,0)
bk_wd_num[bookname] += 1
return [nu, bk_wd_num]
class Util:
def __init__(self, parent=None):
pass
@staticmethod
def get_app_path():
"""Returns the base application path."""
if hasattr(sys, 'frozen'):
# Handles PyInstaller
return os.path.dirname(sys.executable) # 使用pyinstaller打包后的exe目录
return os.path.dirname(__file__) # 没打包前的py目录
if __name__=='__main__':
"""
# XXX move to unitest.kman.py
#books = defaultdict(dict)
km = kMan()
books = km.import_clips()
# remove duplication
km.drop_duplicate(books)
# test search note function
searchnote = km.search_clip(books, '三大都市圈', 'ALL', 'CONTENT')
if searchnote[0] > 0: km.export_notes(searchnote[1], 'searchcontent', ft='MD')
searchnote = km.search_clip(books, '经济', 'ALL', 'TITLE')
if searchnote[0] > 0: km.export_notes(searchnote[1], 'searchtitle', ft='MD')
searchnote = km.search_clip(books, '巴曙松', 'ALL', 'AUTHOR')
if searchnote[0] > 0: km.export_notes(searchnote[1], 'searchauthor', ft='MD')
#print(km.get_bookname_num(books))
#print(km.get_author_num(books))
# add note content to hightlight, then delete note
km.add_note_to_highlight(books)
# test dict json convert
with open(os.path.join(CURRPATH,'xx'), 'w', encoding='utf8', errors='ignore') as fw:
fw.write(km.dict2json(books))
if km.json2dict(os.path.join(CURRPATH,'xx'))==books: print( 'test OK')
km.export_notes(books, OUTPREF, ft='MD')
# print data with json format
logger.debug(json.dumps(books, indent=4, sort_keys=True, ensure_ascii=False))
"""
pass
此处可能存在不合适展示的内容，页面不予展示。您可通过相关编辑功能自查并修改。
如您确认内容无涉及 不当用语 / 纯广告导流 / 暴力 / 低俗色情 / 侵权 / 盗版 / 虚假 / 无价值内容或违法国家有关法律法规的内容，可点击提交进行申诉，我们将尽快为您处理。