feat: line_invoice_fetcher

.gitignore

@@ -15,3 +15,5 @@ __pycache__/
 
 # OS 檔案
 postgres_data/
+
+line_user_data_firefox

app/getLineTokenManual.py

@@ -0,0 +1,54 @@
+from playwright.sync_api import sync_playwright
+import time
+import os
+
+# 設定狀態儲存路徑
+USER_DATA_DIR = os.path.join(os.getcwd(), "line_user_data_firefox")
+
+def run():
+    with sync_playwright() as p:
+        print("🔧 正在啟動 Firefox...")
+        # 1. 確保 headless=False
+        context = p.firefox.launch_persistent_context(
+            user_data_dir=USER_DATA_DIR,
+            headless=False,
+            slow_mo=500, # 讓動作慢一點，方便觀察
+            user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:123.0) Gecko/20100101 Firefox/123.0"
+        )
+        
+        page = context.new_page()
+
+        # 攔截邏輯
+        def intercept_request(request):
+            if "graphql" in request.url:
+                token = request.headers.get("x-parse-session-token")
+                if token:
+                    print(f"\n🔥 🔥 抓到 Token 了！！ 🔥 🔥")
+                    print(f"Token: {token}")
+                    with open(".env.token", "w") as f:
+                        f.write(f"LINE_INVOICE_TOKEN={token}")
+                    print(f"✅ 已存入 .env.token\n")
+
+        page.on("request", intercept_request)
+
+        try:
+            print("🚀 前往 LINE 發票管家...")
+            page.goto("https://invoice.line.me/", wait_until="domcontentloaded")
+            
+            print("⏳ 視窗應該已經彈出！請在視窗內完成登入。")
+            print("注意：登入後請留在發票列表頁面，直到終端機印出 Token。")
+            
+            # 給妳 2 分鐘慢慢登入
+            for i in range(120, 0, -1):
+                if i % 10 == 0:
+                    print(f"倒數 {i} 秒...")
+                time.sleep(1)
+                
+        except Exception as e:
+            print(f"❌ 發生錯誤: {e}")
+        finally:
+            context.close()
+            print("👋 瀏覽器已關閉。")
+
+if __name__ == "__main__":
+    run()

app/lineinvoicefetcher.py

@@ -1,71 +0,0 @@
-import requests
-
-SESSION_TOKEN = "r:df33fc57ed1b792131eb0eb24f8715a3"
-
-headers = {
-    "Cookie": "_ldbrbid=tr__k1y/XGTcLDWeHW1fMcteAg==",
-    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
-    "x-parse-session-token": SESSION_TOKEN,
-    "x-parse-application-id": "line-invoice",
-    "Content-Type": "application/json",
-}
-
-def get_invoices(period_code, end_cursor=None):
-    payload = {
-        "operationName": "GetInvoices",
-        "variables": {
-            "periodCode": period_code,
-            **({"endCursor": end_cursor} if end_cursor else {})
-        },
-        "query": """query GetInvoices($periodCode: String!, $endCursor: String) {
-  invoices(
-    where: {periodCode: {equalTo: $periodCode}, status: {notEqualTo: "ERROR"}}
-    first: 50
-    after: $endCursor
-  ) {
-    edges {
-      node {
-        amount
-        brandName
-        category
-        createdAt
-        invoiceDate
-        lineItems {
-          ... on Element {
-            value
-          }
-        }
-        sellerName
-        serial
-      }
-    }
-    pageInfo {
-      hasNextPage
-      endCursor
-    }
-  }
-}"""
-    }
-    r = requests.post("https://invoice.line.me/graphql", headers=headers, json=payload)
-    return r.json()
-
-# 拿所有發票（自動翻頁）
-def get_all_invoices(period_code):
-    all_invoices = []
-    end_cursor = None
-    while True:
-        data = get_invoices(period_code, end_cursor)
-        edges = data["data"]["invoices"]["edges"]
-        all_invoices.extend([e["node"] for e in edges])
-        page_info = data["data"]["invoices"]["pageInfo"]
-        if not page_info["hasNextPage"]:
-            break
-        end_cursor = page_info["endCursor"]
-    return all_invoices
-
-# 民國期別：11401=1-2月, 11403=3-4月, ..., 11412=11-12月
-invoices = get_all_invoices("11502")
-print(f"共 {len(invoices)} 張發票")
-for inv in invoices:
-    # print(f"{inv}")
-    print(f"{inv['invoiceDate'][:10]} {inv['brandName']} {inv['sellerName']} ${inv['amount']}")

app/syncLineInvoiceData.py

