Refactor: enhance LLM prompt for total amount accuracy and prevent misidentification

incoming_goods_engine.py

@@ -6,6 +6,7 @@ import time
 import re
 import json
 import pandas as pd
+import logging
 from pathlib import Path
 from datetime import datetime
 
@@ -17,6 +18,9 @@ FAILED_DIR = TMP_DIR / "failed"
 COMPLETED_BASE = BASE_DIR / "已完成入貨單"
 QUEUE_FILE = TMP_DIR / "work_queue.json"
 
+# Logging Configuration
+LOG_FILE = Path("/vol1/@apphome/trim.openclaw/data/workspace/incoming_goods_engine.log")
+
 # Files for lookups
 PAYMENT_METHODS_FILE = BASE_DIR / "供應商結算方式.xlsx"
 PAYMENT_METHODS_JSON = Path("/vol1/@apphome/trim.openclaw/data/workspace/memory/payment_methods.json")
@@ -38,6 +42,18 @@ LLM_MODEL = "deepseek-chat"
 SUBMIT_API_URL = "https://www.various-stars.xyz/prod-api/finance/purchaseOrder/add"
 SUBMIT_API_KEY = "sk-finance-2026-d455sd5a5d5s4d4gtyt7"
 
+def setup_logging():
+    """Configures logging to both file and console."""
+    logging.basicConfig(
+        level=logging.INFO,
+        format='%(asctime)s [%(levelname)s] %(message)s',
+        handlers=[
+            logging.FileHandler(LOG_FILE, encoding='utf-8'),
+            logging.StreamHandler()
+        ]
+    )
+    logging.info("=== Incoming Goods Engine Started ===")
+
 def setup_dirs():
     """Ensure required directory structure exists."""
     TMP_DIR.mkdir(parents=True, exist_ok=True)
@@ -53,7 +69,7 @@ def get_pdf_page_count(file_path):
         reader = PdfReader(file_path)
         return len(reader.pages)
     except Exception as e:
-        print(f"  [!] Error reading PDF page count: {e}")
+        logging.error(f"Error reading PDF page count for {file_path.name}: {e}")
         return 1
 
 def sync_payment_methods_json():
@@ -70,14 +86,16 @@ def sync_payment_methods_json():
         PAYMENT_METHODS_JSON.parent.mkdir(parents=True, exist_ok=True)
         with open(PAYMENT_METHODS_JSON, 'w', encoding='utf-8') as f:
             json.dump(methods, f, ensure_ascii=False, indent=2)
+        logging.info("Payment methods successfully synced from Excel to JSON.")
         return True
     except Exception as e:
-        print(f"Failed to sync payment methods: {e}")
+        logging.error(f"Failed to sync payment methods: {e}")
         return False
 
 def load_payment_methods():
     """Loads supplier payment methods with Dual-Track Sync."""
     if not PAYMENT_METHODS_FILE.exists():
+        logging.warning("Payment methods Excel file not found.")
         return {}
 
     needs_sync = False
@@ -87,17 +105,19 @@ def load_payment_methods():
         needs_sync = True
 
     if needs_sync:
+        logging.info("Payment methods cache out of date. Syncing...")
         sync_payment_methods_json()
 
     try:
         with open(PAYMENT_METHODS_JSON, 'r', encoding='utf-8') as f:
             return json.load(f)
-    except:
+    except Exception as e:
+        logging.error(f"Error loading payment methods JSON: {e}")
         return {}
 
 def sync_accounting_from_api():
     """Fetches the latest accounting mapping from the central API."""
-    print("[*] Attempting to sync accounting database from API...")
+    logging.info("Attempting to sync accounting database from API...")
     headers = {"X-API-Key": SUBMIT_API_KEY, "Content-Type": "application/json"}
     try:
         response = requests.get(SYNC_ACCOUNTING_API_URL, headers=headers, timeout=15)
@@ -105,7 +125,6 @@ def sync_accounting_from_api():
             res_data = response.json()
             if res_data.get("code") == 200:
                 items = res_data.get("data", [])
-                # Transform to {material_name: account_subject} - skip incomplete items
                 mapping = {}
                 for item in items:
                     m_name = item.get('material_name')
