Upload 2 files

file_api.py

@@ -0,0 +1,459 @@
+"""
+파일 처리 및 API 관련 함수
+"""
+import os
+import re
+import zlib
+import zipfile
+import tempfile
+import requests
+from pathlib import Path
+from typing import Optional, Tuple, List, Dict, Generator
+from xml.etree import ElementTree as ET
+from datetime import datetime
+
+from utils import (
+    API_URL, API_KEY, GROQ_API_KEY, CATEGORY_CODES,
+    OLEFILE_AVAILABLE, PYPDF2_AVAILABLE, PDFPLUMBER_AVAILABLE, GROQ_AVAILABLE,
+    extract_region_from_text, extract_region_from_hashtags, classify_org_type,
+    parse_deadline, is_ongoing
+)
+
+if OLEFILE_AVAILABLE:
+    import olefile
+if PYPDF2_AVAILABLE:
+    import PyPDF2
+if PDFPLUMBER_AVAILABLE:
+    import pdfplumber
+if GROQ_AVAILABLE:
+    from groq import Groq
+
+import pandas as pd
+from bs4 import BeautifulSoup
+
+
+# ============================================================
+# 파일 텍스트 추출 함수
+# ============================================================
+def extract_text_from_hwpx(file_path: str) -> Tuple[Optional[str], Optional[str]]:
+    """HWPX 파일에서 텍스트 추출"""
+    try:
+        text_parts = []
+        with zipfile.ZipFile(file_path, 'r') as zf:
+            file_list = zf.namelist()
+            section_files = sorted([f for f in file_list if f.startswith('Contents/section') and f.endswith('.xml')])
+            if not section_files:
+                section_files = sorted([f for f in file_list if 'section' in f.lower() and f.endswith('.xml')])
+            for section_file in section_files:
+                try:
+                    with zf.open(section_file) as sf:
+                        content = sf.read()
+                        content_str = content.decode('utf-8')
+                        content_str = re.sub(r'\sxmlns[^"]*"[^"]*"', '', content_str)
+                        content_str = re.sub(r'<[a-zA-Z]+:', '<', content_str)
+                        content_str = re.sub(r'</[a-zA-Z]+:', '</', content_str)
+                        try:
+                            root = ET.fromstring(content_str)
+                            texts = []
+                            for elem in root.iter():
+                                if elem.tag.endswith('t') or elem.tag == 't':
+                                    if elem.text:
+                                        texts.append(elem.text)
+                                elif elem.text and elem.text.strip():
+                                    if any(x in elem.tag.lower() for x in ['text', 'run', 'para', 'char']):
+                                        texts.append(elem.text.strip())
+                            if texts:
+                                text_parts.append(' '.join(texts))
+                        except ET.ParseError:
+                            text_matches = re.findall(r'>([^<]+)<', content.decode('utf-8', errors='ignore'))
+                            clean_texts = [t.strip() for t in text_matches if t.strip() and len(t.strip()) > 1]
+                            if clean_texts:
+                                text_parts.append(' '.join(clean_texts))
+                except:
+                    continue
+        if text_parts:
+            result = '\n\n'.join(text_parts)
+            result = re.sub(r'\s+', ' ', result)
+            result = re.sub(r'\n{3,}', '\n\n', result)
+            return result.strip(), None
+        return None, "HWPX에서 텍스트를 찾을 수 없습니다"
+    except zipfile.BadZipFile:
+        return None, "유효하지 않은 HWPX 파일"
+    except Exception as e:
+        return None, f"HWPX 처리 오류: {str(e)}"
+
+
+def extract_hwp_section_text(data: bytes) -> Optional[str]:
+    """HWP 섹션 데이터에서 텍스트 추출"""
+    texts = []
+    pos = 0
+    while pos < len(data) - 4:
+        try:
+            header = int.from_bytes(data[pos:pos+4], 'little')
+            tag_id = header & 0x3FF
+            size = (header >> 20) & 0xFFF
+            pos += 4
+            if size == 0xFFF:
+                if pos + 4 > len(data):
+                    break
+                size = int.from_bytes(data[pos:pos+4], 'little')
+                pos += 4
+            if pos + size > len(data):
+                break
+            record_data = data[pos:pos+size]
+            pos += size
+            if tag_id == 67 and size > 0:
+                text = decode_para_text(record_data)
+                if text:
+                    texts.append(text)
+        except:
+            pos += 1
+            continue
+    return '\n'.join(texts) if texts else None
+
+
+def decode_para_text(data: bytes) -> Optional[str]:
+    """HWP 문단 텍스트 디코딩"""
+    result = []
+    i = 0
+    while i < len(data) - 1:
+        code = int.from_bytes(data[i:i+2], 'little')
+        if code == 0:
+            pass
+        elif code in [1, 2, 3]:
+            i += 14
+        elif code == 9:
+            result.append('\t')
+        elif code in [10, 13]:
+            result.append('\n')
+        elif code == 24:
+            result.append('-')
+        elif code in [30, 31]:
+            result.append(' ')
+        elif code >= 32:
+            try:
+                char = chr(code)
+                if char.isprintable() or char in '\n\t ':
+                    result.append(char)
+            except:
+                pass
+        i += 2
+    text = ''.join(result).strip()
+    text = re.sub(r'[ \t]+', ' ', text)
+    text = re.sub(r'\n{3,}', '\n\n', text)
+    return text if len(text) > 2 else None
+
+
+def extract_text_from_hwp(file_path: str) -> Tuple[Optional[str], Optional[str]]:
+    """HWP 파일에서 텍스트 추출"""
+    if not OLEFILE_AVAILABLE:
+        return None, "olefile 모듈 없음"
+    try:
+        ole = olefile.OleFileIO(file_path)
+        if not ole.exists('FileHeader'):
+            ole.close()
+            return None, "HWP 파일 헤더 없음"
+        header_data = ole.openstream('FileHeader').read()
+        is_compressed = (header_data[36] & 1) == 1 if len(header_data) > 36 else True
+        all_texts = []
+        for entry in ole.listdir():
+            entry_path = '/'.join(entry)
+            if entry_path.startswith('BodyText/Section'):
+                try:
+                    stream_data = ole.openstream(entry).read()
+                    if is_compressed:
+                        try:
+                            stream_data = zlib.decompress(stream_data, -15)
+                        except:
+                            try:
+                                stream_data = zlib.decompress(stream_data)
+                            except:
+                                pass
+                    section_text = extract_hwp_section_text(stream_data)
+                    if section_text:
+                        all_texts.append(section_text)
+                except:
+                    continue
+        ole.close()
+        if all_texts:
+            return '\n\n'.join(all_texts).strip(), None
+        return None, "텍스트를 찾을 수 없습니다"
+    except Exception as e:
+        return None, f"olefile 오류: {str(e)}"
+
+
+def extract_text_from_pdf(file_path: str) -> Optional[str]:
+    """PDF 파일에서 텍스트 추출"""
+    text_parts = []
+    if PDFPLUMBER_AVAILABLE:
+        try:
+            with pdfplumber.open(file_path) as pdf:
+                for page in pdf.pages:
+                    text = page.extract_text()
+                    if text:
+                        text_parts.append(text)
+            if text_parts:
+                return "\n\n".join(text_parts)
+        except Exception as e:
+            print(f"pdfplumber error: {e}")
+    if PYPDF2_AVAILABLE:
+        try:
+            with open(file_path, 'rb') as f:
+                reader = PyPDF2.PdfReader(f)
+                for page in reader.pages:
+                    text = page.extract_text()
+                    if text:
+                        text_parts.append(text)
+            if text_parts:
+                return "\n\n".join(text_parts)
+        except Exception as e:
+            print(f"PyPDF2 error: {e}")
+    return None
+
+
+def extract_text_from_file(file_path: str) -> Tuple[Optional[str], Optional[str]]:
+    """파일 유형에 따라 텍스트 추출"""
+    ext = Path(file_path).suffix.lower()
+    if ext == '.hwp':
+        return extract_text_from_hwp(file_path)
+    elif ext == '.hwpx':
+        return extract_text_from_hwpx(file_path)
+    elif ext == '.pdf':
+        text = extract_text_from_pdf(file_path)
+        if text:
+            return text, None
+        return None, "PDF 텍스트 추출 실패"
+    elif ext in ['.txt', '.md']:
+        try:
+            with open(file_path, 'r', encoding='utf-8') as f:
+                return f.read(), None
+        except:
+            try:
+                with open(file_path, 'r', encoding='cp949') as f:
+                    return f.read(), None
+            except Exception as e:
+                return None, f"텍스트 파일 읽기 실패: {e}"
+    else:
+        return None, f"지원하지 않는 파일 형식: {ext}"
+
+
+def extract_zip_files(zip_path: str, extract_dir: str) -> List[str]:
+    """ZIP 파일 압축 해제"""
+    extracted_files = []
+    try:
+        with zipfile.ZipFile(zip_path, 'r') as zf:
+            for name in zf.namelist():
+                ext = Path(name).suffix.lower()
+                if ext in ['.hwp', '.hwpx', '.pdf', '.txt', '.doc', '.docx']:
+                    try:
+                        try:
+                            decoded_name = name.encode('cp437').decode('cp949')
+                        except:
+                            decoded_name = name
+                        safe_name = os.path.basename(decoded_name)
+                        extract_path = os.path.join(extract_dir, safe_name)
+                        with zf.open(name) as src, open(extract_path, 'wb') as dst:
+                            dst.write(src.read())
+                        extracted_files.append(extract_path)
+                    except Exception as e:
+                        print(f"ZIP 파일 추출 오류 ({name}): {e}")
+    except Exception as e:
+        print(f"ZIP 처리 오류: {e}")
+    return extracted_files
+
+
+# ============================================================
+# API 관련 함수
+# ============================================================
+def fetch_all_from_api(category: str = "전체", region: str = "전체(지역)", keyword: str = "") -> Tuple[List[Dict], str]:
+    """API에서 전체 데이터를 페이지네이션으로 수집"""
+    if not API_KEY:
+        return [], "❌ API 키가 설정되지 않았습니다. (BIZ_API 환경변수)"
+    all_items = []
+    page_size = 100
+    max_pages = 10
+    headers = {"User-Agent": "Mozilla/5.0", "Accept": "application/json"}
+    hashtags = []
+    if category and category != "전체":
+        hashtags.append(category)
+    if region and region != "전체(지역)":
+        hashtags.append(region)
+    if keyword and keyword.strip():
+        hashtags.append(keyword.strip())
+    
+    for page_idx in range(1, max_pages + 1):
+        try:
+            params = {"crtfcKey": API_KEY, "dataType": "json", "pageUnit": page_size, "pageIndex": page_idx}
+            if category and category != "전체" and category in CATEGORY_CODES:
+                if CATEGORY_CODES[category]:
+                    params["searchLclasId"] = CATEGORY_CODES[category]
+            if hashtags:
+                params["hashtags"] = ",".join(hashtags)
+            response = requests.get(API_URL, params=params, headers=headers, timeout=(15, 60), verify=True)
+            response.raise_for_status()
+            result = response.json()
+            items = []
+            json_array = result.get("jsonArray", result)
+            if isinstance(json_array, dict):
+                items = json_array.get("item", [])
+                if isinstance(items, dict):
+                    items = [items]
+            elif isinstance(json_array, list):
+                items = json_array
+            if not items:
+                break
+            all_items.extend(items)
+            if page_idx == 1 and items:
+                total_cnt = items[0].get("totCnt", 0)
+                try:
+                    total_cnt = int(total_cnt)
+                except:
+                    total_cnt = len(items)
+                needed_pages = (total_cnt + page_size - 1) // page_size
+                max_pages = min(max_pages, needed_pages)
+            if len(items) < page_size:
+                break
+        except requests.exceptions.Timeout:
+            return all_items, f"⏱️ 페이지 {page_idx} 요청 시간 초과"
+        except requests.exceptions.RequestException as e:
+            if all_items:
+                break
+            return [], f"❌ API 요청 오류: {str(e)[:50]}"
+        except Exception as e:
+            if all_items:
+                break
+            return [], f"❌ 오류: {str(e)[:50]}"
+    return all_items, ""
+
+
+def download_file(url: str, save_dir: str, hint_filename: str = None) -> Tuple[Optional[str], Optional[str]]:
+    """파일 다운로드"""
+    try:
+        headers = {"User-Agent": "Mozilla/5.0", "Accept": "*/*", "Referer": "https://www.bizinfo.go.kr/"}
+        response = requests.get(url, headers=headers, timeout=60, stream=True, verify=False, allow_redirects=True)
+        response.raise_for_status()
+        cd = response.headers.get('Content-Disposition', '')
+        filename = None
+        if cd:
+            match = re.search(r"filename\*=(?:UTF-8''|utf-8'')(.+)", cd, re.IGNORECASE)
+            if match:
+                from urllib.parse import unquote
+                filename = unquote(match.group(1))
+            else:
+                match = re.search(r'filename=(["\']?)(.+?)\1(?:;|$)', cd)
+                if match:
+                    filename = match.group(2)
+        if not filename and hint_filename:
+            filename = hint_filename
+        if not filename:
+            from urllib.parse import urlparse
+            parsed = urlparse(url)
+            filename = parsed.path.split('/')[-1]
+            if not filename or '.' not in filename:
+                content_type = response.headers.get('Content-Type', '').lower()
+                if 'pdf' in content_type:
+                    filename = f"document_{hash(url) % 10000}.pdf"
+                elif 'hwp' in content_type:
+                    filename = f"document_{hash(url) % 10000}.hwp"
+                else:
+                    filename = f"file_{hash(url) % 10000}.bin"
+        filename = re.sub(r'[<>:"/\\|?*]', '_', filename)
+        file_path = os.path.join(save_dir, filename)
+        with open(file_path, 'wb') as f:
+            for chunk in response.iter_content(chunk_size=8192):
+                if chunk:
+                    f.write(chunk)
+        if os.path.getsize(file_path) == 0:
+            os.remove(file_path)
+            return None, "빈 파일이 다운로드됨"
+        return file_path, None
+    except Exception as e:
+        return None, f"다운로드 실패: {str(e)}"
+
+
+def call_groq_api_stream(messages: List[Dict]) -> Generator[str, None, None]:
+    """Groq API 스트리밍 호출"""
+    if not GROQ_AVAILABLE:
+        yield "❌ Groq 라이브러리가 설치되지 않았습니다."
+        return
+    if not GROQ_API_KEY:
+        yield "❌ GROQ_API_KEY 환경변수가 설정되지 않았습니다."
+        return
+    try:
+        client = Groq(api_key=GROQ_API_KEY)
+        completion = client.chat.completions.create(
+            model="llama-3.3-70b-versatile",
+            messages=messages,
+            temperature=0.3,
+            max_tokens=4096,
+            stream=True,
+        )
+        for chunk in completion:
+            if chunk.choices[0].delta.content:
+                yield chunk.choices[0].delta.content
+    except Exception as e:
+        yield f"❌ API 오류: {str(e)}"
+
+
+def fetch_announcement_detail(url: str) -> Tuple[str, List[Dict]]:
+    """공고 상세 페이지에서 본문과 첨부파일 정보 추출"""
+    try:
+        headers = {
+            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+            "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
+            "Accept-Language": "ko-KR,ko;q=0.9,en-US;q=0.8,en;q=0.7",
+        }
+        response = requests.get(url, headers=headers, timeout=30, verify=False)
+        response.raise_for_status()
+        html_text = response.text
+        soup = BeautifulSoup(html_text, 'html.parser')
+        content_text = ""
+        tables = soup.find_all('table')
+        for table in tables:
+            text = table.get_text(separator='\n', strip=True)
+            if '사업개요' in text or '지원대상' in text or '신청기간' in text:
+                content_text += text + "\n\n"
+        main_content = soup.find('div', {'id': 'container'}) or soup.find('main') or soup.find('article')
+        if main_content and not content_text:
+            content_text = main_content.get_text(separator='\n', strip=True)
+        attachments = []
+        for a_tag in soup.find_all('a', href=True):
+            href = a_tag.get('href', '')
+            href_clean = re.sub(r';jsessionid=[^?]*', '', href)
+            if 'getImageFile.do' in href_clean or 'fileDown' in href_clean or 'atchFileId' in href_clean:
+                filename = a_tag.get_text(strip=True)
+                if filename in ['다운로드', '바로보기', '내려받기', '']:
+                    parent = a_tag.parent
+                    if parent:
+                        parent_text = parent.get_text(separator='|', strip=True)
+                        parts = [p.strip() for p in parent_text.split('|') if p.strip()]
+                        for part in parts:
+                            if part not in ['다운로드', '바로보기', '내려받기'] and ('.' in part):
+                                filename = part
+                                break
+                    title = a_tag.get('title', '')
+                    if title and '첨부파일' in title:
+                        match = re.search(r'첨부파일\s+(.+?)\s+다운로드', title)
+                        if match:
+                            filename = match.group(1)
+                if not filename or filename in ['다운로드', '바로보기', '내려받기']:
+                    filename = f"첨부파일_{len(attachments)+1}"
+                if href_clean.startswith('/'):
+                    full_url = f"https://www.bizinfo.go.kr{href_clean}"
+                elif href_clean.startswith('http'):
+                    full_url = href_clean
+                else:
+                    continue
+                ext = Path(filename).suffix.lower()
+                if not ext:
+                    ext = '.unknown'
+                if not any(att['url'] == full_url for att in attachments):
+                    attachments.append({
+                        "filename": filename,
+                        "url": full_url,
+                        "type": ext[1:] if ext.startswith('.') else ext
+                    })
+        return content_text, attachments
+    except Exception as e:
+        import traceback
+        return f"상세 정보 조회 실패: {str(e)}\n{traceback.format_exc()}", []

