Update app.py

app.py

@@ -481,49 +481,43 @@ class TextToSQLSystem:
 
         return formatted.strip()
 
-
-    # in class TextToSQLSystem:
-
     # in class TextToSQLSystem:
 
-    def _validate_and_fix_sql(self, question: str, raw_response: str) -> Tuple[Optional[str], str]:
+    def _validate_and_fix_sql(self, sql: str, question: str) -> str:
         """
-        (V17 / 最終決策版)
+        (V19 / Top N 工作單版)
         一個全面、多層次的 SQL 驗證與生成引擎。
-        本函數作為第一決策者，優先匹配用戶問題與專家知識庫。
-        如果匹配成功，則直接使用模板覆寫；若不成功，才解析並修正 AI 的輸出。
-        返回一個元組 (SQL字符串或None, 狀態消息)。
+        新增了對“查詢總金額最高的 N 個工作單”這一高頻查詢的專家模板支持。
         """
-        q_lower = question.lower()
+        if not sql or not self.schema:
+            self._log("SQL 修正被跳過，因輸入為空或 schema 未載入。", "WARNING")
+            return sql
 
+        original_sql = sql
+        fixed_sql = " " + sql.strip() + " "
+        fixes_applied = []
+        q_lower = question.lower()
+        
         # ==============================================================================
         #  第一層：高價值意圖識別與模板覆寫 (Intent Recognition & Templating)
         # ==============================================================================
-
+        
         # --- 預先檢測所有可能的意圖和實體 ---
         job_no_match = re.search(r"(?:工單|jobno)\s*'\"?([A-Z]{2,3}\d+)'\"?", question, re.IGNORECASE)
-
+        
         entity_match_data = None
-        ENTITY_TO_COLUMN_MAP = {
-            '申請廠商': 'sd.ApplicantName', '申請方': 'sd.ApplicantName', 'applicant': 'sd.ApplicantName',
-            '付款廠商': 'sd.InvoiceToName', '付款方': 'sd.InvoiceToName', 'invoiceto': 'sd.InvoiceToName',
-            '代理商': 'sd.AgentName', 'agent': 'sd.AgentName',
-            '買家': 'sd.BuyerName', 'buyer': 'sd.BuyerName', '客戶': 'sd.BuyerName', '品牌': 'tsr.BuyerName'
-        }
+        ENTITY_TO_COLUMN_MAP = {'申請廠商':'sd.ApplicantName','申請方':'sd.ApplicantName','applicant':'sd.ApplicantName','付款廠商':'sd.InvoiceToName','付款方':'sd.InvoiceToName','invoiceto':'sd.InvoiceToName','代理商':'sd.AgentName','agent':'sd.AgentName','買家':'sd.BuyerName','buyer':'sd.BuyerName','客戶':'sd.BuyerName', '品牌': 'tsr.BuyerName'}
         for keyword, column in ENTITY_TO_COLUMN_MAP.items():
             if keyword in q_lower:
                 match = re.search(fr"{re.escape(keyword)}[\s:;\'\"-]*([a-zA-Z0-9&.\s-]+?)(?:\s*的|\s+|為|$)", question, re.IGNORECASE)
-                if match:
-                    entity_match_data = {"type": keyword, "name": match.group(1).strip(), "column": column}
-                    break
-
+                if match: entity_match_data = {"type": keyword, "name": match.group(1).strip(), "column": column}; break
+        
         lab_group_match_data = None
         LAB_GROUP_MAP = {'A':'TA','B':'TB','C':'TC','D':'TD','E':'TE','Y':'TY','TA':'TA','TB':'TB','TC':'TC','TD':'TD','TE':'TE','TY':'TY','WC':'WC','EO':'EO','GCI':'GCI','GCO':'GCO','MI':'MI'}
         lab_group_match = re.findall(r"([A-Z]+)\s*組", question, re.IGNORECASE)
         if lab_group_match:
             codes = [LAB_GROUP_MAP.get(g.upper()) for g in lab_group_match if LAB_GROUP_MAP.get(g.upper())]
