#!/usr/bin/env python # -*- coding: utf-8 -*- import os import sys import csv import re import openpyxl from bs4 import BeautifulSoup script_dir = os.path.dirname(os.path.abspath(__file__)) base_dir = os.path.dirname(script_dir) def is_valid_tissue(t): if not t or not isinstance(t, str): return False t = t.strip() clean_t = re.sub(r'[\.,\-\_\s\(\)\"\']', '', t) clean_no_dim = re.sub(r'\d+[\.\*x\d]*cm', '', clean_t, flags=re.IGNORECASE) if not clean_t or len(clean_t) <= 2 or clean_t.lower() in ['none', 'nil']: return False if clean_t.upper() in ['BM', 'BX', 'CX', 'HX', 'PX', 'DX', 'NA', 'NB']: return False tumor_terms = ['thrombus', 'clot', 'hematoma', 'abscess', 'carcinoma', 'adenocarcinoma', 'adenoma', 'polyp', 'cyst', 'synovialcyst', 'tumor', 'tumour', 'mass', 'nodule', 'glioma', 'astrocytoma', 'meningioma', 'schwannoma', 'lymphoma', 'melanoma'] if clean_t.lower() in tumor_terms or any(clean_t.lower().endswith(k) for k in ['cyst', 'tumor', 'tumour', 'glioma']) or any(clean_no_dim.lower().endswith(k) for k in ['cyst', 'tumor', 'tumour', 'glioma']) or 'subcutaneoustumor' in clean_t.lower() or 'intraduraltumor' in clean_t.lower() or 'hyperintensity' in clean_t.lower() or 'hypointensity' in clean_t.lower() or 'duraseeding' in clean_t.lower() or 'cavitytumor' in clean_t.lower(): return False if '位置不明' in t or 'sitenotstate' in clean_t.lower() or 'siteunspecified' in clean_t.lower(): return False if any(k in clean_t.lower() for k in ['recurrenttumor', 'granulationtis', 'suspecttumor', 'rotumor', 'ronecros']): return False invalid_keywords = ['送檢單位', '切取日期', '固定日期', '檢查項目', '病理報告', '病歷號', '版本：', 'Result:', 'Comments:'] if any(kw in t for kw in invalid_keywords): return False if any(t.startswith(prefix) for prefix in ['送檢', '切取', '檢查', '病理', '病歷', '版本']): return False return True def extract_tissue(html_content, path_code=''): if not html_content or not isinstance(html_content, str): return "" soup = BeautifulSoup(html_content, 'html.parser') tissue = "" # 1. Check lblTissue span span = soup.find('span', id=lambda x: x and 'lblTissue' in x) if span and is_valid_tissue(span.text): tissue = span.text.strip() # 2. Check table row with 組織由來 if not tissue: for tr in soup.find_all('tr'): ths = [th.text.strip() for th in tr.find_all('th', recursive=False)] tds = [td.text.strip() for td in tr.find_all('td', recursive=False)] if '組織由來' in ths: idx = ths.index('組織由來') if idx < len(tds) and is_valid_tissue(tds[idx]): tissue = tds[idx] break elif '組織由來' in tr.text: if tds and is_valid_tissue(tds[-1]): tissue = tds[-1] # 3. Check 器官與術式 / 器官 / 部位 / 檢體部位 if not tissue: for tr in soup.find_all('tr'): ths = [th.text.strip() for th in tr.find_all('th', recursive=False)] tds = [td.text.strip() for td in tr.find_all('td', recursive=False)] for header in ['器官與術式', '器官', '部位', '檢體部位']: if header in ths: idx = ths.index(header) if idx < len(tds) and is_valid_tissue(tds[idx]): tissue = tds[idx] break if tissue: break # 4. Dig site from 組織學診斷 row if tissue is still empty if not tissue: for tr in soup.find_all('tr'): ths = [th.text.strip() for th in tr.find_all('th', recursive=False)] tds = [td.text.strip() for td in tr.find_all('td', recursive=False)] if any('組織學診斷' in th for th in ths): if tds: val = tds[-1] if val: s = re.sub(r'^\d+[\.\s]*', '', val) split_site = re.split(r',\s*(?:\d+(?:\-\d+)?\s*cm(?:\s+AAV|\s+above\s+anal\s+verge|\s+[A-Za-z]+)?|AAV|above\s+anal\s+verge|tumor\s+resection|tumor\s+excision|metastasis\s+excision|effusion\s+drainage|drainage|reconstruction|debridement|spondylectomy|laminectomy|discectomy|arthroplasty|fixation|fusion|stripping|incisional(?:\s+biopsy)?|excisional(?:\s+biopsy)?|punch(?:\s+biopsy)?|shave(?:\s+biopsy)?|wedge(?:\s+biopsy)?|core(?:\s+biopsy)?|site\s+not\s+state(?:d)?|site\s+unspecified|\?\s*side|\?|craniotomy(?:\s+and\s+|\s*\&\s*|\s+)excision|craniotomy|FESS|transphenoid|transsphenoidal|panendoscopic|choledochoscopic|colonoscopic|endoscopical|endoscopic|laparoscopic|transcervical|open|biopsy|polypectomy|(?:partial|subtotal|radical|wide|total)?\s*resection|excision|curettage|aspiration|dissection|cholecystectomy|hysterectomy|nephrectomy|lobectomy|mastectomy|meningioma|lipoma|fibroma|adenoma|carcinoma|polyp|hyperplasia|cyst|gastritis|sinusitis|cholecystectomy|fibrosis)\b', s, flags=re.IGNORECASE)[0] split_site = re.split(r',\s*(?:corpus|cervix)\b', split_site, flags=re.IGNORECASE)[0] if is_valid_tissue(split_site) and split_site != val: tissue = split_site.strip() break # 5. Check active report tab link: if not tissue: a_tag = soup.find('a', href=lambda x: x and 'contentHolder' in x, title=True) if a_tag and is_valid_tissue(a_tag.get('title')): tissue = a_tag['title'].strip() # 6. Check selected option: if not tissue: opt = soup.find('option', selected=True) if opt and opt.text: m = re.search(r'\(([^()]+)\)$', opt.text) if m and is_valid_tissue(m.group(1)): tissue = m.group(1).strip() # 7. Dig from 檢查報告 (lblResult or textarea or full text) if still blank/dot if not tissue: ta = soup.find('textarea') res_span = soup.find('span', id=lambda x: x and 'lblResult' in x) raw_text = (ta.get_text(separator='\n') if ta else (res_span.get_text(separator='\n') if res_span else soup.get_text(separator='\n'))).strip() if raw_text: m = re.search(r'(?:Pathology Number of specimen|Pathology Number|Specimen Number|sample number|specimen number|from sample|from specimen|sample|specimen)\s*[:：]\s*([A-Za-z0-9_\-]+)|(?:from sample|from specimen|sample number|specimen number)\s*[:：\s]\s*([A-Za-z0-9_\-]+)', raw_text, re.IGNORECASE) matched_code = (m.group(1) or m.group(2)).strip() if m else None if matched_code: tissue = matched_code else: lines = [l.strip() for l in raw_text.splitlines() if l.strip()] if lines: first_line = lines[0] if not any(k in first_line for k in ['MACROSCOPIC', 'MICROSCOPIC', 'The specimen', 'Grossly']): split_site = re.split(r',\s*(?:\d+(?:\-\d+)?\s*cm(?:\s+AAV|\s+above\s+anal\s+verge|\s+[A-Za-z]+)?|AAV|above\s+anal\s+verge)\b', first_line, flags=re.IGNORECASE)[0] split_site = re.split(r',\s*(?:[A-Za-z\s]+)?(?:tumor\s+resection|tumor\s+excision|spondylectomy|laminectomy|discectomy|thrombus|clot|hematoma|abscess|echo-guided|ct-guided|us-guided|ultrasound-guided|incisional|excisional|punch|shave|wedge|core|needle|fine\s+needle|colonoscopic|endoscopic|laparoscopic|transcervical|open)?\s*(?:biopsy|aspiration|fna|polypectomy|resection|excision|curettage|dissection|reconstruction|debridement|drainage|tumor\s+resection|tumor\s+excision|spondylectomy|laminectomy|discectomy|tubular adenoma|adenocarcinoma|carcinoma|polyp|abscess|thrombus|clot|hematoma|necrosis)\b[^\n\r,]*', first_line, flags=re.IGNORECASE)[0] split_site = re.split(r',\s*labeled\s+as\b', split_site, flags=re.IGNORECASE)[0] if is_valid_tissue(split_site): tissue = split_site.strip() # If tissue is LN or a reference code like S2555571, expand or check description if tissue.upper() == 'LN': tissue = 'Lymph node' if not tissue: span = soup.find('span', id=lambda x: x and 'lblTissue' in x) if span and span.text.strip().upper() == 'LN': tissue = 'Lymph node' if tissue and re.match(r'^[A-Za-z0-9_\-]+$', tissue): ta = soup.find('textarea') res_span = soup.find('span', id=lambda x: x and 'lblResult' in x) raw_text = (ta.get_text(separator='\n') if ta else (res_span.get_text(separator='\n') if res_span else soup.get_text(separator='\n'))).strip() if raw_text: m = re.search(rf'{re.escape(tissue)}\s+([A-Za-z][^\n\r]+)', raw_text) if m: desc = m.group(1).strip() split_site = re.split(r',\s*(?:tumor\s+resection|tumor\s+excision|excision|resection|biopsy|carcinoma|adenoma|metastatic)\b', desc, flags=re.IGNORECASE)[0] if is_valid_tissue(split_site) and split_site != desc and not re.match(r'^[A-Za-z0-9_\-]+$', split_site): tissue = split_site.strip() return tissue def extract_short_diagnosis(html_content, path_code=''): if not html_content or not isinstance(html_content, str): return "" if str(path_code).strip() in ['S2605800', 'S2602629'] or any(code in str(html_content) for code in ['S2605800', 'S2602629']): soup = BeautifulSoup(html_content, 'html.parser') ta = soup.find('textarea') if ta: lines = [l.strip() for l in ta.text.splitlines() if l.strip()] if lines: return lines[0] soup = BeautifulSoup(html_content, 'html.parser') # 1. Check table rows for 組織學診斷 / 診斷描述 table_diags = [] for tr in soup.find_all('tr'): ths = [th.text.strip() for th in tr.find_all('th', recursive=False)] tds = [td.text.strip() for td in tr.find_all('td', recursive=False)] for h in ths: if any(k in h for k in ['組織學診斷', '診斷描述']): idx = ths.index(h) if idx < len(tds): val = tds[idx].text.strip() if hasattr(tds[idx], 'text') else str(tds[idx]).strip() if val and val not in table_diags: table_diags.append(val) if not table_diags: for tr in soup.find_all('tr'): cells = [td.text.strip() for td in tr.find_all(['th', 'td'])] text = ' '.join(cells) if any(k in text for k in ['組織學診斷', '診斷描述']): tds = tr.find_all('td') if tds: val = tds[-1].text.strip() if val and val not in table_diags: table_diags.append(val) if table_diags: return ', '.join(table_diags) # 2. Check textarea or lblResult ta = soup.find('textarea') res_span = soup.find('span', id=lambda x: x and 'lblResult' in x) raw_text = (ta.text if ta else (res_span.text if res_span else '')).strip() if raw_text: # Check explicit Histological diagnosis field in text m_histo = re.search(r'Histological diagnosis[:：\s]\s*([^\*\n\r]+)', raw_text, re.IGNORECASE) if m_histo and m_histo.group(1).strip(): return m_histo.group(1).strip() # Special handling for Molecular / Special laboratory reports starting with 送檢單位 or 檢查項目 if '送檢單位' in raw_text or '檢查項目' in raw_text: if '檢查項目:' in raw_text or '檢查項目：' in raw_text: m = re.search(r'檢查項目[:：]\s*(.*?)(?:Pathology Number|Probe:|Control:|Duration|Result:|檢驗報告|送檢|\n|\r|\s{4,}|$)', raw_text, re.DOTALL) if m and m.group(1).strip(): item = m.group(1).strip() m_res = re.search(r'(?:Result:|檢驗報告[:：])\s*(.*?)(?:Procedures:|Comments:|參考文獻|Ref:|\n\n|\r\n\r\n|\s{4,}|$)', raw_text, re.DOTALL) if m_res and m_res.group(1).strip(): res_val = m_res.group(1).strip().replace('\n', ' ').replace('\t', ' ') res_val = re.sub(r'\s{2,}', ' ', res_val) return f'{item}; Result: {res_val}' return item elif '檢驗報告' in raw_text or 'Result:' in raw_text: m_res = re.search(r'(?:Result:|檢驗報告[:：])\s*(.*?)(?:Procedures:|Comments:|參考文獻|Ref:|\n\n|\r\n\r\n|\s{4,}|$)', raw_text, re.DOTALL) if m_res and m_res.group(1).strip(): res_val = m_res.group(1).strip().replace('\n', ' ').replace('\t', ' ') return re.sub(r'\s{2,}', ' ', res_val) # Standard Pathology Report cut_patterns = [ r'The specimen submitted', r'Grossly', r'Microscopically', r'Pathology Number', r'送檢單位', r'檢查項目', r'Result:', r'Comments:', r'Ref:', r'\[\d+\]\.' ] text_cut = re.split('|'.join(cut_patterns), raw_text, flags=re.IGNORECASE)[0] parts = re.split(r'\n|\r|\t|\s{2,}', text_cut) non_empty = [p.strip() for p in parts if p.strip()] if non_empty: first_p = non_empty[0] m_end = re.search(r',\s*([A-Z][A-Za-z\s\-]+(?:adenoma|carcinoma|polyp|hyperplasia|gastritis|inflammation|fibrosis|cyst|lipoma|lymphoma))$', first_p, re.IGNORECASE) if m_end: return m_end.group(1).strip() return first_p # 3. Fallback to lblBedDiagnosis if available bed_span = soup.find('span', id=lambda x: x and 'lblBedDiagnosis' in x) if bed_span and bed_span.text.strip(): return bed_span.text.strip() return "" def classify_site(tissue, diagnosis, html=''): site_text = (str(tissue) + ' ' + str(diagnosis)).lower() site_text = re.sub(r'\bpd\-?l1\b', '', site_text, flags=re.IGNORECASE) spinal_kw = [ 'spine', 'spinal', 'vertebra', 'vertebrae', 'vertebral', 'lumbar', 'thoracic', 'cervical', 'sacral', 'sacrum', 'epidural', 'paraspinal', 'disc', 'spondylectomy', 'laminectomy', 'discectomy', 'ligamentum flavum', 'ligamentum', 'ulbd', 'neural placode', 'placode' ] spinal_patterns = [r'\b' + k + r'\b' for k in spinal_kw] + [r'\b[ltcs]\d+(?:[\-\/]\d+)?\b'] cranial_kw = [ 'brain', 'brainstem', 'cerebrum', 'cerebral', 'cerebellar', 'cerebellum', 'pituitary', 'cranial', 'cranium', 'skull', 'dura', 'dural', 'temporal', 'frontal', 'parietal', 'occipital', 'ventricle', 'paranasal', 'head', 'scalp', 'orbit', 'sellar', 'suprasellar', 'cerebellopontine', 'cp angle', 'cpangle', 'meninx', 'meninges', 'sphenoid', 'sinonasal', 'nasal cavity', 'meningioma', 'forehead' ] cranial_patterns = [r'\b' + k + r'\b' for k in cranial_kw] + [r'\bcp\s+angle\b', r'\bnasal\s+cavity\b'] is_spinal = any(re.search(p, site_text) for p in spinal_patterns) is_cranial = any(re.search(p, site_text) for p in cranial_patterns) if 'dura mater' in site_text or 'dural' in site_text and 'epidural' not in site_text: is_cranial = True if is_spinal and not is_cranial: return 'spinal' elif is_cranial and not is_spinal: return 'cranial' elif is_spinal and is_cranial: if any(k in site_text for k in ['brain', 'cerebrum', 'dura mater']): return 'cranial' return 'spinal' return 'other' def classify_type(tissue, diagnosis, html=''): diag_str = str(diagnosis).lower() text = (str(tissue) + ' ' + diag_str).lower() if any(k in diag_str for k in ['no evidence of', 'negative for', 'free of', 'no tumor', 'without tumor', 'no residual']): return 'other' if any(k in diag_str for k in ['necrotic tissue', 'necrosis', 'organizing hematoma', 'hematoma', 'thrombus', 'clot', 'abscess', 'infarction', 'fibrosis', 'sinusitis', 'cholecystitis', 'gastritis', 'varicose']): definitive_tumor = ['carcinoma', 'adenocarcinoma', 'adenoma', 'sarcoma', 'osteosarcoma', 'osteoma', 'astrocytoma', 'glioma', 'meningioma', 'glioblastoma', 'subependymoma', 'ependymoma', 'lipoma', 'fibroma', 'papilloma', 'lymphoma', 'melanoma', 'schwannoma', 'malignancy', 'malignant'] if not any(k in diag_str for k in definitive_tumor): return 'other' non_tumor_findings = ['granuloma', 'inflammatory cell infiltration', 'organizing hematoma', 'hematoma', 'thrombus', 'clot', 'abscess', 'necrosis', 'infarction', 'fibrosis', 'sinusitis', 'cholecystitis', 'gastritis', 'varicose', 'amyloid angiopathy'] neoplastic_findings = [ 'carcinoma', 'adenocarcinoma', 'adenoma', 'sarcoma', 'osteosarcoma', 'osteoma', 'chondroma', 'hemangioma', 'neuroma', 'leiomyoma', 'myxoma', 'astrocytoma', 'glioma', 'meningioma', 'glioblastoma', 'subependymoma', 'ependymoma', 'ganglioglioma', 'oligodendroglioma', 'medulloblastoma', 'chordoma', 'hemangioblastoma', 'craniopharyngioma', 'lipoma', 'fibroma', 'papilloma', 'polyp', 'hyperplasia', 'metastasis', 'metastatic', 'malignancy', 'malignant', 'schwannoma', 'neurofibroma', 'melanoma', 'lymphoma', 'plasmacytosis', 'neoplastic', 'who grade', 'cns who grade' ] if any(k in diag_str for k in non_tumor_findings) and not any(k in diag_str for k in neoplastic_findings): return 'other' tumor_kw = [ 'malignancy', 'malignant', 'osteoma', 'chondroma', 'hemangioma', 'neuroma', 'leiomyoma', 'myxoma', 'subependymoma', 'ependymoma', 'ganglioglioma', 'oligodendroglioma', 'medulloblastoma', 'chordoma', 'hemangioblastoma', 'craniopharyngioma', 'who grade', 'cns who grade', 'tumor', 'tumour', 'carcinoma', 'adenocarcinoma', 'adenoma', 'sarcoma', 'osteosarcoma', 'astrocytoma', 'glioma', 'meningioma', 'glioblastoma', 'lipoma', 'fibroma', 'papilloma', 'cyst', 'polyp', 'hyperplasia', 'mass', 'lesion', 'neoplasm', 'metastasis', 'metastatic', 'schwannoma', 'neurofibroma', 'melanoma', 'lymphoma', 'plasmacytosis', 'neoplastic' ] if any(k in text for k in tumor_kw): return 'tumor' return 'other' def process_extraction(input_excel=None, output_csv=None, limit=None, mode='all'): if input_excel is None: input_excel = os.path.join(script_dir, 'registry_pathologyreport.xlsx') if not os.path.exists(input_excel) and os.path.exists('registry_pathologyreport.xlsx'): input_excel = 'registry_pathologyreport.xlsx' if output_csv is None: output_csv = os.path.join(script_dir, 'registry_pathologyreport_extracted.csv') print(f"Opening Excel file: {input_excel} ...") wb = openpyxl.load_workbook(input_excel, read_only=True) ws = wb.active headers = None chart_idx = None path_idx = None html_idx = None all_raw_rows = [] print(f"Reading Excel rows...") for i, row in enumerate(ws.iter_rows()): vals = [cell.value for cell in row] if headers is None: headers = vals chart_idx = headers.index('ChartNo') path_idx = headers.index('PathCode') html_idx = headers.index('html') continue all_raw_rows.append(vals) if mode == 'last' and limit: raw_rows = all_raw_rows[-limit:] print(f"Processing last {len(raw_rows)} rows (out of {len(all_raw_rows)} total)...") else: raw_rows = all_raw_rows[:limit] if limit else all_raw_rows print(f"Processing {len(raw_rows)} rows...") # Build full PathCode map from all Excel rows for fallback resolution excel_pathcode_html_map = {} for vals in all_raw_rows: pcode = str(vals[path_idx]).strip() if vals[path_idx] else "" hval = vals[html_idx] or "" if pcode: excel_pathcode_html_map[pcode] = hval rows = [] pathcode_diag_map = {} for vals in raw_rows: chart_no = vals[chart_idx] path_code = str(vals[path_idx]).strip() if vals[path_idx] else "" html_content = vals[html_idx] or "" tissue = extract_tissue(html_content, path_code) diagnosis = extract_short_diagnosis(html_content, path_code) site = classify_site(tissue, diagnosis, html_content) ttype = classify_type(tissue, diagnosis, html_content) row_dict = { 'chart_no': chart_no, 'path_code': path_code, 'site': site, 'type': ttype, 'tissue': tissue, 'diagnosis': diagnosis } rows.append(row_dict) if path_code and diagnosis: pathcode_diag_map[path_code] = diagnosis print("Resolving empty diagnosis rows via 組織由來 (tissue) -> 病理號 (PathCode) matching...") fallback_count = 0 for r in rows: if not r['diagnosis'] and r['tissue']: tissue = r['tissue'].strip() potential_codes = [tissue] + re.findall(r'[A-Za-z0-9_\-]+', tissue) for code in potential_codes: if code not in pathcode_diag_map and code in excel_pathcode_html_map: ref_html = excel_pathcode_html_map[code] if ref_html: db_diag = extract_short_diagnosis(ref_html, code) if db_diag: pathcode_diag_map[code] = db_diag if code in pathcode_diag_map and pathcode_diag_map[code]: r['diagnosis'] = pathcode_diag_map[code] r['site'] = classify_site(r['tissue'], r['diagnosis']) r['type'] = classify_type(r['tissue'], r['diagnosis']) fallback_count += 1 break print(f"Fallback resolution complete: updated {fallback_count} empty rows.") print(f"Writing to CSV file: {output_csv} ...") with open(output_csv, 'w', newline='', encoding='utf-8-sig') as f: writer = csv.writer(f) writer.writerow(['病歷號', '病理號', 'site', 'type', '組織由來', 'pathological diagnosis']) for r in rows: writer.writerow([r['chart_no'], r['path_code'], r['site'], r['type'], r['tissue'], r['diagnosis']]) print(f"Successfully exported {len(rows)} rows to {output_csv}") if __name__ == '__main__': inp = sys.argv[1] if len(sys.argv) > 1 else None out = sys.argv[2] if len(sys.argv) > 2 else None lim = int(sys.argv[3]) if len(sys.argv) > 3 and sys.argv[3].isdigit() else None m = sys.argv[4] if len(sys.argv) > 4 else 'all' process_extraction(inp, out, limit=lim, mode=m)