utils.py

@@ -0,0 +1,288 @@
+import gradio as gr
+import requests
+import pandas as pd
+from typing import Optional, Tuple, List, Dict, Generator
+import os
+import tempfile
+import zipfile
+import re
+import json
+import zlib
+from pathlib import Path
+from bs4 import BeautifulSoup
+import urllib3
+from datetime import datetime, timedelta
+urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+
+try:
+    import olefile
+    OLEFILE_AVAILABLE = True
+except ImportError:
+    OLEFILE_AVAILABLE = False
+
+try:
+    import PyPDF2
+    PYPDF2_AVAILABLE = True
+except ImportError:
+    PYPDF2_AVAILABLE = False
+
+try:
+    import pdfplumber
+    PDFPLUMBER_AVAILABLE = True
+except ImportError:
+    PDFPLUMBER_AVAILABLE = False
+
+try:
+    from groq import Groq
+    GROQ_AVAILABLE = True
+except ImportError:
+    GROQ_AVAILABLE = False
+
+from xml.etree import ElementTree as ET
+
+# ============================================================
+# 환경변수 및 API 설정
+# ============================================================
+API_URL = "https://www.bizinfo.go.kr/uss/rss/bizinfoApi.do"
+API_KEY = os.getenv("BIZ_API", "")
+GROQ_API_KEY = os.getenv("GROQ_API_KEY", "")
+
+# ============================================================
+# 상수 정의
+# ============================================================
+CATEGORY_CODES = {
+    "전체": "", "금융": "01", "기술": "02", "인력": "03", "수출": "04",
+    "내수": "05", "창업": "06", "경영": "07", "기타": "09"
+}
+
+REGION_LIST = [
+    "전체(지역)", "서울", "부산", "대구", "인천", "광주", "대전", "울산", "세종",
+    "경기", "강원", "충북", "충남", "전북", "전남", "경북", "경남", "제주"
+]
+
+SIDO_LIST = [
+    "서울특별시", "부산광역시", "대구광역시", "인천광역시", "광주광역시",
+    "대전광역시", "울산광역시", "세종특별자치시", "경기도", "강원특별자치도",
+    "충청북도", "충청남도", "전북특별자치도", "전라남도", "경상북도", "경상남도", "제주특별자치도"
+]
+
+ORG_TYPE_OPTIONS = ["전체", "중앙부처", "지자체"]
+SORT_OPTIONS = ["등록일순", "마감일순"]
+STATUS_OPTIONS = ["진행중", "전체(지난공고포함)"]
+
+COMPANY_TYPE_OPTIONS = ["개인사업자", "법인사업자", "비영리법인", "조합", "협동조합", "사회적기업"]
+CORP_TYPE_OPTIONS = ["해당없음(개인)", "주식회사", "유한회사", "유한책임회사", "합자회사", "합명회사"]
+COMPANY_SIZE_OPTIONS = ["소기업", "중기업", "중견기업", "대기업"]
+
+INDUSTRY_MAJOR_OPTIONS = [
+    "A. 농업, 임업 및 어업", "B. 광업", "C. 제조업",
+    "D. 전기, 가스, 증기 및 공기조절 공급업", "E. 수도, 하수 및 폐기물 처리, 원료 재생업",
+    "F. 건설업", "G. 도매 및 소매업", "H. 운수 및 창고업", "I. 숙박 및 음식점업",
+    "J. 정보통신업", "K. 금융 및 보험업", "L. 부동산업", "M. 전문, 과학 및 기술 서비스업",
+    "N. 사업시설 관리, 사업 지원 및 임대 서비스업", "P. 교육 서비스업",
+    "Q. 보건업 및 사회복지 서비스업", "R. 예술, 스포츠 및 여가관련 서비스업"
+]
+
+CORE_INDUSTRY_OPTIONS = [
+    "인공지능(AI)", "반도체", "바이오/헬스케어", "미래자동차(전기차/자율주행)",
+    "로봇", "이차전지", "디스플레이", "5G/6G 통신", "양자기술", "우주항공"
+]
+
+NATIONAL_STRATEGIC_TECH = [
+    "반도체·디스플레이", "이차전지", "첨단모빌리티", "차세대 원자력", "첨단바이오",
+    "우주항공·해양", "수소", "사이버보안", "인공지능", "차세대 통신", "첨단로봇·제조", "양자"
+]
+
+CREDIT_GRADE_OPTIONS = ["AAA", "AA+", "AA", "AA-", "A+", "A", "A-", "BBB+", "BBB", "BBB-",
+                        "BB+", "BB", "BB-", "B+", "B", "B-", "CCC", "CC", "C", "D", "미평가"]
+TCB_GRADE_OPTIONS = ["T1", "T2", "T3", "T4", "T5", "T6", "T7", "T8", "T9", "T10", "미평가"]
+
+ISO_CERT_OPTIONS = [
+    "ISO 9001 (품질경영)", "ISO 14001 (환경경영)", "ISO 27001 (정보보안)",
+    "ISO 45001 (안전보건)", "ISO 22000 (식품안전)", "ISO 13485 (의료기기)", "IATF 16949 (자동차)"
+]
+
+CENTRAL_GOV_KEYWORDS = [
+    "부", "처", "청", "원", "위원회", "중소벤처기업부", "산업통상", "과학기술", "고용노동",
+    "농림축산", "해양수산", "환경부", "국토교통", "문화체육", "보건복지", "행정안전",
+    "기획재정", "외교부", "법무부", "국방부", "교육부", "통일부", "여성가족",
+    "특허청", "관세청", "조달청", "통계청", "기상청", "소방청", "경찰청",
+    "산림청", "농촌진흥청", "식품의약품안전처", "방위사업청", "병무청", "국세청",
+    "대한무역투자진흥공사", "한국산업기술진흥원", "한국에너지공단", "한국인터넷진흥원",
+    "정보통신산업진흥원", "한국콘텐츠진흥원", "한국관광공사", "대한상공회의소",
+    "중소기업중앙회", "기��보증기금", "신용보증기금", "한국산업안전보건공단"
+]
+
+LOCAL_GOV_KEYWORDS = [
+    "도", "시", "군", "구청", "특별시", "광역시", "특별자치", "자치도", "자치시",
+    "서울", "부산", "대구", "인천", "광주", "대전", "울산", "세종",
+    "경기", "강원", "충북", "충남", "전북", "전남", "경북", "경남", "제주",
+    "테크노파크", "TP", "경제진흥원", "창조경제", "지역혁신", "소상공인진흥"
+]
+
+REGION_MATCH_KEYWORDS = {
+    "서울": ["서울", "서울특별시", "서울시", "SBA", "서울산업진흥원", "서울경제진흥원",
+             "서울신용보증재단", "서울테크노파크", "서울창업허브", "SH공사"],
+    "부산": ["부산", "부산광역시", "부산시", "부산경제진흥원", "부산테크노파크"],
+    "대구": ["대구", "대구광역시", "대구시", "대구경북", "대구테크노파크"],
+    "인천": ["인천", "인천광역시", "인천시", "인천경제자유구역", "인천테크노파크"],
+    "광주": ["광주", "광주광역시", "광주시", "광주전남", "광주테크노파크"],
+    "대전": ["대전", "대전광역시", "대전시", "대전테크노파크", "대전경제통상진흥원"],
+    "울산": ["울산", "울산광역시", "울산시", "울산테크노파크", "울산경제진흥원"],
+    "세종": ["세종", "세종특별자치시", "세종시", "세종테크노파크"],
+    "경기": ["경기", "경기도", "경기TP", "경기테크노파크", "경기경제과학진흥원",
+             "수원", "성남", "고양", "용인", "부천", "안산", "안양",
+             "남양주", "화성", "평택", "의정부", "시흥", "파주", "김포", "광명", "광주시",
+             "군포", "하남", "오산", "이천", "안성", "의왕", "양평", "여주", "과천",
+             "구리", "포천", "동두천", "가평", "연천"],
+    "강원": ["강원", "강원도", "강원특별자치도", "강원TP", "강원테크노파크", "강원경제진흥원",
+             "춘천", "원주", "강릉", "동해", "태백",
+             "속초", "삼척", "홍천", "횡성", "영월", "평창", "정선", "철원", "화천",
+             "양구", "인제", "고성", "양양"],
+    "충북": ["충북", "충청북도", "충북TP", "충북테크노파크", "충청북도기업진흥원",
+             "청주", "충주", "제천", "보은", "옥천", "영동", "증평",
+             "진천", "괴산", "음성", "단양"],
+    "충남": ["충남", "충청남도", "충남TP", "충남테크노파크", "충남경제진흥원",
+             "천안", "공주", "보령", "아산", "서산", "논산", "계룡",
+             "당진", "금산", "부여", "서천", "청양", "홍성", "예산", "태안"],
+    "전북": ["전북", "전라북도", "전북특별자치도", "전북TP", "전북테크노파크", "전북경제통상진흥원",
+             "전주", "군산", "익산", "정읍", "남원",
+             "김제", "완주", "진안", "무주", "장수", "임실", "순창", "고창", "부안"],
+    "전남": ["전남", "전라남도", "전남TP", "전남테크노파크", "전남경제통상진흥원",
+             "목포", "여수", "순천", "나주", "광양", "담양", "곡성",
+             "구례", "고흥", "보성", "화순", "장흥", "강진", "해남", "영암", "무안",
+             "함평", "영광", "장성", "완도", "진도", "신안"],
+    "경북": ["경북", "경상북도", "경북TP", "경북테크노파크", "경북경제진흥원",
+             "포항", "경주", "김천", "안동", "구미", "영주", "영천",
+             "상주", "문경", "경산", "군위", "의성", "청송", "영양", "영덕", "청도",
+             "고령", "성주", "칠곡", "예천", "봉화", "울진", "울릉"],
+    "경남": ["경남", "경상남도", "경남TP", "경남테크노파크", "경남경제진흥원",
+             "창원", "진주", "통영", "사천", "김해", "밀양", "거제",
+             "양산", "의령", "함안", "창녕", "고성", "남해", "하동", "산청", "함양",
+             "거창", "합천"],
+    "제주": ["제주", "제주도", "제주특별자치도", "제주시", "서귀포", "제주TP", "제주테크노파크"]
+}
+
+
+# ============================================================
+# 유틸리티 함수
+# ============================================================
+def calculate_age(birth_date: str) -> int:
+    """생년월일로 만나이 계산"""
+    if not birth_date:
+        return 0
+    try:
+        birth = datetime.strptime(birth_date, "%Y-%m-%d")
+        today = datetime.now()
+        age = today.year - birth.year
+        if (today.month, today.day) < (birth.month, birth.day):
+            age -= 1
+        return age
+    except:
+        return 0
+
+
+def calculate_company_age(establish_date: str) -> Tuple[int, int]:
+    """설립일로 업력 계산 (년, 월)"""
+    if not establish_date:
+        return 0, 0
+    try:
+        establish = datetime.strptime(establish_date, "%Y-%m-%d")
+        today = datetime.now()
+        years = today.year - establish.year
+        months = today.month - establish.month
+        if months < 0:
+            years -= 1
+            months += 12
+        return years, months
+    except:
+        return 0, 0
+
+
+def extract_region_from_text(text: str) -> Optional[str]:
+    """텍스트에서 지역 정보 추출"""
+    if not text:
+        return None
+    text = text.strip()
+    bracket_match = re.match(r'\[([가-힣]+)\]', text)
+    if bracket_match:
+        region_text = bracket_match.group(1)
+        for region, keywords in REGION_MATCH_KEYWORDS.items():
+            for kw in keywords:
+                if kw in region_text:
+                    return region
+    for region in ["서울", "부산", "대구", "인천", "광주", "대전", "울산", "세종"]:
+        keywords = REGION_MATCH_KEYWORDS.get(region, [])
+        for kw in keywords:
+            if kw in text:
+                if region == "광주" and ("경기" in text or "경기도" in text):
+                    continue
+                return region
+    for region in ["경기", "강원", "충북", "충남", "전북", "전남", "경북", "경남", "제주"]:
+        keywords = REGION_MATCH_KEYWORDS.get(region, [])
+        for kw in keywords:
+            if kw in text:
+                return region
+    return None
+
+
+def extract_region_from_hashtags(hash_tags: str) -> Optional[str]:
+    """해시태그에서 지역 정보 추출"""
+    if not hash_tags:
+        return None
+    tags = [t.strip() for t in hash_tags.split(",")]
+    region_names = list(REGION_MATCH_KEYWORDS.keys())
+    for tag in tags:
+        if tag in region_names:
+            return tag
+        for region in region_names:
+            if region in tag:
+                return region
+    return None
+
+
+def classify_org_type(author: str) -> str:
+    """소관기관명을 기반으로 기관유형 분류"""
+    if not author:
+        return "기타"
+    author = author.strip()
+    if author.startswith("[") and "]" in author:
+        return "지자체"
+    for keyword in LOCAL_GOV_KEYWORDS:
+        if keyword in author:
+            is_central = any(central_kw in author and len(central_kw) > 2 for central_kw in CENTRAL_GOV_KEYWORDS)
+            if not is_central:
+                return "지자체"
+    for keyword in CENTRAL_GOV_KEYWORDS:
+        if keyword in author:
+            return "중앙부처"
+    return "중앙부처"
+
+
+def parse_deadline(req_dt: str) -> Optional[datetime]:
+    """신청기간에서 마감일 추출"""
+    if not req_dt:
+        return None
+    try:
+        if "~" in req_dt:
+            end_date_str = req_dt.split("~")[-1].strip()
+        else:
+            end_date_str = req_dt.strip()
+        end_date_str = re.sub(r'[^0-9]', '', end_date_str)
+        if len(end_date_str) >= 8:
+            return datetime.strptime(end_date_str[:8], "%Y%m%d")
+    except:
+        pass
+    return None
+
+
+def is_ongoing(req_dt: str) -> bool:
+    """진행중 공고인지 확인"""
+    if not req_dt:
+        return True
+    if any(kw in req_dt for kw in ["소진", "추후", "상시", "별도"]):
+        return True
+    deadline = parse_deadline(req_dt)
+    if deadline:
+        return deadline >= datetime.now()
+    return True