feat: invoice_fetcher

setting DB connection

app/invoice_fetcher.py

@@ -30,6 +30,17 @@ EINVOICE_USER = os.getenv("EINVOICE_USER")
 EINVOICE_PASS = os.getenv("EINVOICE_PASS")
 GROQ_API_KEY = os.getenv("GROQ_API_KEY")
 MY_USER_ID = os.getenv("LINE_USER_ID")
+DB_HOST = os.getenv("DB_HOST")
+DB_NAME = os.getenv("DB_NAME")
+DB_USER = os.getenv("DB_USER")
+DB_PASSWORD = os.getenv("DB_PASSWORD")
+
+DB_CONFIG = {
+    "host": DB_HOST,
+    "database": DB_NAME,
+    "user": DB_USER,
+    "password": DB_PASSWORD 
+}
 
 cloudinary.config(
     cloud_name=os.getenv("CLOUDINARY_CLOUD_NAME"),
@@ -38,19 +49,20 @@ cloudinary.config(
 )
 
 # 本地直接連 localhost
-# DATABASE_URL = os.getenv("LOCAL_DATABASE_URL")
+DATABASE_URL = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}/{DB_NAME}"
-# engine = create_engine(DATABASE_URL)
+engine = create_engine(DATABASE_URL)
-# SessionLocal = sessionmaker(bind=engine)
+SessionLocal = sessionmaker(bind=engine)
 Base = declarative_base()
 
-# class Transaction(Base):
+class Transaction(Base):
-#     __tablename__ = "transactions"
+    __tablename__ = "transactions"
-#     id = Column(Integer, primary_key=True, index=True)
+    id = Column(Integer, primary_key=True)
-#     user_id = Column(String)
+    date = Column(DateTime, nullable=False)
-#     category = Column(String)
+    description = Column(String)
-#     amount = Column(Float)
+    amount = Column(Float, nullable=False)
-#     note = Column(String, nullable=True)
+    category = Column(String)
-#     created_at = Column(DateTime, default=datetime.now)
+    invoice_number = Column(String, unique=True)
+    created_at = Column(DateTime, server_default=text("CURRENT_TIMESTAMP"))
 
 async def solve_captcha(img_b64: str) -> str:
 
@@ -267,41 +279,44 @@ async def fetch_invoices(token: str, days: int = 7) -> list:
     return invoice_list
 
 def save_invoices(invoices: list):
-    # db = SessionLocal()
+    db = SessionLocal()
     saved = 0
+    skipped = 0
     try:
         for inv in invoices:
-            inv_date = inv.get("invoiceDate", "未知日期")
+            inv_num = inv.get("invoiceNumber")
+            
+            # 檢查是否已存在 (避免重複)
+            existing = db.query(Transaction).filter(Transaction.invoice_number == inv_num).first()
+            if existing:
+                skipped += 1
+                continue
+            
+            # 簡單分類
             seller = inv.get("sellerName", "未知店家")
-            amount = inv.get("totalAmount", 0)
+            category = "其他"
-            inv_num = inv.get("invoiceNumber", "無號碼")
+            if "統一超商" in seller or "全家" in seller: category = "餐飲/超商"
-            # existing = db.query(Transaction).filter(
+            elif "和德昌" in seller: category = "外食"
-            #     Transaction.note == inv["invoiceNumber"]
+
-            # ).first()
+            # 建立新資料列
-            # if existing:
+            new_inv = Transaction(
-                # continue
+                date=datetime.fromtimestamp(inv.get("invoiceDate") / 1000),
-            # db.add(Transaction(
+                description=seller,
-            #     user_id="auto_import",
+                amount=float(inv.get("totalAmount", 0)),
-            #     category=inv["sellerName"],
+                category=category,
-            #     amount=inv["totalAmount"],
+                invoice_number=inv_num
-            #     note=inv["invoiceNumber"],
+            )
-            #     created_at=datetime.fromisoformat(
+            db.add(new_inv)
-            #         inv["invoiceDate"].replace("Z", "+00:00")
-            #     )
-            # ))
-            # 美化輸出格式
-            print(f"新增發票 | 日期: {inv_date[:10]} | 店家: {seller[:15]:<15} | 金額: {amount:>6} | 號碼: {inv_num}")
             saved += 1
-        # db.commit()
+            
+        db.commit()
         print("-" * 30)
-        print(f"✅ 模擬處理完成：預計新增 {saved} 筆，總計來源 {len(invoices)} 筆")
+        print(f"✅ 存入完成：新增 {saved} 筆，略過 {skipped} 筆 (已存在)，總計來源 {len(invoices)} 筆")
     except Exception as e:
-        print("❌ 儲存發票失敗:", e)
+        print("❌ 儲存發票至資料庫失敗:", e)
-        if 'inv' in locals():
+        db.rollback()
-            print(f"錯誤發票內容: {inv}")
+    finally:
-        # db.rollback()
+        db.close()
-    # finally:
-        # db.close()
 
 async def main():
     print("開始抓取發票...")