初始化 ziwei-power:紫微磁场管理系统(Flask + SQLite + 登录鉴权)
This commit is contained in:
94
database.py
Normal file
94
database.py
Normal file
@@ -0,0 +1,94 @@
|
||||
# -*- 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')
|
||||
|
||||
|
||||
def get_conn():
|
||||
conn = sqlite3.connect(DB_PATH)
|
||||
conn.row_factory = sqlite3.Row
|
||||
return conn
|
||||
|
||||
|
||||
def init_db():
|
||||
"""初始化数据库表"""
|
||||
conn = get_conn()
|
||||
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
|
||||
)
|
||||
''')
|
||||
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
|
||||
Reference in New Issue
Block a user