@@ -0,0 +1,143 @@
+import requests
+from datetime import datetime
+import psycopg2
+from psycopg2.extras import execute_values
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+# 1. 設定區 (把妳拿到的 Token 貼在這裡，或讀取 .env.token)
+TOKEN = "r:407b1c9de10b67e8ad107b850d2edba0" # 妳剛才抓到的那串
+POSTGRES_USER = os.getenv("POSTGRES_USER")
+POSTGRES_PASSWORD = os.getenv("POSTGRES_PASSWORD")
+POSTGRES_HOST = os.getenv("POSTGRES_HOST")
+POSTGRES_PORT = os.getenv("POSTGRES_PORT")
+DB_CONFIG = {
+    "dbname": "myfinance",
+    "user": POSTGRES_USER,   # 💡 檢查 1Panel 裡的使用者名稱
+    "password": POSTGRES_PASSWORD,   # 💡 檢查密碼
+    "host": POSTGRES_HOST, # 妳的 Oracle IP
+    "port": POSTGRES_PORT, # 妳的 Oracle Port
+}
+
+# 2. GraphQL 查詢 (妳之前逆向出來的精華)
+GRAPHQL_URL = "https://invoice.line.me/graphql"
+QUERY = """
+query GetInvoices($periodCode: String!) {
+  invoices(where: {periodCode: {equalTo: $periodCode}}, first: 50) {
+    edges {
+      node {
+        amount
+        brandName
+        invoiceDate
+        sellerName
+        serial
+        lineItems { ... on Element { value } }
+      }
+    }
+  }
+}
+"""
+
+def get_merchant_dict(cur):
+    """從資料庫讀取商家對照表"""
+    cur.execute("SELECT pattern, display_name, default_category_id FROM merchant_mapping")
+    return cur.fetchall()
+
+def sync():
+    # A. 抓取 LINE 資料
+    headers = {
+        "Content-Type": "application/json",
+        "x-parse-application-id": "line-invoice",
+        "x-parse-session-token": TOKEN,
+        "User-Agent": "Mozilla/5.0..."
+    }
+    
+    # 假設抓 11502 (115年2月)
+    payload = {"variables": {"periodCode": "11504"}, "query": QUERY}
+    res = requests.post(GRAPHQL_URL, headers=headers, json=payload)
+    
+    if res.status_code != 200:
+        print(f"❌ 抓取失敗: {res.status_code}")
+        return
+
+    invoices = res.json()['data']['invoices']['edges']
+    
+    conn = psycopg2.connect(**DB_CONFIG)
+    cur = conn.cursor()
+
+    merchant_rules = get_merchant_dict(cur)
+
+    for inv in invoices:
+        node = inv['node']
+        serial = node['serial']
+        seller = node.get('sellerName', '')
+        brand = node.get('brandName', '')
+
+        # 預設值
+        display_name = brand or seller or "電子發票"
+        category_id = 1 # 預設分類
+
+        # 自動翻譯邏輯：如果資料庫有定義規則，就覆蓋掉
+        for pattern, mapping_name, mapping_cat in merchant_rules:
+            if pattern in seller:
+                display_name = mapping_name
+                if mapping_cat:
+                    category_id = mapping_cat
+                break
+
+        # 1. 檢查主表是否已存在
+        cur.execute("SELECT id FROM expenses WHERE invoice_number = %s", (serial,))
+        if cur.fetchone():
+            continue
+
+        # 2. 寫入主表並取得 ID
+        # 注意：我們直接用總額 node['amount'] 作為支出金額
+        sql_main = """
+        INSERT INTO expenses (user_id, category_id, amount, seller_name, item_name, invoice_number, date)
+        VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id
+        """
+
+        cur.execute(sql_main, (
+            2, 
+            category_id,    # 這裡改用翻譯後的 category_id
+            node['amount'], 
+            node['sellerName'], 
+            display_name,   # 這裡改用翻譯後的 display_name
+            serial, 
+            node['invoiceDate']
+        ))
+        new_expense_id = cur.fetchone()[0]
+
+        # 3. 處理子表 (lineItems)
+        if node.get('lineItems'):
+            item_data = []
+            for item in node['lineItems']:
+                val = item.get('value', {})
+                if not val: continue
+                
+                # 準備批量插入的數據
+                item_data.append((
+                    new_expense_id,
+                    val.get('name', '未知品項'),
+                    val.get('quantity', 1),
+                    val.get('unitPrice', 0),
+                    val.get('amount', 0),
+                    val.get('category', 'others')
+                ))
+
+            if item_data:
+                sql_items = """
+                INSERT INTO expense_items (expense_id, name, quantity, unit_price, total_price, category)
+                VALUES %s
+                """
+                execute_values(cur, sql_items, item_data)
+
+    conn.commit()
+    cur.close()
+    conn.close()
+    print(f"✅ 深度同步完成！")
+
+if __name__ == "__main__":
+    sync()