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>
123 lines
4.1 KiB
Python
123 lines
4.1 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
驗證匯入的資料是否正確
|
|
"""
|
|
|
|
import sys
|
|
import os
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from database.vendor_manager import VendorDatabaseManager
|
|
from database.vendor_models import VendorStockout
|
|
import re
|
|
|
|
def verify_import():
|
|
"""驗證匯入資料"""
|
|
|
|
db = VendorDatabaseManager()
|
|
session = db.get_session()
|
|
|
|
try:
|
|
# 查詢所有記錄
|
|
records = session.query(VendorStockout).order_by(VendorStockout.id).all()
|
|
|
|
print("=" * 80)
|
|
print(f"驗證匯入資料")
|
|
print("=" * 80)
|
|
|
|
if not records:
|
|
print("\n❌ 沒有資料！請確認是否成功匯入")
|
|
return
|
|
|
|
print(f"\n✅ 找到 {len(records)} 筆記錄")
|
|
|
|
# 檢查第一筆記錄
|
|
record = records[0]
|
|
|
|
print("\n" + "=" * 80)
|
|
print(f"第一筆記錄詳細資訊")
|
|
print("=" * 80)
|
|
|
|
print(f"\n📅 日期資訊：")
|
|
print(f" import_date: {record.import_date} (類型: {type(record.import_date)})")
|
|
if str(record.import_date) == "2026-01-05":
|
|
print(f" ✅ 日期正確！")
|
|
else:
|
|
print(f" ❌ 日期錯誤！應該是 2026-01-05")
|
|
|
|
print(f"\n👤 組織資訊：")
|
|
print(f" 處別: {record.department}")
|
|
print(f" 科別: {record.section}")
|
|
print(f" PM姓名: {record.pm_name}")
|
|
|
|
print(f"\n📦 商品資訊：")
|
|
print(f" 商品ID: {record.product_code}")
|
|
print(f" 商品名稱: {record.product_name[:50]}...")
|
|
print(f" 單品/組合商品: {record.product_spec}")
|
|
|
|
print(f"\n🏢 廠商資訊：")
|
|
print(f" 來源供應商編號: {record.vendor_code}")
|
|
print(f" 來源供應商名稱: {record.vendor_name}")
|
|
|
|
print(f"\n📊 數量資訊：")
|
|
print(f" 商品可賣量: {record.monthly_sales_qty}")
|
|
print(f" 缺貨日期: {record.current_stock}")
|
|
print(f" 缺貨商品前30天業績: {record.monthly_sales_amount}")
|
|
print(f" 最近30天銷售量: {record.daily_avg_sales}")
|
|
print(f" 庫存水位: {record.safe_stock_days}")
|
|
|
|
print(f"\n📝 備註資訊（解析前）：")
|
|
print(f" {record.notes}")
|
|
|
|
# 解析 notes 中的額外欄位
|
|
if record.notes:
|
|
print(f"\n📝 備註資訊（解析後）：")
|
|
zone_id_match = re.search(r'區ID:\s*([^,]*)', record.notes)
|
|
zone_name_match = re.search(r'區名稱:\s*([^,]*)', record.notes)
|
|
borrow_match = re.search(r'借採轉:\s*([^,]*)', record.notes)
|
|
stockout_days_match = re.search(r'缺貨天數:\s*([^,]*)', record.notes)
|
|
|
|
if zone_id_match:
|
|
print(f" 區ID: {zone_id_match.group(1).strip()}")
|
|
if zone_name_match:
|
|
print(f" 區名稱: {zone_name_match.group(1).strip()}")
|
|
if borrow_match:
|
|
print(f" 借採轉: {borrow_match.group(1).strip()}")
|
|
if stockout_days_match:
|
|
print(f" 缺貨天數: {stockout_days_match.group(1).strip()}")
|
|
|
|
# 檢查所有記錄的日期
|
|
print("\n" + "=" * 80)
|
|
print(f"檢查所有記錄的日期")
|
|
print("=" * 80)
|
|
|
|
date_counts = {}
|
|
for r in records:
|
|
date_str = str(r.import_date)
|
|
date_counts[date_str] = date_counts.get(date_str, 0) + 1
|
|
|
|
for date_str, count in sorted(date_counts.items()):
|
|
icon = "✅" if date_str == "2026-01-05" else "⚠️"
|
|
print(f" {icon} {date_str}: {count} 筆")
|
|
|
|
print("\n" + "=" * 80)
|
|
print("✅ 驗證完成")
|
|
print("=" * 80)
|
|
|
|
if all(str(r.import_date) == "2026-01-05" for r in records):
|
|
print("\n🎉 所有資料的日期都正確！")
|
|
print("現在可以發送測試郵件了！")
|
|
else:
|
|
print("\n⚠️ 部分資料的日期不正確，請檢查")
|
|
|
|
except Exception as e:
|
|
print(f"\n❌ 錯誤: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
finally:
|
|
session.close()
|
|
|
|
if __name__ == '__main__':
|
|
verify_import()
|