314 lines
11 KiB
Python
314 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""ziwei-power SQLite 数据库操作层"""
|
||
|
||
import sqlite3
|
||
import json
|
||
import os
|
||
from datetime import datetime
|
||
|
||
DB_DIR = os.path.join(os.path.expanduser('~'), '.workbuddy', 'data', 'ziwei-power')
|
||
os.makedirs(DB_DIR, exist_ok=True)
|
||
DB_PATH = os.path.join(DB_DIR, 'ziwei_power.db')
|
||
|
||
# 当前数据库 schema 版本 —— 改表结构时必须 +1 并补迁移逻辑
|
||
CURRENT_SCHEMA_VERSION = 4
|
||
|
||
|
||
def get_conn():
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
return conn
|
||
|
||
|
||
def _get_schema_version(conn):
|
||
"""读取当前数据库的 schema 版本，无表时返回 0"""
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS schema_version (
|
||
version INTEGER NOT NULL
|
||
)
|
||
''')
|
||
row = conn.execute('SELECT version FROM schema_version').fetchone()
|
||
return row['version'] if row else 0
|
||
|
||
|
||
def _set_schema_version(conn, version):
|
||
"""写入 schema 版本"""
|
||
conn.execute('DELETE FROM schema_version')
|
||
conn.execute('INSERT INTO schema_version (version) VALUES (?)', (version,))
|
||
|
||
|
||
def init_db():
|
||
"""初始化数据库表 & 自动迁移"""
|
||
conn = get_conn()
|
||
current = _get_schema_version(conn)
|
||
|
||
# ── 迁移步骤（按版本号递增）────────────────────────
|
||
if current < 1:
|
||
# v1: 初始表结构
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS checkins (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
date TEXT UNIQUE NOT NULL,
|
||
data TEXT NOT NULL,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
)
|
||
''')
|
||
|
||
if current < 2:
|
||
# v2: 心愿清单
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS wishes (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL,
|
||
priority TEXT NOT NULL DEFAULT '中',
|
||
deadline TEXT NOT NULL DEFAULT '',
|
||
done INTEGER NOT NULL DEFAULT 0,
|
||
sort_order INTEGER NOT NULL DEFAULT 0,
|
||
created_at TEXT NOT NULL
|
||
)
|
||
''')
|
||
|
||
if current < 3:
|
||
# v3: 优先级改为四象限
|
||
conn.execute("ALTER TABLE wishes ADD COLUMN quadrant TEXT NOT NULL DEFAULT '重要不紧急'")
|
||
conn.execute("UPDATE wishes SET quadrant = CASE priority WHEN '高' THEN '重要紧急' WHEN '中' THEN '重要不紧急' WHEN '低' THEN '不紧急不重要' ELSE '重要不紧急' END WHERE quadrant = '重要不紧急'")
|
||
|
||
if current < 4:
|
||
# v4: 日打卡 → 周打卡
|
||
# 备份旧表 → 聚合 → 新建周表
|
||
existing = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='checkins'").fetchone()
|
||
if existing:
|
||
# 检查是否已有备份表（幂等）
|
||
backup_exists = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='checkins_daily_backup'").fetchone()
|
||
if not backup_exists:
|
||
conn.execute("ALTER TABLE checkins RENAME TO checkins_daily_backup")
|
||
# 重新创建 checkins 表（结构不变，date 语义变为周号）
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS checkins (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
date TEXT UNIQUE NOT NULL,
|
||
data TEXT NOT NULL,
|
||
created_at TEXT NOT NULL,
|
||
updated_at TEXT NOT NULL
|
||
)
|
||
''')
|
||
# 聚合日数据到周
|
||
if not backup_exists:
|
||
_aggregate_daily_to_weekly(conn)
|
||
|
||
# ── 将来加字段/改表在此扩展 ──
|
||
# if current < 2:
|
||
# conn.execute('ALTER TABLE checkins ADD COLUMN tags TEXT DEFAULT ""')
|
||
# # 可选：对已有行做数据补全
|
||
# conn.execute("UPDATE checkins SET tags = '[]' WHERE tags IS NULL")
|
||
|
||
# ── 写入最新版本号 ──
|
||
_set_schema_version(conn, CURRENT_SCHEMA_VERSION)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_checkin(date_str):
|
||
"""获取某天的打卡记录，返回 dict 或 None"""
|
||
conn = get_conn()
|
||
row = conn.execute('SELECT * FROM checkins WHERE date = ?', (date_str,)).fetchone()
|
||
conn.close()
|
||
if row:
|
||
return {
|
||
'id': row['id'],
|
||
'date': row['date'],
|
||
'data': json.loads(row['data']),
|
||
'created_at': row['created_at'],
|
||
'updated_at': row['updated_at']
|
||
}
|
||
return None
|
||
|
||
|
||
def save_checkin(date_str, data_dict):
|
||
"""保存或更新打卡记录"""
|
||
now = datetime.now().isoformat()
|
||
conn = get_conn()
|
||
existing = conn.execute('SELECT id FROM checkins WHERE date = ?', (date_str,)).fetchone()
|
||
json_data = json.dumps(data_dict, ensure_ascii=False)
|
||
if existing:
|
||
conn.execute(
|
||
'UPDATE checkins SET data = ?, updated_at = ? WHERE date = ?',
|
||
(json_data, now, date_str)
|
||
)
|
||
else:
|
||
conn.execute(
|
||
'INSERT INTO checkins (date, data, created_at, updated_at) VALUES (?, ?, ?, ?)',
|
||
(date_str, json_data, now, now)
|
||
)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_checkin(date_str):
|
||
"""删除某天的打卡记录"""
|
||
conn = get_conn()
|
||
conn.execute('DELETE FROM checkins WHERE date = ?', (date_str,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_all_checkins():
|
||
"""获取所有打卡记录，按日期倒序"""
|
||
conn = get_conn()
|
||
rows = conn.execute('SELECT * FROM checkins ORDER BY date DESC').fetchall()
|
||
conn.close()
|
||
results = []
|
||
for row in rows:
|
||
results.append({
|
||
'id': row['id'],
|
||
'date': row['date'],
|
||
'data': json.loads(row['data']),
|
||
'created_at': row['created_at'],
|
||
'updated_at': row['updated_at']
|
||
})
|
||
return results
|
||
|
||
|
||
# ── 心愿清单 CRUD ──────────────────────────────────
|
||
|
||
def get_wishes():
|
||
"""获取所有心愿，按 sort_order 排序"""
|
||
conn = get_conn()
|
||
rows = conn.execute('SELECT * FROM wishes ORDER BY sort_order').fetchall()
|
||
conn.close()
|
||
return [dict(row) for row in rows]
|
||
|
||
|
||
def save_wish(name, quadrant, deadline):
|
||
"""新增一条心愿"""
|
||
now = datetime.now().isoformat()
|
||
conn = get_conn()
|
||
max_order = conn.execute('SELECT COALESCE(MAX(sort_order), -1) + 1 AS n FROM wishes').fetchone()['n']
|
||
conn.execute(
|
||
'INSERT INTO wishes (name, quadrant, deadline, done, sort_order, created_at) VALUES (?, ?, ?, 0, ?, ?)',
|
||
(name, quadrant, deadline, max_order, now)
|
||
)
|
||
conn.commit()
|
||
wish_id = conn.execute('SELECT last_insert_rowid()').fetchone()[0]
|
||
conn.close()
|
||
return wish_id
|
||
|
||
|
||
def update_wish(wish_id, **kwargs):
|
||
"""更新心愿字段"""
|
||
allowed = ['name', 'quadrant', 'deadline', 'done']
|
||
updates = {k: v for k, v in kwargs.items() if k in allowed}
|
||
if not updates:
|
||
return
|
||
conn = get_conn()
|
||
sets = ', '.join(f'{k} = ?' for k in updates)
|
||
vals = list(updates.values()) + [wish_id]
|
||
conn.execute(f'UPDATE wishes SET {sets} WHERE id = ?', vals)
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def delete_wish(wish_id):
|
||
"""删除心愿"""
|
||
conn = get_conn()
|
||
conn.execute('DELETE FROM wishes WHERE id = ?', (wish_id,))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def reorder_wishes(order_list):
|
||
"""批量更新排序：order_list = [id1, id2, ...]"""
|
||
conn = get_conn()
|
||
for idx, wid in enumerate(order_list):
|
||
conn.execute('UPDATE wishes SET sort_order = ? WHERE id = ?', (idx, wid))
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def get_weeks_list():
|
||
"""返回所有已有打卡记录的周号列表"""
|
||
conn = get_conn()
|
||
rows = conn.execute('SELECT DISTINCT date FROM checkins ORDER BY date').fetchall()
|
||
conn.close()
|
||
return [r['date'] for r in rows]
|
||
|
||
|
||
def get_daily_backup(week_str):
|
||
"""获取某周的原始日记录（从备份表）"""
|
||
from datetime import datetime as _dt, timedelta
|
||
if '-W' not in week_str:
|
||
return []
|
||
year_part, week_part = week_str.split('-W')
|
||
try:
|
||
year = int(year_part)
|
||
week = int(week_part)
|
||
except (ValueError, IndexError):
|
||
return []
|
||
jan1 = _dt(year, 1, 1)
|
||
delta = timedelta(days=(week - 1) * 7 - jan1.weekday())
|
||
monday = jan1 + delta
|
||
dates = [monday.strftime('%Y-%m-%d')]
|
||
for i in range(1, 7):
|
||
dates.append((monday + timedelta(days=i)).strftime('%Y-%m-%d'))
|
||
conn = get_conn()
|
||
placeholders = ','.join('?' for _ in dates)
|
||
exists = conn.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='checkins_daily_backup'").fetchone()
|
||
if not exists:
|
||
conn.close()
|
||
return []
|
||
rows = conn.execute(
|
||
f"SELECT date, data FROM checkins_daily_backup WHERE date IN ({placeholders}) ORDER BY date",
|
||
dates
|
||
).fetchall()
|
||
conn.close()
|
||
results = []
|
||
for row in rows:
|
||
results.append({'date': row['date'], 'data': json.loads(row['data'])})
|
||
return results
|
||
|
||
|
||
def _aggregate_daily_to_weekly(conn):
|
||
"""将 checkins_daily_backup 的日数据聚合为周数据，写入 checkins"""
|
||
from datetime import datetime as _dt
|
||
rows = conn.execute('SELECT date, data FROM checkins_daily_backup ORDER BY date').fetchall()
|
||
weekly = {}
|
||
for row in rows:
|
||
try:
|
||
d = _dt.strptime(row['date'], '%Y-%m-%d')
|
||
except ValueError:
|
||
continue
|
||
iso_year, iso_week, _ = d.isocalendar()
|
||
week_key = f"{iso_year}-W{iso_week:02d}"
|
||
if week_key not in weekly:
|
||
weekly[week_key] = {'morning': [], 'evening': [], 'study': []}
|
||
data = json.loads(row['data'])
|
||
for mi in data.get('morning', []):
|
||
text = mi if isinstance(mi, str) else mi.get('text', '')
|
||
if text:
|
||
existing_texts = {x if isinstance(x, str) else x.get('text', '') for x in weekly[week_key]['morning']}
|
||
if text not in existing_texts:
|
||
weekly[week_key]['morning'].append(mi)
|
||
for ei in data.get('evening', []):
|
||
mst = ei.get('mistake', '') if isinstance(ei, dict) else ''
|
||
if mst:
|
||
existing = {x.get('mistake', '') if isinstance(x, dict) else x for x in weekly[week_key]['evening']}
|
||
if mst not in existing:
|
||
weekly[week_key]['evening'].append(ei if isinstance(ei, dict) else {'mistake': ei, 'improvement': ''})
|
||
existing_names = {}
|
||
for si in weekly[week_key]['study']:
|
||
nm = si.get('name', '') if isinstance(si, dict) else str(si)
|
||
if nm:
|
||
existing_names[nm] = si
|
||
for si in data.get('study', []):
|
||
nm = si.get('name', '') if isinstance(si, dict) else str(si)
|
||
if nm and nm not in existing_names:
|
||
weekly[week_key]['study'].append(si)
|
||
existing_names[nm] = si
|
||
now = _dt.now().isoformat()
|
||
for week_key in sorted(weekly.keys()):
|
||
conn.execute(
|
||
'INSERT INTO checkins (date, data, created_at, updated_at) VALUES (?, ?, ?, ?)',
|
||
(week_key, json.dumps(weekly[week_key], ensure_ascii=False), now, now)
|
||
)
|