1b4f3a7bbe
Some checks failed
CD Pipeline / deploy (push) Failing after 59s
- 建立 Gitea Actions CD pipeline (.gitea/workflows/cd.yaml) - 部署模式: rsync Python 檔案至 188 → docker restart (volume mount) - Dockerfile/requirements 變動時自動重建 Docker image - 部署通知: Telegram (開始/成功/失敗) - 健康檢查: https://mo.wooo.work/health (最多 5 次重試) - 同步最新 CLAUDE.md / ADR-008 / memory (2026-04-19) Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
116 lines
4.5 KiB
Python
116 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
匯入進度追蹤模型
|
||
"""
|
||
|
||
from sqlalchemy import Column, Integer, String, DateTime, Text, Float
|
||
from sqlalchemy.ext.declarative import declarative_base
|
||
from datetime import datetime
|
||
import pytz
|
||
|
||
# 台北時區
|
||
TAIPEI_TZ = pytz.timezone('Asia/Taipei')
|
||
|
||
|
||
def taipei_now():
|
||
"""取得台北時區的當前時間（無時區資訊，用於資料庫存儲）"""
|
||
return datetime.now(TAIPEI_TZ).replace(tzinfo=None)
|
||
|
||
|
||
Base = declarative_base()
|
||
|
||
|
||
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,
|
||
}
|