import sys import json import pdfplumber import hashlib import re import os from datetime import datetime from pathlib import Path from collections import Counter from typing import List from dotenv import load_dotenv import mysql.connector import requests import subprocess from ollama import Client from datetime import datetime from collections import defaultdict # ============================================================================ # CONFIGURATION # ============================================================================ load_dotenv() # Database configuration dbhost = os.getenv("DB_HOST") database = os.getenv("DATABASE") database_2 = os.getenv("DATABASE_2") dbuser = os.getenv("DB_USER") dbpwd = os.getenv("DB_PWD") # PDF cache directory PDF_CACHE_DIR = 'pdf_cache_cli' # LLM configuration api_endpoint = os.getenv("API_ENDPOINT") # ollama server llm_model = os.getenv("LLM_MODEL") num_ctx = int(os.getenv("CONTEXT_LENGTH")) client = Client(host=api_endpoint, headers={'Content-Type': 'application/json'}) # --- Prompt A: 針對表格 --- # 將「解析表格」的輸出要求改為 JSON PROMPT_TABLE = """【角色】你是一個高階資料解析助理。 【任務】將輸入的文字內容（包含標題、表格、備註）轉換為 JSON Array。 【重要原則】 為了配合系統格式，所有內容（包含標題、表格數據、備註）都必須封裝在同一個 JSON Array 中。**嚴禁**在 JSON Array 之外輸出任何文字。 【JSON 欄位定義】 每一筆資料包含兩個欄位： 1. **section_title**: 章節或項目名稱。 2. **content**: 該項目的完整內容描述。 【處理邏輯 (請依序執行)】 **步驟 1：處理文件標題 (若存在)** - 檢查表格上方是否有標題文字。 - 若有，請建立一個物件： {{ "section_title": "表單資訊", "content": "標題內容..." }} - 若無，則跳過此步驟。 **步驟 2：處理表格內容 (核心)** - 逐行解析 Markdown 表格。 - **section_title**: 取自表格的第一欄（例如：婚假、事假）。 - **content**: 將該列其餘欄位合併，並加上欄位名稱描述（例如：給假日數：8日；工資：照給）。 - **合併規則**：若同一項目被切成多行，請合併為單一物件。 **步驟 3：處理備註 (若存在)** - 檢查表格下方是否有備註或說明文字。 - 若有，請建立一個物件： {{ "section_title": "表單備註", "content": "備註內容..." }} - 若無，則跳過此步驟。 【JSON 輸出範例】 [ {{ "section_title": "表單資訊", "content": "國立成功大學教師請假一覽表" }}, {{ "section_title": "婚假", "content": "給假日數：8日，需檢附證明；工資給與：照給。" }}, {{ "section_title": "事假", "content": "給假日數：14日..." }}, {{ "section_title": "表單備註", "content": "1. 本表適用於編制內人員。 2. 請假需透過系統申請。" }} ] 【待處理內容】： {content} """ # --- Prompt B: 通用純文字專用 --- PROMPT_TEXT = """你是一個「通用型文件結構切分器」。 你的唯一功能是：依據視覺與格式標記，將長文切分為獨立的 JSON 區塊。 你不是作家、不是編輯、絕對不進行摘要或改寫。 【最高原則（違反即失敗）】 1. **100% 忠於原文**：輸出內容必須完全來自輸入，嚴禁新增、修改、潤飾任何字詞。 2. **禁止腦補標題**：`section_title` 必須是原文中真實存在的標題行。若原文該段落沒有標題，該欄位填入空字串 `""`，嚴禁自行創建或推測。 3. **禁止外部知識**：即使文中提到特定法規或事件，若無詳細內容，禁止自行補充。 【切分與合併邏輯（優先順序）】 請嚴格依據以下順序識別區塊： 1. **強結構切分與吞噬 (法規/公文)**： - **錨點識別**：只有當段落開頭明確出現「第X條」、「一、」時，才視為新區塊的開始。 - **範圍吞噬 (Scope Aggregation) [關鍵規則]**： - 在出現「下一個編號」之前，所有後續的段落（無論是否有換行），**全部屬於當前條文**。 - **嚴禁**因為文中出現換行或句號就將其切斷。 - 範例：第五條有三段文字，中間沒有新編號，這三段必須全部合併在「第五條」的 content 中。 2. **過短區塊合併規則（僅限強結構內容）**： - 僅適用於「情境 A：強結構內容」（如法規、辦法、條文）。 - 僅限於「同一文件、同一頁面」中連續條文。 - 合併條件（需同時滿足）： 1. 連續出現的條文（如：第九條、第十條、第十一條）。 2. 每一條文內容短於 80 個中文字。 3. 條文內容屬於程序性、附則性或說明性規定，無明顯主題轉換。 - 合併方式： - 將符合條件的連續條文合併為「單一 JSON 區塊」。 - `content` 必須完整保留所有原文條文內容與條號，順序不得更動。 - section_title 填寫規則： - 若合併多條條文，`section_title` 填入「第一條的原始條號」。 - 嚴禁自行產生總結型或概括型標題。 - 不可合併情況： - 條文不連續、跨頁、任一條文超過 80 字、或內容涉及不同主題。 3. **零碎段落聚合 (Fragment Merge) -- [關鍵規則]**： - 若連續出現多個「無標題」且內容短少（少於 100 字）的段落，請務必將它們合併為同一個區塊，直到總字數接近 300~500 字，或遇到新的標題為止。 4. **弱結構切分 (一般文章)**： - 若非上述情況，才以「雙換行 (Empty Line)」分隔自然段落。 【section_title 規則（含繼承邏輯）】 1. **首要規則**：若區塊開頭包含原文標題（如「第二條」），請直接填入。 2. **繼承規則**：若該區塊是「上一區塊的延續」（因為字數過長而強制切分），且屬於同一條文，**請填入與上一區塊相同的標題**。 - 範例：上一塊是「第二條」，這一塊是接續內容，標題應填「第二條 (續)」。 3. **無標題規則**：若非上述情況（包含合併後的零碎段落），則填入空字串 `""`。 【長度控制（保守模式）】 - 若單一區塊超過 400 個中文字： - 請在「句號」或「分段符號」處進行切分。 - **切分後的後半段，必須依據上述「繼承規則」填寫 section_title，嚴禁留空。** 【雜訊刪除（嚴格限制）】 僅刪除以下「非內文」資訊，其餘保留： 1. **頁碼與頁首尾**：如 "Page 1"、"- 1 -"、"國立成功大學 ooo 文件"。 2. **孤立的行政日期 (歷史流水帳)**：如單獨成行的「中華民國xx年xx月制定」、「xx年xx月修訂」、「xx年xx月xx日函核定」。例外保護：若日期出現在完整句子中（例如：本校於112年舉辦校慶），嚴禁刪除。 3. **無意義的重複字詞 (重要)**：若遇到連續重複且無完整語意的詞彙（如：「通過 通過 通過」、「修正 修正」），請直接刪除該片段。 【輸出格式】 - 輸出純 JSON Array。 - 欄位：`section_title` (字串), `content` (字串)。 - 不要包含 Markdown (` ```json `)。 【輸入內容】 {content} """ PROMPT_METADATA_EXTRACT = """ 你是一個「文件資訊提取專家」。請根據提供的文件內容（文件首頁），提取標題、日期，並判斷**文件類型**。 【任務一：提取標題 (doc_title)】 1. 標題通常位於第一頁最上方，字體較大。 2. 若無明顯標題，請歸納簡短標題。 3. **【重要】週期性文件標題正規化**： - 若文件為每學年/每學期更新的性質（如「學雜費收費標準」、「行事曆」），請**移除標題中的年份或學年度**，統一輸出為通用標題。 - 範例：「112學年度學雜費收費標準」 -> 輸出 "國立成功大學學雜費收費標準" - 範例：「113學年度行事曆」 -> 輸出 "國立成功大學行事曆" - (目的：讓系統能識別這是同一份文件的不同版本，以便進行舊換新) 4. **【重要】格式要求**：標題中請勿包含任何空白字元 (空格)，請將「國立 成功 大學」輸出為「國立成功大學」。 【任務二：提取日期 (doc_date)】 1. 找出文件中最晚的生效/修訂/發布日期。 2. **【重要】學年度換算規則**： - 優先尋找具體日期 (YYYY-MM-DD)。 - 若無具體日期，僅標示「學年度」或「學期」，請依下列規則換算： A. **【全學年 或 上學期】** (即：標示為第1學期，或**未標示學期**)： - 設定為該學年度起始日。 - 公式：(民國年 + 1911) 年 08 月 01 日 - 範例：「110學年度」、「110上」、「110-1」 -> 2021-08-01 B. **【下學期】** (即：標示為第2學期)： - 設定為該學年度下半起始日 (年份需+1)。 - 公式：(民國年 + 1911 + 1) 年 02 月 01 日 - 範例：「110學年度下學期」、「110下」、「110-2」 -> 2022-02-01 3. 轉換為西元整數 (YYYY-MM-DD)，無日期回傳 0000-00-00。 【任務三：判斷文件類型 (doc_type)】 請根據內容判斷文件屬性，回傳下列其中之一： - "REGULATION": 法規、辦法、要點、細則、章程 (需入庫) - "MEETING": 會議紀錄、議程 (不入庫) - "ARTICLE": 專題報導、人物專訪、活動紀實、校友通訊、新聞稿 (不入庫) - "LIST": 捐款名冊、芳名錄、徵信錄、錄取名單、簽到表 (不入庫) - "OTHER": 其他公告、簡介、手冊 (需入庫) 【判斷邏輯 (關鍵)】 1. **ARTICLE (文章/報導/訪談/刊物)**： - **定義**：敘事體 (Narrative) 的內容，包含人物故事、特定事件的報導、校友活動花絮、或**期刊/電子報/通訊/刊物**形式的內容。 - **關鍵字**：標題或內文包含「專訪」、「採訪」、「報導」、「紀實」、「花絮」、「動態」、「簡訊」、「通訊」、「專題」、「特刊」、**「電子報」、「刊物」、「第\d+期」**。 - **內容特徵**：像在說故事或報新聞，**或是由多篇文章組成的合輯**，而非條列式的法規或純名單。 2. **LIST (名單/名冊)**： - **捐款類**：標題或內文包含「捐款」、「捐贈」、「芳名錄」、「徵信錄」、「致謝」。內容特徵為大量的「姓名/單位」搭配「金額」的表格或列表。 - **一般類**：僅包含人名列表（如錄取名單、簽到表），缺乏完整的文章敘述或規範條文。 3. **MEETING (會議)**： - 包含「會議紀錄」、「議程」、「提案」、「決議」、「出席人員」。 4. **REGULATION (法規)**： - 標題通常以「法」、「辦法」、「細則」、「要點」結尾，內容為條列式規範。 【輸出格式 (JSON)】 請輸出純 JSON： {{ "doc_title": "逆風飛翔—王小明學長專訪", "doc_date": 20230614, "doc_type": "INTERVIEW", "reason": "內容為敘述王學長創業過程的人物特寫，屬於訪談類" }} 【文件內容預覽】 {text_sample} """ NOISE_PATTERNS = [ # --- 1. 頁碼與頁首尾 --- r'^Page\s*\d+$', r'^-\s*\d+\s*-$', r'^\d+\s*/\s*\d+$', r'^第\s*\d+\s*頁.*$', # --- 2. 修法沿革 (最常見的兇手) --- # 邏輯：以日期開頭 (中華民國 or 數字年) + 任意文字 + 結尾是行政動作 r'^\s*中華民國\d+年.*(修正|訂定|公布|核定|備查|通過|廢止|函|增訂).*$', r'^\s*\d{2,3}年\d{1,2}月\d{1,2}日.*(修正|訂定|公布|核定|備查|通過|廢止|函|增訂).*$', r'^\s*\d{2,3}\.\d{1,2}\.\d{1,2}.*(修正|訂定|公布|核定|備查|通過|廢止|函|增訂).*$', # --- 3. 針對你遇到的「函核定」特例 --- # 抓取包含 "教育部" + "號函" 的流水帳 r'.*教育部.*號函.*', # --- 4. 針對殘留的「通過 通過」或「修正通過」 --- # 邏輯：整行只有重複的「通過」或「修正」字眼，沒有其他內容 r'^\s*(修正|通過|修正通過|核定|備查|延會|會議)(\s+(修正|通過|修正通過|核定|備查|延會|會議))*\s*$', # --- 5. 其他 --- r'^[-─＿_＊*]{5,}$', # 分隔線 r'^\s*（本頁空白）\s*$' # --- 6. 英文版修法沿革 (English Legislative History) --- # 針對: "Amended and approved at the 2nd meeting..." # 邏輯: 以 Amended/Approved/Passed 開頭 + 中間包含 meeting/committee + 結尾通常有日期或年份 r'^\s*(Amended|Approved|Passed|Adopted|Ratified).*at\s+the\s+.*(meeting|Committee|Council|Board).*$', # 針對: "Approved by..." 開頭的格式 r'^\s*(Amended|Approved|Passed|Adopted|Ratified)\s+by\s+the\s+.*(Committee|Council|Board).*$', # 針對單純日期在後的格式 (以防萬一) # e.g., "Approved on Feb. 4, 2025" r'^\s*(Amended|Approved|Passed|Adopted).*on\s+(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec).*20\d{2}.*$' ] SKIP_KEYWORDS = [ "會議紀錄", "會議記錄", # 雖然之前說可以看年份，但如果你想直接不處理，就加在這 "捐款芳名錄", "捐贈名單","捐款名單", "簽到表", "出席表", # 這種通常沒內容 "議程", # 單純議程通常沒價值 "致詞", # 只有致詞稿通常無參考價值 "附件", # 許多附件若無主檔名，通常難以辨識，可選擇性跳過 "名單", ] # INTERVIEW (訪談) 或 MEETING (會議紀錄) 或 LIST (名單) SKIP_TYPES = ["ARTICLE", "MEETING", "LIST"] # 【URL 黑名單】特定問題 PDF 檔案，文字提取失敗或無法處理 SKIP_PDF_URLS = [ "https://cid-acad.ncku.edu.tw/var/file/42/1042/img/731/course", # 課程委員會會議紀錄 "https://cashier-ufo.ncku.edu.tw/var/file/96/1096/img/212/withholding", #薪資所得扣繳稅額表 ] # ============================================================================ # PART 1: DATABASE AND PDF DOWNLOAD FUNCTIONS (from retrieve_pdf.py) # ============================================================================ def get_db_connection(db_name): """Establish database connection""" try: db = mysql.connector.connect( host=dbhost, database=db_name, user=dbuser, password=dbpwd ) return db except mysql.connector.Error as e: print(f"Database connection error: {e}") raise def download_pdf(url, save_dir=PDF_CACHE_DIR): """Download PDF from URL and save to local cache""" # Create directory if it doesn't exist os.makedirs(save_dir, exist_ok=True) try: headers = { 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36' } response = requests.get(url, headers=headers, timeout=30, allow_redirects=True) # Check HTTP status code if response.status_code == 404: raise Exception(f"URL 不存在 (404): {url}") if response.status_code >= 400: raise Exception(f"HTTP 錯誤 ({response.status_code}): {url}") # Check if it's actually a PDF content_type = response.headers.get('content-type', '').lower() if 'pdf' not in content_type: raise Exception(f"Not a PDF or content-type check failed: {url}") data = response.content if not data: raise Exception(f"Download failed or empty response: {url}") # Generate MD5 hash of the file content to ensure uniqueness file_hash = hashlib.md5(data).hexdigest() local_path = os.path.join(save_dir, f"{file_hash}.pdf") # If file already exists, skip download if os.path.exists(local_path): return local_path # Write the PDF file with open(local_path, 'wb') as f: f.write(data) return local_path except requests.exceptions.RequestException as e: raise Exception(f"Download error: {str(e)}") # ============================================================================ # PART 2: PDF PARSING FUNCTIONS (from parse_pdf.py) # ============================================================================ def is_skip_document(title): """ 檢查標題是否包含黑名單關鍵字 """ if not title: return False for kw in SKIP_KEYWORDS: if kw in title: return True return False def get_doc_metadata_via_llm(pdf, metadata_title=None, pdf_path=None): # 1. 設定保底標題 fallback_title = "未命名文件" if metadata_title and metadata_title.strip() not in ["", "未命名文件", "Untitled"]: fallback_title = metadata_title # 2. 判斷文字方向（直式或橫式） orientation = "horizontal" # 預設為橫式 if pdf_path: try: orientation = detect_text_orientation(pdf_path) except Exception as e: print(f"[警告] 方向檢測失敗: {e}，預設為橫式") orientation = "horizontal" # 3. 抓取文字 (只看第一頁 + 稍微補強) text_sample = "" try: if len(pdf.pages) > 0: # 根據方向選擇提取方法 if orientation == "vertical": # 直式：使用 extract_vertical_text() try: text_sample = extract_vertical_text(pdf.pages[0]) or "" except Exception as e: print(f"[警告] 提取第一頁直排文字失敗: {e}，切換到橫排") text_sample = pdf.pages[0].extract_text() or "" else: # 橫式：使用一般的 extract_text() try: text_sample = pdf.pages[0].extract_text() or "" except Exception as e: print(f"[警告] 提取第一頁文字失敗: {e}") text_sample = "" if len(text_sample) < 100 and len(pdf.pages) > 1: try: if orientation == "vertical": text_sample += "\n" + (extract_vertical_text(pdf.pages[1]) or "") else: text_sample += "\n" + (pdf.pages[1].extract_text() or "") except Exception as e: print(f"[警告] 提取第二頁文字失敗: {e}") pass except Exception as e: print(f"[警告] 文字提取過程出錯: {e}") pass # 如果抓不到字，直接回傳保底 if len(text_sample) < 10: return fallback_title, "0000-00-00", "OTHER" # 截斷內容 (1500 字通常足夠) text_sample = text_sample[:1500] # ========================================== # 3. [優先] Regex 暴力判斷 (秒殺標準公文) # ========================================== regex_date = "0000-00-00" regex_title = None # [Regex A] 抓日期: "時間：106年11月30日" # 支援: 時間：106年... 或 時間:106年... date_match = re.search(r"時間[:：]\s*(\d{2,3})\s*年\s*(\d{1,2})\s*月\s*(\d{1,2})\s*日", text_sample) if date_match: try: y, m, d = date_match.groups() year = int(y) month = int(m) day = int(d) # 簡單判斷：如果是民國年 (2位或3位數)，加 1911 if year < 1911: year += 1911 regex_date = f"{year:04d}-{month:02d}-{day:02d}" except: pass # [Regex B] 抓標題: 第一行如果包含 "會議紀錄"，它就是標題 first_line = text_sample.split('\n')[0].strip() if "會議紀錄" in first_line: regex_title = first_line # ★★★ 如果 Regex 兩個都抓到了，直接回傳★★★ if regex_title and regex_date != "0000-00-00": #print(f"[秒殺] Regex 直接命中 -> 標題: {regex_title}, 日期: {regex_date}") return regex_title, regex_date, "MEETING" # 【重要】防止 .format() 崩潰 # 如果 PDF 內文含有 { 或 } (例如程式碼或 CSS)，會導致 prompt template 報錯 # 所以要將內文的 { 替換為 {{， } 替換為 }} safe_text_sample = text_sample.replace("{", "{{").replace("}", "}}") #print(f"[DEBUG]safe_text={safe_text_sample}") try: response = client.chat( model=llm_model, format='json', # 強制開啟 JSON 模式，保證穩定性 options={'num_ctx': 2048, 'temperature': 0.0}, # 溫度 0 確保精準 messages=[{ 'role': 'user', # 使用全域變數 PROMPT_METADATA_EXTRACT 'content': PROMPT_METADATA_EXTRACT.format(text_sample=safe_text_sample) }], stream=False # 關閉 Stream ) # 取得完整回應 result_text = response['message']['content'] #print(f"[DEBUG] LLM 回應: {result_text}") # 需要除錯時可打開 # 4. 解析 JSON data = {} try: data = json.loads(result_text) except json.JSONDecodeError: # 萬一回傳的不是標準 JSON，嘗試用 clean_and_parse_json (如果你有引入的話) # 這裡做一個簡單的 fallback print(f"[錯誤] JSON 解析失敗: {result_text}") return fallback_title, 0 # 5. 提取欄位並驗證 llm_title = data.get("doc_title", "").strip() # 處理日期 (轉換為 YYYY-MM-DD 格式) llm_date = "0000-00-00" try: date_value = data.get("doc_date", 0) if isinstance(date_value, int) and date_value > 0: # 如果是整數格式 YYYYMMDD，轉換為字串格式 YYYY-MM-DD date_str = str(date_value).zfill(8) year = date_str[0:4] month = date_str[4:6] day = date_str[6:8] llm_date = f"{year}-{month}-{day}" elif isinstance(date_value, str) and date_value.strip(): # 如果已經是字串格式，直接使用 (假設格式正確) llm_date = date_value.strip() except: pass # 取得 doc_type doc_type = data.get("doc_type", "OTHER").upper() # 轉大寫防呆 # 標題防呆檢查 final_title = llm_title invalid_keywords = ["none", "null", "unknown", "n/a", "無", "找不到標題", "未命名文件", "無標題", "doc_title"] if not final_title or final_title.lower() in invalid_keywords: # 如果 LLM 失敗，嘗試抓取內文第一行非空文字 lines = [l.strip() for l in text_sample.split('\n') if l.strip()] if lines: final_title = lines[0][:50] else: final_title = fallback_title #print(f"[Metadata 結果] 標題: {final_title}, 日期: {llm_date}") #return final_title, llm_date if final_title: # 轉成字串並去除空白 final_title = str(final_title).replace(" ", "").replace("　", "").strip() return final_title, llm_date, doc_type except Exception as e: print(f"[Metadata] 分析失敗: {e}") return fallback_title, "0000-00-00", "OTHER" def remove_noise_patterns(text, patterns): """ 【修正版】逐行清洗雜訊 針對每一行進行 Regex 檢查，若符合雜訊特徵則移除該行。 """ if not text: return "" lines = text.split('\n') cleaned_lines = [] for line in lines: line_stripped = line.strip() # 1. 如果是空行，先暫時保留 (之後再一次整理) if not line_stripped: # 但如果連續空行太多，最後再由 join 處理，這裡先保留結構 cleaned_lines.append("") continue is_noise = False for pattern in patterns: # 使用 re.search 或 re.match # 這裡加上 re.IGNORECASE 忽略大小寫 if re.search(pattern, line_stripped, re.IGNORECASE): # print(f"[DEBUG-清洗] 刪除雜訊行: {line_stripped}") # 除錯用 is_noise = True break if not is_noise: cleaned_lines.append(line) # 重新組合 result = "\n".join(cleaned_lines) # 最後整理：把連續 3 個以上的換行變成 2 個 (段落分明) result = re.sub(r'\n{3,}', '\n\n', result) return result.strip() def post_process_merge(json_list, max_merge_length=500): """ 【後處理神器】 邏輯： 1. 如果當前區塊「沒有標題」 (section_title == "") 2. 且把它合併到「上一塊」之後，總字數還算安全 (< max_merge_length) 3. 那就合併！(因為這通常代表它是上一條文的後半段) 4. 如果合併後會太長，那就改用「繼承標題」 (title = last_title + " (續)") """ if not json_list: return [] merged_results = [] # 初始化：先放入第一項 if json_list: current_master = json_list[0] # 防呆：第一項如果就沒標題，給個預設 if not current_master['section_title']: current_master['section_title'] = "前言/總則" merged_results.append(current_master) # 從第二項開始遍歷 for i in range(1, len(json_list)): item = json_list[i] title = item.get('section_title', '').strip() content = item.get('content', '').strip() # 取得最後一個已存入的 master chunk last_master = merged_results[-1] # --- 判斷邏輯 --- # 狀況 A: 這是一塊「無標題」的內容 (也就是你的問題所在) if not title: # 檢查合併後長度是否還行 (設 800 是因為 embedding 模型會截斷，但至少語意完整) if len(last_master['content']) + len(content) < max_merge_length: # 【執行合併】 # print(f"[自動修復] 將無標題段落合併至: {last_master['section_title']}") last_master['content'] += "\n\n" + content else: # 【太長了，不能合 -> 改為繼承標題】 # print(f"[自動修復] 內容過長，改為繼承標題: {last_master['section_title']} (續)") item['section_title'] = last_master['section_title'] + " (續)" merged_results.append(item) # 狀況 B: 這是一塊「有標題」的新內容 (如：第六條) else: # 直接存入新的 master merged_results.append(item) return merged_results def merge_sequential_short_chunks(json_list, short_threshold=150, max_total_length=500): """ 【後處理第二關：短條文聚合】 針對連續出現的「有標題」但「內容很短」的區塊進行合併。 例如：第九條(50字) + 第十條(40字) + 第十一條(30字) -> 合併為一塊 """ if not json_list: return [] merged_results = [] # 緩衝區 (Accumulator) buffer = None for item in json_list: current_title = item.get('section_title', '').strip() current_content = item.get('content', '').strip() current_len = len(current_content) # 初始化緩衝區 if buffer is None: buffer = item continue # --- 判斷邏輯 --- buffer_len = len(buffer['content']) # 條件 1: 當前區塊是「短區塊」 (例如 < 150 字) # 條件 2: 緩衝區加上當前區塊，不會超過總長度限制 (例如 600 字) # 條件 3: 當前區塊必須有內容 (防呆) if (current_len < short_threshold) and \ (buffer_len + current_len < max_total_length): # 【執行合併】 # 1. 合併內容 buffer['content'] += "\n\n" + current_content # 2. 更新標題 (變更為範圍表示法) # 邏輯：取緩衝區原本標題的「開頭」 + 當前標題的「結尾」 # 例如: "第九條" + "第十條" -> "第九條 ～ 第十條" # 如果緩衝區已經是範圍 (e.g. "第九條 ～ 第十條")，要先拆開取頭 start_title = buffer['section_title'].split(' ～ ')[0].split(' (')[0] # 簡單清洗 end_title = current_title.split(' (')[0] # 簡單清洗 # 避免標題重複 (如: 第九條 ～ 第九條) if start_title != end_title: buffer['section_title'] = f"{start_title} ～ {end_title}" # print(f"[自動聚合] 合併短條文: {buffer['section_title']}") else: # 不符合合併條件 (例如遇到長條文了)，結算緩衝區 merged_results.append(buffer) buffer = item # 當前項成為新的緩衝區 # 迴圈結束，存入最後一項 if buffer: merged_results.append(buffer) return merged_results # === 輔助函式：清理 LLM 回傳的 JSON 字串 === def clean_and_parse_json(llm_output): """ 嘗試解析 LLM 回傳的 JSON，處理 Markdown code block 和不完整的格式。 """ try: # 1. 移除 Markdown code blocks (```json ... ```) clean_text = re.sub(r'```json\s*', '', llm_output, flags=re.IGNORECASE) clean_text = re.sub(r'```', '', clean_text) clean_text = clean_text.strip() # 2. 嘗試解析 JSON return json.loads(clean_text) except json.JSONDecodeError as e: print(f"JSON 解析失敗: {e}") # 如果失敗，印出原始文字以供除錯 # print(f"原始輸出片段: {llm_output[:200]}...") return [] def detect_text_orientation(pdf_path, check_pages=3, min_chars_per_page=15): """ 偵測 PDF 文字排版方向：直排 (vertical) 或 橫排 (horizontal) 參數： pdf_path: PDF 檔案路徑 check_pages: 檢查前 N 頁（預設 3 頁） min_chars_per_page: 每頁最少字符數（預設 15） 返回： "vertical" (直排) 或 "horizontal" (橫排) """ vertical_score = 0 horizontal_score = 0 valid_pairs = 0 # 有效的字對數量 try: with pdfplumber.open(pdf_path) as pdf: # 檢查前幾頁 for page in pdf.pages[:check_pages]: chars = page.chars if len(chars) < min_chars_per_page: continue # ================================================== # 線性掃描：比較相鄰字符的位置 # ================================================== for i in range(len(chars) - 1): c1 = chars[i] c2 = chars[i + 1] # 取字體大小（預設 12） font_size = c1.get("size", 12) if font_size <= 0: font_size = 12 # 計算相鄰字的水平和垂直距離 dx = abs(c2["x0"] - c1["x0"]) dy = abs(c2["top"] - c1["top"]) # -------------------------------------------------- # 過濾異常情況：距離太遠的字對（可能是換段或跳行） # -------------------------------------------------- dist_limit = font_size * 2.5 # 容許 2.5 倍字高的距離 if dx > dist_limit and dy > dist_limit: # 兩個軸都太遠，說明不連續 continue # 橫向跳躍（換欄）：X 軸跳躍但 Y 軸穩定 if dx > dist_limit and dy < font_size * 0.8: continue # 縱向跳躍（換行/段落）：Y 軸跳躍但 X 軸穩定 if dy > dist_limit and dx < font_size * 0.8: continue # -------------------------------------------------- # 判定方向 # 容許值：0.6 倍字高（較寬鬆，容許更多波動） # -------------------------------------------------- tol = font_size * 0.6 # 橫排特徵：Y 軸幾乎不動，X 軸明顯移動 if dy < tol and dx > font_size * 0.3: horizontal_score += 1 valid_pairs += 1 # 直排特徵：X 軸幾乎不動，Y 軸明顯移動 elif dx < tol and dy > font_size * 0.3: vertical_score += 1 valid_pairs += 1 # 輸出診斷資訊 #print(f"[檢測結果] 檢查字對數: {valid_pairs}, 直排: {vertical_score}, 橫排: {horizontal_score}") # 邊界情況：如果沒有有效的判斷，預設為橫排 if valid_pairs < 10: #print(f"[警告] 樣本過少 ({valid_pairs} 對)，可能判斷不准確") return "horizontal" if vertical_score == 0 and horizontal_score == 0: return "horizontal" # 判定邏輯：取積分較高的方向 # 不考慮混合排版，只判定主要方向 result = "vertical" if vertical_score > horizontal_score else "horizontal" #confidence = max(vertical_score, horizontal_score) / valid_pairs if valid_pairs > 0 else 0 # 檢查是否為混合排版（兩者積分都接近） #ratio = min(vertical_score, horizontal_score) / max(vertical_score, horizontal_score) if max(vertical_score, horizontal_score) > 0 else 0 #if ratio > 0.3: # print(f"[注意] 檢測到可能的混合排版（比例: {ratio:.1%}），但只判定主要方向") #print(f"[可信度] {confidence:.1%}") return result except Exception as e: error_msg = str(e) if "FontBBox" in error_msg or "cannot be parsed as 4 floats" in error_msg: print(f"[跳過] 檔案字型損壞 (FontBBox error): {pdf_path}") return "horizontal" elif "font descriptor" in error_msg.lower(): print(f"[跳過] 字型描述符損壞: {pdf_path}") return "horizontal" else: print(f"[警告] 偵測方向失敗，預設為橫排. 錯誤: {error_msg}") return "horizontal" def extract_vertical_text(page): """ 只處理單一頁面的直排文字提取 :param page: pdfplumber 的 page 物件 """ try: chars = page.chars if not chars: return "" # 1. 關鍵排序邏輯：直排 (從右到左，從上到下) # 優先序: X軸 (由大到小 -c['x0']) -> Y軸 (由小到大 c['top']) # 這裡加入一點 X 軸的容錯分群 (Bucket) 避免些微歪斜導致亂序 # 但為了效能與簡潔，我們先用簡單排序 # 為了避免同一行文字因為些微高低差導致順序錯亂 (例如標點符號)， # 我們可以稍微放寬排序條件，但針對這份整齊的刊物，直接排通常沒問題。 # ============== 【防護】字型損壞檢查 ============== # 檢查字符中是否有無效的座標值（可能導致 FontBBox 錯誤） valid_chars = [] for char in chars: try: # 檢查座標是否為有效的數字 x0 = char.get('x0') top = char.get('top') if x0 is None or top is None: continue if not isinstance(x0, (int, float)) or not isinstance(top, (int, float)): continue valid_chars.append(char) except (KeyError, TypeError, ValueError): # 跳過有問題的字符 continue if not valid_chars: return "" chars_sorted = sorted(valid_chars, key=lambda c: (-c['x0'], c['top'])) page_text = "" for char in chars_sorted: page_text += char.get('text', '') return page_text except Exception as e: error_msg = str(e) if "FontBBox" in error_msg or "cannot be parsed" in error_msg: print(f"[警告] 字型損壞，無法提取直排文字: {error_msg}") return "" else: print(f"[警告] 提取直排文字失敗: {error_msg}") return "" def extract_mixed_content(page, tables): """ 混合提取頁面內容： 1. 提取不在表格內的文字。 2. 提取表格內容 (轉換為 JSON/Markdown)。 3. 依照 Y 軸 (top) 順序合併，還原閱讀流。 """ # 用來儲存所有的內容區塊 (Text 或 Table) # 格式: {'top': y_coord, 'type': 'text/table', 'content': string} content_objects = [] # --------------------------------------------------------- # 1. 處理表格 (加入防呆機制) # --------------------------------------------------------- for table in tables: try: # 【關鍵修復】檢查表格 Bounding Box 是否出界 # 取得頁面邊界 p_width = page.width p_height = page.height # table.bbox 格式: (x0, top, x1, bottom) t_bbox = table.bbox # 檢查：如果表格座標完全在頁面外，直接跳過 if (t_bbox[0] > p_width or t_bbox[1] > p_height or t_bbox[2] < 0 or t_bbox[3] < 0): print(f"[忽略] 表格座標完全在頁面外: {t_bbox}") continue # 嘗試提取表格資料 # 使用 x_tolerance / y_tolerance 增加容錯率 table_data = table.extract(x_tolerance=5, y_tolerance=5) if not table_data: continue # 將表格轉為 Markdown 格式 (LLM 較易讀懂) 或 JSON # 這裡簡單轉為 JSON 字串 table_str = json.dumps(table_data, ensure_ascii=False) # 加入物件列表，以表格的頂部 (top) 作為排序依據 content_objects.append({ 'top': t_bbox[1], 'type': 'table', 'content': f"\n[表格資料開始]\n{table_str}\n[表格資料結束]\n" }) except ValueError as e: # 這是捕捉 "Bounding box not fully within parent page" 的關鍵 print(f"[警告] 表格解析失敗 (座標溢位)，已略過。錯誤: {e}") continue except Exception as e: print(f"[警告] 表格處理發生未預期錯誤: {e}") continue # --------------------------------------------------------- # 2. 處理文字 (剔除表格內的字) # --------------------------------------------------------- try: # 取得頁面所有文字物件 (包含座標) try: words = page.extract_words() except Exception as e: if "FontBBox" in str(e) or "cannot be parsed" in str(e): print(f"[警告] 字型損壞，無法提取單詞: {e}") words = [] else: print(f"[警告] 提取單詞失敗: {e}") words = [] # 建立一個檢查函式：判斷某個字是否「掉進」了任何一個表格裡 def is_inside_any_table(word_bbox, table_list): try: wx0, wtop, wx1, wbottom = word_bbox w_center_x = (wx0 + wx1) / 2 w_center_y = (wtop + wbottom) / 2 for t in table_list: tx0, ttop, tx1, tbottom = t.bbox # 簡單判定：如果字的中心點在表格範圍內，就算是在表格裡 if tx0 <= w_center_x <= tx1 and ttop <= w_center_y <= tbottom: return True except (TypeError, ValueError): pass return False current_text_block = [] current_block_top = 0 # 簡單的分群邏輯：將連續的文字組合成一個區塊 for word in words: word_bbox = (word['x0'], word['top'], word['x1'], word['bottom']) # 如果這個字在表格裡，就跳過 (因為表格已經在上面處理過了) if is_inside_any_table(word_bbox, tables): # 如果之前有累積的文字，先存起來 (因為遇到表格被打斷了) if current_text_block: text_content = " ".join([w['text'] for w in current_text_block]) content_objects.append({ 'top': current_block_top, # 使用該區塊第一個字的 top 'type': 'text', 'content': text_content }) current_text_block = [] continue # 如果是新的區塊，記錄起始高度 if not current_text_block: current_block_top = word['top'] current_text_block.append(word) # 迴圈結束後，別忘了存最後一段文字 if current_text_block: text_content = " ".join([w['text'] for w in current_text_block]) content_objects.append({ 'top': current_block_top, 'type': 'text', 'content': text_content }) except Exception as e: print(f"[警告] 文字提取發生錯誤: {e}，改用純全頁文字提取") return page.extract_text() # --------------------------------------------------------- # 3. 合併與排序 # --------------------------------------------------------- # 依照 'top' (Y座標) 由小到大排序，還原閱讀順序 content_objects.sort(key=lambda x: x['top']) # 組合最終字串 final_content = "" for obj in content_objects: final_content += obj['content'] + "\n" return final_content # === 主函式：自動分流處理器 === def process_pdf_auto_router(pdf_path): """ 自動分流 PDF 處理器 包含 PDF 驗證、修復和容錯機制 """ all_json_results = [] # 【防護】檢查檔案是否存在 if not os.path.exists(pdf_path): print(f"[跳過] 檔案不存在: {pdf_path}") return {"skipped": True, "reason": "File not found"} if not os.path.isfile(pdf_path): print(f"[跳過] 路徑不是檔案: {pdf_path}") return {"skipped": True, "reason": "Path is not a file"} # 【防護】檢查檔案大小 file_size = os.path.getsize(pdf_path) if file_size == 0: print(f"[跳過] 檔案為空: {pdf_path}") return {"skipped": True, "reason": "Empty file"} table_settings = { "vertical_strategy": "lines", "horizontal_strategy": "lines", "snap_tolerance": 4, "join_tolerance": 4, } try: # 【防護】先嘗試驗證 PDF 是否能打開 with pdfplumber.open(pdf_path) as pdf: #print(f"--- 開始處理檔案: {pdf_path} ---") # 【防護】檢查頁數限制（超過 50 頁則跳過） num_pages = len(pdf.pages) if num_pages > 50: print(f"[跳過] 檔案頁數過多 ({num_pages} 頁 > 50 頁)，跳過處理") return {"skipped": True, "reason": f"Too many pages ({num_pages} > 50)"} # ============================================================== # 步驟 -1: 全域文字量快篩 (針對 QR Code、純圖片檔) # ============================================================== total_valid_chars = 0 # 只檢查前 3 頁。只要前 3 頁累積超過 100 字，就當作是正常文件。 check_pages = pdf.pages[:3] for p in check_pages: try: txt = p.extract_text() or "" total_valid_chars += len(txt.strip()) except Exception as e: if "FontBBox" in str(e) or "cannot be parsed" in str(e): print(f"[警告] 頁面字型損壞，跳過此頁: {e}") continue else: print(f"[警告] 提取文字失敗: {e}") continue # 如果字數夠多了，就不用再檢查，直接認定為有效文件，跳出檢查迴圈 if total_valid_chars > 100: break # 判斷結果：如果字數太少，直接回傳 skipped 訊號 if total_valid_chars < 50: #print(f"[跳過] 檔案內容過少 (Total: {total_valid_chars} chars)，判定為純圖片/QR Code。") return { "skipped": True, "doc_title": os.path.basename(pdf_path), "reason": f"Low text density (Total chars: {total_valid_chars})" } doc_title = pdf.metadata.get("Title") or "未命名文件" #current_section_title = doc_title # 呼叫 LLM 校正標題及取得日期 final_doc_title, doc_date, doc_type = get_doc_metadata_via_llm(pdf, doc_title, pdf_path) print(f"[debug]final_doc_title={final_doc_title} date:{doc_date} type={doc_type}") # 過濾黑名單檔案 if is_skip_document(final_doc_title): print(f"[過濾] 偵測到忽略關鍵字，跳過檔案: {final_doc_title}") # 回傳一個明確的 JSON 結構，而不是空 list return { "skipped": True, "doc_title": final_doc_title, "reason": "Match blacklist keyword" } # 類型黑名單 # 如果 LLM 認為這是 INTERVIEW (訪談) 或 MEETING (會議紀錄) 或 LIST (名單) if doc_type in SKIP_TYPES: print(f"[跳過] LLM 判定文件類型為: {doc_type}") return { "skipped": True, "doc_title": final_doc_title, "reason": f"Doc Type is {doc_type}" } for page_num, page in enumerate(pdf.pages, start=1): #print(f"\n正在分析第 {page_num} 頁...") # --- 步驟 0: 偵測文字排版方向（直排或橫排） --- # 為了避免重複檢測，只在前 3 頁檢測，取得全文的排版方向 if page_num == 1: text_orientation = detect_text_orientation(pdf_path, check_pages=3) #print(f"[排版偵測] 此 PDF 為: {text_orientation}") # 如果是直排，用特殊的提取方式 if text_orientation == "vertical": # 直排：從右到左、由上到下 content_for_llm = extract_vertical_text(page) selected_prompt = PROMPT_TEXT # 直排通常是文件內容，用文字提示詞 #print(f"[第 {page_num} 頁] 使用直排提取模式") else: # 橫排：正常流程（原有邏輯） # --- 步驟 1: 偵測有效表格 --- raw_tables = page.find_tables(table_settings) valid_tables = [] # 過濾雜訊 (只保留夠大的表格) for table in raw_tables: data = table.extract() if not data: continue num_cols = len(data[0]) num_rows = len(data) # 判斷標準：欄位>=2 且 (欄位>=4 或 行數>=2) if num_cols >= 2 and (num_cols >= 4 or num_rows >= 2): valid_tables.append(table) # --- 步驟 2: 分流 (Router) --- has_table = len(valid_tables) > 0 content_for_llm = "" selected_prompt = "" if has_table: #print(f"模式: [表格修復模式] - 偵測到 {len(valid_tables)} 個表格") # 呼叫上面的輔助函式，傳入 page 和 table 物件 content_for_llm = extract_mixed_content(page, valid_tables) selected_prompt = PROMPT_TABLE else: #print(f"模式: [純文字模式] - 無表格，執行法規解析") # 直接抓純文字 content_for_llm = page.extract_text() selected_prompt = PROMPT_TEXT content_for_llm = remove_noise_patterns(content_for_llm, NOISE_PATTERNS) #print("[DEBUG]content_for_llm="+content_for_llm) # --- 步驟 3: 呼叫 LLM (針對該頁面) --- if content_for_llm and content_for_llm.strip(): try: try: response = client.chat( model=llm_model, options={'num_ctx':num_ctx, 'temperature':0.1}, messages=[{'role':'user', 'content': selected_prompt.format(content=content_for_llm)}], stream=False ) result_text = response["message"]["content"] except Exception as e: print(f"[LLM 錯誤] 第 {page_num} 頁 LLM 調用失敗: {e}") continue #print("[DEBUG]LLM result:"+result_text) # --- 步驟 4: 解析 JSON 並印出 --- parsed_json = clean_and_parse_json(result_text) # 2. 修復被切斷的長文 parsed_json = post_process_merge(parsed_json, max_merge_length=500) # 3. 聚合過碎的短文 parsed_json = merge_sequential_short_chunks(parsed_json, short_threshold=200, max_total_length=500) parsed_json = post_process_merge(parsed_json) # 確保回傳的是 List，如果是單個 Dict 就包成 List #if isinstance(parsed_list, dict): # parsed_list = [parsed_list] # --- D. 注入統一標題 (Data Injection) --- for item in parsed_json: item['doc_title'] = final_doc_title item['doc_date'] = doc_date item['page_num'] = page_num # 加入列表 all_json_results.append(item) except json.JSONDecodeError: print(f"[錯誤] 第 {page_num} 頁解析 JSON 失敗，跳過。") # 實務上這裡可以記錄 raw_content 以便 debug except Exception as e: print(f"[錯誤] 發生未預期錯誤: {e}") except Exception as e: error_msg = str(e) if "FontBBox" in error_msg or "cannot be parsed as 4 floats" in error_msg: print(f"[跳過] 檔案字型損壞 (FontBBox error): {pdf_path}") return {"skipped": True, "reason": "Malformed PDF FontBBox"} elif "font descriptor" in error_msg.lower(): print(f"[跳過] 字型描述符損壞: {pdf_path}") return {"skipped": True, "reason": "Font descriptor error"} else: print(f"[錯誤] 其他錯誤: {error_msg}") return {"skipped": True, "reason": error_msg} # 檢查：如果所有頁面都失敗，all_json_results 為空 if not all_json_results: print(f"[警告] 檔案所有頁面解析都失敗，回傳空結果") return [] return all_json_results # ============================================================================ # PART 3: MAIN PROCESSING FUNCTION (MERGED) # ============================================================================ def build_knowledge_database(): """Main function to build knowledge database from PDFs (Refactored for Atomic Updates)""" # Connect to databases conn_rag = get_db_connection(database) conn_ipsystem = get_db_connection(database_2) # 定義日期裁判函式 (用於決定是否更新) def should_update_record(existing_date_val, new_date_val): s_new = str(new_date_val).strip().replace('-', '').replace('/', '') s_old = str(existing_date_val).strip().replace('-', '').replace('/', '') invalid_list = ['00000000', '0', 'None', '', 'nan', 'null'] is_new_invalid = s_new in invalid_list is_old_invalid = s_old in invalid_list if is_new_invalid: if not is_old_invalid: return False, "New date invalid, preserving existing valid date" return True, "Both dates invalid, updating content change" # 兩者都無日期但內容變了，更新 if is_old_invalid: return True, "Updating from invalid date to valid date" if s_new >= s_old: return True, f"Update allowed ({s_new} >= {s_old})" return False, f"Skipping older version ({s_new} < {s_old})" try: cursor_rag = conn_rag.cursor(dictionary=True) cursor_ipsystem = conn_ipsystem.cursor(dictionary=True) # Query to get URLs to process #sql = "SELECT id, url, url_redirect, status FROM ws_sites_urls_20250924 WHERE id = %s ORDER BY id ASC" sql = "SELECT id, url, url_redirect, status FROM ws_sites_urls_20250924 WHERE ext = %s ORDER BY id ASC" # Test with specific ID #test_id = 10526 # 國立成功大學教師出國講學及國內外研究進修申請作業要點 #test_id = 323 # 教育部捐資獎牌申請須知, 文字+表格, 版面:直式、橫式 #test_id= 10233 #國立成功大學專任教師兼任營利事業機構獨立董事、董事或監察人名單 #test_id= 10977 #國立成功大學適用勞動基準法人員給假一覽表 #test_id= 11037 #國立成功大學教師評審委員會設置辦法 #test_id= 403 #受贈文物資產審鑑委員會 106 年度第 1 次會議紀錄 test_id= 594 num = 0 insert_num = 0 update_num = 0 #cursor_ipsystem.execute(sql, (test_id,)) cursor_ipsystem.execute(sql, ('pdf',)) pages = cursor_ipsystem.fetchall() for page in pages: num += 1 pdf_url = page.get('url_redirect') or page.get('url') # --- 黑名單檢查 --- skip_this_pdf = any(skip_pattern in pdf_url for skip_pattern in SKIP_PDF_URLS) if skip_this_pdf: print(f"[{page['id']}] {pdf_url} - 已列入黑名單，跳過處理") continue # --- 下載 PDF --- try: local_pdf = download_pdf(pdf_url) except Exception as e: error_msg = str(e) print(f"[{page['id']}] {pdf_url} - download failed Reason: {error_msg}") # 下載失敗記錄邏輯 (保持原樣) sql_check_download_fail = "SELECT id FROM pdf_ncku WHERE url = %s AND embedding_status = 'skip' LIMIT 1" cursor_rag.execute(sql_check_download_fail, (pdf_url,)) existing_record = cursor_rag.fetchone() if not existing_record: try: sql_insert_fail = """INSERT INTO pdf_ncku (url, title, file_hash, page_num, section_title, content, url_available, embedding_status, doc_date, last_fetched) VALUES (%s, '', '', 1, '', %s, 0, 'skip', '0000-00-00', %s)""" cursor_rag.execute(sql_insert_fail, [pdf_url, error_msg, datetime.now().strftime('%Y-%m-%d %H:%M:%S')]) conn_rag.commit() except: conn_rag.rollback() continue # --- 計算 File Hash --- with open(local_pdf, 'rb') as f: file_hash = hashlib.sha256(f.read()).hexdigest() # --- 第一關：Hash 快篩 --- sql_check_hash = "SELECT file_hash FROM pdf_ncku WHERE file_hash = %s LIMIT 1" cursor_rag.execute(sql_check_hash, (file_hash,)) existing_hash_record = cursor_rag.fetchone() if existing_hash_record: print(f"[{page['id']}] {pdf_url} - file_hash 相同，檔案內容未變") if os.path.exists(local_pdf): os.remove(local_pdf) continue # --- 解析 PDF --- print(f"[{page['id']}] {pdf_url} - 檔案內容變更或首次處理") rows = process_pdf_auto_router(local_pdf) # --- 解析結果處理 (黑名單或錯誤) --- if isinstance(rows, dict) and rows.get('skipped'): # (保持原有的黑名單寫入邏輯) print(f" -> 識別為黑名單/Skip") # 檢查該 URL 是否已存在 sql_check_blacklist = "SELECT id FROM pdf_ncku WHERE url = %s AND embedding_status = 'skip' LIMIT 1" cursor_rag.execute(sql_check_blacklist, (pdf_url,)) existing_record = cursor_rag.fetchone() if not existing_record or not existing_record.get('id'): sql_insert_skip = """ INSERT INTO pdf_ncku (url, title, file_hash, page_num, section_title, content, url_available, embedding_status, doc_date, last_fetched) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ try: cursor_rag.execute(sql_insert_skip, [pdf_url, rows.get('doc_title', ''),file_hash,1,'','',0,'skip','0000-00-00',datetime.now().strftime('%Y-%m-%d %H:%M:%S')]) conn_rag.commit() except Exception as e: conn_rag.rollback() else: print(f"[{page['id']}] {pdf_url} - 黑名單記錄已存在，跳過插入") # 清理下載的 PDF 檔案 if os.path.exists(local_pdf): os.remove(local_pdf) continue # Case B: File parsing result is empty or error if not rows or (isinstance(rows, dict) and 'error' in rows): print(f" -> 檔案解析結果為空或異常") # 檢查該 URL 是否已存在 sql_check_parse_fail = "SELECT id FROM pdf_ncku WHERE url = %s AND embedding_status = 'skip' LIMIT 1" cursor_rag.execute(sql_check_parse_fail, (pdf_url,)) existing_record = cursor_rag.fetchone() if not existing_record or not existing_record.get('id'): # 存入資料庫，標記為解析失敗 sql_insert_parse_fail = """ INSERT INTO pdf_ncku (url, title, file_hash, page_num, section_title, content, url_available, embedding_status, doc_date, last_fetched) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ try: error_reason = rows.get('reason', '未知錯誤') if isinstance(rows, dict) else '所有頁面解析失敗' cursor_rag.execute(sql_insert_parse_fail, [pdf_url,'',file_hash,1,'','',1,'skip','0000-00-00',datetime.now().strftime('%Y-%m-%d %H:%M:%S')]) conn_rag.commit() except Exception as insert_error: conn_rag.rollback() else: print(f"[{page['id']}] {pdf_url} - 解析失敗記錄已存在，跳過插入") # 清理下載的 PDF 檔案 if os.path.exists(local_pdf): os.remove(local_pdf) continue # ============================================================================== # 重構核心：頁面級原子更新 (Page-Level Atomic Update) # ============================================================================== # 1. 將資料按頁碼分組 (Group by Page Number) pages_data = defaultdict(list) for row in rows: p_num = row.get('page_num', 1) pages_data[p_num].append(row) # 2. 逐頁處理 for page_num, page_rows in pages_data.items(): if not page_rows: continue # 提取該頁共用資訊 (取第一筆即可) first_row = page_rows[0] doc_title = first_row.get('doc_title', '') new_doc_date = first_row.get('doc_date', '') # 檢查資料庫該頁現況 (只查一筆代表) sql_check_page = "SELECT id, doc_date, file_hash FROM pdf_ncku WHERE title = %s AND page_num = %s LIMIT 1" cursor_rag.execute(sql_check_page, (doc_title, page_num)) existing_rec = cursor_rag.fetchone() should_process = False action_desc = "SKIP" if not existing_rec: # [情況 A] 全新頁面 -> 插入 should_process = True action_desc = "INSERT" else: # [情況 B] 頁面已存在 -> 日期裁判 existing_date = existing_rec.get('doc_date', '0000-00-00') do_update, reason = should_update_record(existing_date, new_doc_date) if do_update: should_process = True action_desc = f"UPDATE ({reason})" else: print(f"[SKIP][{page['id']}] Page {page_num} 日期較舊: {reason}") try: # 只要存一筆 if page_num == 1: sql_insert_old = """ INSERT INTO pdf_ncku (url, title, file_hash, page_num, section_title, content, url_available, embedding_status, doc_date, last_fetched) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cursor_rag.execute(sql_insert_old, [pdf_url, doc_title,file_hash, page_num,'','', 0,'skip', new_doc_date,datetime.now().strftime('%Y-%m-%d %H:%M:%S') ]) conn_rag.commit() except Exception as e_skip: # 這裡如果出錯 (例如重複插入)，不要讓程式崩潰，Rollback 後繼續跑下一個 conn_rag.rollback() continue # 跳過此頁 if should_process: try: # [原子操作 Step 1] 刪除該頁所有舊資料 (防止 Section 變更導致殘留) if existing_rec: sql_delete = "DELETE FROM pdf_ncku WHERE title = %s AND page_num = %s" cursor_rag.execute(sql_delete, (doc_title, page_num)) # [原子操作 Step 2] 準備批次插入資料 insert_data_list = [] now_str = datetime.now().strftime('%Y-%m-%d %H:%M:%S') url_available = 0 if page.get('status') == 'delete' else 1 for row in page_rows: insert_data_list.append(( pdf_url, doc_title, file_hash, page_num, row.get('section_title', ''), row.get('content', ''), url_available, 'pending', new_doc_date, now_str )) # [原子操作 Step 3] 執行批次插入 if insert_data_list: sql_insert_batch = """ INSERT INTO pdf_ncku (url, title, file_hash, page_num, section_title, content, url_available, embedding_status, doc_date, last_fetched) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s) """ cursor_rag.executemany(sql_insert_batch, insert_data_list) if action_desc == "INSERT": insert_num += 1 else: update_num += 1 print(f"[{action_desc}][{page['id']}] {pdf_url} (Page {page_num}) - {len(insert_data_list)} sections") # [原子操作 Step 4] 提交交易 conn_rag.commit() except Exception as e: conn_rag.rollback() print(f"[ERROR][{page['id']}] Page {page_num} 更新失敗: {str(e)}") # Clean up if os.path.exists(local_pdf): os.remove(local_pdf) # Print summary print("="*50) print(f"Total Processed: {num}") print(f"Pages Inserted/Updated: {insert_num + update_num}") print("="*50) except Exception as e: print(f"Critical Error: {str(e)}") conn_rag.rollback() raise finally: cursor_ipsystem.close() cursor_rag.close() conn_ipsystem.close() conn_rag.close() if __name__ == '__main__': try: build_knowledge_database() except Exception as e: print(f"Fatal error: {str(e)}") sys.exit(1)