-            if codes:
-                lab_group_match_data = {"codes": codes, "identifiers": lab_group_match}
+            if codes: lab_group_match_data = {"codes": codes, "identifiers": lab_group_match}
 
         is_tat_query = any(k in q_lower for k in ['平均', 'average']) and any(k in q_lower for k in ['時間', '時長', '多久', '天', 'tat', 'turnaround'])
 
@@ -532,39 +526,56 @@ class TextToSQLSystem:
         if job_no_match and any(kw in q_lower for kw in ['工作日', 'workday']):
             job_no = job_no_match.group(1).upper()
             self._log(f"🔄 檢測到計算【工單 {job_no}】工作日TAT的意圖，啟用模板。", "INFO")
-            template_sql = f"WITH span AS (SELECT date(jt.JobCreation) AS d1, date(jt.ReportAuthorization) AS d2 FROM JobTimeline jt WHERE jt.JobNo = '{job_no}'), days AS (SELECT 1 FROM calendar_days, span WHERE date BETWEEN d1 AND d2 AND is_workday = 1) SELECT COUNT(*) FROM days;"
-            return self._finalize_sql(template_sql, f"模板覆寫: {job_no} 的工作日天數")
+            fixed_sql = f"""WITH span AS (SELECT date(jt.JobCreation) AS d1, date(jt.ReportAuthorization) AS d2 FROM JobTimeline jt WHERE jt.JobNo = '{job_no}'), days AS (SELECT 1 FROM calendar_days, span WHERE date BETWEEN d1 AND d2 AND is_workday = 1) SELECT COUNT(*) FROM days;"""
+            fixes_applied.append(f"模板覆寫: {job_no} 的工作日天數")
 
-        if job_no_match and any(kw in q_lower for kw in ['總處理時長', '時長', '多少天']):
+        elif job_no_match and any(kw in q_lower for kw in ['總處理���長', '時長', '多少天']):
             job_no = job_no_match.group(1).upper()
             self._log(f"🔄 檢測到計算【工單 {job_no}】日曆日TAT的意圖，啟用模板。", "INFO")
-            template_sql = f"SELECT ROUND(julianday(ReportAuthorization) - julianday(JobCreation), 2) AS days FROM JobTimeline WHERE JobNo = '{job_no}';"
-            return self._finalize_sql(template_sql, f"模板覆寫: {job_no} 的日曆日總時長")
+            fixed_sql = f"""SELECT ROUND(julianday(ReportAuthorization) - julianday(JobCreation), 2) AS days FROM JobTimeline WHERE JobNo = '{job_no}';"""
+            fixes_applied.append(f"模板覆寫: {job_no} 的日曆日總時長")
 
-        if job_no_match and any(kw in q_lower for kw in ['總金額', '金額', '業績', 'total amount']):
+        elif job_no_match and any(kw in q_lower for kw in ['總金額', '金額', '業績', 'total amount']):
             job_no = job_no_match.group(1).upper()
             self._log(f"🔄 檢測到對【單一工作單 '{job_no}'】的【標準金額計算】意圖，啟用模板。", "INFO")
