import sys import json import pdfplumber import hashlib import re from collections import Counter from typing import List from ollama import Client import os from dotenv import load_dotenv load_dotenv() 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. 若無明顯標題，請歸納簡短標題。 【任務二：提取日期 (doc_date)】 1. 找出文件中最晚的生效/修訂/發布日期。 2. 轉換為西元整數 (YYYYMMDD)，無日期回傳 0。 【任務三：判斷文件類型 (doc_type)】 請根據內容判斷文件屬性，回傳下列其中之一： - "REGULATION": 法規、辦法、要點、細則、章程 (需入庫) - "MEETING": 會議紀錄、議程 (不入庫) - "INTERVIEW": 人物專訪、校友訪談、口述歷史、個人故事、人物特寫 (不入庫) - "LIST": 名單、芳名錄、簽到表 (不入庫) - "OTHER": 其他公告、簡介、手冊 (需入庫) 【判斷邏輯 (關鍵)】 1. **INTERVIEW (訪談/人物)**： - **格式不限**：不一定要有「問/答」。若是**「敘事體 (Narrative)」**，講述特定人物的求學回憶、職涯歷程、創業故事或人生觀點，即屬於此類。 - **關鍵字**：標題或內文包含「專訪」、「採訪」、「特寫」、「側寫」、「報導」、「校友故事」、「傑出校友」。 - **內容特徵**：大量出現特定人名、學長/姐稱謂、畢業級數 (e.g., 65級)、系別，且內容聚焦於該人物的經歷。 2. **MEETING (會議)**： - 包含「會議紀錄」、「議程」、「提案」、「決議」、「出席人員」。 3. **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*$' ] SKIP_KEYWORDS = [ "會議紀錄", "會議記錄", # 雖然之前說可以看年份，但如果你想直接不處理，就加在這 "捐款芳名錄", "捐贈名單","捐款名單", "簽到表", "出席表", # 這種通常沒內容 "議程", # 單純議程通常沒價值 "致詞", # 只有致詞稿通常無參考價值 "附件", # 許多附件若無主檔名，通常難以辨識，可選擇性跳過 "名單", ] # INTERVIEW (訪談) 或 MEETING (會議紀錄) 或 LIST (名單) SKIP_TYPES = ["INTERVIEW", "MEETING", "LIST"] 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() text_sample = extract_vertical_text(pdf.pages[0]) or "" else: # 橫式：使用一般的 extract_text() text_sample = pdf.pages[0].extract_text() or "" # 如果第一頁字太少 (可能是封面或目錄)，多抓第二頁 if len(text_sample) < 100 and len(pdf.pages) > 1: if orientation == "vertical": text_sample += "\n" + extract_vertical_text(pdf.pages[1]) else: text_sample += "\n" + pdf.pages[1].extract_text() except: pass # 如果抓不到字，直接回傳保底 if len(text_sample) < 10: return fallback_title, 0, "OTHER" # 截斷內容 (1500 字通常足夠) text_sample = text_sample[:1500] # ========================================== # 3. [優先] Regex 暴力判斷 (秒殺標準公文) # ========================================== regex_date = 0 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) # 簡單判斷：如果是民國年 (2位或3位數)，加 1911 if year < 1911: year += 1911 regex_date = year * 10000 + int(m) * 100 + int(d) 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 > 0: #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]text_sample={safe_text_sample}") #print("[debug]LLM to get title and date") 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() # 處理日期 (轉型防呆) try: llm_date = int(data.get("doc_date", 0)) except: llm_date = 0 # 取得 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 return final_title, llm_date, doc_type except Exception as e: print(f"[Metadata] 分析失敗: {e}") return fallback_title, 0, "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: #print(f"[檢測失敗] {e}，預設使用橫排") return "horizontal" def extract_vertical_text(page): """ 只處理單一頁面的直排文字提取 :param page: pdfplumber 的 page 物件 """ chars = page.chars if not chars: return "" # 1. 關鍵排序邏輯：直排 (從右到左，從上到下) # 優先序: X軸 (由大到小 -c['x0']) -> Y軸 (由小到大 c['top']) # 這裡加入一點 X 軸的容錯分群 (Bucket) 避免些微歪斜導致亂序 # 但為了效能與簡潔，我們先用簡單排序 # 為了避免同一行文字因為些微高低差導致順序錯亂 (例如標點符號)， # 我們可以稍微放寬排序條件，但針對這份整齊的刊物，直接排通常沒問題。 chars_sorted = sorted(chars, key=lambda c: (-c['x0'], c['top'])) page_text = "" for char in chars_sorted: page_text += char.get('text', '') return page_text 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: # 取得頁面所有文字物件 (包含座標) words = page.extract_words() # 建立一個檢查函式：判斷某個字是否「掉進」了任何一個表格裡 def is_inside_any_table(word_bbox, table_list): 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 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): all_json_results = [] table_settings = { "vertical_strategy": "lines", "horizontal_strategy": "lines", "snap_tolerance": 4, "join_tolerance": 4, } try: with pdfplumber.open(pdf_path) as pdf: #print(f"--- 開始處理檔案: {pdf_path} ---") # ============================================================== # 步驟 -1: 全域文字量快篩 (針對 QR Code、純圖片檔) # ============================================================== total_valid_chars = 0 # 只檢查前 3 頁。只要前 3 頁累積超過 100 字，就當作是正常文件。 check_pages = pdf.pages[:3] for p in check_pages: txt = p.extract_text() or "" total_valid_chars += len(txt.strip()) # 如果字數夠多了，就不用再檢查，直接認定為有效文件，跳出檢查迴圈 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): # 回傳一個明確的 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+1} 頁...") # --- 步驟 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: 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)}] ) result_text = response["message"]["content"] #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"[錯誤] Chunk {i+1} 解析 JSON 失敗，跳過。") # 實務上這裡可以記錄 raw_content 以便 debug except Exception as e: print(f"[錯誤] 發生未預期錯誤: {e}") except Exception as e: return f"PDF Error: {str(e)}" return all_json_results if __name__ == "__main__": if len(sys.argv) < 2: print("Usage: python parse_pdf.py ") sys.exit(1) try: # 2. 執行你的解析邏輯 final_data = process_pdf_auto_router(sys.argv[1]) # ensure_ascii=False 讓中文正常顯示，不會變成 \uXXXX # indent=2 雖然好讀，但在正式串接時建議拿掉以節省頻寬，這裡為了 Debug 先保留也行 print(json.dumps(final_data, ensure_ascii=False)) sys.exit(0) """ print("\n" + "="*50) print(f"【處理完成】共取得 {len(final_data)} 筆結構化資料") print("="*50 + "\n") for index, record in enumerate(final_data): print(f"資料 ID: {index + 1}") print(f"文件標題: {record.get('doc_title')}") print(f"章節標題: {record.get('section_title')}") print(f"頁碼: {record.get('page_num')}") print(f"內 容: {record.get('content')}...") print("-" * 30) """ except Exception as e: # 捕捉所有錯誤，並以 JSON 格式回傳錯誤訊息給 PHP error_msg = {"error": str(e), "type": type(e).__name__} print(json.dumps(error_msg, ensure_ascii=False)) sys.exit(1) # 告訴 PHP 執行失敗