feat: 전체 PKM 스크립트 일괄 작성 — 분류/법령/메일/다이제스트/임베딩
- scripts/pkm_utils.py: 공통 유틸 (로거, dotenv, osascript 래퍼) - scripts/prompts/classify_document.txt: Ollama 분류 프롬프트 - applescript/auto_classify.scpt: Inbox → AI 분류 → DB 이동 - applescript/omnifocus_sync.scpt: Projects → OmniFocus 작업 생성 - scripts/law_monitor.py: 법령 변경 모니터링 + DEVONthink 임포트 - scripts/mailplus_archive.py: MailPlus IMAP → Archive DB - scripts/pkm_daily_digest.py: 일일 다이제스트 + OmniFocus 액션 - scripts/embed_to_chroma.py: GPU 서버 벡터 임베딩 → ChromaDB - launchd/*.plist: 3개 스케줄 (07:00, 07:00+18:00, 20:00) - docs/deploy.md: Mac mini 배포 가이드 - docs/devonagent-setup.md: 검색 세트 9종 설정 가이드 - tests/test_classify.py: 5종 문서 분류 테스트 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
Hyungi Ahn
184
scripts/law_monitor.py
Normal file
184
scripts/law_monitor.py
Normal file
@@ -0,0 +1,184 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
법령 모니터링 스크립트
|
||||
- 국가법령정보센터 OpenAPI (open.law.go.kr) 폴링
|
||||
- 산업안전보건법, 중대재해처벌법 등 변경 추적
|
||||
- 변경 감지 시 DEVONthink 04_Industrial Safety 자동 임포트
|
||||
※ API 승인 대기중 — 스크립트만 작성, 실제 호출은 승인 후
|
||||
"""
|
||||
|
||||
import os
|
||||
import sys
|
||||
import json
|
||||
import requests
|
||||
import xml.etree.ElementTree as ET
|
||||
from datetime import datetime, timedelta
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).parent))
|
||||
from pkm_utils import setup_logger, load_credentials, run_applescript_inline, PROJECT_ROOT, DATA_DIR
|
||||
|
||||
logger = setup_logger("law_monitor")
|
||||
|
||||
# 모니터링 대상 법령
|
||||
MONITORED_LAWS = [
|
||||
{"name": "산업안전보건법", "law_id": "001789", "category": "법률"},
|
||||
{"name": "산업안전보건법 시행령", "law_id": "001790", "category": "대통령령"},
|
||||
{"name": "산업안전보건법 시행규칙", "law_id": "001791", "category": "부령"},
|
||||
{"name": "중대재해 처벌 등에 관한 법률", "law_id": "019005", "category": "법률"},
|
||||
{"name": "중대재해 처벌 등에 관한 법률 시행령", "law_id": "019006", "category": "대통령령"},
|
||||
{"name": "화학물질관리법", "law_id": "012354", "category": "법률"},
|
||||
{"name": "위험물안전관리법", "law_id": "001478", "category": "법률"},
|
||||
]
|
||||
|
||||
# 마지막 확인 일자 저장 파일
|
||||
LAST_CHECK_FILE = DATA_DIR / "law_last_check.json"
|
||||
LAWS_DIR = DATA_DIR / "laws"
|
||||
LAWS_DIR.mkdir(exist_ok=True)
|
||||
|
||||
|
||||
def load_last_check() -> dict:
|
||||
"""마지막 확인 일자 로딩"""
|
||||
if LAST_CHECK_FILE.exists():
|
||||
with open(LAST_CHECK_FILE, "r") as f:
|
||||
return json.load(f)
|
||||
return {}
|
||||
|
||||
|
||||
def save_last_check(data: dict):
|
||||
"""마지막 확인 일자 저장"""
|
||||
with open(LAST_CHECK_FILE, "w") as f:
|
||||
json.dump(data, f, ensure_ascii=False, indent=2)
|
||||
|
||||
|
||||
def fetch_law_info(law_oc: str, law_id: str) -> dict | None:
|
||||
"""법령 정보 조회 (법령 API)"""
|
||||
url = "https://www.law.go.kr/DRF/lawSearch.do"
|
||||
params = {
|
||||
"OC": law_oc,
|
||||
"target": "law",
|
||||
"type": "JSON",
|
||||
"MST": law_id,
|
||||
}
|
||||
try:
|
||||
resp = requests.get(url, params=params, timeout=30)
|
||||
resp.raise_for_status()
|
||||
data = resp.json()
|
||||
if "LawSearch" in data and "law" in data["LawSearch"]:
|
||||
laws = data["LawSearch"]["law"]
|
||||
if isinstance(laws, list):
|
||||
return laws[0] if laws else None
|
||||
return laws
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"법령 조회 실패 [{law_id}]: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def fetch_law_text(law_oc: str, law_mst: str) -> str | None:
|
||||
"""법령 본문 XML 다운로드"""
|
||||
url = "https://www.law.go.kr/DRF/lawService.do"
|
||||
params = {
|
||||
"OC": law_oc,
|
||||
"target": "law",
|
||||
"type": "XML",
|
||||
"MST": law_mst,
|
||||
}
|
||||
try:
|
||||
resp = requests.get(url, params=params, timeout=60)
|
||||
resp.raise_for_status()
|
||||
return resp.text
|
||||
except Exception as e:
|
||||
logger.error(f"법령 본문 다운로드 실패 [{law_mst}]: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def save_law_file(law_name: str, content: str) -> Path:
|
||||
"""법령 XML 저장"""
|
||||
today = datetime.now().strftime("%Y%m%d")
|
||||
safe_name = law_name.replace(" ", "_").replace("/", "_")
|
||||
filepath = LAWS_DIR / f"{safe_name}_{today}.xml"
|
||||
with open(filepath, "w", encoding="utf-8") as f:
|
||||
f.write(content)
|
||||
logger.info(f"법령 저장: {filepath}")
|
||||
return filepath
|
||||
|
||||
|
||||
def import_to_devonthink(filepath: Path, law_name: str, category: str):
|
||||
"""DEVONthink 04_Industrial Safety로 임포트"""
|
||||
script = f'''
|
||||
tell application id "DNtp"
|
||||
set targetDB to missing value
|
||||
repeat with db in databases
|
||||
if name of db is "04_Industrial safety" then
|
||||
set targetDB to db
|
||||
exit repeat
|
||||
end if
|
||||
end repeat
|
||||
|
||||
if targetDB is not missing value then
|
||||
set targetGroup to create location "/10_Legislation/Law" in targetDB
|
||||
set theRecord to import POSIX path "{filepath}" to targetGroup
|
||||
set tags of theRecord to {{"#주제/산업안전/법령", "$유형/법령", "{category}"}}
|
||||
add custom meta data "law_monitor" for "sourceChannel" to theRecord
|
||||
add custom meta data "external" for "dataOrigin" to theRecord
|
||||
add custom meta data (current date) for "lastAIProcess" to theRecord
|
||||
end if
|
||||
end tell
|
||||
'''
|
||||
try:
|
||||
run_applescript_inline(script)
|
||||
logger.info(f"DEVONthink 임포트 완료: {law_name}")
|
||||
except Exception as e:
|
||||
logger.error(f"DEVONthink 임포트 실패 [{law_name}]: {e}")
|
||||
|
||||
|
||||
def run():
|
||||
"""메인 실행"""
|
||||
logger.info("=== 법령 모니터링 시작 ===")
|
||||
|
||||
creds = load_credentials()
|
||||
law_oc = creds.get("LAW_OC")
|
||||
if not law_oc:
|
||||
logger.error("LAW_OC 인증키가 설정되지 않았습니다. credentials.env를 확인하세요.")
|
||||
sys.exit(1)
|
||||
|
||||
last_check = load_last_check()
|
||||
changes_found = 0
|
||||
|
||||
for law in MONITORED_LAWS:
|
||||
law_name = law["name"]
|
||||
law_id = law["law_id"]
|
||||
category = law["category"]
|
||||
|
||||
logger.info(f"확인 중: {law_name} ({law_id})")
|
||||
|
||||
info = fetch_law_info(law_oc, law_id)
|
||||
if not info:
|
||||
continue
|
||||
|
||||
# 시행일자 또는 공포일자로 변경 감지
|
||||
announce_date = info.get("공포일자", info.get("시행일자", ""))
|
||||
prev_date = last_check.get(law_id, "")
|
||||
|
||||
if announce_date and announce_date != prev_date:
|
||||
logger.info(f"변경 감지: {law_name} — 공포일자 {announce_date} (이전: {prev_date or '없음'})")
|
||||
|
||||
# 법령 본문 다운로드
|
||||
law_mst = info.get("법령MST", law_id)
|
||||
text = fetch_law_text(law_oc, law_mst)
|
||||
if text:
|
||||
filepath = save_law_file(law_name, text)
|
||||
import_to_devonthink(filepath, law_name, category)
|
||||
changes_found += 1
|
||||
|
||||
last_check[law_id] = announce_date
|
||||
else:
|
||||
logger.debug(f"변경 없음: {law_name}")
|
||||
|
||||
save_last_check(last_check)
|
||||
logger.info(f"=== 법령 모니터링 완료 — {changes_found}건 변경 감지 ===")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run()
|
||||
Reference in New Issue
Block a user