-            template_sql = f"WITH JobTotalAmount AS (SELECT JobNo, SUM(LocalAmount) AS TotalAmount FROM (SELECT DISTINCT JobNo, InvoiceCreditNoteNo, LocalAmount FROM TSR53Invoice) GROUP BY JobNo) SELECT TotalAmount FROM JobTotalAmount WHERE JobNo = '{job_no}';"
-            return self._finalize_sql(template_sql, f"模板覆寫: 工作單 {job_no} 的標準總金額 (CTE去重)")
+            fixed_sql = f"""WITH JobTotalAmount AS (SELECT JobNo, SUM(LocalAmount) AS TotalAmount FROM (SELECT DISTINCT JobNo, InvoiceCreditNoteNo, LocalAmount FROM TSR53Invoice) GROUP BY JobNo) SELECT TotalAmount FROM JobTotalAmount WHERE JobNo = '{job_no}';"""
+            fixes_applied.append(f"模板覆寫: 工作單 {job_no} 的標準總金額 (CTE去重)")
+            
+        elif '總金額最高' in q_lower and '工作單' in q_lower:
+            limit_match = re.search(r'(\d+)', question)
+            limit = limit_match.group(1) if limit_match else '10'
+            self._log(f"🔄 檢測到查詢【Top {limit} 工作單金額】意圖，啟用模板。", "INFO")
+            fixed_sql = f"""
+WITH JobTotalAmount AS (
+  SELECT JobNo, SUM(LocalAmount) AS TotalAmount
+  FROM (SELECT DISTINCT JobNo, InvoiceCreditNoteNo, LocalAmount FROM TSR53Invoice)
+  GROUP BY JobNo
+)
+SELECT JobNo, TotalAmount 
+FROM JobTotalAmount 
+ORDER BY TotalAmount DESC 
+LIMIT {limit};
+"""
+            fixes_applied.append(f"模板覆寫: Top {limit} 工作單金額")
 
-        if entity_match_data and any(kw in q_lower for kw in ['業績', '營收', '金額', 'sales', 'revenue']):
+        elif entity_match_data and any(kw in q_lower for kw in ['業績', '營收', '金額', 'sales', 'revenue']):
             entity_type, entity_name, column_name = entity_match_data["type"], entity_match_data["name"], entity_match_data["column"]
             year = (re.search(r'(\d{4})\s*年?', question) or ['', datetime.now().strftime('%Y')])[1]
             self._log(f"🔄 檢測到查詢【{entity_type} '{entity_name}' 在 {year} 年的總業績】意圖，啟用模板。", "INFO")
-            template_sql = f"WITH JobTotalAmount AS (SELECT JobNo, SUM(LocalAmount) AS TotalAmount FROM (SELECT DISTINCT JobNo, InvoiceCreditNoteNo, LocalAmount FROM TSR53Invoice) GROUP BY JobNo) SELECT SUM(jta.TotalAmount) AS total_revenue FROM TSR53SampleDescription AS sd JOIN JobTotalAmount AS jta ON sd.JobNo = jta.JobNo WHERE {column_name} LIKE '%{entity_name}%' AND strftime('%Y', sd.FirstReportAuthorizedDate) = '{year}';"
-            return self._finalize_sql(template_sql, f"模板覆寫: 查詢 {entity_type}='{entity_name}' ({year}年) 的總業績")
+            fixed_sql = f"""WITH JobTotalAmount AS (SELECT JobNo, SUM(LocalAmount) AS TotalAmount FROM (SELECT DISTINCT JobNo, InvoiceCreditNoteNo, LocalAmount FROM TSR53Invoice) GROUP BY JobNo) SELECT SUM(jta.TotalAmount) AS total_revenue FROM TSR53SampleDescription AS sd JOIN JobTotalAmount AS jta ON sd.JobNo = jta.JobNo WHERE {column_name} LIKE '%{entity_name}%' AND strftime('%Y', sd.FirstReportAuthorizedDate) = '{year}';"""
+            fixes_applied.append(f"模板覆寫: 查詢 {entity_type}='{entity_name}' ({year}年) 的總業績")
 
-        if not entity_match_data and any(kw in q_lower for kw in ['業績', '營收', '金額', 'sales', 'revenue']):
+        elif not entity_match_data and any(kw in q_lower for kw in ['業績', '營收', '金額', 'sales', 'revenue']):
             year_match, month_match = re.search(r'(\d{4})\s*年?', question), re.search(r'(\d{1,2})\s*月', question)
             time_condition, time_log = "", "總"
             if year_match:
                 year = year_match.group(1); time_condition = f"WHERE strftime('%Y', InvoiceCreditNoteDate) = '{year}'"; time_log = f"{year}年"
                 if month_match: month = month_match.group(1).zfill(2); time_condition += f" AND strftime('%m', InvoiceCreditNoteDate) = '{month}'"; time_log += f"{month}月"
             self._log(f"🔄 檢測到查詢【{time_log}全局總業績】意圖，啟用模板。", "INFO")
