diff --git a/.gitignore b/.gitignore index 708c8fc..394c16b 100755 --- a/.gitignore +++ b/.gitignore @@ -15,3 +15,5 @@ __pycache__/ # OS 檔案 postgres_data/ + +line_user_data_firefox diff --git a/app/getLineTokenManual.py b/app/getLineTokenManual.py new file mode 100644 index 0000000..a2f98e9 --- /dev/null +++ b/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() \ No newline at end of file diff --git a/app/lineinvoicefetcher.py b/app/lineinvoicefetcher.py deleted file mode 100644 index 6f6ebe0..0000000 --- a/app/lineinvoicefetcher.py +++ /dev/null @@ -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']}") \ No newline at end of file diff --git a/app/syncLineInvoiceData.py b/app/syncLineInvoiceData.py new file mode 100644 index 0000000..c53521f --- /dev/null +++ b/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() \ No newline at end of file