from __future__ import annotations
"""전제조건 진단 엔진 (아키텍처 §3.15, 온보딩 §4·§11.2).
온보딩(R6)·복구(R7)·CLI ``doctor`` 가 공유하는 **단일 진실**이다(ADR-24).
GUI 는 판단하지 않는다 — 판단은 전부 여기에 있고 화면은 렌더러일 뿐이다.
이 모듈이 지키는 규칙 4가지.
1. **어떤 상황에서도 예외로 죽지 않는다.** 진단기가 죽으면 진단이 불가능해진다.
개별 체크가 던지는 예외는 ``_safe()`` 가 흡수해 실패 결과값으로 바꾼다.
2. **``fix_action`` 없는 체크는 등록을 거부한다**(요구 R7.7). 화면에 실패로 뜨는데
누를 버튼이 없는 항목은 구조적으로 존재할 수 없다.
3. **아직 만들어지지 않은 모듈에 강하게 의존하지 않는다.** ``storage.repo`` ·
``watchdog`` · ``keycheck`` 가 있으면 쓰고, 없으면 같은 판정을 하는 내부 폴백을
쓴다. 진단 화면은 파이프라인이 미완성인 상태에서도 떠야 한다.
4. **인증키 원문은 어디에도 남기지 않는다.** 화면에는 식별번호(sha256 앞 8자)만 쓴다.
체크 12종의 실행 순서가 곧 화면 표시 순서다.
"""
import json
import logging
import os
import re
import sqlite3
import subprocess
import sys
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Optional, Sequence
from dmf_crawler.models import CheckResult, Severity
from dmf_crawler.paths import (
AGY_EXE,
AGY_TOKEN,
CONFIG_PATH,
HEARTBEAT_PATH,
PROJECT_ROOT,
VENV_PYTHONW,
db_path,
free_space_gb,
reports_dir,
)
__all__ = [
"CheckResult",
"Severity",
"CHECK_KEYS",
"REPROBE_AFTER",
"TASK_NAMES",
"run_all",
"run_one",
"verdict",
"can_run",
"summary_text",
"read_heartbeat",
"touch_heartbeat",
"last_task_result",
"next_run_text",
"KeyCheck",
"validate_service_key",
"explain_key_check",
"normalize_service_key",
"PORTAL_DATASET",
"PORTAL_MYPAGE",
"MYPAGE_HINT",
]
logger = logging.getLogger(__name__)
# Windows 에서만 의미가 있는 플래그. 다른 OS 에서는 0 이어야 subprocess 가 죽지 않는다.
_NO_WINDOW = getattr(subprocess, "CREATE_NO_WINDOW", 0)
# 작업 스케줄러에 등록되는 작업 3종(ADR-10, 온보딩 §10.1).
TASK_NAMES: tuple[str, ...] = (
r"\DMF Crawler\Daily",
r"\DMF Crawler\Agent",
r"\DMF Crawler\AgyUpdate",
)
# 런타임 의존성 3종(아키텍처 §8.1). (import 이름, 배포 이름)
_REQUIRED_PACKAGES: tuple[tuple[str, str], ...] = (
("httpx", "httpx"),
("xlsxwriter", "XlsxWriter"),
("jsonschema", "jsonschema"),
)
# 인증키 실검증 캐시 수명. 마법사를 열 때마다 API 를 때리면 하루 한도가 아깝다.
_KEY_CACHE_HOURS = 12.0
# heartbeat 신선도 기본 한계. `notify.watchdog_stale_minutes`(120분)는
# '오늘 아침 배치가 돌았는가' 를 판정하는 워치독 전용 값이라 그대로 쓰면
# 오후에 마법사를 열 때마다 경고가 뜬다. 진단은 하루 단위로 본다(온보딩 §4.2 ⑫).
_DEFAULT_STALE_HOURS = 30.0
# ==========================================================================
# 1. 인증키 검증 — keycheck 모듈이 있으면 그것을, 없으면 내부 구현을 쓴다
# ==========================================================================
PORTAL_DATASET = "https://www.data.go.kr/data/15057075/openapi.do"
PORTAL_MYPAGE = "https://www.data.go.kr/"
MYPAGE_HINT = "마이페이지 > 데이터 활용 > Open API > 활용신청 현황"
_DEFAULT_API_URL = (
"https://apis.data.go.kr/1471000/MdcDmfInfoService01/getMdcDmfList01"
)
from dmf_crawler.secrets_dpapi import ( # noqa: E402 - 순서 의존(경로 상수 뒤에 온다)
SECRET_PATH,
key_fingerprint,
load_service_key,
normalize_service_key,
)
@dataclass(frozen=True, slots=True)
class KeyCheck:
"""인증키 1회 조회의 구조화된 진단 결과.
``kind`` 는 응답 봉투의 종류다 — 공공데이터포털은 게이트웨이 오류와
서비스 오류의 JSON 구조가 완전히 다르다(온보딩 §7.4).
"""
ok: bool # 지금 이 키로 조회가 되는가
kind: str # service | gateway | network | unparsable | empty
status: Optional[int] # HTTP 상태 코드
code: Optional[str] # resultCode 또는 returnReasonCode
msg: Optional[str] # resultMsg 또는 errMsg 원문
total_count: Optional[int] # 전체 등록 건수(정상 응답일 때만)
def _api_url(cfg: Any = None) -> str:
"""검증에 쓸 엔드포인트를 고른다.
Args:
cfg: 설정 객체. ``source.base_url`` 을 반영한다.
Returns:
str: 엔드포인트 URL.
"""
url = getattr(getattr(cfg, "source", None), "base_url", "") or ""
return str(url).strip() or _DEFAULT_API_URL
def _probe(key: str, cfg: Any = None, timeout: float = 20.0) -> KeyCheck:
"""키 하나로 1건만 조회한다. 예외를 던지지 않고 구조화된 값을 돌려준다.
Args:
key: Decoding 형태의 인증키.
cfg: 설정 객체(엔드포인트 결정용).
timeout: 응답 대기 한계(초).
Returns:
KeyCheck: 진단 결과. 인증키 원문은 포함하지 않는다.
"""
params = {"serviceKey": key, "pageNo": 1, "numOfRows": 1, "type": "json"}
url = _api_url(cfg)
try:
import httpx # 지연 import — 부품 미설치 상태에서도 이 모듈은 import 돼야 한다
# params= 가 값을 정확히 1회 인코딩한다(ADR-05). 키는 Decoding 형태여야 한다.
response = httpx.get(
url,
params=params,
timeout=timeout,
headers={"User-Agent": "DMF-Crawler/onboarding"},
)
status, body = response.status_code, response.text
except ImportError:
return KeyCheck(
False,
"network",
None,
None,
"필요한 부품(httpx)이 아직 설치되지 않았습니다",
None,
)
except Exception as exc: # noqa: BLE001 - 네트워크 실패는 값으로 돌려준다
# 예외 문자열에 URL 이 섞여 들어가면 인증키가 파일에 남는다. 반드시 마스킹한다.
return KeyCheck(
False, "network", None, None, _mask(f"{type(exc).__name__}: {exc}"), None
)
# --- (A) 게이트웨이 오류 봉투 : OpenAPI_ServiceResponse / cmmMsgHeader ---
# 실측: 키 누락 -> HTTP 401, 잘못된 키 -> HTTP 403, 오퍼레이션 오타 -> HTTP 400.
# 정상 봉투와 경로가 완전히 다르므로 반드시 먼저 판정한다.
if "OpenAPI_ServiceResponse" in body:
code: Optional[str] = None
msg: Optional[str] = None
try:
header = json.loads(body)["OpenAPI_ServiceResponse"]["cmmMsgHeader"]
raw_code = header.get("returnReasonCode")
code = str(raw_code) if raw_code is not None else None
msg = header.get("errMsg")
except Exception: # noqa: BLE001 - type=json 이어도 XML 로 답하는 경우가 있다
m_code = re.search(r"([^<]*)", body)
m_msg = re.search(r"([^<]*)", body)
code = m_code.group(1) if m_code else None
msg = m_msg.group(1) if m_msg else None
return KeyCheck(False, "gateway", status, code, msg, None)
# --- (B) 정상/서비스 봉투 : response.header.resultCode 또는 header.resultCode ---
try:
payload = json.loads(body)
except ValueError:
return KeyCheck(False, "unparsable", status, None, _mask(body[:200]), None)
if not isinstance(payload, dict):
return KeyCheck(False, "unparsable", status, None, _mask(body[:200]), None)
wrapper = payload.get("response") if isinstance(payload.get("response"), dict) else {}
header = payload.get("header") or wrapper.get("header") or {}
body_obj = payload.get("body") or wrapper.get("body") or {}
if not isinstance(header, dict):
header = {}
if not isinstance(body_obj, dict):
body_obj = {}
raw_code = header.get("resultCode")
code = str(raw_code).zfill(2) if raw_code is not None else None
total = body_obj.get("totalCount")
try:
total_count = int(total) if total is not None else None
except (TypeError, ValueError):
total_count = None
return KeyCheck(
code == "00", "service", status, code, header.get("resultMsg"), total_count
)
def _mask(text: str) -> str:
"""문자열에 섞인 serviceKey 를 지운다.
Args:
text: 원문.
Returns:
str: 인증키가 ``***`` 로 바뀐 문자열.
"""
return re.sub(r"(?i)(serviceKey=)[^&\s'\"]+", r"\1***", text or "")
def validate_service_key(
raw: str, cfg: Any = None
) -> tuple[Optional[str], KeyCheck]:
"""붙여넣은 값에서 '실제로 되는' 키를 골라 돌려준다.
· 정규화본을 먼저, 원문을 그 다음으로 시도한다.
· 네트워크 실패면 후보를 더 시도하지 않는다(쿼터 낭비 방지).
· 쿼터 초과(22/23)는 '키가 유효하다' 는 증거이므로 반드시 저장한다.
이걸 실패로 처리하면 사용자가 멀쩡한 키를 하루 종일 다시 입력하게 된다.
Args:
raw: 사용자가 붙여넣은 원문.
cfg: 설정 객체(엔드포인트 결정용).
Returns:
tuple: (저장할 Decoding 형태 키 또는 None, 마지막 진단).
"""
if not raw or not raw.strip():
return None, KeyCheck(
False, "empty", None, None, "인증키가 비어 있습니다", None
)
candidates: list[str] = []
for candidate in (normalize_service_key(raw), raw.strip()):
if candidate and candidate not in candidates:
candidates.append(candidate)
last = KeyCheck(False, "empty", None, None, "인증키가 비어 있습니다", None)
for candidate in candidates:
last = _probe(candidate, cfg)
if last.ok:
return candidate, last
if last.kind == "network":
return None, last
if last.code in ("22", "23"):
return candidate, last
return None, last
_HUMAN: dict[str, str] = {
"20": "인증키가 비어 있거나 이 자료에 대한 사용 신청이 완료되지 않았습니다.\n"
f"{MYPAGE_HINT} 에서 상태를 확인해 주세요.",
"21": f"인증키가 일시적으로 중지된 상태입니다.\n{MYPAGE_HINT} 에서 확인해 주세요.",
"22": "오늘 조회 횟수를 다 썼습니다. 인증키 자체는 정상입니다.\n"
"자정이 지나면 자동으로 초기화됩니다.",
"23": "잠시 뒤 다시 시도해 주세요. 인증키 자체는 정상입니다.",
"29": "이 컴퓨터의 인터넷 주소가 차단되어 있습니다.\n"
"공공데이터포털 활용지원센터(1566-0025)로 문의해 주세요.",
"30": "등록되지 않은 인증키입니다.\n"
"포털에서 인증키를 새로 발급하면 예전 키는 자동으로 사라집니다.\n"
f"{MYPAGE_HINT} 에서 현재 인증키를 다시 복사해 주세요.",
"31": f"인증키 사용 기간이 끝났습니다.\n{MYPAGE_HINT} 에서 '활용연장신청' 을 해주세요.",
"32": "신청할 때 등록한 인터넷 주소와 지금 주소가 다릅니다.\n"
f"{MYPAGE_HINT} 에서 변경신청을 해주세요.",
"10": "자료 제공 방식이 바뀐 것 같습니다. 프로그램 업데이트가 필요할 수 있습니다.",
"11": "자료 제공 방식이 바뀐 것 같습니다. 프로그램 업데이트가 필요할 수 있습니다.",
"12": "자료 제공 방식이 바뀐 것 같습니다. 프로그램 업데이트가 필요할 수 있습니다.",
}
def explain_key_check(res: KeyCheck) -> str:
"""GUI 에 그대로 띄울 한국어 문장을 만든다. 인증키 원문은 절대 포함하지 않는다.
Args:
res: ``validate_service_key`` 가 돌려준 진단.
Returns:
str: 사용자에게 보여줄 문구.
"""
if res.ok:
count = f"{res.total_count:,}" if res.total_count else "?"
return f"정상 확인되었습니다. (전체 {count}건)"
if res.kind == "empty":
return "인증키를 입력해 주세요."
if res.kind == "network":
return (
"인터넷에 연결되어 있지 않거나 회사 방화벽이 접속을 막고 있습니다.\n"
"회사 PC 라면 IT 담당자에게 apis.data.go.kr (443) 허용을 요청해 주세요."
)
if res.kind == "unparsable":
return "서버가 예상과 다른 응답을 보냈습니다. 잠시 뒤 다시 시도해 주세요."
return _HUMAN.get(res.code or "", "잠시 문제가 있었습니다. 다시 시도해 주세요.")
# ==========================================================================
# 2. 상태 파일(heartbeat) 접근 — watchdog/state 모듈이 생기면 그쪽에 위임한다
# ==========================================================================
def read_heartbeat() -> dict[str, Any]:
"""``state/heartbeat.json`` 을 읽는다. 없거나 깨졌으면 빈 사전.
Returns:
dict[str, Any]: heartbeat 내용.
"""
delegate = _delegate("read_heartbeat")
if delegate is not None:
try:
value = delegate()
if isinstance(value, dict):
return value
except Exception: # noqa: BLE001 - 폴백으로 내려간다
logger.debug("위임 read_heartbeat 실패. 파일을 직접 읽는다.", exc_info=True)
try:
raw = HEARTBEAT_PATH.read_text(encoding="utf-8")
except (OSError, UnicodeDecodeError):
return {}
try:
data = json.loads(raw)
except ValueError:
return {}
return data if isinstance(data, dict) else {}
def touch_heartbeat(**patch: Any) -> dict[str, Any]:
"""heartbeat 에 키 몇 개만 병합해 기록한다.
온보딩이 ``api_key_verified_at`` 같은 보조 필드를 남기는 용도다.
실행 결과(``last_success_at`` 등)는 파이프라인이 기록한다.
Args:
**patch: 병합할 키·값.
Returns:
dict[str, Any]: 병합된 최종 내용.
"""
delegate = _delegate("touch_heartbeat")
if delegate is not None:
try:
value = delegate(**patch)
if isinstance(value, dict):
return value
return read_heartbeat()
except Exception: # noqa: BLE001 - 폴백으로 내려간다
logger.debug("위임 touch_heartbeat 실패. 파일에 직접 쓴다.", exc_info=True)
data = read_heartbeat()
data.update(patch)
try:
HEARTBEAT_PATH.parent.mkdir(parents=True, exist_ok=True)
tmp = HEARTBEAT_PATH.with_name(HEARTBEAT_PATH.name + ".tmp")
tmp.write_text(
json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8"
)
os.replace(tmp, HEARTBEAT_PATH)
except OSError:
logger.warning("heartbeat 를 기록하지 못했습니다: %s", HEARTBEAT_PATH)
return data
def _delegate(name: str) -> Optional[Callable[..., Any]]:
"""다른 모듈이 같은 이름의 함수를 이미 제공하면 그것을 돌려준다.
``watchdog.py`` · ``state.py`` 는 다른 담당자의 파일이라 아직 없을 수 있다.
있으면 그쪽이 정본이고, 없으면 이 모듈의 파일 직접 접근이 폴백이다.
Args:
name: 찾을 함수 이름.
Returns:
Optional[Callable]: 찾은 함수. 없으면 None.
"""
import importlib
for module_name in ("dmf_crawler.state", "dmf_crawler.watchdog"):
try:
module = importlib.import_module(module_name)
except Exception: # noqa: BLE001 - 아직 없는 모듈이다
continue
fn = getattr(module, name, None)
if callable(fn):
return fn
return None
def _repo_call(name: str, default: Any, **kwargs: Any) -> Any:
"""``storage.repo`` 의 함수를 있으면 부르고, 없으면 기본값을 돌려준다.
Args:
name: 함수 이름.
default: 모듈·함수가 없거나 실패했을 때 돌려줄 값.
**kwargs: 함수 인자.
Returns:
Any: 호출 결과 또는 기본값.
"""
import importlib
try:
repo = importlib.import_module("dmf_crawler.storage.repo")
except Exception: # noqa: BLE001 - 아직 없는 모듈이다
return default
fn = getattr(repo, name, None)
if not callable(fn):
return default
try:
return fn(**kwargs)
except Exception: # noqa: BLE001 - 진단기는 죽지 않는다
logger.debug("storage.repo.%s 호출 실패", name, exc_info=True)
return default
# ==========================================================================
# 3. 시간 · 파일 보조 함수
# ==========================================================================
def _parse_iso(value: str) -> Optional[datetime]:
"""ISO-8601 문자열을 시간대 있는 datetime 으로 바꾼다.
Args:
value: ISO 문자열.
Returns:
Optional[datetime]: 변환 결과. 실패하면 None.
"""
if not value:
return None
text = str(value).strip().replace("Z", "+00:00")
try:
parsed = datetime.fromisoformat(text)
except ValueError:
return None
if parsed.tzinfo is None:
parsed = parsed.replace(tzinfo=timezone.utc)
return parsed
def _age_hours(value: str) -> float:
"""주어진 시각으로부터 지금까지의 시간(시간 단위).
Args:
value: ISO 문자열.
Returns:
float: 경과 시간. 해석 실패 시 ``inf``.
"""
parsed = _parse_iso(value)
if parsed is None:
return float("inf")
delta = datetime.now(timezone.utc) - parsed
return delta.total_seconds() / 3600.0
def _fmt_ago(value: str) -> str:
"""'3시간 전' 같은 사람이 읽는 경과 시간 문구를 만든다.
Args:
value: ISO 문자열.
Returns:
str: 한국어 경과 표현.
"""
hours = _age_hours(value)
if hours == float("inf"):
return "시각 불명"
if hours < 1:
minutes = max(int(hours * 60), 0)
return f"{minutes}분 전"
if hours < 48:
return f"{int(hours)}시간 전"
return f"{int(hours / 24)}일 전"
def _is_locked(path: Path) -> bool:
"""파일이 다른 프로그램(Excel)에 잠겨 있는지 본다.
Args:
path: 검사할 파일.
Returns:
bool: 열 수 없으면 True.
"""
try:
with open(path, "r+b"):
return False
except OSError:
return True
def _stale_hours(cfg: Any) -> float:
"""heartbeat 신선도 한계를 시간 단위로 고른다.
Args:
cfg: 설정 객체.
Returns:
float: 한계 시간.
"""
value = getattr(getattr(cfg, "watchdog", None), "stale_hours", None)
try:
if value:
return float(value)
except (TypeError, ValueError):
pass
return _DEFAULT_STALE_HOURS
def _min_free_gb(cfg: Any) -> float:
"""최소 여유 공간 기준을 고른다.
``general.min_free_gb`` 는 설정 스키마에 없다. 정본은 ``backup.min_free_gb`` 다.
Args:
cfg: 설정 객체.
Returns:
float: 최소 여유 공간(GB).
"""
for section, key in (("general", "min_free_gb"), ("backup", "min_free_gb")):
value = getattr(getattr(cfg, section, None), key, None)
try:
if value:
return float(value)
except (TypeError, ValueError):
continue
return 2.0
# ==========================================================================
# 4. 체크 12종
# ==========================================================================
def _check_python_venv(cfg: Any = None) -> CheckResult:
"""① 실행 환경 — 전용 venv 안에서 Python 3.11+ 로 돌고 있는가."""
ver = sys.version_info
in_venv = sys.prefix != sys.base_prefix
ok = ver >= (3, 11) and in_venv and VENV_PYTHONW.exists()
if ok:
detail = f"Python {ver.major}.{ver.minor}.{ver.micro} · 전용 환경 사용 중"
else:
reason = []
if ver < (3, 11):
reason.append("파이썬 버전이 낮습니다")
if not in_venv:
reason.append("전용 실행 환경 밖에서 실행 중입니다")
if not VENV_PYTHONW.exists():
reason.append("전용 실행 환경이 만들어지지 않았습니다")
detail = (
f"Python {ver.major}.{ver.minor} · " + ", ".join(reason)
if reason
else "전용 실행 환경이 준비되지 않았습니다"
)
return CheckResult(
key="python_venv",
title=_TITLES["python_venv"],
ok=ok,
detail=detail,
fix_hint="프로그램 폴더의 bootstrap.cmd 를 한 번 더 실행하면 자동으로 준비됩니다.",
fix_action="open_bootstrap_help",
severity=Severity.CRITICAL,
)
def _check_dependencies(cfg: Any = None) -> CheckResult:
"""② 필요한 부품 — 런타임 의존성 3종이 import 되는가."""
import importlib
missing: list[str] = []
for module_name, dist_name in _REQUIRED_PACKAGES:
try:
importlib.import_module(module_name)
except Exception: # noqa: BLE001 - import 실패 원인은 구분하지 않는다
missing.append(dist_name)
ok = not missing
return CheckResult(
key="dependencies",
title=_TITLES["dependencies"],
ok=ok,
detail="모두 설치되어 있습니다" if ok else f"빠진 것: {', '.join(missing)}",
fix_hint="[설치하기] 를 누르면 자동으로 내려받아 설치합니다. (약 30초~2분)",
fix_action="install_deps",
severity=Severity.CRITICAL,
)
def _check_config(cfg: Any = None) -> CheckResult:
"""③ 설정 파일 — config.toml 이 파싱되고 검증을 통과하는가."""
from dmf_crawler.config import load_config, validate
from dmf_crawler.errors import ConfigError
if not CONFIG_PATH.exists():
return CheckResult(
key="config_valid",
title=_TITLES["config_valid"],
ok=False,
detail=f"설정 파일이 없습니다. ({CONFIG_PATH})",
fix_hint="[설정 열기] 를 누르면 기본값으로 만들어 메모장으로 엽니다.",
fix_action="open_config",
severity=Severity.CRITICAL,
)
try:
loaded = load_config()
except (ConfigError, OSError, ValueError) as exc:
return CheckResult(
key="config_valid",
title=_TITLES["config_valid"],
ok=False,
detail=f"설정 파일에 문제가 있습니다: {exc}",
fix_hint="[설정 열기] 를 눌러 해당 줄을 고치거나 기본값으로 되돌리세요.",
fix_action="open_config",
severity=Severity.CRITICAL,
)
problems = validate(loaded)
if problems:
return CheckResult(
key="config_valid",
title=_TITLES["config_valid"],
ok=False,
detail="설정 값에 문제가 있습니다: " + " / ".join(problems[:3]),
fix_hint="[설정 열기] 를 눌러 해당 줄을 고친 뒤 [다시 검사] 를 눌러 주세요.",
fix_action="open_config",
severity=Severity.CRITICAL,
)
return CheckResult(
key="config_valid",
title=_TITLES["config_valid"],
ok=True,
detail=f"정상 · 실행 시각 {loaded.schedule.daily_time}",
fix_hint="",
fix_action="open_config",
severity=Severity.CRITICAL,
)
def _check_api_key_present(cfg: Any = None) -> CheckResult:
"""④ 인증키 존재 — 공식 API 경로를 켤 수 있는 선택 항목인가."""
fingerprint = key_fingerprint()
if fingerprint:
return CheckResult(
key="api_key_present",
title=_TITLES["api_key_present"],
ok=True,
detail=f"등록됨 (식별번호 {fingerprint})",
fix_hint="",
fix_action="enter_api_key",
severity=Severity.WARN,
)
detail = (
"저장된 인증키를 읽을 수 없습니다. 공식 API 자동 수집만 비활성화되고 리포트는 기존 자료로 계속 만들어집니다."
if SECRET_PATH.exists()
else "선택 항목입니다. 아직 등록되지 않아 공식 API 자동 수집만 건너뜁니다."
)
return CheckResult(
key="api_key_present",
title=_TITLES["api_key_present"],
ok=False,
detail=detail,
fix_hint="공식 API 경로를 쓰고 싶을 때만 [키 입력] 으로 등록하세요. 키가 없어도 기존 기준선 리포트 생성은 막지 않습니다.",
fix_action="enter_api_key",
severity=Severity.WARN,
)
def _check_api_key_valid(cfg: Any = None, force: bool = False) -> CheckResult:
"""⑤ 인증키 사용 가능 여부 — 저장된 키로 실제 1건을 조회한다.
하루 10,000 호출 중 1회를 쓰므로 12시간 캐시를 둔다.
``force=True``([다시 검사])면 캐시를 무시한다.
"""
heartbeat = read_heartbeat()
last_verified = heartbeat.get("api_key_verified_at")
if not force and last_verified and _age_hours(str(last_verified)) < _KEY_CACHE_HOURS:
return CheckResult(
key="api_key_valid",
title=_TITLES["api_key_valid"],
ok=True,
detail=f"{_fmt_ago(str(last_verified))} 확인됨",
fix_hint="",
fix_action="enter_api_key",
severity=Severity.WARN,
)
key = load_service_key()
if not key:
return CheckResult(
key="api_key_valid",
title=_TITLES["api_key_valid"],
ok=False,
detail="선택 항목입니다. 키가 없어서 공식 API 실호출 확인을 건너뜁니다.",
fix_hint="공식 API 자동 수집이 필요할 때만 인증키를 등록해 주세요.",
fix_action="enter_api_key",
severity=Severity.WARN,
)
res = _probe(key, cfg)
# 쿼터 초과(22/23)는 '키가 유효하다' 는 증거다. 실패로 취급하지 않는다.
ok = res.ok or res.code in ("22", "23")
if ok:
patch: dict[str, Any] = {"api_key_verified_at": _utcnow_iso()}
if not heartbeat.get("api_key_first_success_at"):
patch["api_key_first_success_at"] = _utcnow_iso()
touch_heartbeat(**patch)
detail = explain_key_check(res)
return CheckResult(
key="api_key_valid",
title=_TITLES["api_key_valid"],
ok=ok,
detail=detail.replace("\n", " "),
fix_hint="" if res.ok else "[키 입력] 을 눌러 인증키를 다시 등록해 주세요. 공식 API가 필요하지 않다면 건너뛰어도 됩니다.",
fix_action="enter_api_key",
severity=Severity.WARN,
)
def _agy_binary(cfg: Any = None) -> Path:
"""agy 실행 파일 경로를 고른다.
Args:
cfg: 설정 객체. ``agy.binary_path`` 를 반영한다.
Returns:
Path: agy.exe 절대경로.
"""
configured = getattr(getattr(cfg, "agy", None), "binary_path", "") or ""
if str(configured).strip():
return Path(str(configured).strip())
return AGY_EXE
def _agy_required_reason(cfg: Any = None) -> str:
"""현재 설정에서 AGY가 필요한 이유를 돌려준다. 필요 없으면 빈 문자열."""
if bool(getattr(getattr(cfg, "agy", None), "enabled", False)):
return "AI 요약"
mode = str(getattr(getattr(cfg, "source", None), "mode", "auto") or "auto").lower()
if mode == "browser":
return "브라우저 수집 모드"
if mode == "auto":
try:
from dmf_crawler import secrets_dpapi
if not secrets_dpapi.load_service_key():
return "공식 API 키 없음 — headful 브라우저 수집"
except Exception: # noqa: BLE001 - 키를 못 읽으면 headful 수집 준비가 필요하다
return "공식 API 키 확인 불가 — headful 브라우저 수집"
return ""
def _check_agy_installed(cfg: Any = None) -> CheckResult:
"""⑥ AGY — 수집용 headful 또는 요약용 agy.exe 가 실물로 있고 실행되는가(WARN)."""
reason = _agy_required_reason(cfg)
if not reason:
return CheckResult(
key="agy_installed",
title=_TITLES["agy_installed"],
ok=True,
detail="AI 요약/브라우저 수집 사용 안 함 — 설치 필요 없음",
fix_hint="",
fix_action=None,
severity=Severity.WARN,
)
binary = _agy_binary(cfg)
ok = False
version: Optional[str] = None
# 절대 경로로만 판정한다. where.exe 는 winget Links 심볼릭까지 잡아 두 경로가 나온다.
if binary.exists() and binary.stat().st_size > 1_000_000:
env = {**os.environ, "AGY_CLI_DISABLE_AUTO_UPDATE": "true"}
try:
proc = subprocess.run(
[str(binary), "--version"],
capture_output=True,
text=True,
encoding="utf-8",
errors="replace",
timeout=30,
env=env,
creationflags=_NO_WINDOW,
)
ok = proc.returncode == 0 and bool((proc.stdout or "").strip())
if ok:
version = (proc.stdout or "").strip().splitlines()[0]
except Exception: # noqa: BLE001 - 실행 실패는 '설치 안 됨' 과 같다
ok = False
return CheckResult(
key="agy_installed",
title=_TITLES["agy_installed"],
ok=ok,
detail=f"설치됨 (버전 {version})" if ok else f"{reason}에 필요한 AGY가 설치되어 있지 않습니다.",
fix_hint=(
"[설치하기] 를 누르면 자동 설치합니다. (약 190 MB, 1~4분)\n"
"공식 API 키가 없거나 브라우저 수집 모드이면 최신 자료를 받기 위해 AGY headful Chrome 이 필요합니다. "
"설치하지 않아도 기존 성공 자료 리포트는 만들 수 있습니다."
),
fix_action="install_agy",
severity=Severity.WARN,
)
def _check_agy_auth(cfg: Any = None) -> CheckResult:
"""⑦ Google 로그인 — 최근 agy_calls 판정 + 토큰 파일(WARN).
``agy -p`` 로 실제 호출해 확인하지 않는다. 첫 호출 오버헤드가
input 28,317 토큰 / 33.7초다(agy SSOT §4.3 실측). ADR-16 의 원칙 —
"인증 상태는 별도 헬스체크가 아니라 실제 작업 호출의 결과로 판정한다".
"""
reason = _agy_required_reason(cfg)
if not reason:
return CheckResult(
key="agy_auth",
title=_TITLES["agy_auth"],
ok=True,
detail="AI 요약/브라우저 수집 사용 안 함 — Google 로그인 필요 없음",
fix_hint="",
fix_action=None,
severity=Severity.WARN,
)
recent = _repo_call("last_agy_error_kind", None, within_days=3)
if recent is not None and str(recent).upper().endswith("AUTH"):
return CheckResult(
key="agy_auth",
title=_TITLES["agy_auth"],
ok=False,
detail="지난 실행에서 로그인이 만료된 것으로 확인되었습니다.",
fix_hint="[로그인] 을 눌러 다시 로그인해 주세요. (최초 1회 방식과 동일)",
fix_action="login_agy",
severity=Severity.WARN,
)
ok = AGY_TOKEN.exists() and AGY_TOKEN.stat().st_size > 50
return CheckResult(
key="agy_auth",
title=_TITLES["agy_auth"],
ok=ok,
detail="로그인되어 있습니다" if ok else f"{reason}에 필요한 Google 로그인이 없습니다.",
fix_hint=(
"[로그인] 을 누르면 검은 창과 브라우저가 열립니다.\n"
"공식 API 키가 없거나 브라우저 수집 모드이면 최신 자료를 받기 위해 AGY headful Chrome 로그인이 필요합니다. "
"로그인하지 않아도 기존 성공 자료 리포트는 만들 수 있습니다."
),
fix_action="login_agy",
severity=Severity.WARN,
)
def _expected_schema_version() -> int:
"""코드가 기대하는 스키마 버전을 알아낸다.
``storage.db`` 가 있으면 그 상수를, 없으면 마이그레이션 파일 개수를 쓴다.
Returns:
int: 기대 버전. 알 수 없으면 0.
"""
import importlib
try:
db_module = importlib.import_module("dmf_crawler.storage.db")
except Exception: # noqa: BLE001 - 아직 없는 모듈이다
db_module = None
if db_module is not None:
for name in ("EXPECTED_SCHEMA_VERSION", "SCHEMA_VERSION", "LATEST_VERSION"):
value = getattr(db_module, name, None)
if isinstance(value, int):
return value
from dmf_crawler.paths import MIGRATIONS_DIR
try:
numbers = [
int(m.group(1))
for m in (re.match(r"(\d+)_", p.name) for p in MIGRATIONS_DIR.glob("*.sql"))
if m
]
except OSError:
return 0
return max(numbers) if numbers else 0
def _current_schema_version(conn: sqlite3.Connection) -> int:
"""DB 에 적용된 스키마 버전을 읽는다.
Args:
conn: 열린 연결.
Returns:
int: 적용된 최대 버전. 테이블이 없으면 0.
"""
row = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name='schema_version'"
).fetchone()
if row is None:
return 0
got = conn.execute("SELECT COALESCE(MAX(version), 0) FROM schema_version").fetchone()
return int(got[0]) if got else 0
def _check_database(cfg: Any = None) -> CheckResult:
"""⑧ 자료 보관소 — 존재·integrity_check·스키마 버전."""
path = db_path(cfg)
if not path.exists():
return CheckResult(
key="database",
title=_TITLES["database"],
ok=False,
detail="아직 만들어지지 않았습니다. (처음 실행 시 자동 생성)",
fix_hint="[복구하기] 를 누르면 지금 만듭니다.",
fix_action="repair_db",
severity=Severity.CRITICAL,
)
try:
conn = sqlite3.connect(f"file:{path}?mode=ro", uri=True, timeout=5.0)
except sqlite3.Error as exc:
return CheckResult(
key="database",
title=_TITLES["database"],
ok=False,
detail=f"자료 파일을 열 수 없습니다: {type(exc).__name__}",
fix_hint="[복구하기] 를 눌러 주세요. 최근 백업본으로 되돌립니다.",
fix_action="repair_db",
severity=Severity.CRITICAL,
)
try:
try:
row = conn.execute("PRAGMA integrity_check").fetchone()
except sqlite3.Error as exc:
return CheckResult(
key="database",
title=_TITLES["database"],
ok=False,
detail=f"자료 파일을 검사할 수 없습니다: {type(exc).__name__}",
fix_hint="[복구하기] 를 눌러 주세요.",
fix_action="repair_db",
severity=Severity.CRITICAL,
)
if not row or str(row[0]).lower() != "ok":
return CheckResult(
key="database",
title=_TITLES["database"],
ok=False,
detail="자료 파일이 손상되었습니다.",
fix_hint=(
"[복구하기] 를 누르면 가장 최근 백업본으로 되돌립니다.\n"
"되돌린 뒤 [지금 실행] 을 누르면 오늘 자료를 다시 받습니다."
),
fix_action="repair_db",
severity=Severity.CRITICAL,
)
version = _current_schema_version(conn)
finally:
conn.close()
expected = _expected_schema_version()
if expected and version < expected:
return CheckResult(
key="database",
title=_TITLES["database"],
ok=False,
detail=f"자료 형식 갱신이 필요합니다. (현재 {version} → 필요 {expected})",
fix_hint="[복구하기] 를 누르면 백업을 먼저 뜬 뒤 갱신합니다.",
fix_action="repair_db",
severity=Severity.CRITICAL,
)
return CheckResult(
key="database",
title=_TITLES["database"],
ok=True,
detail=f"정상 (형식 {version})",
fix_hint="",
fix_action="repair_db",
severity=Severity.CRITICAL,
)
def _query_task_xml(task_name: str) -> tuple[int, str]:
"""작업 스케줄러에서 작업 정의(XML)를 읽는다.
Args:
task_name: ``\\DMF Crawler\\Daily`` 형태의 전체 이름.
Returns:
tuple: (종료 코드, XML 문자열).
"""
try:
proc = subprocess.run(
["schtasks.exe", "/Query", "/TN", task_name, "/XML"],
capture_output=True,
timeout=30,
creationflags=_NO_WINDOW,
)
except Exception: # noqa: BLE001 - schtasks 가 없는 환경도 있다
return (1, "")
# schtasks /XML 은 UTF-16LE 로 출력한다.
raw = proc.stdout or b""
for encoding in ("utf-16-le", "utf-8", "cp949"):
try:
text = raw.decode(encoding, errors="replace")
except LookupError:
continue
if "<" in text:
return (proc.returncode, text)
return (proc.returncode, raw.decode("utf-8", errors="replace"))
def _check_tasks(cfg: Any = None) -> CheckResult:
"""⑨ 자동 실행 등록 — 3종이 있고 현재 폴더를 가리키는가(WARN).
폴더를 옮기면 작업은 남아 있지만 유령이 된다. 매일 06:00 에 조용히 실패하고
아무도 모른다. 그래서 경로 일치까지 확인한다.
"""
missing: list[str] = []
stale: list[str] = []
root = str(PROJECT_ROOT).lower()
for task_name in TASK_NAMES:
code, xml = _query_task_xml(task_name)
short = task_name.rsplit("\\", 1)[-1]
if code != 0:
missing.append(short)
continue
if short != "AgyUpdate" and root not in xml.lower():
stale.append(short)
if missing:
return CheckResult(
key="tasks_registered",
title=_TITLES["tasks_registered"],
ok=False,
detail=f"등록되지 않았습니다. (빠진 것: {', '.join(missing)})",
fix_hint="[등록하기] 를 누르면 바로 설정됩니다. 관리자 권한은 필요 없습니다.",
fix_action="install_tasks",
severity=Severity.WARN,
)
if stale:
return CheckResult(
key="tasks_registered",
title=_TITLES["tasks_registered"],
ok=False,
detail="예전 폴더를 가리키고 있습니다. 프로그램 폴더를 옮기신 것 같습니다.",
fix_hint="[등록하기] 를 눌러 지금 폴더로 갱신해 주세요.",
fix_action="install_tasks",
severity=Severity.WARN,
)
return CheckResult(
key="tasks_registered",
title=_TITLES["tasks_registered"],
ok=True,
detail=next_run_text(),
fix_hint="",
fix_action="install_tasks",
severity=Severity.WARN,
)
def _check_disk(cfg: Any = None) -> CheckResult:
"""⑩ 저장 공간 — 프로젝트 볼륨의 여유 공간."""
free = free_space_gb(PROJECT_ROOT)
need = _min_free_gb(cfg)
ok = free >= need
return CheckResult(
key="disk_space",
title=_TITLES["disk_space"],
ok=ok,
detail=f"여유 {free:.1f} GB" + ("" if ok else f" (최소 {need:.0f} GB 필요)"),
fix_hint="[디스크 정리] 를 눌러 공간을 확보한 뒤 [다시 검사] 를 눌러 주세요.",
fix_action="open_cleanmgr",
severity=Severity.CRITICAL,
)
def _check_report_writable(cfg: Any = None) -> CheckResult:
"""⑪ 리포트 저장 폴더 — 쓰기 가능한가, 오늘 파일이 Excel 에 잠겨 있지 않은가."""
directory = reports_dir(cfg)
try:
directory.mkdir(parents=True, exist_ok=True)
probe = directory / f".w_{os.getpid()}.tmp"
probe.write_text("ok", encoding="utf-8")
probe.unlink()
except OSError as exc:
return CheckResult(
key="report_writable",
title=_TITLES["report_writable"],
ok=False,
detail=f"폴더에 파일을 만들 수 없습니다: {type(exc).__name__}",
fix_hint="[폴더 열기] 로 위치를 확인하고, 쓰기가 가능한 곳으로 프로그램을 옮겨 주세요.",
fix_action="open_reports_dir",
severity=Severity.CRITICAL,
)
pattern = getattr(
getattr(cfg, "report", None), "filename_pattern", "DMF_리포트_{date}.xlsx"
)
today_name = str(pattern).format(date=datetime.now().strftime("%Y-%m-%d"))
today = directory / today_name
if today.exists() and _is_locked(today):
return CheckResult(
key="report_writable",
title=_TITLES["report_writable"],
ok=False,
detail=f"오늘 자 리포트가 Excel 에서 열려 있습니다.\n{today.name}",
fix_hint=(
"Excel 을 닫은 뒤 [다시 검사] 를 눌러 주세요.\n"
"닫지 않아도 실행은 되지만 다른 이름으로 저장됩니다."
),
fix_action="open_reports_dir",
severity=Severity.CRITICAL,
)
return CheckResult(
key="report_writable",
title=_TITLES["report_writable"],
ok=True,
detail=str(directory),
fix_hint="",
fix_action="open_reports_dir",
severity=Severity.CRITICAL,
)
def _check_recent_runs(cfg: Any = None) -> CheckResult:
"""⑫ 최근 실행 상태 — heartbeat 신선도와 연속 실패 횟수(WARN).
06:00 배치가 조용히 실패하는 것이 이 시스템의 가장 위험한 실패 모드다.
사용자가 마법사를 열었을 때 즉시 보이게 한다.
"""
heartbeat = read_heartbeat()
last_success = heartbeat.get("last_success_at")
if not last_success:
return CheckResult(
key="recent_runs",
title=_TITLES["recent_runs"],
ok=True,
detail="아직 한 번도 실행하지 않았습니다.",
fix_hint="",
fix_action="open_last_log",
severity=Severity.WARN,
)
age = _age_hours(str(last_success))
fails = _repo_call("consecutive_failures", 0)
try:
fails = int(fails or 0)
except (TypeError, ValueError):
fails = 0
limit = _stale_hours(cfg)
if age > limit:
return CheckResult(
key="recent_runs",
title=_TITLES["recent_runs"],
ok=False,
detail=f"마지막 성공이 {age / 24:.1f}일 전입니다. 연속 실패 {fails}회.",
fix_hint="[로그 보기] 로 원인을 확인하거나 [지금 실행] 으로 직접 돌려 보세요.",
fix_action="open_last_log",
severity=Severity.WARN,
)
return CheckResult(
key="recent_runs",
title=_TITLES["recent_runs"],
ok=True,
detail=f"마지막 성공 {_fmt_ago(str(last_success))}",
fix_hint="",
fix_action="open_last_log",
severity=Severity.WARN,
)
# ==========================================================================
# 5. 작업 스케줄러 상태 읽기
# ==========================================================================
_TASK_RESULT_HUMAN: dict[int, str] = {
0: "성공",
1: "실패 (일시적 원인일 수 있음)",
2: "확인 필요 (인증키·설정 문제)",
130: "사용자가 중단함",
267009: "실행 중", # SCHED_S_TASK_RUNNING
267011: "아직 실행된 적 없음", # SCHED_S_TASK_HAS_NOT_RUN
267014: "사용자가 작업을 종료함",
2147942402: "실행 파일을 찾지 못함 (폴더를 옮기셨나요?)",
}
def last_task_result() -> tuple[str, Optional[str]]:
"""``Daily`` 작업의 마지막 결과와 다음 실행 시각을 읽는다.
Returns:
tuple: (사람이 읽는 마지막 결과, 다음 실행 시각 문자열 또는 None).
"""
try:
proc = subprocess.run(
["schtasks.exe", "/Query", "/TN", TASK_NAMES[0], "/FO", "LIST", "/V"],
capture_output=True,
timeout=30,
creationflags=_NO_WINDOW,
)
except Exception: # noqa: BLE001 - 진단기는 죽지 않는다
return ("확인할 수 없음", None)
if proc.returncode != 0:
return ("등록되어 있지 않음", None)
text = ""
for encoding in ("cp949", "utf-8", "utf-16-le"):
try:
candidate = (proc.stdout or b"").decode(encoding, errors="replace")
except LookupError:
continue
if ":" in candidate:
text = candidate
break
result_text = "확인할 수 없음"
next_run: Optional[str] = None
for line in text.splitlines():
if ":" not in line:
continue
label, _, value = line.partition(":")
label = label.strip().lower()
value = value.strip()
if not value:
continue
if "last result" in label or "마지막 실행 결과" in label:
try:
code = int(value, 0)
except ValueError:
result_text = value
else:
result_text = _TASK_RESULT_HUMAN.get(code, f"코드 {code}")
elif "next run time" in label or "다음 실행 시간" in label:
next_run = value
return (result_text, next_run)
def next_run_text() -> str:
"""``tasks_registered`` 통과 시 보여줄 한 줄 문구를 만든다.
Returns:
str: 다음 실행 시각과 마지막 결과.
"""
result, next_run = last_task_result()
if next_run:
return f"다음 실행 {next_run} · 마지막 결과 {result}"
return f"등록됨 · 마지막 결과 {result}"
# ==========================================================================
# 6. 등록부 · 실행기
# ==========================================================================
_TITLES: dict[str, str] = {
"python_venv": "실행 환경",
"dependencies": "필요한 부품",
"config_valid": "설정 파일",
"api_key_present": "공식 API 인증키(선택)",
"api_key_valid": "공식 API 사용 가능 여부(선택)",
"agy_installed": "AGY 수집/AI 요약(선택)",
"agy_auth": "AGY 로그인(선택)",
"database": "자료 보관소",
"tasks_registered": "자동 실행 등록",
"disk_space": "저장 공간",
"report_writable": "리포트 저장 폴더",
"recent_runs": "최근 실행 상태",
}
_SEVERITY: dict[str, Severity] = {
"python_venv": Severity.CRITICAL,
"dependencies": Severity.CRITICAL,
"config_valid": Severity.CRITICAL,
"api_key_present": Severity.WARN,
"api_key_valid": Severity.WARN,
"agy_installed": Severity.WARN,
"agy_auth": Severity.WARN,
"database": Severity.CRITICAL,
"tasks_registered": Severity.WARN,
"disk_space": Severity.CRITICAL,
"report_writable": Severity.CRITICAL,
"recent_runs": Severity.WARN,
}
# 이 체크가 실패하면 뒤의 어떤 체크를 '판정 보류' 로 만드는가.
_GATES: frozenset[str] = frozenset({"python_venv", "dependencies"})
_DEPENDS_ON: dict[str, tuple[str, ...]] = {
"python_venv": (
"dependencies",
"config_valid",
"database",
),
"dependencies": ("database",),
}
# GUI 가 액션을 마친 뒤 어떤 key 만 다시 검사할지(상태 머신 §6.3-2).
REPROBE_AFTER: dict[str, tuple[str, ...]] = {
"enter_api_key": ("api_key_present", "api_key_valid"),
"install_deps": ("dependencies",),
"install_agy": ("agy_installed", "agy_auth"),
"login_agy": ("agy_auth",),
"install_tasks": ("tasks_registered",),
"repair_db": ("database", "recent_runs"),
"open_config": ("config_valid", "tasks_registered"),
"open_cleanmgr": ("disk_space",),
"open_reports_dir": ("report_writable",),
"open_last_log": ("recent_runs",),
"open_bootstrap_help": ("python_venv", "dependencies"),
}
# '앞 단계 확인 후 판정' 상태를 알아보는 표식. GUI 가 회색 처리에 쓴다.
PENDING_PREFIX = "앞 단계"
_REGISTRY: list[tuple[str, Callable[..., CheckResult], str]] = []
def _register(key: str, fn: Callable[..., CheckResult], fix_action: str) -> None:
"""체크를 등록한다.
``fix_action`` 이 비어 있으면 등록을 거부한다. 이것이 '막다른 골목 금지'
(요구 R7.7)를 계약으로 강제하는 지점이다. 화면에 실패로 뜨는데 누를 버튼이
없는 항목은 존재할 수 없다.
Args:
key: 체크 식별자.
fn: 체크 함수. ``cfg`` 를 첫 인자로 받는다.
fix_action: GUI 버튼이 호출할 액션 키.
Raises:
ValueError: ``fix_action`` 이 비었거나 제목·등급 메타데이터가 없을 때.
"""
if not fix_action:
raise ValueError(
f"check '{key}' 에 fix_action 이 없습니다. "
"모든 체크는 사용자가 취할 행동을 반드시 제공해야 합니다."
)
if key not in _TITLES or key not in _SEVERITY:
raise ValueError(f"check '{key}' 의 제목·등급 메타데이터가 없습니다.")
_REGISTRY.append((key, fn, fix_action))
def _safe(
key: str,
fn: Callable[..., CheckResult],
fix_action: str,
*args: Any,
**kwargs: Any,
) -> CheckResult:
"""체크 자체가 던지는 예외를 흡수해 결과값으로 바꾼다.
Args:
key: 체크 식별자.
fn: 체크 함수.
fix_action: 실패 시 제시할 액션.
*args: 체크 함수 인자.
**kwargs: 체크 함수 키워드 인자.
Returns:
CheckResult: 정상 결과 또는 예외를 담은 실패 결과.
"""
try:
return fn(*args, **kwargs)
except Exception as exc: # noqa: BLE001 - 진단기는 죽으면 안 된다
logger.warning("체크 '%s' 가 예외로 실패했습니다.", key, exc_info=True)
return CheckResult(
key=key,
title=_TITLES.get(key, key),
ok=False,
detail=f"확인하는 중 문제가 생겼습니다: {type(exc).__name__}: {exc}",
fix_hint=(
"[다시 검사] 를 눌러 보고, 계속되면 [진단 결과 복사하기] 로 "
"내용을 복사해 도움을 요청해 주세요."
),
fix_action=fix_action,
severity=_SEVERITY.get(key, Severity.CRITICAL),
)
def _pending(key: str, fix_action: str) -> CheckResult:
"""앞 단계가 막혀 아직 판정할 수 없는 항목을 만든다.
Args:
key: 체크 식별자.
fix_action: 해당 체크의 액션.
Returns:
CheckResult: 보류 상태 결과.
"""
return CheckResult(
key=key,
title=_TITLES[key],
ok=False,
detail=f"{PENDING_PREFIX} 확인 후 판정합니다.",
fix_hint="",
fix_action=fix_action,
severity=_SEVERITY[key],
)
def run_all(cfg: Any = None, *, force: bool = False) -> list[CheckResult]:
"""12종을 순서대로 실행한다. 앞 항목이 실패하면 뒤 항목을 보류로 표시한다.
Args:
cfg: 설정 객체. 없으면 기본값으로 판정한다.
force: True 면 인증키 실검증 캐시를 무시한다([다시 검사]).
Returns:
list[CheckResult]: 등록 순서대로의 결과 12건.
"""
results: list[CheckResult] = []
blocked: set[str] = set()
for key, fn, fix_action in _REGISTRY:
if key in blocked:
results.append(_pending(key, fix_action))
continue
kwargs: dict[str, Any] = {"force": True} if (force and key == "api_key_valid") else {}
result = _safe(key, fn, fix_action, cfg, **kwargs)
results.append(result)
if not result.ok and result.severity is Severity.CRITICAL and key in _GATES:
blocked.update(_DEPENDS_ON.get(key, ()))
return results
def run_one(key: str, cfg: Any = None, **kwargs: Any) -> CheckResult:
"""단일 체크만 다시 실행한다. GUI 의 부분 재검사가 쓴다.
Args:
key: 체크 식별자.
cfg: 설정 객체.
**kwargs: 체크 함수에 전달할 추가 인자(``force`` 등).
Returns:
CheckResult: 실행 결과.
Raises:
KeyError: 등록되지 않은 체크일 때.
"""
for registered_key, fn, fix_action in _REGISTRY:
if registered_key == key:
return _safe(registered_key, fn, fix_action, cfg, **kwargs)
raise KeyError(f"알 수 없는 체크: {key}")
def verdict(results: Sequence[CheckResult]) -> int:
"""``doctor`` 종료 코드를 계산한다.
Args:
results: 체크 결과 목록.
Returns:
int: 0=전부 통과, 1=WARN 존재, 2=CRITICAL 존재.
"""
if any(not r.ok and r.severity is Severity.CRITICAL for r in results):
return 2
if any(not r.ok for r in results):
return 1
return 0
def can_run(results: Sequence[CheckResult]) -> bool:
"""[지금 실행] 버튼 활성화 여부. WARN 은 실행을 막지 않는다.
Args:
results: 체크 결과 목록.
Returns:
bool: CRITICAL 실패가 없으면 True.
"""
return not any(not r.ok and r.severity is Severity.CRITICAL for r in results)
def summary_text(results: Sequence[CheckResult]) -> str:
"""진단 결과를 콘솔·클립보드용 여러 줄 텍스트로 만든다.
Args:
results: 체크 결과 목록.
Returns:
str: 사람이 읽는 요약.
"""
if not results:
return "진단 결과가 없습니다."
width = max(len(r.title) for r in results)
lines: list[str] = []
for r in results:
mark = "OK " if r.ok else ("WARN" if r.severity is Severity.WARN else "FAIL")
lines.append(f"[{mark}] {r.title:<{width}} {r.detail}")
if not r.ok and r.fix_hint:
for hint_line in r.fix_hint.splitlines():
lines.append(f" → {hint_line}")
return "\n".join(lines)
def _utcnow_iso() -> str:
"""현재 시각을 UTC ISO-8601 문자열로 돌려준다.
Returns:
str: ``2026-09-03T01:02:03+00:00`` 형태.
"""
return datetime.now(timezone.utc).isoformat()
# --- 등록 (실행 순서 = 화면 표시 순서) ------------------------------------
_register("python_venv", _check_python_venv, "open_bootstrap_help")
_register("dependencies", _check_dependencies, "install_deps")
_register("config_valid", _check_config, "open_config")
_register("api_key_present", _check_api_key_present, "enter_api_key")
_register("api_key_valid", _check_api_key_valid, "enter_api_key")
_register("agy_installed", _check_agy_installed, "install_agy")
_register("agy_auth", _check_agy_auth, "login_agy")
_register("database", _check_database, "repair_db")
_register("tasks_registered", _check_tasks, "install_tasks")
_register("disk_space", _check_disk, "open_cleanmgr")
_register("report_writable", _check_report_writable, "open_reports_dir")
_register("recent_runs", _check_recent_runs, "open_last_log")
CHECK_KEYS: tuple[str, ...] = tuple(key for key, _, _ in _REGISTRY)