-            template_sql = f"SELECT SUM(LocalAmount) AS total_revenue FROM TSR53Invoice {time_condition};"
-            return self._finalize_sql(template_sql, f"模板覆寫: {time_log}全局總業績查詢")
+            fixed_sql = f"""SELECT SUM(LocalAmount) AS total_revenue FROM TSR53Invoice {time_condition};"""
+            fixes_applied.append(f"模板覆寫: {time_log}全局總業績查詢")
 
-        if lab_group_match_data and any(kw in q_lower for kw in ['測試項目', 'test item']):
+        elif lab_group_match_data and any(kw in q_lower for kw in ['測試項目', 'test item']):
             lab_group_code = lab_group_match_data["codes"][0]
             target_table = f"JobTimeline_{lab_group_code}"
             year = (re.search(r'(\d{4})\s*年', question) or ['', datetime.now().strftime('%Y')])[1]
@@ -576,10 +587,10 @@ class TextToSQLSystem:
                 time_condition += f" AND strftime('%m', end_time) = '{month}'"
                 month_str = f"{month}月"
             self._log(f"🔄 檢測到查詢【{lab_group_code}組】完成的【測試項目數】意圖，啟用專屬模板。", "INFO")
-            template_sql = f"SELECT COUNT(JobItemKey) AS test_item_count FROM {target_table} WHERE end_time IS NOT NULL AND {time_condition};"
-            return self._finalize_sql(template_sql, f"模板覆寫: 查詢 {lab_group_code}組 在 {year}年{month_str} 完成的測試項目數")
+            fixed_sql = f"""SELECT COUNT(JobItemKey) AS test_item_count FROM {target_table} WHERE end_time IS NOT NULL AND {time_condition};"""
+            fixes_applied.append(f"模板覆寫: 查詢 {lab_group_code}組 在 {year}年{month_str} 完成的測試項目數")
 
-        if '報告' in q_lower and any(kw in q_lower for kw in ['幾份', '多少', '數量', '總數']) and not entity_match_data and not lab_group_match_data:
+        elif '報告' in q_lower and any(kw in q_lower for kw in ['幾份', '多少', '數量', '總數']) and not entity_match_data and not lab_group_match_data:
             year_match = re.search(r'(\d{4})\s*年?', question)
             time_condition, time_log = "", "總"
             if year_match:
@@ -589,59 +600,68 @@ class TextToSQLSystem:
             else:
                 time_condition = "WHERE ReportAuthorization IS NOT NULL"
             self._log(f"🔄 檢測到查詢【{time_log}全局報告總數】意圖，啟用模板。", "INFO")
-            template_sql = f"SELECT COUNT(DISTINCT JobNo) AS report_count FROM JobTimeline {time_condition};"
-            return self._finalize_sql(template_sql, f"模板覆寫: {time_log}全局報告總數查詢")
+            fixed_sql = f"SELECT COUNT(DISTINCT JobNo) AS report_count FROM JobTimeline {time_condition};"
+            fixes_applied.append(f"模板覆寫: {time_log}全局報告總數查詢")
+
+        if not fixes_applied:
+            self._log("未觸發任何模板，執行常規修正流程...", "DEBUG")
+            
+            # ==============================================================================
+            #  第二層：常規修正流程 (Fallback Corrections)
+            # ==============================================================================
+            
+            dialect_corrections = {
+                r'YEAR\s*\(([^)]+)\)': r"strftime('%Y', \1)",
+                r"(strftime\('%Y',\s*[^)]+\))\s*=\s*(\d{4})": r"\1 = '\2'",
+                r"EXTRACT\s*\(\s*YEAR\s+FROM\s+([^)]+)\s*\)": r"strftime('%Y', \1)"
+            }
+            for pattern, replacement in dialect_corrections.items():
+                if re.search(pattern, fixed_sql, re.IGNORECASE):
+                    fixed_sql = re.sub(pattern, replacement, fixed_sql, flags=re.IGNORECASE)
+                    fixes_applied.append(f"修正方言: {pattern}")
+
+            schema_corrections = {
+                'TSR53ReportAuthorization': 'TSR53SampleDescription', 'TSR53TestResult': 'TSR53SampleDescription',
+                'JobInvoice': 'TSR53Invoice', 'JobInvoiceAuthorization': 'TSR53Invoice', 'JobInvoiceCreditNote': 'TSR53Invoice',
+                'Customer': 'TSR53SampleDescription', 'Customers': 'TSR53SampleDescription',
+                'Invoice': 'TSR53Invoice', 'Invoices': 'TSR53Invoice', 'Job': 'JobTimeline', 'Jobs': 'JobsInProgress',
+                'Tests': 'TSR53MarsItem', 'TestsLog': 'JobItemsInProgress',
+                'AuthorizationDate': 'ReportAuthorization', 'ReportAuthorizationDate': 'ReportAuthorization',
+                'LegalAuthorization': 'OverallRating', 'LegalAuthorizationDate': 'ReportAuthorization',
+                'TestResult': 'OverallRating', 'Rating': 'OverallRating', 'CustomerName': 'BuyerName', 'InvoiceTo': 'InvoiceToName',
+                'Applicant': 'ApplicantName', 'Agent': 'AgentName', 'JobNumber': 'JobNo', 'ReportNo': 'JobNo', 'TestName': 'ItemInvoiceDescriptionJob',
+                'CreationDate': 'JobCreation', 'CreateDate': 'JobCreation', 'CompletedDate': 'ReportAuthorization',
+                'InvoiceCreditNoteAmount': 'LocalAmount',
+                'Amount': 'LocalAmount', 'Price': 'LocalAmount', 'Lab': 'LabGroup'
+            }
+            for wrong, correct in schema_corrections.items():
+                pattern = r'\b' + re.escape(wrong) + r'\b'
+                if re.search(pattern, fixed_sql, re.IGNORECASE):
+                    fixed_sql = re.sub(pattern, correct, fixed_sql, flags=re.IGNORECASE)
+                    fixes_applied.append(f"映射 Schema: '{wrong}' -> '{correct}'")
+
+            if any(kw in q_lower for kw in ['幾份', '多少', 'how many', 'count', '數量']) and 'select ' in fixed_sql.lower() and 'count' not in fixed_sql.lower() and 'group by' not in fixed_sql.lower():
+                fixed_sql = re.sub(r'SELECT\s+.*?FROM', 'SELECT COUNT(*) FROM', fixed_sql, count=1, flags=re.IGNORECASE)
+                fixes_applied.append("修正邏輯: 補全 COUNT(*)")
 
         # ==============================================================================
-        #  第二層：常規修正流程 (Fallback Corrections)
+        #  第三層：清理與完成 (Finalization)
         # ==============================================================================
-        self._log("未觸發任何模板，嘗試解析並修正 AI 輸出...", "INFO")
-
-        parsed_sql = parse_sql_from_response(raw_response)
-        if not parsed_sql:
-            self._log(f"❌ 未能從模型回應中解析出任何 SQL。原始回應: {raw_response}", "ERROR")
-            return None, f"無法解析SQL。原始回應:\n{raw_response}"
-
-        self._log(f"📊 解析出的原始 SQL: {parsed_sql}", "DEBUG")
-
-        fixed_sql = " " + parsed_sql.strip() + " "
-        fixes_applied_fallback = []
+        fixed_sql = fixed_sql.strip()
+        if not fixed_sql.endswith(';'):
+            fixed_sql += ';'
+        fixed_sql = re.sub(r'\s+', ' ', fixed_sql).strip()
+
+        if fixes_applied:
+            self._log("🔄 SQL 已被自動修正:", "INFO")
+            self._log(f"  - 原始 SQL: {original_sql}", "DEBUG")
+            for fix in fixes_applied:
+                self._log(f"  - 應用規則: {fix}", "DEBUG")
+            self._log(f"  - 修正後 SQL: {fixed_sql}", "INFO")
+        else:
+            self._log("✅ SQL 驗證通過，無需常規修正。", "INFO")
 
-        dialect_corrections = {
-            r'YEAR\s*\(([^)]+)\)': r"strftime('%Y', \1)",
-            r"(strftime\('%Y',\s*[^)]+\))\s*=\s*(\d{4})": r"\1 = '\2'",
-            r"EXTRACT\s*\(\s*YEAR\s+FROM\s+([^)]+)\s*\)": r"strftime('%Y', \1)"
-        }
-        for pattern, replacement in dialect_corrections.items():
-            if re.search(pattern, fixed_sql, re.IGNORECASE):
-                fixed_sql = re.sub(pattern, replacement, fixed_sql, flags=re.IGNORECASE)
-                fixes_applied_fallback.append(f"修正方言: {pattern}")
-
-        schema_corrections = {
-            'TSR53ReportAuthorization': 'TSR53SampleDescription', 'TSR53TestResult': 'TSR53SampleDescription',
-            'JobInvoice': 'TSR53Invoice', 'JobInvoiceAuthorization': 'TSR53Invoice', 'JobInvoiceCreditNote': 'TSR53Invoice',
-            'Customer': 'TSR53SampleDescription', 'Customers': 'TSR53SampleDescription',
-            'Invoice': 'TSR53Invoice', 'Invoices': 'TSR53Invoice', 'Job': 'JobTimeline', 'Jobs': 'JobsInProgress',
-            'Tests': 'TSR53MarsItem', 'TestsLog': 'JobItemsInProgress',
-            'AuthorizationDate': 'ReportAuthorization', 'ReportAuthorizationDate': 'ReportAuthorization',
-            'LegalAuthorization': 'OverallRating', 'LegalAuthorizationDate': 'ReportAuthorization',
-            'TestResult': 'OverallRating', 'Rating': 'OverallRating', 'CustomerName': 'BuyerName', 'InvoiceTo': 'InvoiceToName',
-            'Applicant': 'ApplicantName', 'Agent': 'AgentName', 'JobNumber': 'JobNo', 'ReportNo': 'JobNo', 'TestName': 'ItemInvoiceDescriptionJob',
-            'CreationDate': 'JobCreation', 'CreateDate': 'JobCreation', 'CompletedDate': 'ReportAuthorization',
-            'Amount': 'LocalAmount', 'Price': 'LocalAmount', 'Lab': 'LabGroup'
-        }
-        for wrong, correct in schema_corrections.items():
-            pattern = r'\b' + re.escape(wrong) + r'\b'
-            if re.search(pattern, fixed_sql, re.IGNORECASE):
-                fixed_sql = re.sub(pattern, correct, fixed_sql, flags=re.IGNORECASE)
-                fixes_applied_fallback.append(f"映射 Schema: '{wrong}' -> '{correct}'")
-
-        if any(kw in q_lower for kw in ['幾份', '多少', 'how many', 'count', '數量']) and 'select ' in fixed_sql.lower() and 'count' not in fixed_sql.lower() and 'group by' not in fixed_sql.lower():
-            fixed_sql = re.sub(r'SELECT\s+.*?FROM', 'SELECT COUNT(*) FROM', fixed_sql, count=1, flags=re.IGNORECASE)
-            fixes_applied_fallback.append("修正邏輯: 補全 COUNT(*)")
-
-        log_msg = "AI 生成並成功修正" if fixes_applied_fallback else "AI 生成且無需修正"
-        return self._finalize_sql(fixed_sql, log_msg)
+        return fixed_sql
 
     def _finalize_sql(self, sql: str, log_message: str) -> Tuple[str, str]:
         """一個輔助函數，用於清理最終的SQL並記錄成功日誌。"""