SakurasanandClaude Opus 4.7
5153fa28de
- docker-compose.yml: bind mount → named volume (auv_data:/app/data) - backend/database.py: DB_PATH 改为 data/stock_data.db，自动迁移旧数据库 - .gitignore: 忽略 *.db 文件 Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
74 lines
2.1 KiB
Python
74 lines
2.1 KiB
Python
import sqlite3
|
||
import os
|
||
|
||
DB_PATH = os.path.join(os.path.dirname(__file__), "data", "stock_data.db")
|
||
|
||
SCHEMA_SQL = """
|
||
CREATE TABLE IF NOT EXISTS stock_collections (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
name TEXT NOT NULL,
|
||
description TEXT,
|
||
user_id TEXT NOT NULL DEFAULT '',
|
||
last_accessed_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
|
||
updated_at TEXT NOT NULL DEFAULT (datetime('now','localtime'))
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS collection_stocks (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
collection_id INTEGER NOT NULL REFERENCES stock_collections(id) ON DELETE CASCADE,
|
||
stock_code TEXT NOT NULL,
|
||
stock_name TEXT NOT NULL,
|
||
added_price REAL,
|
||
added_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
|
||
UNIQUE(collection_id, stock_code)
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS share_links (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
collection_id INTEGER NOT NULL REFERENCES stock_collections(id) ON DELETE CASCADE,
|
||
short_code TEXT NOT NULL UNIQUE,
|
||
created_at TEXT NOT NULL DEFAULT (datetime('now','localtime')),
|
||
expires_at TEXT
|
||
);
|
||
|
||
CREATE TABLE IF NOT EXISTS cache (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT NOT NULL,
|
||
expires_at TEXT NOT NULL
|
||
);
|
||
"""
|
||
|
||
|
||
def get_connection() -> sqlite3.Connection:
|
||
os.makedirs(os.path.dirname(DB_PATH), exist_ok=True)
|
||
conn = sqlite3.connect(DB_PATH)
|
||
conn.row_factory = sqlite3.Row
|
||
conn.execute("PRAGMA foreign_keys = ON")
|
||
return conn
|
||
|
||
|
||
def init_db():
|
||
# 迁移旧版 DB（backend/stock_data.db → backend/data/stock_data.db）
|
||
old_path = os.path.join(os.path.dirname(__file__), "stock_data.db")
|
||
if os.path.exists(old_path) and os.path.isfile(old_path):
|
||
import shutil
|
||
shutil.move(old_path, DB_PATH)
|
||
print(f"[database] 已迁移 stock_data.db → data/stock_data.db")
|
||
|
||
conn = get_connection()
|
||
conn.executescript(SCHEMA_SQL)
|
||
|
||
# 清理过期缓存
|
||
from services.cache import clean_expired
|
||
clean_expired()
|
||
|
||
conn.commit()
|
||
conn.close()
|
||
|
||
|
||
def dict_from_row(row: sqlite3.Row) -> dict:
|
||
if row is None:
|
||
return None
|
||
return dict(row)
|