@@ -116,26 +135,26 @@ def sync_accounting_from_api():
                 ACCOUNTING_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
                 with open(ACCOUNTING_DB_PATH, 'w', encoding='utf-8') as f:
                     json.dump(mapping, f, ensure_ascii=False, indent=2)
-                print(f"  [+] API Sync Successful. {len(mapping)} valid mappings updated.")
+                logging.info(f"API Sync Successful. {len(mapping)} valid mappings updated.")
                 return True
             else:
-                print(f"  [!] API returned error code: {res_data.get('code')}")
+                logging.error(f"API returned error code: {res_data.get('code')}")
         else:
-            print(f"  [!] API request failed. Status: {response.status_code}")
+            logging.error(f"API request failed. Status: {response.status_code}")
     except Exception as e:
-        print(f"  [!] API Sync Error: {e}")
+        logging.error(f"API Sync Error: {e}")
     return False
 
 def sync_accounting_from_excel():
     """Fallback: Converts local Excel to JSON cache."""
-    print("[*] Attempting to sync accounting database from local Excel...")
+    logging.info("Attempting to sync accounting database from local Excel fallback...")
     try:
         if not ACCOUNTING_EXCEL_FILE.exists():
+            logging.error(f"Accounting Excel file not found at {ACCOUNTING_EXCEL_FILE}")
             return False
             
         df = pd.read_excel(ACCOUNTING_EXCEL_FILE)
         mapping = {}
-        # Expected columns: ['會計科目', '貨品名稱']
         for _, row in df.iterrows():
             subject = str(row['會計科目']).strip()
             material = str(row['貨品名稱']).strip()
@@ -145,9 +164,302 @@ def sync_accounting_from_excel():
         ACCOUNTING_DB_PATH.parent.mkdir(parents=True, exist_ok=True)
         with open(ACCOUNTING_DB_PATH, 'w', encoding='utf-8') as f:
             json.dump(mapping, f, ensure_ascii=False, indent=2)
-        print(f"  [+] Excel Sync Successful. {len(mapping)} mappings updated.")
+        logging.info(f"Excel Sync Successful. {len(mapping)} mappings updated.")
         return True
     except Exception as e:
+        logging.error(f"Excel Sync Error: {e}")
+        return False
+
+def load_accounting_db():
+    """Loads the accounting subject mapping with Hybrid Sync (API -> Excel fallback)."""
+    if sync_accounting_from_api():
+        pass 
+    elif ACCOUNTING_EXCEL_FILE.exists():
+        if not ACCOUNTING_DB_PATH.exists() or \
+           ACCOUNTING_EXCEL_FILE.stat().st_mtime > ACCOUNTING_DB_PATH.stat().st_mtime:
+            sync_accounting_from_excel()
+    
+    if not ACCOUNTING_DB_PATH.exists():
+        logging.error("No accounting database found (API and Excel failed).")
+        return {}
+    try:
+        with open(ACCOUNTING_DB_PATH, 'r', encoding='utf-8') as f:
+            return json.load(f)
+    except Exception as e:
+        logging.error(f"Error loading accounting JSON: {e}")
+        return {}
+
+def call_ocr_with_retry(file_path, retries=3, interval=3):
+    """Calls the OCR API with a retry mechanism and dynamic polling time based on page count."""
+    file_ext = file_path.suffix.lower()
+    file_type = "pdf" if file_ext == ".pdf" else "image"
+    
+    page_count = get_pdf_page_count(file_path)
+    max_wait_seconds = 30 + (page_count * 50)
+    max_polls = max(1, max_wait_seconds // 2)
+    
+    logging.info(f"OCR Task: {file_path.name} | Pages: {page_count} | Max Wait: {max_wait_seconds}s")
+
+    try:
+        with open(file_path, "rb") as f:
+            encoded_string = base64.b64encode(f.read()).decode('utf-8')
+    except Exception as e:
+        logging.error(f"Error reading file for OCR ({file_path.name}): {e}")
+        return None
+    
+    payload = {"file_base64": encoded_string, "file_type": file_type, "output_type": "ocr_only"}
+    headers = {"X-API-Key": OCR_API_KEY, "Content-Type": "application/json"}
+    
+    for attempt in range(retries):
+        try:
+            response = requests.post(OCR_API_URL, json=payload, headers=headers, timeout=30)
+            if response.status_code == 200:
+                data = response.json()
+                task_id = data.get("task_id")
+                if not task_id:
+                    logging.error(f"OCR submission failed: No task_id returned. Response: {data}")
+                    return None
+                
+                logging.info(f"  [*] OCR Task {task_id} submitted. Polling...")
+                for poll_count in range(max_polls): 
+                    time.sleep(2)
+                    res = requests.get(f"{OCR_API_URL.rsplit('/', 1)[0]}/result/{task_id}", headers=headers, timeout=10)
+                    if res.status_code == 200:
+                        res_data = res.json()
+                        if res_data.get("status") == "completed":
+                            ocr_text = res_data.get("result", {}).get("ocr_text", "")
+                            logging.info(f"  [+] OCR completed for {file_path.name}. Text length: {len(ocr_text)}")
+                            logging.debug(f"RAW OCR TEXT for {file_path.name}:\n{ocr_text}")
+                            return ocr_text
+                        elif res_data.get("status") == "failed":
+                            logging.error(f"  [!] OCR task {task_id} failed on server.")
+                            return None
+                    elif res.status_code != 202:
+                        if res.status_code >= 500:
+                            logging.warning(f"  [!] OCR server error ({res.status_code}). Retrying poll...")
+                            time.sleep(5)
+                            continue
+                        logging.error(f"  [!] OCR polling error: Status {res.status_code}")
+                        return None
+                logging.error(f"  [!] OCR polling timeout for task {task_id}.")
+                return None
+            else:
+                logging.error(f"OCR submission failed. Status: {response.status_code} | Response: {response.text}")
+        except Exception as e:
+            logging.error(f"Attempt {attempt + 1} error: {e}")
+        time.sleep(interval)
+    return None
+
+def chunk_ocr_text(ocr_text, pages_per_chunk=5):
+    """Splits OCR text into chunks based on [Page X] markers using regex lookahead."""
+    chunks = re.split(r'(?=\[Page \d+\])', ocr_text)
+    chunks = [c.strip() for c in chunks if c.strip()]
+    
+    final_chunks = []
+    for i in range(0, len(chunks), pages_per_chunk):
+        chunk_content = "\n".join(chunks[i : i + pages_per_chunk])
+        final_chunks.append(chunk_content)
+    return final_chunks
+
+def parse_ocr_chunks(chunks, accounting_db, location):
+    """Sends text chunks to LLM to extract invoice data."""
+    all_invoices = []
+    mapping_str = "\n".join([f"- {k} -> {v}" for k, v in accounting_db.items()])
+    
+    system_prompt = f"""
+    You are a high-precision data extraction specialist for a finance system. 
+    Your task is to parse OCR text from an incoming goods invoice into a strict JSON format.
+
+    ### RULES:
+    1. **LANGUAGE CONVERSION (STRICT)**: 
+       - Every single field in the JSON (supplierName, materialName, etc.) MUST be in **TRADITIONAL CHINESE (繁體中文)**.
+    2. **SUPPLIER IDENTIFICATION**:
+       - The "supplierName" must be the clean COMPANY NAME. Strip out document type descriptors like "發貨單", "Invoice", etc.
+    3. **DATE**:
+       - Identify the "Order Date". Format: `YYYY-MM-DD`.
+    4. **LOCATION**: "receiveLocation" MUST be exactly "{location}".
+    5. **ACCOUNTING SUBJECT**: 
+       - Use the provided [MAPPING] to map items. 
+       - IMPORTANT: If you are uncertain about an 'accountSubject', append '**' to the end of the string (e.g., "Food Supplies**").
+    6. **OUTPUT**: Return ONLY a valid JSON object with a key "items" containing an array of invoice objects.
+
+    ### [MAPPING]:
+    {mapping_str}
+    """
+    
+    headers = {"Authorization": f"Bearer {LLM_API_KEY}", "Content-Type": "application/json"}
+
+    for i, chunk in enumerate(chunks):
+        if not chunk: continue
+        
+        payload = {
+            "model": LLM_MODEL,
+            "messages": [
+                {"role": "system", "content": system_prompt},
+                {"role": "user", "content": f"OCR Text:\n{chunk}"}
+            ],
+            "response_format": { "type": "json_object" },
+            "temperature": 0.1
+        }
+        
+        try:
+            logging.info(f"  [*] Sending chunk {i+1}/{len(chunks)} to LLM...")
+            res = requests.post(LLM_API_URL, json=payload, headers=headers, timeout=60)
+            res.raise_for_status()
+            content = res.json()['choices'][0]['message']['content'].strip()
+            
+            logging.debug(f"LLM Response for chunk {i+1}:\n{content}")
+            
+            data = json.loads(content)
+            all_invoices.extend(data.get("items", []))
+        except Exception as e:
+            logging.error(f"  [!] LLM chunk error (chunk {i+1}): {e}")
+            
+    return all_invoices
+
+def submit_to_api(invoice):
+    """Submits a single invoice payload to the Finance API."""
+    headers = {"X-API-Key": SUBMIT_API_KEY, "Content-Type": "application/json"}
+    try:
+        res = requests.post(SUBMIT_API_URL, json=invoice, headers=headers, timeout=30)
+        if res.status_code == 200 and res.json().get("code") == 200:
+            logging.info(f"    [+] ERP Submission Successful for {invoice.get('supplierName', 'Unknown')}.")
+            return True
+        else:
+            logging.error(f"    [!] ERP Submission Failed for {invoice.get('supplierName', 'Unknown')}. Status: {res.status_code} | Response: {res.text}")
+            return False
+    except Exception as e:
+        logging.error(f"    [!] ERP Submission Error for {invoice.get('supplierName', 'Unknown')}: {e}")
+        return False
+
+def ingestion_phase():
+    """Phase 1: Scan directories, move files to tmp, and build work_queue.json."""
+    logging.info("=== [PHASE 1] Ingestion Phase ===")
+    setup_dirs()
+    queue = []
+    if QUEUE_FILE.exists():
+        try:
+            with open(QUEUE_FILE, 'r', encoding='utf-8') as f:
+                queue = json.load(f)
+            logging.info(f"[*] Resuming from existing queue ({len(queue)} tasks).")
+            return queue
+        except Exception as e:
+            logging.error(f"Error reading existing queue: {e}")
+
+    logging.info(f"[*] Scanning {BASE_DIR}...")
+    for loc_dir in [d for d in BASE_DIR.iterdir() if d.is_dir() and d.name not in EXCLUDE_DIRS]:
+        loc_name = loc_dir.name
+        for original_file in loc_dir.glob("*"):
+            if not original_file.is_file() or original_file.suffix.lower() not in ['.pdf', '.jpg', '.jpeg', '.png']:
+                continue
+            if (time.time() - original_file.stat().st_mtime) < 120:
+                continue
+            
+            tmp_path = TMP_DIR / f"{loc_name}_{original_file.name}"
+            try:
+                shutil.move(str(original_file), str(tmp_path))
+                queue.append({"original_name": original_file.name, "location": loc_name, "tmp_path": str(tmp_path)})
+                logging.info(f"  [+] Ingested: {original_file.name} -> {tmp_path.name}")
+            except Exception as e:
+                logging.error(f"  [!] Error ingesting {original_file.name}: {e}")
+
+    with open(QUEUE_FILE, 'w', encoding='utf-8') as f:
+        json.dump(queue, f, ensure_ascii=False, indent=2)
+    return queue
+
+def execution_phase(queue, full_db, payment_methods):
+    """Phase 2: Process the queue with logical chunking and all-or-nothing archiving."""
+    logging.info("\n=== [PHASE 2] Execution Phase ===")
+    
+    while queue:
+        task = queue.pop(0)
+        tmp_path = Path(task['tmp_path'])
+        loc_name = task['location']
+        orig_name = task['original_name']
+        
+        logging.info(f"[Processing] {orig_name} (Location: {loc_name})")
+        if not tmp_path.exists():
+            logging.warning(f"  [!] File missing from tmp: {tmp_path}. Skipping task.")
+            continue
+
+        # 1. OCR
+        ocr_text = call_ocr_with_retry(tmp_path)
+        if not ocr_text:
+            logging.error(f"  [!] OCR failed for {orig_name}. Moving to failed.")
+            shutil.move(str(tmp_path), FAILED_DIR / tmp_path.name)
+            with open(QUEUE_FILE, 'w', encoding='utf-8') as f: json.dump(queue, f, ensure_ascii=False, indent=2)
+            continue
+
+        # 2. Chunking & Parsing
+        chunks = chunk_ocr_text(ocr_text)
+        logging.info(f"  [*] Split into {len(chunks)} chunk(s).")
+        invoices = parse_ocr_chunks(chunks, full_db, loc_name)
+        logging.info(f"  [*] Extracted {len(invoices)} invoices.")
+
+        if not invoices:
+            logging.error(f"  [!] No invoices found in OCR text for {orig_name}. Moving to failed.")
+            shutil.move(str(tmp_path), FAILED_DIR / tmp_path.name)
+            with open(QUEUE_FILE, 'w', encoding='utf-8') as f: json.dump(queue, f, ensure_ascii=False, indent=2)
+            continue
+
+        # 3. Submission (All-or-nothing)
+        all_success = True
+        for inv in invoices:
+            supplier = inv.get("supplierName", "")
+            inv["paymentMethod"] = payment_methods.get(supplier, "現金找付")
+            if not submit_to_api(inv):
+                all_success = False
+                break
+        
+        if all_success:
+            # Archive with date prefix for better organization
+            date_str = datetime.now().strftime("%Y%m%d")
+            dest_dir = COMPLETED_BASE / loc_name / date_str
+            dest_dir.mkdir(parents=True, exist_ok=True)
+            
+            new_filename = f"{date_str}_{orig_name}"
+            
+            if tmp_path.exists():
+                shutil.move(str(tmp_path), dest_dir / new_filename)
+                logging.info(f"  [✅] Success! Archived to {dest_dir}/{new_filename}")
+            else:
+                logging.warning(f"  [⚠️] Success, but file was missing from tmp: {tmp_path}")
+        else:
+            logging.error(f"  [❌] One or more invoices failed submission for {orig_name}. Moving to failed.")
+            if tmp_path.exists():
+                shutil.move(str(tmp_path), FAILED_DIR / tmp_path.name)
+            else:
+                logging.warning(f"  [⚠️] Failed, but file was already missing from tmp: {tmp_path}")
+
+        # Update queue file
+        with open(QUEUE_FILE, 'w', encoding='utf-8') as f:
+            json.dump(queue, f, ensure_ascii=False, indent=2)
+
+def run_process():
+    setup_logging()
+    setup_dirs()
+    
+    # Load accounting db using Hybrid Sync
+    full_db = load_accounting_db()
+    payment_methods = load_payment_methods()
+    
+    queue = ingestion_phase()
+    if queue:
+        execution_phase(queue, full_db, payment_methods)
+    else:
+        logging.info("No tasks to process.")
+    
+    if not queue and QUEUE_FILE.exists():
+        try:
+            os.remove(QUEUE_FILE)
+            logging.info("Work queue cleared.")
+        except: pass
+    
+    logging.info("=== Engine Process Finished ===")
+
+if __name__ == "__main__":
+    run_process()
+:
         print(f"  [!] Excel Sync Error: {e}")
         return False
 
@@ -384,15 +696,24 @@ def execution_phase(queue, full_db, payment_methods):
                 break
         
         if all_success:
-            # Archive
+            # Archive with date prefix for better organization
             date_str = datetime.now().strftime("%Y%m%d")
             dest_dir = COMPLETED_BASE / loc_name / date_str
             dest_dir.mkdir(parents=True, exist_ok=True)
-            shutil.move(str(tmp_path), dest_dir / orig_name)
-            print(f"  [✅] Success! Archived to {dest_dir}")
+            
+            new_filename = f"{date_str}_{orig_name}"
+            
+            if tmp_path.exists():
+                shutil.move(str(tmp_path), dest_dir / new_filename)
+                print(f"  [✅] Success! Archived to {dest_dir}/{new_filename}")
+            else:
+                print(f"  [⚠️] Success, but file was missing from tmp: {tmp_path}")
         else:
             print(f"  [❌] One or more invoices failed. Moving file to failed.")
-            shutil.move(str(tmp_path), FAILED_DIR / tmp_path.name)
+            if tmp_path.exists():
+                shutil.move(str(tmp_path), FAILED_DIR / tmp_path.name)
+            else:
+                print(f"  [⚠️] Failed, but file was already missing from tmp: {tmp_path}")
 
         # Update queue file
         with open(QUEUE_FILE, 'w', encoding='utf-8') as f: