361 lines
15 KiB
Python
361 lines
15 KiB
Python
"""
|
||
数据库连接和初始化模块
|
||
"""
|
||
|
||
import sqlite3
|
||
import os
|
||
from datetime import datetime
|
||
|
||
|
||
def get_db_connection(db_path):
|
||
"""获取数据库连接"""
|
||
return sqlite3.connect(db_path, timeout=30)
|
||
|
||
|
||
class DatabaseManager:
|
||
"""数据库管理类"""
|
||
|
||
def __init__(self, db_path):
|
||
self.db_path = db_path
|
||
self.init_db()
|
||
|
||
def get_conn(self):
|
||
"""获取数据库连接"""
|
||
return get_db_connection(self.db_path)
|
||
|
||
def init_db(self):
|
||
"""初始化数据库表结构"""
|
||
try:
|
||
with self.get_conn() as conn:
|
||
# 原料表
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS fabrics (
|
||
model TEXT PRIMARY KEY,
|
||
category TEXT DEFAULT '未分类',
|
||
fabric_type TEXT,
|
||
supplier TEXT,
|
||
color TEXT,
|
||
width REAL,
|
||
gsm REAL,
|
||
retail_price REAL,
|
||
bulk_price REAL,
|
||
unit TEXT DEFAULT '米',
|
||
timestamp TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
# 为fabrics表添加索引
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_fabrics_category ON fabrics(category)')
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_fabrics_fabric_type ON fabrics(fabric_type)')
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_fabrics_supplier ON fabrics(supplier)')
|
||
|
||
# 衣服款号表
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS garments (
|
||
style_number TEXT PRIMARY KEY,
|
||
image_path TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
# 衣服材料用量表
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS garment_materials (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
style_number TEXT,
|
||
category TEXT,
|
||
fabric_type TEXT,
|
||
model TEXT,
|
||
usage_per_piece REAL,
|
||
unit TEXT DEFAULT '米',
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
FOREIGN KEY (style_number) REFERENCES garments(style_number)
|
||
)
|
||
''')
|
||
|
||
# 为garment_materials表添加索引
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_garment_materials_style ON garment_materials(style_number)')
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_garment_materials_category ON garment_materials(category)')
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_garment_materials_model ON garment_materials(model)')
|
||
|
||
# 添加新字段（如果不存在）
|
||
try:
|
||
conn.execute("ALTER TABLE garment_materials ADD COLUMN fabric_type TEXT")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE garment_materials ADD COLUMN model TEXT")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE fabrics ADD COLUMN fabric_type TEXT")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE fabrics ADD COLUMN created_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE fabrics ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE garments ADD COLUMN created_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE garments ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE garment_materials ADD COLUMN created_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE garment_materials ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE admin_settings ADD COLUMN created_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE admin_settings ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE fabric_stock_in ADD COLUMN created_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE fabric_stock_in ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE fabric_consumption ADD COLUMN created_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
try:
|
||
conn.execute("ALTER TABLE fabric_consumption ADD COLUMN updated_at DATETIME DEFAULT CURRENT_TIMESTAMP")
|
||
except:
|
||
pass
|
||
|
||
# 数据迁移：将fabrics表中category字段的"类目-类型"格式拆分成两个字段
|
||
try:
|
||
cursor = conn.execute("SELECT model, category FROM fabrics WHERE category LIKE '%-%' AND (fabric_type IS NULL OR fabric_type = '')")
|
||
rows = cursor.fetchall()
|
||
for model, category in rows:
|
||
if '-' in category:
|
||
parts = category.split('-', 1)
|
||
new_category = parts[0]
|
||
new_fabric_type = parts[1]
|
||
conn.execute("UPDATE fabrics SET category = ?, fabric_type = ? WHERE model = ?",
|
||
(new_category, new_fabric_type, model))
|
||
conn.commit()
|
||
except:
|
||
pass
|
||
|
||
# 管理员设置表
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS admin_settings (
|
||
key TEXT PRIMARY KEY,
|
||
value TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP
|
||
)
|
||
''')
|
||
|
||
# 原料入库表
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS fabric_stock_in (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
model TEXT,
|
||
quantity REAL,
|
||
unit TEXT,
|
||
purchase_date TEXT,
|
||
note TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
FOREIGN KEY (model) REFERENCES fabrics(model)
|
||
)
|
||
''')
|
||
|
||
# 为fabric_stock_in表添加索引
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_fabric_stock_in_model ON fabric_stock_in(model)')
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_fabric_stock_in_date ON fabric_stock_in(purchase_date)')
|
||
|
||
# 原料消耗表
|
||
conn.execute('''
|
||
CREATE TABLE IF NOT EXISTS fabric_consumption (
|
||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||
style_number TEXT,
|
||
model TEXT,
|
||
single_usage REAL,
|
||
quantity_made INTEGER,
|
||
loss_rate REAL,
|
||
consume_quantity REAL,
|
||
consume_date TEXT,
|
||
unit TEXT,
|
||
created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,
|
||
FOREIGN KEY (style_number) REFERENCES garments(style_number),
|
||
FOREIGN KEY (model) REFERENCES fabrics(model)
|
||
)
|
||
''')
|
||
|
||
# 为fabric_consumption表添加索引
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_fabric_consumption_style ON fabric_consumption(style_number)')
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_fabric_consumption_model ON fabric_consumption(model)')
|
||
conn.execute('CREATE INDEX IF NOT EXISTS idx_fabric_consumption_date ON fabric_consumption(consume_date)')
|
||
|
||
# 添加库存计算视图
|
||
conn.execute('''
|
||
CREATE VIEW IF NOT EXISTS fabric_stock_view AS
|
||
SELECT
|
||
f.model,
|
||
f.category,
|
||
f.supplier,
|
||
f.color,
|
||
f.unit,
|
||
COALESCE(stock_in.total_in, 0) as total_stock_in,
|
||
COALESCE(consumption.total_consumed, 0) as total_consumed,
|
||
COALESCE(stock_in.total_in, 0) - COALESCE(consumption.total_consumed, 0) as current_stock
|
||
FROM fabrics f
|
||
LEFT JOIN (
|
||
SELECT model, SUM(quantity) as total_in
|
||
FROM fabric_stock_in
|
||
GROUP BY model
|
||
) stock_in ON f.model = stock_in.model
|
||
LEFT JOIN (
|
||
SELECT model, SUM(consume_quantity) as total_consumed
|
||
FROM fabric_consumption
|
||
GROUP BY model
|
||
) consumption ON f.model = consumption.model
|
||
''')
|
||
|
||
# 初始化默认密码
|
||
if not self.get_setting("admin_password"):
|
||
conn.execute("INSERT INTO admin_settings (key, value) VALUES ('admin_password', ?)", ("123456",))
|
||
if not self.get_setting("user_password"):
|
||
conn.execute("INSERT INTO admin_settings (key, value) VALUES ('user_password', ?)", ("123456",))
|
||
|
||
conn.commit()
|
||
except Exception as e:
|
||
raise Exception(f"无法初始化数据库：{str(e)}")
|
||
|
||
def get_setting(self, key):
|
||
"""获取设置值"""
|
||
try:
|
||
with self.get_conn() as conn:
|
||
cursor = conn.execute("SELECT value FROM admin_settings WHERE key = ?", (key,))
|
||
row = cursor.fetchone()
|
||
return row[0] if row else None
|
||
except:
|
||
return None
|
||
|
||
def set_setting(self, key, value):
|
||
"""设置配置值"""
|
||
try:
|
||
with self.get_conn() as conn:
|
||
conn.execute("INSERT OR REPLACE INTO admin_settings (key, value) VALUES (?, ?)", (key, value))
|
||
conn.commit()
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def get_fabric_categories(db_path):
|
||
"""获取所有面料类目"""
|
||
try:
|
||
with get_db_connection(db_path) as conn:
|
||
cursor = conn.execute("""
|
||
SELECT DISTINCT category
|
||
FROM fabrics
|
||
WHERE category IS NOT NULL AND category != ''
|
||
ORDER BY category
|
||
""")
|
||
categories = set()
|
||
for row in cursor.fetchall():
|
||
if row[0] and row[0].strip():
|
||
categories.add(row[0])
|
||
|
||
# 添加默认类目
|
||
categories.update(["布料", "辅料", "其他"])
|
||
return sorted(categories)
|
||
except:
|
||
return ["布料", "辅料", "其他"]
|
||
|
||
|
||
def get_fabric_types_by_category(db_path, category):
|
||
"""根据类目获取面料类型"""
|
||
try:
|
||
with get_db_connection(db_path) as conn:
|
||
cursor = conn.execute("""
|
||
SELECT DISTINCT fabric_type
|
||
FROM fabrics
|
||
WHERE category = ? AND fabric_type IS NOT NULL AND fabric_type != ''
|
||
ORDER BY fabric_type
|
||
""", (category,))
|
||
|
||
types = []
|
||
for row in cursor.fetchall():
|
||
if row[0] and row[0].strip():
|
||
types.append(row[0])
|
||
return types
|
||
except:
|
||
return []
|
||
|
||
|
||
def get_fabric_models_by_category_type(db_path, category, fabric_type):
|
||
"""根据类目和类型获取面料型号"""
|
||
try:
|
||
with get_db_connection(db_path) as conn:
|
||
cursor = conn.execute("""
|
||
SELECT model, color, unit
|
||
FROM fabrics
|
||
WHERE category = ? OR category = ? OR category LIKE ?
|
||
ORDER BY model
|
||
""", (category, f"{category}-{fabric_type}", f"{category}-{fabric_type}-%"))
|
||
|
||
models = []
|
||
for row in cursor.fetchall():
|
||
model, color, unit = row
|
||
display_text = model
|
||
if color and color.strip():
|
||
display_text = f"{model}-{color}"
|
||
models.append((display_text, model, unit))
|
||
return models
|
||
except:
|
||
return []
|
||
|
||
|
||
def get_password(db_path, password_type):
|
||
"""获取密码设置"""
|
||
try:
|
||
with get_db_connection(db_path) as conn:
|
||
cursor = conn.execute(
|
||
"SELECT value FROM admin_settings WHERE key = ?",
|
||
(f"{password_type}_password",)
|
||
)
|
||
row = cursor.fetchone()
|
||
return row[0] if row else "123456"
|
||
except:
|
||
return "123456"
|
||
|
||
|
||
def update_password(db_path, password_type, new_password):
|
||
"""更新密码"""
|
||
try:
|
||
with get_db_connection(db_path) as conn:
|
||
conn.execute(
|
||
"UPDATE admin_settings SET value = ?, updated_at = CURRENT_TIMESTAMP WHERE key = ?",
|
||
(new_password, f"{password_type}_password")
|
||
)
|
||
conn.commit()
|
||
return True
|
||
except:
|
||
return False |