e611702bb9
- database/import_models.py: 移除 ext.declarative.declarative_base，改用 from database.models import Base - database/notification_models.py: 同上 - database/ppt_reports.py: 移除 orm.declarative_base，改用共用 Base - database/vendor_models.py: 同上 - database/manager.py: 加入 4 個模型的 noqa import，確保 Base.metadata 完整管理所有資料表 Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
113 lines
4.5 KiB
Python
113 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
匯入進度追蹤模型
|
||
"""
|
||
|
||
from sqlalchemy import Column, Integer, String, DateTime, Text, Float
|
||
from database.models import Base
|
||
from datetime import datetime
|
||
import pytz
|
||
|
||
# 台北時區
|
||
TAIPEI_TZ = pytz.timezone('Asia/Taipei')
|
||
|
||
|
||
def taipei_now():
|
||
"""取得台北時區的當前時間（無時區資訊，用於資料庫存儲）"""
|
||
return datetime.now(TAIPEI_TZ).replace(tzinfo=None)
|
||
|
||
|
||
class ImportJob(Base):
|
||
"""匯入任務模型"""
|
||
__tablename__ = 'import_jobs'
|
||
|
||
id = Column(Integer, primary_key=True, autoincrement=True, comment='匯入任務 ID')
|
||
|
||
# 任務資訊
|
||
job_type = Column(String(50), nullable=False, comment='任務類型：daily_sales, vendor_stockout')
|
||
status = Column(String(20), nullable=False, default='pending', comment='狀態：pending, downloading, importing, completed, failed')
|
||
|
||
# Google Drive 檔案資訊
|
||
drive_file_id = Column(String(200), comment='Google Drive 檔案 ID')
|
||
drive_file_name = Column(String(500), comment='Google Drive 檔案名稱')
|
||
drive_file_size = Column(Integer, comment='檔案大小（bytes）')
|
||
|
||
# 本地檔案資訊
|
||
local_file_path = Column(String(500), comment='本地檔案路徑')
|
||
|
||
# 進度資訊
|
||
progress_percent = Column(Float, default=0.0, comment='進度百分比 (0-100)')
|
||
current_step = Column(String(200), comment='當前步驟描述')
|
||
total_rows = Column(Integer, comment='總行數')
|
||
processed_rows = Column(Integer, default=0, comment='已處理行數')
|
||
success_rows = Column(Integer, default=0, comment='成功匯入行數')
|
||
error_rows = Column(Integer, default=0, comment='錯誤行數')
|
||
|
||
# 時間記錄 (2026-01-30 修正：使用台北時區)
|
||
created_at = Column(DateTime, default=taipei_now, comment='建立時間')
|
||
started_at = Column(DateTime, comment='開始時間')
|
||
completed_at = Column(DateTime, comment='完成時間')
|
||
|
||
# 結果資訊
|
||
error_message = Column(Text, comment='錯誤訊息')
|
||
import_summary = Column(Text, comment='匯入摘要（JSON 格式）')
|
||
|
||
def __repr__(self):
|
||
return f"<ImportJob(id={self.id}, type={self.job_type}, status={self.status}, progress={self.progress_percent}%)>"
|
||
|
||
def to_dict(self):
|
||
"""轉換為字典格式"""
|
||
return {
|
||
'id': self.id,
|
||
'job_type': self.job_type,
|
||
'status': self.status,
|
||
'drive_file_id': self.drive_file_id,
|
||
'drive_file_name': self.drive_file_name,
|
||
'drive_file_size': self.drive_file_size,
|
||
'local_file_path': self.local_file_path,
|
||
'progress_percent': self.progress_percent,
|
||
'current_step': self.current_step,
|
||
'total_rows': self.total_rows,
|
||
'processed_rows': self.processed_rows,
|
||
'success_rows': self.success_rows,
|
||
'error_rows': self.error_rows,
|
||
'created_at': self.created_at.isoformat() if self.created_at else None,
|
||
'started_at': self.started_at.isoformat() if self.started_at else None,
|
||
'completed_at': self.completed_at.isoformat() if self.completed_at else None,
|
||
'error_message': self.error_message,
|
||
'import_summary': self.import_summary,
|
||
}
|
||
|
||
|
||
class ImportConfig(Base):
|
||
"""匯入配置模型"""
|
||
__tablename__ = 'import_config'
|
||
|
||
id = Column(Integer, primary_key=True, autoincrement=True)
|
||
|
||
# 配置項目
|
||
config_key = Column(String(100), unique=True, nullable=False, comment='配置鍵')
|
||
config_value = Column(Text, comment='配置值')
|
||
config_type = Column(String(50), comment='配置類型：string, int, bool, json')
|
||
description = Column(String(500), comment='配置說明')
|
||
|
||
# 時間記錄 (2026-01-30 修正：使用台北時區)
|
||
created_at = Column(DateTime, default=taipei_now, comment='建立時間')
|
||
updated_at = Column(DateTime, default=taipei_now, onupdate=taipei_now, comment='更新時間')
|
||
|
||
def __repr__(self):
|
||
return f"<ImportConfig(key={self.config_key}, value={self.config_value})>"
|
||
|
||
def to_dict(self):
|
||
"""轉換為字典格式"""
|
||
return {
|
||
'id': self.id,
|
||
'config_key': self.config_key,
|
||
'config_value': self.config_value,
|
||
'config_type': self.config_type,
|
||
'description': self.description,
|
||
'created_at': self.created_at.isoformat() if self.created_at else None,
|
||
'updated_at': self.updated_at.isoformat() if self.updated_at else None,
|
||
}
|