# -*- coding: utf-8 -*-
"""
Server Manager v2.5 BATCH LOG CENTER
추가 내용:
- 오늘 수집/상세수집/초안/발행 KPI 표시
- Apache 80 포트 상태 표시
- FileZilla FTP 21 포트 상태 표시
- 5000/9000 API 응답시간 표시
- 9200 자체 상태 표시
- 전체 서버 건강도 점수 표시
"""
import os
import sys
import json
import time
import socket
import threading
import subprocess
import traceback
import webbrowser
import urllib.request
import ctypes
from datetime import datetime, timedelta
from pathlib import Path
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
BASE_DIR = Path(__file__).resolve().parent
CONFIG_PATH = BASE_DIR / "server_manager_config.json"
LOG_DIR = BASE_DIR / "logs"
STATUS_DIR = BASE_DIR / "status"
LOG_DIR.mkdir(parents=True, exist_ok=True)
STATUS_DIR.mkdir(parents=True, exist_ok=True)
processes = {}
job_process = None
job_lock = threading.Lock()
event_log = []
# STEP101-12 scheduler state
scheduler_state = {
"enabled": True,
"last_tick": "",
"last_started": {},
"next_runs": [],
"last_skipped": "",
}
def now_text():
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def console(message):
print(message, flush=True)
def error_log(message):
try:
with open(LOG_DIR / "manager_error.log", "a", encoding="utf-8") as f:
f.write(f"[{now_text()}] {message}\n")
except Exception:
pass
def load_config():
if not CONFIG_PATH.exists():
raise FileNotFoundError(f"config file not found: {CONFIG_PATH}")
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.load(f)
def append_event(level, message):
item = {"time": now_text(), "level": level, "message": str(message)}
event_log.insert(0, item)
del event_log[100:]
line = f"[{item['time']}] [{level}] {message}"
console(line)
try:
with open(LOG_DIR / "manager.log", "a", encoding="utf-8") as f:
f.write(line + "\n")
except Exception:
pass
def is_port_open(port, host="127.0.0.1", timeout=1.0):
started = time.time()
try:
with socket.create_connection((host, int(port)), timeout=timeout):
elapsed_ms = int((time.time() - started) * 1000)
return True, elapsed_ms
except Exception:
elapsed_ms = int((time.time() - started) * 1000)
return False, elapsed_ms
def check_http_url(url, timeout=2.5):
if not url:
return False, 0, ""
started = time.time()
try:
req = urllib.request.Request(url, headers={"User-Agent": "HongheeServerManager/1.2"})
with urllib.request.urlopen(req, timeout=timeout) as res:
status = getattr(res, "status", 0)
elapsed_ms = int((time.time() - started) * 1000)
return 200 <= int(status) < 500, elapsed_ms, str(status)
except Exception as e:
elapsed_ms = int((time.time() - started) * 1000)
return False, elapsed_ms, str(e)[:120]
def run_cmd_capture(cmd):
try:
p = subprocess.run(
cmd,
capture_output=True,
text=True,
encoding="utf-8",
errors="ignore",
shell=True
)
return p.stdout.strip(), p.stderr.strip(), p.returncode
except Exception as e:
return "", str(e), 1
def get_pid_by_port(port):
"""
Windows netstat 기준으로 LISTENING PID를 찾는다.
"""
try:
port = int(port)
except Exception:
return None
out, err, code = run_cmd_capture(f'netstat -ano | findstr LISTENING | findstr :{port}')
if not out:
return None
for line in out.splitlines():
parts = line.split()
if len(parts) < 5:
continue
local_addr = parts[1]
state = parts[3]
pid_text = parts[4]
if state.upper() != "LISTENING":
continue
# 0.0.0.0:5000, 127.0.0.1:5000, [::]:5000 대응
if local_addr.endswith(f":{port}"):
try:
return int(pid_text)
except Exception:
return None
return None
def get_process_info(pid):
if not pid:
return {
"pid": None,
"name": "",
"command_line": "",
"executable_path": "",
"exists": False,
}
ps_cmd = (
'powershell -NoProfile -Command '
f'"Get-CimInstance Win32_Process -Filter \\"ProcessId={int(pid)}\\" | '
'Select-Object ProcessId,Name,ExecutablePath,CommandLine | ConvertTo-Json -Compress"'
)
out, err, code = run_cmd_capture(ps_cmd)
if not out:
return {
"pid": int(pid),
"name": "",
"command_line": "",
"executable_path": "",
"exists": False,
"error": err,
}
try:
data = json.loads(out)
if isinstance(data, list):
data = data[0] if data else {}
return {
"pid": int(pid),
"name": data.get("Name") or "",
"command_line": data.get("CommandLine") or "",
"executable_path": data.get("ExecutablePath") or "",
"exists": True,
}
except Exception:
return {
"pid": int(pid),
"name": "",
"command_line": out[:500],
"executable_path": "",
"exists": bool(out),
"error": err,
}
def is_manager_owned_process(key, pid):
if not pid:
return False
proc_info = processes.get(key)
if not proc_info or not proc_info.get("process"):
return False
p = proc_info["process"]
return p.poll() is None and int(p.pid) == int(pid)
def force_kill_pid(pid):
try:
pid = int(pid)
except Exception:
return False, "invalid pid"
if pid <= 0:
return False, "invalid pid"
# 9200 manager 자신을 죽이는 실수 방지
if pid == os.getpid():
return False, "cannot kill server_manager itself"
out, err, code = run_cmd_capture(f'taskkill /PID {pid} /F')
if code == 0:
return True, out or f"killed pid={pid}"
return False, err or out or f"taskkill failed code={code}"
def force_stop_service_by_key(key):
svc = get_service(key)
if not svc:
return False, f"service not found: {key}"
port = int(svc.get("port") or 0)
pid = get_pid_by_port(port)
if not pid:
return False, f"no pid for port={port}"
ok, msg = force_kill_pid(pid)
if ok:
# manager가 들고 있던 프로세스 정보가 있으면 정리
proc_info = processes.get(key)
if proc_info and proc_info.get("process") and int(proc_info["process"].pid) == int(pid):
try:
proc_info.get("log_file").close()
except Exception:
pass
processes.pop(key, None)
return ok, msg
def restart_service(key):
svc = get_service(key)
if not svc:
return False, f"service not found: {key}"
# 먼저 포트 점유 프로세스 강제 종료
port = int(svc.get("port") or 0)
pid = get_pid_by_port(port)
if pid:
ok, msg = force_kill_pid(pid)
append_event("warning" if ok else "error", f"{svc.get('name')} 재시작 전 종료 pid={pid}, result={msg}")
time.sleep(2)
# manager 내부 프로세스 정보 정리
proc_info = processes.get(key)
if proc_info:
try:
proc_info.get("log_file").close()
except Exception:
pass
processes.pop(key, None)
return start_service(key), "restart requested"
def get_service(key):
cfg = load_config()
for svc in cfg.get("services", []):
if svc.get("key") == key:
return svc
return None
def start_service(key):
svc = get_service(key)
if not svc:
append_event("error", f"서비스를 찾을 수 없습니다: {key}")
return False
port = int(svc.get("port") or 0)
port_ok, _ = is_port_open(port)
if port and port_ok:
append_event("info", f"{svc.get('name')} 이미 실행 중 port={port}")
return True
cfg = load_config()
python_exe = cfg.get("python_exe")
script = svc.get("script")
cwd = svc.get("cwd") or os.path.dirname(script)
if not os.path.exists(python_exe):
append_event("error", f"python.exe 없음: {python_exe}")
return False
if not os.path.exists(script):
append_event("error", f"script 없음: {script}")
return False
log_path = LOG_DIR / f"{key}.log"
log_file = open(log_path, "a", encoding="utf-8", errors="ignore")
try:
p = subprocess.Popen(
[python_exe, script],
cwd=cwd,
stdout=log_file,
stderr=subprocess.STDOUT,
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
)
processes[key] = {"process": p, "log_file": log_file, "started_at": now_text()}
append_event("info", f"{svc.get('name')} 시작 pid={p.pid}, port={port}")
return True
except Exception as e:
append_event("error", f"{svc.get('name')} 시작 실패: {e}")
error_log(traceback.format_exc())
try:
log_file.close()
except Exception:
pass
return False
def stop_service(key):
svc = get_service(key)
proc_info = processes.get(key)
if proc_info and proc_info.get("process"):
p = proc_info["process"]
try:
p.terminate()
append_event("warning", f"{svc.get('name') if svc else key} 종료 요청 pid={p.pid}")
time.sleep(2)
if p.poll() is None:
p.kill()
append_event("warning", f"{svc.get('name') if svc else key} 강제 종료 pid={p.pid}")
except Exception as e:
append_event("error", f"{key} 종료 실패: {e}")
error_log(traceback.format_exc())
try:
proc_info.get("log_file").close()
except Exception:
pass
processes.pop(key, None)
return True
append_event("warning", f"{key}는 manager가 시작한 프로세스가 아니거나 이미 종료됨")
return False
def service_status():
cfg = load_config()
rows = []
for svc in cfg.get("services", []):
key = svc.get("key")
port = int(svc.get("port") or 0)
port_ok, port_ms = is_port_open(port) if port else (False, 0)
health_url = svc.get("health_url") or ""
health_ok = False
health_ms = 0
health_note = ""
if health_url and port_ok:
health_ok, health_ms, health_note = check_http_url(health_url)
proc_info = processes.get(key)
manager_pid = None
started_at = ""
if proc_info and proc_info.get("process"):
p = proc_info["process"]
if p.poll() is None:
manager_pid = p.pid
started_at = proc_info.get("started_at") or ""
else:
processes.pop(key, None)
port_pid = get_pid_by_port(port) if port else None
pid = port_pid or manager_pid
process_info = get_process_info(pid) if pid else {
"pid": None,
"name": "",
"command_line": "",
"executable_path": "",
"exists": False,
}
manager_owned = is_manager_owned_process(key, pid)
rows.append({
"key": key,
"name": svc.get("name"),
"script": svc.get("script"),
"port": port,
"port_ok": port_ok,
"port_ms": port_ms,
"health_url": health_url,
"health_ok": health_ok if health_url else port_ok,
"health_ms": health_ms if health_url else port_ms,
"health_note": health_note,
"auto_start": bool(svc.get("auto_start")),
"pid": pid,
"manager_pid": manager_pid,
"port_pid": port_pid,
"process_name": process_info.get("name") or "",
"command_line": process_info.get("command_line") or "",
"executable_path": process_info.get("executable_path") or "",
"process_exists": bool(process_info.get("exists")),
"manager_owned": manager_owned,
"owner_label": "Manager" if manager_owned else ("External" if pid else "-"),
"started_at": started_at,
"managed": True,
})
return rows
def external_port_status():
cfg = load_config()
rows = []
# external ports
for item in cfg.get("external_ports", []):
host = item.get("host") or "127.0.0.1"
port = int(item.get("port") or 0)
protocol = item.get("protocol") or "tcp"
url = item.get("url") or ""
ok, ms = is_port_open(port, host=host)
health_ok = ok
health_ms = ms
note = ""
if protocol == "http" and url and ok:
health_ok, health_ms, note = check_http_url(url)
rows.append({
"key": item.get("key"),
"name": item.get("name"),
"host": host,
"port": port,
"protocol": protocol,
"port_ok": ok,
"port_ms": ms,
"health_ok": health_ok,
"health_ms": health_ms,
"health_note": note,
"managed": False,
})
# self manager port
cfg_port = int(cfg.get("manager_port") or 9200)
ok, ms = is_port_open(cfg_port)
rows.append({
"key": "server_manager",
"name": "Server Manager 상황판",
"host": "127.0.0.1",
"port": cfg_port,
"protocol": "http",
"port_ok": ok,
"port_ms": ms,
"health_ok": ok,
"health_ms": ms,
"health_note": "self",
"managed": False,
})
return rows
def all_health_rows():
return external_port_status() + service_status()
def health_score(rows):
if not rows:
return 0
ok_count = sum(1 for r in rows if r.get("health_ok") or r.get("port_ok"))
return int(round((ok_count / len(rows)) * 100))
def normalize_time_text(value):
try:
parts = str(value or "").strip().split(":")
if len(parts) < 2:
return ""
return f"{int(parts[0]):02d}:{int(parts[1]):02d}"
except Exception:
return ""
# STEP142:
# Server Manager의 내부 job_key는 과거 이름(day_pipeline/night_publish)을 유지하되,
# 화면/로그/진단에서는 실제 운영 의미로 통일한다.
JOB_DISPLAY_DEFAULTS = {
"day_pipeline": "야간 수집/상세/초안 1회",
"night_publish": "주간 서버 발행 07:00~23:00",
"server_publish": "주간 서버 발행 07:00~23:00",
"test_2_realtors": "2개 테스트",
}
def get_job_display_name(job_key, cfg=None):
cfg = cfg or load_config()
job_key = str(job_key or "")
for item in cfg.get("schedules") or []:
if str(item.get("key") or "") == job_key and item.get("name"):
return str(item.get("name"))
return JOB_DISPLAY_DEFAULTS.get(job_key, job_key)
def cmd_text(cmd):
if isinstance(cmd, list):
return " ".join(str(x) for x in cmd)
return str(cmd or "")
def publish_queue_snapshot(execution_target="server"):
"""
발행 작업 전/후 큐 변화를 확인하기 위한 스냅샷.
publish_worker가 0코드로 끝나도 큐가 그대로면 no-op으로 경고한다.
"""
row, err = db_query_one("""
SELECT
COUNT(*) AS total_pending,
COUNT(DISTINCT realtor_id) AS realtor_count,
MIN(id) AS min_id,
MAX(id) AS max_id
FROM blog_publish_queue
WHERE queue_status IN ('pending','waiting')
AND COALESCE(NULLIF(execution_target,''), 'server') = %s
""", [execution_target])
if err:
return {"ok": False, "error": err, "execution_target": execution_target}
return {
"ok": True,
"execution_target": execution_target,
"pending": int((row or {}).get("total_pending") or 0),
"realtors": int((row or {}).get("realtor_count") or 0),
"min_id": (row or {}).get("min_id"),
"max_id": (row or {}).get("max_id"),
}
def job_kind_from_key_cmd(job_key, steps):
text = (str(job_key or "") + " " + " ".join(
(str(s.get("name") or "") + " " + cmd_text(s.get("cmd"))) for s in (steps or [])
)).lower()
if "publish_worker" in text or "night_publish" in str(job_key or "") or "server_publish" in str(job_key or ""):
return "server_publish"
if "daily_pipeline" in text or "schedule_daily" in text or "generate_blog_drafts" in text or "naver_response_article_fetcher" in text or "day_pipeline" in str(job_key or ""):
return "collect_draft"
return "unknown"
def validate_runtime_config(cfg=None):
"""
스케줄/작업 정의가 서로 꼬였는지 표시하기 위한 진단.
실제 실행을 막지는 않고 관제센터와 로그에 경고만 남긴다.
"""
cfg = cfg or load_config()
issues = []
jobs = cfg.get("jobs") or {}
for key, steps in jobs.items():
if not isinstance(steps, list) or not steps:
issues.append({"level": "error", "key": key, "message": "jobs 정의가 비어 있습니다."})
continue
for idx, step in enumerate(steps, start=1):
cmd = step.get("cmd")
if not isinstance(cmd, list) or not cmd:
issues.append({"level": "error", "key": key, "message": f"{idx}번째 step cmd가 비어 있거나 list가 아닙니다."})
continue
cmdline = cmd_text(cmd).lower()
if "publish" in str(key).lower() and "publish_worker" not in cmdline and "v2_publish_worker" not in cmdline:
issues.append({"level": "warning", "key": key, "message": f"발행 job처럼 보이지만 cmd에 publish_worker가 없습니다: {cmd_text(cmd)}"})
if "day_pipeline" == str(key) and ("publish_worker" in cmdline or "v2_publish_worker" in cmdline):
issues.append({"level": "warning", "key": key, "message": "day_pipeline에 발행 worker가 연결되어 있습니다. 수집/초안 job과 발행 job이 바뀌었을 수 있습니다."})
for item in cfg.get("schedules") or []:
key = item.get("key")
if key and key not in jobs:
issues.append({"level": "error", "key": key, "message": "schedule에는 있지만 jobs 정의가 없습니다."})
name = str(item.get("name") or "")
if key == "night_publish" and "수집" in name:
issues.append({"level": "warning", "key": key, "message": "night_publish의 표시명이 수집으로 되어 있습니다."})
if key == "day_pipeline" and "발행" in name:
issues.append({"level": "warning", "key": key, "message": "day_pipeline의 표시명이 발행으로 되어 있습니다."})
return issues
def render_config_issue_table(issues):
if not issues:
return '스케줄/작업 설정 진단 이상 없음
'
html = """
레벨작업내용
"""
for i in issues[:20]:
level = i.get("level") or "warning"
color = "#dc2626" if level == "error" else "#d97706"
html += f"""
{html_escape(level)}
{html_escape(i.get('key'))}
{html_escape(i.get('message'))}
"""
html += "
"
return html
def build_schedule_plan(cfg=None):
cfg = cfg or load_config()
schedules = cfg.get("schedules") or []
plan = []
now = datetime.now()
for item in schedules:
if not item.get("enabled", True):
continue
key = item.get("key")
name = item.get("name") or get_job_display_name(key, cfg)
times = item.get("times") or []
if not key or not isinstance(times, list):
continue
next_dt = None
clean_times = []
for t in times:
tt = normalize_time_text(t)
if not tt:
continue
clean_times.append(tt)
hh, mm = [int(x) for x in tt.split(":")]
candidate = now.replace(hour=hh, minute=mm, second=0, microsecond=0)
if candidate <= now:
candidate = candidate + timedelta(days=1)
if next_dt is None or candidate < next_dt:
next_dt = candidate
if next_dt:
steps = (cfg.get("jobs") or {}).get(key) or []
plan.append({
"key": key,
"name": name,
"times": clean_times,
"next_run": next_dt.strftime("%Y-%m-%d %H:%M:%S"),
"max_run_minutes": int(item.get("max_run_minutes") or 0),
"kind": job_kind_from_key_cmd(key, steps),
"steps": len(steps),
"cmd_preview": cmd_text((steps[0] or {}).get("cmd"))[:180] if steps else "",
})
plan.sort(key=lambda x: x.get("next_run") or "")
return plan
def scheduler_loop():
append_event("info", "scheduler loop started")
while True:
try:
cfg = load_config()
scheduler_state["last_tick"] = now_text()
scheduler_state["next_runs"] = build_schedule_plan(cfg)
now = datetime.now()
hhmm = now.strftime("%H:%M")
today = now.strftime("%Y-%m-%d")
for item in cfg.get("schedules") or []:
if not item.get("enabled", True):
continue
key = item.get("key")
name = item.get("name") or get_job_display_name(key, cfg)
times = [normalize_time_text(t) for t in (item.get("times") or [])]
times = [t for t in times if t]
if not key or hhmm not in times:
continue
run_token = f"{today} {hhmm} {key}"
if scheduler_state["last_started"].get(key) == run_token:
continue
if job_process and job_process.poll() is None:
scheduler_state["last_skipped"] = f"{now_text()} {name} skipped: previous job running pid={job_process.pid}"
append_event("warning", f"[SCHEDULER SKIP] {name} 예약시간={hhmm}, 이전 작업 실행중 pid={job_process.pid}")
scheduler_state["last_started"][key] = run_token
continue
append_event("info", f"[SCHEDULER RUN] {name} 예약시간={hhmm}")
ok = run_job(key)
scheduler_state["last_started"][key] = run_token
if not ok:
scheduler_state["last_skipped"] = f"{now_text()} {name} failed to start"
except Exception as e:
append_event("error", f"scheduler error: {e}")
error_log(traceback.format_exc())
try:
sleep_sec = int(load_config().get("scheduler_tick_seconds") or 20)
except Exception:
sleep_sec = 20
time.sleep(max(5, sleep_sec))
def monitor_loop():
append_event("info", "monitor loop started")
while True:
try:
cfg = load_config()
for svc in cfg.get("services", []):
key = svc.get("key")
port = int(svc.get("port") or 0)
auto_start = bool(svc.get("auto_start"))
port_ok, _ = is_port_open(port)
if auto_start and port and not port_ok:
append_event("warning", f"{svc.get('name')} port={port} 응답 없음 → 자동 시작")
start_service(key)
write_status_file()
except Exception as e:
append_event("error", f"monitor error: {e}")
error_log(traceback.format_exc())
time.sleep(int(load_config().get("check_interval_seconds") or 10))
def run_job(job_key):
global job_process
cfg = load_config()
steps = (cfg.get("jobs") or {}).get(job_key)
display_name = get_job_display_name(job_key, cfg)
if not steps:
append_event("error", f"작업 정의 없음: {job_key}")
return False
if not isinstance(steps, list):
append_event("error", f"작업 정의 형식 오류: {job_key} steps must be list")
return False
with job_lock:
if job_process and job_process.poll() is None:
append_event("warning", "이미 실행 중인 작업이 있어 새 작업을 시작하지 않습니다.")
return False
def worker():
global job_process
job_kind = job_kind_from_key_cmd(job_key, steps)
started_ts = time.time()
before_snapshot = publish_queue_snapshot("server") if job_kind == "server_publish" else None
append_event("info", f"작업 시작: {display_name} ({job_key})")
if before_snapshot:
append_event(
"info",
f"[{job_key}] 서버 발행 시작 전 pending={before_snapshot.get('pending')} / realtor={before_snapshot.get('realtors')}"
)
for idx, step in enumerate(steps, start=1):
name = step.get("name") or f"step {idx}"
cmd = step.get("cmd")
cwd = step.get("cwd") or cfg.get("project_root")
timeout_minutes = int(step.get("timeout_minutes") or step.get("max_run_minutes") or 0)
if not isinstance(cmd, list) or not cmd:
append_event("error", f"[{job_key}] {name} cmd 정의 오류. 이후 단계 중단")
job_process = None
return
append_event("info", f"[{job_key}] {idx}/{len(steps)} {name} 시작")
append_event("info", f"[{job_key}] CMD: {cmd_text(cmd)[:220]}")
log_path = LOG_DIR / f"job_{job_key}.log"
with open(log_path, "a", encoding="utf-8", errors="ignore") as log_file:
log_file.write(f"\n\n[{now_text()}] START {idx}/{len(steps)} {name}\n")
log_file.write("JOB: " + str(job_key) + " / " + str(display_name) + "\n")
log_file.write("CWD: " + str(cwd) + "\n")
log_file.write("CMD: " + cmd_text(cmd) + "\n")
try:
job_process = subprocess.Popen(
cmd,
cwd=cwd,
stdout=log_file,
stderr=subprocess.STDOUT,
creationflags=subprocess.CREATE_NEW_PROCESS_GROUP
)
if timeout_minutes > 0:
try:
code = job_process.wait(timeout=timeout_minutes * 60)
except subprocess.TimeoutExpired:
append_event("error", f"[{job_key}] {name} timeout {timeout_minutes}분 초과 → 종료")
try:
job_process.terminate()
time.sleep(2)
if job_process.poll() is None:
job_process.kill()
except Exception:
pass
code = 124
else:
code = job_process.wait()
except Exception as e:
append_event("error", f"[{job_key}] {name} 실행 예외: {e}")
error_log(traceback.format_exc())
job_process = None
return
log_file.write(f"[{now_text()}] END {name} code={code}\n")
if code != 0:
append_event("error", f"[{job_key}] {name} 실패 code={code}. 이후 단계 중단")
job_process = None
return
append_event("info", f"[{job_key}] {name} 완료")
elapsed = int(time.time() - started_ts)
if job_kind == "server_publish":
after_snapshot = publish_queue_snapshot("server")
before_pending = int((before_snapshot or {}).get("pending") or 0)
after_pending = int((after_snapshot or {}).get("pending") or 0)
if before_pending > 0 and after_pending >= before_pending:
append_event(
"error",
f"[{job_key}] 서버 발행 no-op 의심: 시작 전 pending={before_pending}, 종료 후 pending={after_pending}, elapsed={elapsed}s. publish_worker 옵션/대상조건 확인 필요"
)
elif before_pending > 0:
append_event(
"info",
f"[{job_key}] 서버 발행 큐 변화: pending {before_pending} → {after_pending}, elapsed={elapsed}s"
)
else:
append_event("warning", f"[{job_key}] 서버 발행 대상 pending=0 상태로 실행됨")
job_process = None
append_event("info", f"작업 완료: {display_name} ({job_key})")
threading.Thread(target=worker, daemon=True).start()
return True
def stop_job():
global job_process
with job_lock:
if job_process and job_process.poll() is None:
try:
job_process.terminate()
append_event("warning", f"작업 종료 요청 pid={job_process.pid}")
time.sleep(2)
if job_process.poll() is None:
job_process.kill()
append_event("warning", f"작업 강제 종료 pid={job_process.pid}")
job_process = None
return True
except Exception as e:
append_event("error", f"작업 종료 실패: {e}")
error_log(traceback.format_exc())
return False
append_event("info", "실행 중인 작업 없음")
return False
def find_python_process_by_keyword(keyword):
"""
Windows python.exe 프로세스 중 CommandLine에 keyword가 포함된 PID를 찾는다.
Server Manager가 직접 실행하지 않은 CMD 수동 실행 작업도 감지하기 위한 보조 함수.
"""
keyword = str(keyword or "").lower().replace("/", "\\")
if not keyword:
return None
ps_cmd = (
'powershell -NoProfile -Command '
'"Get-CimInstance Win32_Process | '
'Where-Object {$_.Name -eq ''python.exe''} | '
'Select-Object ProcessId,CommandLine | ConvertTo-Json -Compress"'
)
out, err, code = run_cmd_capture(ps_cmd)
if not out:
return None
try:
data = json.loads(out)
if isinstance(data, dict):
data = [data]
for item in data or []:
cmd = str(item.get("CommandLine") or "")
cmd_norm = cmd.lower().replace("/", "\\")
if keyword in cmd_norm:
return {
"pid": int(item.get("ProcessId") or 0),
"command_line": cmd,
}
except Exception:
return None
return None
def get_latest_running_pipeline():
"""
DB 기준 실행 중인 파이프라인 1건을 가져온다.
Server Manager가 직접 실행한 작업이 아니어도 관제센터 현재 배치에 표시한다.
"""
rows, err = db_query_all("""
SELECT
id,
run_type,
status,
started_at,
TIMESTAMPDIFF(MINUTE, started_at, NOW()) AS running_minutes,
message
FROM blog_pipeline_runs
WHERE status = 'running'
ORDER BY id DESC
LIMIT 1
""")
if err or not rows:
return None
return rows[0]
def get_running_python_for_run_type(run_type):
"""
run_type을 대표 스크립트명으로 변환하여 실행 중인 python.exe PID를 찾는다.
"""
run_type = str(run_type or "")
keyword_map = {
"publish_worker": "workers\\publish_worker.py",
"private_queue_worker": "workers\\private_queue_worker.py",
"schedule_daily_articles": "jobs\\schedule_daily_articles.py",
"fetch_article_detail": "naver_response_article_fetcher",
"generate_blog_drafts": "jobs\\generate_blog_drafts.py",
"daily_pipeline_runner": "jobs\\daily_pipeline_runner.py",
"check_removed_articles": "check_removed_articles",
}
keyword = keyword_map.get(run_type, run_type)
if not keyword:
return None
return find_python_process_by_keyword(keyword)
def current_job_status():
"""
현재 배치 상태.
우선순위:
1) server_manager가 직접 실행한 job_process
2) DB blog_pipeline_runs.status='running'
3) 실행 중인 python.exe CommandLine 보조 탐색
"""
if job_process and job_process.poll() is None:
return {
"running": True,
"pid": job_process.pid,
"source": "manager",
"label": "Manager 실행중",
"detail": f"pid {job_process.pid}",
}
running = get_latest_running_pipeline()
if running:
run_type = str(running.get("run_type") or "")
found = get_running_python_for_run_type(run_type)
pid = found.get("pid") if found else None
cmd = found.get("command_line") if found else ""
running_minutes = int(running.get("running_minutes") or 0)
# STEP113:
# DB에는 running으로 남아 있지만 실제 python.exe가 없으면 stale run으로 본다.
# 관제센터가 계속 "실행 중/장애"처럼 보이지 않도록 화면상 대기로 처리한다.
# DB 정리는 별도 recover_stale_pipeline_runs 또는 수동 SQL로 처리한다.
if not pid and running_minutes >= 10:
return {
"running": False,
"pid": None,
"source": "db_stale",
"run_id": running.get("id"),
"run_type": run_type,
"label": f"DB stale: {run_type or '-'}",
"detail": f"run_id {running.get('id')} / {running_minutes}분 / 실제 프로세스 없음",
"command_line": "",
}
return {
"running": True,
"pid": pid,
"source": "db",
"run_id": running.get("id"),
"run_type": run_type,
"label": f"DB 실행중: {run_type or '-'}",
"detail": f"run_id {running.get('id')} / {running_minutes}분",
"command_line": cmd or "",
}
return {
"running": False,
"pid": None,
"source": "",
"label": "대기",
"detail": "",
"command_line": "",
}
def write_status_file():
rows = all_health_rows()
data = {
"updated_at": now_text(),
"health_score": health_score(rows),
"health_rows": rows,
"services": service_status(),
"external_ports": external_port_status(),
"job": current_job_status(),
"scheduler": {
"last_tick": scheduler_state.get("last_tick"),
"next_runs": scheduler_state.get("next_runs", []),
"last_started": scheduler_state.get("last_started", {}),
"last_skipped": scheduler_state.get("last_skipped", ""),
},
"events": event_log[:30],
}
with open(STATUS_DIR / "server_status.json", "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
def html_escape(text):
return str(text or "").replace("&", "&").replace("<", "<").replace(">", ">")
def pct(done, total):
try:
total = int(total or 0)
done = int(done or 0)
if total <= 0:
return 0
return max(0, min(100, int(round((done / total) * 100))))
except Exception:
return 0
def db_query_one(sql, params=None):
cfg = load_config()
project_root = cfg.get("project_root") or ""
if project_root and project_root not in sys.path:
sys.path.append(project_root)
try:
from db import get_conn
except Exception as e:
return None, f"db import failed: {e}"
conn = None
try:
conn = get_conn()
with conn.cursor() as cur:
cur.execute(sql, params or [])
return cur.fetchone(), ""
except Exception as e:
return None, str(e)
finally:
try:
if conn:
conn.close()
except Exception:
pass
def db_query_all(sql, params=None):
cfg = load_config()
project_root = cfg.get("project_root") or ""
if project_root and project_root not in sys.path:
sys.path.append(project_root)
try:
from db import get_conn
except Exception as e:
return [], f"db import failed: {e}"
conn = None
try:
conn = get_conn()
with conn.cursor() as cur:
cur.execute(sql, params or [])
return cur.fetchall(), ""
except Exception as e:
return [], str(e)
finally:
try:
if conn:
conn.close()
except Exception:
pass
def db_execute(sql, params=None):
cfg = load_config()
project_root = cfg.get("project_root") or ""
if project_root and project_root not in sys.path:
sys.path.append(project_root)
try:
from db import get_conn
except Exception as e:
return 0, f"db import failed: {e}"
conn = None
try:
conn = get_conn()
with conn.cursor() as cur:
cur.execute(sql, params or [])
affected = cur.rowcount
conn.commit()
return affected, ""
except Exception as e:
try:
if conn:
conn.rollback()
except Exception:
pass
return 0, str(e)
finally:
try:
if conn:
conn.close()
except Exception:
pass
def reset_publish_queue_item(queue_id, mode):
try:
queue_id = int(queue_id)
except Exception:
return False, "queue_id invalid"
mode = str(mode or "").strip()
if mode == "failed_retry":
affected, err = db_execute("""
UPDATE blog_publish_queue
SET
queue_status = 'pending',
worker_status = 'manual_retry',
next_retry_at = NULL,
started_at = NULL,
finished_at = NULL,
error_message = CONCAT('[MANUAL RETRY FROM MANAGER] ', IFNULL(error_message, '')),
updated_at = NOW()
WHERE id = %s
AND queue_status = 'failed'
""", [queue_id])
elif mode == "hold_retry":
affected, err = db_execute("""
UPDATE blog_publish_queue
SET
queue_status = 'pending',
worker_status = 'manual_hold_release',
next_retry_at = NULL,
started_at = NULL,
finished_at = NULL,
error_message = CONCAT('[MANUAL HOLD RELEASE FROM MANAGER] ', IFNULL(error_message, '')),
updated_at = NOW()
WHERE id = %s
AND queue_status = 'hold'
AND COALESCE(worker_status,'') NOT IN ('auto_publish_disabled','daily_limit','manual_hold','session_hold')
AND COALESCE(last_error_type,'') NOT IN ('auto_publish_disabled','daily_limit','manual_hold','session_hold')
AND LOWER(COALESCE(error_message,'')) NOT LIKE '%auto_publish_disabled%'
AND LOWER(COALESCE(error_message,'')) NOT LIKE '%daily_post_limit=0%'
AND LOWER(COALESCE(error_message,'')) NOT LIKE '%자동발행 off%'
""", [queue_id])
elif mode == "stale_retry":
affected, err = db_execute("""
UPDATE blog_publish_queue
SET
queue_status = 'pending',
worker_status = 'manual_stale_reset',
started_at = NULL,
finished_at = NULL,
error_message = CONCAT('[MANUAL STALE RESET FROM MANAGER] ', IFNULL(error_message, '')),
updated_at = NOW()
WHERE id = %s
AND queue_status = 'processing'
""", [queue_id])
else:
return False, "unknown reset mode"
if err:
return False, err
return affected > 0, f"affected={affected}"
def get_today_pipeline_kpi():
"""
오늘 기준 운영 KPI.
DB 장애가 있어도 server_manager 자체는 죽지 않도록 모든 오류를 kpi_error로 반환한다.
"""
kpi = {
"ok": True,
"error": "",
"collect": {
"title": "오늘 후보수집",
"total": 0,
"done": 0,
"pending": 0,
"processing": 0,
"failed": 0,
"remain": 0,
"progress": 0,
"link": "daily_queue_status.php?date=today",
},
"detail": {
"title": "오늘 상세수집",
"total": 0,
"done": 0,
"pending": 0,
"processing": 0,
"failed": 0,
"remain": 0,
"progress": 0,
"link": "realtor_articles.php?date=today",
},
"draft": {
"title": "오늘 초안생성",
"total": 0,
"done": 0,
"pending": 0,
"processing": 0,
"failed": 0,
"remain": 0,
"progress": 0,
"link": "blog_drafts.php?date=today",
},
"publish": {
"title": "오늘 블로그발행",
"total": 0,
"done": 0,
"pending": 0,
"processing": 0,
"failed": 0,
"hold": 0,
"remain": 0,
"progress": 0,
"link": "publish_queue.php?date=today",
},
"sessions": {
"normal": 0,
"expired": 0,
"login_required": 0,
"write_blocked": 0,
"check_failed": 0,
"unknown": 0,
"risk_7": 0,
"risk_14": 0,
"risk_30": 0,
"problem": 0,
},
"session_problem_rows": [],
"session_risk_rows": [],
"publish_problem": {
"failed": 0,
"hold": 0,
"processing_stale": 0,
"retry_ready": 0,
"total_problem": 0,
},
"publish_problem_rows": [],
"client_publish": {
"pending": 0,
"processing": 0,
"published_today": 0,
"failed": 0,
"hold": 0,
"total_active": 0,
},
"client_publish_rows": [],
"ops_summary": {
"today_runs": 0,
"running_runs": 0,
"success_runs": 0,
"failed_runs": 0,
"warning_logs": 0,
"error_logs": 0,
"today_publish_total": 0,
"today_publish_done": 0,
"today_publish_failed": 0,
"today_publish_pending": 0,
},
"recent_runs": [],
"last_errors": [],
"recent_logs": [],
}
# 1) 후보수집 work queue
row, err = db_query_one("""
SELECT
COUNT(*) AS total,
SUM(CASE WHEN work_status IN ('done','completed','success') THEN 1 ELSE 0 END) AS done,
SUM(CASE WHEN work_status IN ('pending','waiting') THEN 1 ELSE 0 END) AS pending,
SUM(CASE WHEN work_status IN ('processing','running') THEN 1 ELSE 0 END) AS processing,
SUM(CASE WHEN work_status IN ('failed','error') THEN 1 ELSE 0 END) AS failed
FROM blog_article_work_queue
WHERE work_type = 'new_article'
AND DATE(created_at) = CURDATE()
""")
if err:
kpi["ok"] = False
kpi["error"] = err
elif row:
total = int(row.get("total") or 0)
done = int(row.get("done") or 0)
pending = int(row.get("pending") or 0)
processing = int(row.get("processing") or 0)
failed = int(row.get("failed") or 0)
kpi["collect"].update({
"total": total,
"done": done,
"pending": pending,
"processing": processing,
"failed": failed,
"remain": max(0, total - done - failed),
"progress": pct(done, total),
})
# 2) 상세수집 - 오늘 생성/수정된 매물 기준
row, err = db_query_one("""
SELECT
COUNT(*) AS total,
SUM(CASE WHEN detail_collected = 1 THEN 1 ELSE 0 END) AS done,
SUM(CASE WHEN detail_collected = 0 AND (detail_error IS NULL OR detail_error = '') THEN 1 ELSE 0 END) AS pending,
0 AS processing,
SUM(CASE WHEN detail_error IS NOT NULL AND detail_error <> '' THEN 1 ELSE 0 END) AS failed
FROM blog_realtor_articles
WHERE DATE(created_at) = CURDATE()
OR DATE(updated_at) = CURDATE()
""")
if not err and row:
total = int(row.get("total") or 0)
done = int(row.get("done") or 0)
pending = int(row.get("pending") or 0)
processing = int(row.get("processing") or 0)
failed = int(row.get("failed") or 0)
kpi["detail"].update({
"total": total,
"done": done,
"pending": pending,
"processing": processing,
"failed": failed,
"remain": max(0, total - done - failed),
"progress": pct(done, total),
})
# 3) 초안생성
row, err = db_query_one("""
SELECT
COUNT(*) AS total,
SUM(CASE WHEN d.id IS NOT NULL THEN 1 ELSE 0 END) AS done,
SUM(CASE WHEN d.id IS NULL AND a.detail_collected = 1 THEN 1 ELSE 0 END) AS pending,
0 AS processing,
0 AS failed
FROM blog_realtor_articles a
LEFT JOIN blog_article_drafts d
ON CONVERT(a.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci
= CONVERT(d.article_no USING utf8mb4) COLLATE utf8mb4_unicode_ci
WHERE DATE(a.created_at) = CURDATE()
OR DATE(a.updated_at) = CURDATE()
OR DATE(d.created_at) = CURDATE()
""")
if not err and row:
total = int(row.get("total") or 0)
done = int(row.get("done") or 0)
pending = int(row.get("pending") or 0)
processing = int(row.get("processing") or 0)
failed = int(row.get("failed") or 0)
kpi["draft"].update({
"total": total,
"done": done,
"pending": pending,
"processing": processing,
"failed": failed,
"remain": max(0, total - done - failed),
"progress": pct(done, total),
})
# 4) 발행
row, err = db_query_one("""
SELECT
COUNT(*) AS total,
SUM(CASE WHEN q.queue_status IN ('published','done','success') THEN 1 ELSE 0 END) AS done,
SUM(CASE WHEN q.queue_status IN ('pending','waiting') THEN 1 ELSE 0 END) AS pending,
SUM(CASE WHEN q.queue_status = 'processing' THEN 1 ELSE 0 END) AS processing,
SUM(CASE WHEN q.queue_status = 'failed' THEN 1 ELSE 0 END) AS failed,
SUM(CASE WHEN q.queue_status = 'hold' THEN 1 ELSE 0 END) AS hold
FROM blog_publish_queue q
WHERE DATE(q.created_at) = CURDATE()
OR DATE(q.updated_at) = CURDATE()
OR DATE(q.started_at) = CURDATE()
OR DATE(q.finished_at) = CURDATE()
""")
if not err and row:
total = int(row.get("total") or 0)
done = int(row.get("done") or 0)
pending = int(row.get("pending") or 0)
processing = int(row.get("processing") or 0)
failed = int(row.get("failed") or 0)
hold = int(row.get("hold") or 0)
kpi["publish"].update({
"total": total,
"done": done,
"pending": pending,
"processing": processing,
"failed": failed,
"hold": hold,
"remain": max(0, total - done - failed - hold),
"progress": pct(done, total),
})
# 5) 세션 상태 + 위험도
row, err = db_query_one("""
SELECT
SUM(CASE WHEN status IN ('login_ok') THEN 1 ELSE 0 END) AS normal,
SUM(CASE WHEN status = 'session_expired' THEN 1 ELSE 0 END) AS expired,
SUM(CASE WHEN status = 'login_required' THEN 1 ELSE 0 END) AS login_required,
SUM(CASE WHEN status = 'write_blocked' THEN 1 ELSE 0 END) AS write_blocked,
SUM(CASE WHEN status = 'check_failed' THEN 1 ELSE 0 END) AS check_failed,
SUM(CASE WHEN status = 'unknown' THEN 1 ELSE 0 END) AS unknown,
SUM(CASE
WHEN status = 'login_ok'
AND last_login_ok_at IS NOT NULL
AND last_login_ok_at < DATE_SUB(NOW(), INTERVAL 7 DAY)
THEN 1 ELSE 0 END
) AS risk_7,
SUM(CASE
WHEN status = 'login_ok'
AND last_login_ok_at IS NOT NULL
AND last_login_ok_at < DATE_SUB(NOW(), INTERVAL 14 DAY)
THEN 1 ELSE 0 END
) AS risk_14,
SUM(CASE
WHEN status = 'login_ok'
AND last_login_ok_at IS NOT NULL
AND last_login_ok_at < DATE_SUB(NOW(), INTERVAL 30 DAY)
THEN 1 ELSE 0 END
) AS risk_30
FROM blog_realtor_session_status
""")
if not err and row:
normal = int(row.get("normal") or 0)
expired = int(row.get("expired") or 0)
login_required = int(row.get("login_required") or 0)
write_blocked = int(row.get("write_blocked") or 0)
check_failed = int(row.get("check_failed") or 0)
unknown = int(row.get("unknown") or 0)
risk_7 = int(row.get("risk_7") or 0)
risk_14 = int(row.get("risk_14") or 0)
risk_30 = int(row.get("risk_30") or 0)
kpi["sessions"].update({
"normal": normal,
"expired": expired,
"login_required": login_required,
"write_blocked": write_blocked,
"check_failed": check_failed,
"unknown": unknown,
"risk_7": risk_7,
"risk_14": risk_14,
"risk_30": risk_30,
"problem": expired + login_required + write_blocked + check_failed,
})
# 5-1) 문제 세션 TOP20
rows, err = db_query_all("""
SELECT
s.realtor_id,
s.naver_id,
s.status,
s.last_checked_at,
s.last_login_ok_at,
s.last_error_message,
r.office_name,
r.representative_name,
r.office_phone,
r.mobile_phone
FROM blog_realtor_session_status s
LEFT JOIN blog_realtors r
ON r.id = s.realtor_id
WHERE s.status IN ('session_expired','login_required','write_blocked','check_failed')
ORDER BY
CASE s.status
WHEN 'write_blocked' THEN 1
WHEN 'login_required' THEN 2
WHEN 'session_expired' THEN 3
WHEN 'check_failed' THEN 4
ELSE 9
END,
s.last_checked_at DESC
LIMIT 20
""")
if not err:
kpi["session_problem_rows"] = rows
# 5-2) 정상이나 장기 미확인/장기 미로그인 위험 세션 TOP20
rows, err = db_query_all("""
SELECT
s.realtor_id,
s.naver_id,
s.status,
s.last_checked_at,
s.last_login_ok_at,
DATEDIFF(NOW(), s.last_login_ok_at) AS days_from_login_ok,
s.last_error_message,
r.office_name,
r.representative_name,
r.office_phone,
r.mobile_phone
FROM blog_realtor_session_status s
LEFT JOIN blog_realtors r
ON r.id = s.realtor_id
WHERE s.status = 'login_ok'
AND s.last_login_ok_at IS NOT NULL
AND s.last_login_ok_at < DATE_SUB(NOW(), INTERVAL 7 DAY)
ORDER BY s.last_login_ok_at ASC
LIMIT 20
""")
if not err:
kpi["session_risk_rows"] = rows
# 5-3) 발행 실패/보류/고착 큐 관제
row, err = db_query_one("""
SELECT
SUM(CASE WHEN q.queue_status = 'failed' THEN 1 ELSE 0 END) AS failed,
SUM(CASE
WHEN q.queue_status='hold'
AND COALESCE(q.worker_status,'') NOT IN ('auto_publish_disabled','daily_limit','manual_hold','session_hold')
AND COALESCE(q.last_error_type,'') NOT IN ('auto_publish_disabled','daily_limit','manual_hold','session_hold')
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%auto_publish_disabled%'
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%daily_post_limit=0%'
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%자동발행 off%'
AND COALESCE(ps.auto_publish_enabled,0)=1
AND COALESCE(ps.daily_post_limit,0)>0
AND COALESCE(na.login_status,'') NOT IN ('password_error','naver_block','disconnected')
THEN 1 ELSE 0 END) AS hold,
SUM(CASE
WHEN q.queue_status = 'processing'
AND q.started_at IS NOT NULL
AND q.started_at < DATE_SUB(NOW(), INTERVAL 90 MINUTE)
THEN 1 ELSE 0 END
) AS processing_stale,
SUM(CASE
WHEN q.queue_status = 'failed'
AND q.next_retry_at IS NOT NULL
AND q.next_retry_at <= NOW()
THEN 1 ELSE 0 END
) AS retry_ready
FROM blog_publish_queue q
LEFT JOIN blog_publish_settings ps
ON ps.realtor_id=q.realtor_id
LEFT JOIN blog_naver_accounts na
ON na.realtor_id=q.realtor_id
WHERE COALESCE(NULLIF(q.execution_target, ''), 'server') = 'server'
""")
if not err and row:
failed = int(row.get("failed") or 0)
hold = int(row.get("hold") or 0)
processing_stale = int(row.get("processing_stale") or 0)
retry_ready = int(row.get("retry_ready") or 0)
kpi["publish_problem"].update({
"failed": failed,
"hold": hold,
"processing_stale": processing_stale,
"retry_ready": retry_ready,
"total_problem": failed + hold + processing_stale,
})
rows, err = db_query_all("""
SELECT
q.id,
q.realtor_id,
q.draft_id,
q.article_no,
q.queue_status,
q.worker_status,
q.retry_count,
q.next_retry_at,
q.last_error_type,
q.error_message,
q.execution_target,
q.started_at,
q.finished_at,
q.updated_at,
r.office_name,
r.representative_name
FROM blog_publish_queue q
LEFT JOIN blog_realtors r
ON r.id = q.realtor_id
LEFT JOIN blog_publish_settings ps
ON ps.realtor_id=q.realtor_id
LEFT JOIN blog_naver_accounts na
ON na.realtor_id=q.realtor_id
WHERE COALESCE(NULLIF(q.execution_target, ''), 'server') = 'server'
AND (
q.queue_status='failed'
OR (
q.queue_status='hold'
AND COALESCE(q.worker_status,'') NOT IN ('auto_publish_disabled','daily_limit','manual_hold','session_hold')
AND COALESCE(q.last_error_type,'') NOT IN ('auto_publish_disabled','daily_limit','manual_hold','session_hold')
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%auto_publish_disabled%'
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%daily_post_limit=0%'
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%자동발행 off%'
AND COALESCE(ps.auto_publish_enabled,0)=1
AND COALESCE(ps.daily_post_limit,0)>0
AND COALESCE(na.login_status,'') NOT IN ('password_error','naver_block','disconnected')
)
OR (
q.queue_status='processing'
AND q.started_at IS NOT NULL
AND q.started_at < DATE_SUB(NOW(), INTERVAL 90 MINUTE)
)
)
ORDER BY
CASE
WHEN q.queue_status = 'processing' THEN 1
WHEN q.queue_status = 'hold' THEN 2
WHEN q.queue_status = 'failed' THEN 3
ELSE 9
END,
q.updated_at DESC
LIMIT 20
""")
if not err:
kpi["publish_problem_rows"] = filter_real_problem_rows(rows)
kpi["publish_problem"]["hold"] = sum(
1 for r in kpi["publish_problem_rows"]
if str(r.get("queue_status") or "") == "hold"
)
kpi["publish_problem"]["failed"] = sum(
1 for r in kpi["publish_problem_rows"]
if str(r.get("queue_status") or "") == "failed"
)
kpi["publish_problem"]["processing_stale"] = sum(
1 for r in kpi["publish_problem_rows"]
if str(r.get("queue_status") or "") == "processing"
)
kpi["publish_problem"]["total_problem"] = (
int(kpi["publish_problem"].get("failed") or 0)
+ int(kpi["publish_problem"].get("hold") or 0)
+ int(kpi["publish_problem"].get("processing_stale") or 0)
)
# 5-3-1) STEP113 클라이언트 발행 큐 관제
row, err = db_query_one("""
SELECT
SUM(CASE WHEN q.queue_status IN ('pending','waiting') THEN 1 ELSE 0 END) AS pending,
SUM(CASE WHEN q.queue_status = 'processing' THEN 1 ELSE 0 END) AS processing,
SUM(CASE
WHEN q.queue_status IN ('published','done','success')
AND q.finished_at >= CURDATE()
AND q.finished_at < DATE_ADD(CURDATE(), INTERVAL 1 DAY)
THEN 1 ELSE 0 END) AS published_today,
SUM(CASE WHEN q.queue_status = 'failed' THEN 1 ELSE 0 END) AS failed,
SUM(CASE
WHEN q.queue_status = 'hold'
AND COALESCE(q.worker_status,'') NOT IN ('auto_publish_disabled','daily_limit','manual_hold','session_hold')
AND COALESCE(q.last_error_type,'') NOT IN ('auto_publish_disabled','daily_limit','manual_hold','session_hold')
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%auto_publish_disabled%'
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%daily_post_limit=0%'
AND LOWER(COALESCE(q.error_message,'')) NOT LIKE '%자동발행 off%'
THEN 1 ELSE 0 END) AS hold
FROM blog_publish_queue q
WHERE q.execution_target = 'client'
""")
if not err and row:
pending = int(row.get("pending") or 0)
processing = int(row.get("processing") or 0)
published_today = int(row.get("published_today") or 0)
failed = int(row.get("failed") or 0)
hold = int(row.get("hold") or 0)
kpi["client_publish"].update({
"pending": pending,
"processing": processing,
"published_today": published_today,
"failed": failed,
"hold": hold,
"total_active": pending + processing + failed + hold,
})
rows, err = db_query_all("""
SELECT
q.id,
q.realtor_id,
q.draft_id,
q.article_no,
q.queue_status,
q.worker_status,
q.retry_count,
q.next_retry_at,
q.last_error_type,
q.error_message,
q.execution_target,
q.started_at,
q.finished_at,
q.updated_at,
r.office_name,
r.representative_name
FROM blog_publish_queue q
LEFT JOIN blog_realtors r
ON r.id = q.realtor_id
WHERE q.execution_target = 'client'
AND q.queue_status IN ('pending','processing','failed','hold')
ORDER BY
CASE
WHEN q.queue_status = 'processing' THEN 1
WHEN q.queue_status = 'pending' THEN 2
WHEN q.queue_status = 'hold' THEN 3
WHEN q.queue_status = 'failed' THEN 4
ELSE 9
END,
q.updated_at DESC
LIMIT 20
""")
if not err:
kpi["client_publish_rows"] = filter_real_problem_rows(rows)
kpi["client_publish"]["hold"] = sum(
1 for r in kpi["client_publish_rows"]
if str(r.get("queue_status") or "") == "hold"
)
kpi["client_publish"]["failed"] = sum(
1 for r in kpi["client_publish_rows"]
if str(r.get("queue_status") or "") == "failed"
)
kpi["client_publish"]["total_active"] = (
int(kpi["client_publish"].get("pending") or 0)
+ int(kpi["client_publish"].get("processing") or 0)
+ int(kpi["client_publish"].get("failed") or 0)
+ int(kpi["client_publish"].get("hold") or 0)
)
# 5-4) 오늘 운영 요약
row, err = db_query_one("""
SELECT
COUNT(*) AS today_runs,
SUM(CASE WHEN status = 'running' THEN 1 ELSE 0 END) AS running_runs,
SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) AS success_runs,
SUM(CASE WHEN status = 'failed' THEN 1 ELSE 0 END) AS failed_runs
FROM blog_pipeline_runs
WHERE DATE(created_at) = CURDATE()
OR DATE(started_at) = CURDATE()
OR DATE(finished_at) = CURDATE()
""")
if not err and row:
kpi["ops_summary"].update({
"today_runs": int(row.get("today_runs") or 0),
"running_runs": int(row.get("running_runs") or 0),
"success_runs": int(row.get("success_runs") or 0),
"failed_runs": int(row.get("failed_runs") or 0),
})
row, err = db_query_one("""
SELECT
SUM(CASE WHEN level = 'warning' THEN 1 ELSE 0 END) AS warning_logs,
SUM(CASE WHEN level = 'error' THEN 1 ELSE 0 END) AS error_logs
FROM blog_pipeline_logs
WHERE DATE(created_at) = CURDATE()
""")
if not err and row:
kpi["ops_summary"].update({
"warning_logs": int(row.get("warning_logs") or 0),
"error_logs": int(row.get("error_logs") or 0),
})
row, err = db_query_one("""
SELECT
COUNT(*) AS total,
SUM(CASE WHEN queue_status IN ('published','done','success') THEN 1 ELSE 0 END) AS done,
SUM(CASE WHEN queue_status = 'failed' THEN 1 ELSE 0 END) AS failed,
SUM(CASE WHEN queue_status IN ('pending','waiting') THEN 1 ELSE 0 END) AS pending
FROM blog_publish_queue
WHERE DATE(created_at) = CURDATE()
OR DATE(updated_at) = CURDATE()
OR DATE(started_at) = CURDATE()
OR DATE(finished_at) = CURDATE()
""")
if not err and row:
kpi["ops_summary"].update({
"today_publish_total": int(row.get("total") or 0),
"today_publish_done": int(row.get("done") or 0),
"today_publish_failed": int(row.get("failed") or 0),
"today_publish_pending": int(row.get("pending") or 0),
})
# 6) 최근 배치 실행
rows, err = db_query_all("""
SELECT run_type, status, started_at, finished_at, total_count, success_count, fail_count, message
FROM blog_pipeline_runs
ORDER BY id DESC
LIMIT 10
""")
if not err:
kpi["recent_runs"] = rows
# 7) 최근 오류
rows, err = db_query_all("""
SELECT level, step_name, article_no, realtor_id, message, created_at
FROM blog_pipeline_logs
WHERE level IN ('warning','error')
ORDER BY id DESC
LIMIT 10
""")
if not err:
kpi["last_errors"] = rows
# 8) 최근 통합 로그 20건
rows, err = db_query_all("""
SELECT
l.level,
l.step_name,
l.article_no,
l.realtor_id,
l.message,
l.created_at,
r.run_type,
r.status AS run_status
FROM blog_pipeline_logs l
LEFT JOIN blog_pipeline_runs r
ON r.id = l.run_id
ORDER BY l.id DESC
LIMIT 20
""")
if not err:
kpi["recent_logs"] = rows
return kpi
def run_type_label(value):
labels = {
"schedule_daily_articles": "후보수집(schedule_daily_articles)",
"fetch_article_detail": "상세수집(fetch_article_detail)",
"generate_blog_drafts": "초안생성(generate_blog_drafts)",
"publish_worker": "블로그발행(publish_worker)",
}
return labels.get(str(value or ""), str(value or ""))
def status_label(value):
labels = {
"pending": "대기",
"running": "실행중",
"success": "성공",
"failed": "실패",
"stopped": "중단",
"skipped": "건너뜀",
}
return labels.get(str(value or ""), str(value or ""))
def render_progress_card(item, color="#2563eb"):
total = int(item.get("total") or 0)
done = int(item.get("done") or 0)
pending = int(item.get("pending") or 0)
processing = int(item.get("processing") or 0)
failed = int(item.get("failed") or 0)
hold = int(item.get("hold") or 0)
remain = int(item.get("remain") or 0)
progress = int(item.get("progress") or 0)
title = html_escape(item.get("title") or "")
link = html_escape(item.get("link") or "#")
hold_html = ""
if hold:
hold_html = f'보류{hold}'
return f"""
"""
def session_status_label(value):
labels = {
"unknown": "미확인",
"login_ok": "정상",
"session_expired": "세션만료",
"login_required": "로그인필요",
"write_blocked": "글쓰기차단",
"check_failed": "확인실패",
}
return labels.get(str(value or ""), str(value or ""))
def render_session_table(rows, empty_text="표시할 계정이 없습니다."):
if not rows:
return f'{html_escape(empty_text)}
'
html = """
중개사
상태
최근확인
최근정상
오류
"""
for r in rows:
office = r.get("office_name") or f"realtor_id={r.get('realtor_id')}"
rep = r.get("representative_name") or ""
naver_id = r.get("naver_id") or ""
title = f"{office}"
if rep:
title += f" / {rep}"
if naver_id:
title += f" / {naver_id}"
html += f"""
{html_escape(title)}
{html_escape(session_status_label(r.get('status')))}
{html_escape(r.get('last_checked_at') or '-')}
{html_escape(r.get('last_login_ok_at') or '-')}
{html_escape(str(r.get('last_error_message') or '')[:80])}
"""
html += "
"
return html
def render_session_risk_table(rows):
if not rows:
return '7일 이상 장기 미확인 정상 세션이 없습니다.
'
html = """
중개사
네이버ID
마지막 정상
경과일
위험도
"""
for r in rows:
days = int(r.get("days_from_login_ok") or 0)
if days >= 30:
risk = "빨강"
elif days >= 14:
risk = "주황"
else:
risk = "노랑"
html += f"""
{html_escape(r.get('office_name') or f"realtor_id={r.get('realtor_id')}")}
{html_escape(r.get('naver_id') or '-')}
{html_escape(r.get('last_login_ok_at') or '-')}
{days}일
{risk}
"""
html += "
"
return html
def is_auto_policy_hold(row):
row = row or {}
status = str(row.get("queue_status") or "").lower()
if status != "hold":
return False
worker_status = str(row.get("worker_status") or "").lower()
last_error_type = str(row.get("last_error_type") or "").lower()
error_message = str(row.get("error_message") or "").lower()
text = " ".join([worker_status, last_error_type, error_message])
tokens = [
"auto_publish_disabled",
"[auto_publish_disabled]",
"daily_limit",
"오늘 1건",
"하루 1건",
"자동발행 off",
"daily_post_limit=0",
"정책보류",
"manual_hold",
"session_hold",
]
return any(token in text for token in tokens)
def filter_real_problem_rows(rows):
return [r for r in (rows or []) if not is_auto_policy_hold(r)]
def render_publish_problem_table(rows):
if not rows:
return '문제 발행 큐가 없습니다.
'
html = """
큐/매물
중개사
상태
재시도
최근오류
처리
"""
for r in rows:
qid = int(r.get("id") or 0)
status = str(r.get("queue_status") or "")
article_no = r.get("article_no") or "-"
office = r.get("office_name") or f"realtor_id={r.get('realtor_id')}"
err_type = r.get("last_error_type") or r.get("worker_status") or "-"
retry_count = int(r.get("retry_count") or 0)
error_message = str(r.get("error_message") or "")[:90]
if is_auto_policy_hold(r):
action = '
정책보류'
elif status == "failed":
action = f'
재시도'
elif status == "hold":
action = f'
보류복구'
elif status == "processing":
action = f'
고착복구'
else:
action = "-"
html += f"""
#{qid}
{html_escape(article_no)}
{html_escape(office)}
{html_escape(status_label(status))}
{html_escape(err_type)}
{retry_count}회
{html_escape(r.get('next_retry_at') or '')}
{html_escape(error_message)}
{action}
상세
"""
html += "
"
return html
def render_client_publish_table(rows):
if not rows:
return '클라이언트 발행 대기/문제 큐가 없습니다.
'
html = """
큐/매물
중개사
상태
재시도
최근오류
처리
"""
for r in rows:
qid = int(r.get("id") or 0)
status = str(r.get("queue_status") or "")
article_no = r.get("article_no") or "-"
office = r.get("office_name") or f"realtor_id={r.get('realtor_id')}"
err_type = r.get("last_error_type") or r.get("worker_status") or "-"
retry_count = int(r.get("retry_count") or 0)
error_message = str(r.get("error_message") or "")[:90]
if is_auto_policy_hold(r):
action = '
정책보류'
elif status == "pending":
action = '
매물HOME 대기'
elif status == "processing":
action = '
PC 발행중'
elif status == "failed":
action = f'
확인'
elif status == "hold":
action = f'
보류확인'
else:
action = "-"
html += f"""
#{qid}
{html_escape(article_no)}
{html_escape(office)}
{html_escape(status_label(status))}
{html_escape(err_type)}
{retry_count}회
{html_escape(r.get('next_retry_at') or '')}
{html_escape(error_message)}
{action}
상세
"""
html += "
"
return html
def render_recent_log_table(rows):
if not rows:
return '최근 파이프라인 로그가 없습니다.
'
html = """
시간
단계
레벨
매물
메시지
"""
for r in rows:
level = str(r.get("level") or "")
level_class = "lv-error" if level == "error" else ("lv-warning" if level == "warning" else "lv-info")
step = r.get("step_name") or run_type_label(r.get("run_type"))
msg = str(r.get("message") or "")[:120]
html += f"""
{html_escape(r.get('created_at') or '-')}
{html_escape(step or '-')}
{html_escape(level or '-')}
{html_escape(r.get('article_no') or '-')}
{html_escape(msg)}
"""
html += "
"
return html
def render_scheduler_table(rows):
if not rows:
return '등록된 자동 스케줄이 없습니다.
'
html = """
작업
예약시간
다음 실행
제한
실행명령
"""
for r in rows:
kind_label = {"server_publish": "서버발행", "collect_draft": "수집/초안", "unknown": "미확인"}.get(r.get("kind"), r.get("kind") or "-")
html += f"""
{html_escape(r.get('name') or r.get('key'))}
{html_escape(r.get('key'))} / {html_escape(kind_label)} / step {int(r.get('steps') or 0)}
{html_escape(', '.join(r.get('times') or []))}
{html_escape(r.get('next_run') or '-')}
{int(r.get('max_run_minutes') or 0)}분
{html_escape(r.get('cmd_preview') or '-')}
"""
html += "
"
if scheduler_state.get("last_skipped"):
html += f'최근 SKIP: {html_escape(scheduler_state.get("last_skipped"))}
'
return html
def render_dashboard():
cfg = load_config()
ext_rows = external_port_status()
svc_rows = service_status()
rows = ext_rows + svc_rows
score = health_score(rows)
job = current_job_status()
kpi = get_today_pipeline_kpi()
score_color = "#047857" if score >= 90 else ("#d97706" if score >= 70 else "#dc2626")
sess = kpi.get("sessions") or {}
publish_problem = kpi.get("publish_problem") or {}
client_publish = kpi.get("client_publish") or {}
session_problem_count = int(sess.get("problem") or 0)
publish_problem_count = int(publish_problem.get("total_problem") or 0)
overall_state = "정상"
overall_icon = "🟢"
overall_note = "주요 서비스 정상"
if score < 70 or session_problem_count >= 10 or publish_problem_count >= 10:
overall_state = "장애"
overall_icon = "🔴"
overall_note = "즉시 확인 필요"
elif score < 90 or session_problem_count > 0 or publish_problem_count > 0:
overall_state = "주의"
overall_icon = "🟡"
overall_note = "일부 확인 필요"
infra_cards = ""
for r in ext_rows:
ok = r.get("health_ok") or r.get("port_ok")
dot = "🟢" if ok else "🔴"
state = "정상" if ok else "중지"
infra_cards += f"""
{html_escape(r.get('name'))}
{dot} {state}
port {r.get('port')}
{r.get('health_ms', 0)}ms
"""
service_cards = ""
for svc in svc_rows:
ok = svc.get("health_ok") or svc.get("port_ok")
dot = "🟢" if ok else "🔴"
badge = "정상" if ok else "중지"
owner = svc.get("owner_label") or "-"
cmd_line = svc.get("command_line") or ""
short_cmd = cmd_line
if len(short_cmd) > 80:
short_cmd = short_cmd[:80] + "..."
service_cards += f"""
{html_escape(svc.get('name'))}
{dot} {badge}
PID {svc.get('pid') or '-'} / {html_escape(owner)}
{html_escape(short_cmd or '-')}
port {svc.get('port')}
{svc.get('health_ms', 0)}ms
"""
kpi_warning = ""
if not kpi.get("ok"):
kpi_warning = f"""
DB KPI 조회 경고
⚠ 일부 조회 실패
{html_escape(kpi.get('error'))}
"""
pipeline_cards = (
render_progress_card(kpi["collect"], "#047857") +
render_progress_card(kpi["detail"], "#2563eb") +
render_progress_card(kpi["draft"], "#7c3aed") +
render_progress_card(kpi["publish"], "#dc2626")
)
job_badge = "🟢 실행 중" if job.get("running") else "⚪ 대기"
job_pid = job.get("pid") or "-"
job_detail = job.get("label") or ""
if job.get("detail"):
job_detail = f"{job_detail}
{html_escape(job.get('detail'))}"
schedule_plan = build_schedule_plan(cfg)
config_issues = validate_runtime_config(cfg)
next_schedule = schedule_plan[0] if schedule_plan else {}
next_schedule_name = next_schedule.get("name") or "-"
next_schedule_time = next_schedule.get("next_run") or "-"
last_tick = scheduler_state.get("last_tick") or "-"
server_q = publish_queue_snapshot("server")
client_q = publish_queue_snapshot("client")
server_pending_text = f"server pending {server_q.get('pending', 0)} / 중개사 {server_q.get('realtors', 0)}"
client_pending_text = f"client pending {client_q.get('pending', 0)} / 중개사 {client_q.get('realtors', 0)}"
top_cards = f"""
전체 상태
{overall_icon} {overall_state}
{html_escape(overall_note)}
서버 건강도
{score}점
Apache / FTP / API
현재 배치
{job_badge}
pid {job_pid}
{job_detail}
{html_escape(server_pending_text)}
다음 예약
{html_escape(next_schedule_name)}
{html_escape(next_schedule_time)}
tick {html_escape(last_tick)}
{html_escape(client_pending_text)}
"""
ops = kpi.get("ops_summary") or {}
ops_summary_card = f"""
오늘 배치 실행 요약
실행 {int(ops.get('today_runs') or 0)}건
관리자
"""
today_publish_card = f"""
오늘 발행 요약
{int(ops.get('today_publish_done') or 0)} / {int(ops.get('today_publish_total') or 0)}
발행큐
"""
session_card = f"""
네이버 세션 관제
문제 {session_problem_count}건
목록
세션 위험도
"""
publish_problem_card = f"""
서버 발행 문제 관제
문제 {publish_problem_count}건
큐
"""
client_publish_card = f"""
클라이언트 발행 관제
대기 {int(client_publish.get('pending') or 0)}건
클라이언트 큐
"""
recent_runs_html = ""
for r in kpi.get("recent_runs") or []:
recent_runs_html += f"""
{html_escape(r.get('started_at'))}
{html_escape(run_type_label(r.get('run_type')))}
{html_escape(status_label(r.get('status')))} / 성공 {int(r.get('success_count') or 0)} / 실패 {int(r.get('fail_count') or 0)}
"""
last_errors_html = ""
for r in kpi.get("last_errors") or []:
last_errors_html += f"""
{html_escape(r.get('created_at'))}
{html_escape(r.get('level'))}
{html_escape(r.get('step_name'))} / {html_escape(r.get('article_no'))} / {html_escape(r.get('message'))}
"""
events_html = ""
for ev in event_log[:20]:
events_html += f"""
{html_escape(ev.get('time'))}
{html_escape(ev.get('level'))}
{html_escape(ev.get('message'))}
"""
recent_logs_html = render_recent_log_table(kpi.get("recent_logs") or [])
return f"""
Server Manager
{top_cards}
자동 스케줄러
예약 실행 상태
Server Manager 내부 스케줄러가 주간 발행/야간 수집·초안을 자동 실행합니다.
{render_scheduler_table(schedule_plan)}
스케줄/작업 설정 진단
작업명, 실제 명령, 발행/수집 매핑이 꼬였는지 확인합니다.
{render_config_issue_table(config_issues)}
{ops_summary_card}
{today_publish_card}
API 서비스 상태
{service_cards}
{kpi_warning}
오늘 파이프라인
{pipeline_cards}
{session_card}
{publish_problem_card}
{client_publish_card}
문제 세션 TOP20
{render_session_table(kpi.get('session_problem_rows') or [], '문제 세션이 없습니다.')}
서버 문제 발행 큐 TOP20
{render_publish_problem_table(kpi.get('publish_problem_rows') or [])}
클라이언트 발행 큐 TOP20
{render_client_publish_table(kpi.get('client_publish_rows') or [])}
최근 파이프라인 로그 20건
배치 실행 로그 센터
{recent_logs_html}
최근 배치 실행
{recent_runs_html or '
최근 실행 기록 없음
'}
최근 파이프라인 경고/오류
{last_errors_html or '
최근 오류 없음
'}
Server Manager 이벤트
{events_html or '
이벤트 없음
'}
"""
class ManagerHandler(BaseHTTPRequestHandler):
def do_GET(self):
parsed = urlparse(self.path)
if parsed.path == "/ping":
payload = b"OK"
self.send_response(200)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
return
if parsed.path == "/":
html = render_dashboard().encode("utf-8")
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(html)))
self.end_headers()
try:
self.wfile.write(html)
except (ConnectionAbortedError, BrokenPipeError):
pass
return
if parsed.path == "/status.json":
write_status_file()
payload = (STATUS_DIR / "server_status.json").read_bytes()
self.send_response(200)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
return
if parsed.path == "/action":
qs = parse_qs(parsed.query)
action_type = (qs.get("type") or [""])[0]
key = (qs.get("key") or [""])[0]
if action_type == "start_service":
start_service(key)
elif action_type == "stop_service":
stop_service(key)
elif action_type == "force_stop_service":
ok, msg = force_stop_service_by_key(key)
append_event("warning" if ok else "error", f"서비스 강제종료 요청 key={key}, result={msg}")
elif action_type == "restart_service":
ok, msg = restart_service(key)
append_event("info" if ok else "error", f"서비스 재시작 요청 key={key}, result={msg}")
elif action_type == "run_job":
run_job(key)
elif action_type == "stop_job":
stop_job()
elif action_type == "reset_publish_queue":
queue_id = (qs.get("id") or [""])[0]
mode = (qs.get("mode") or [""])[0]
ok, msg = reset_publish_queue_item(queue_id, mode)
append_event("info" if ok else "error", f"발행 큐 복구 요청 queue_id={queue_id}, mode={mode}, result={msg}")
else:
append_event("warning", f"알 수 없는 action: {action_type}")
self.send_response(302)
self.send_header("Location", "/")
self.end_headers()
return
self.send_response(404)
self.end_headers()
def log_message(self, fmt, *args):
return
def main():
try:
cfg = load_config()
console("=" * 70)
console("Honghee Server Manager v2.5 BATCH LOG CENTER")
console(f"BASE_DIR = {BASE_DIR}")
console(f"CONFIG = {CONFIG_PATH}")
console("=" * 70)
append_event("info", "Server Manager started")
append_event("info", "STEP147 policy hold filter active")
for svc in cfg.get("services", []):
if svc.get("auto_start"):
start_service(svc.get("key"))
threading.Thread(target=monitor_loop, daemon=True).start()
threading.Thread(target=scheduler_loop, daemon=True).start()
host = cfg.get("manager_host") or "0.0.0.0"
port = int(cfg.get("manager_port") or 9200)
append_event("info", f"Dashboard binding host={host}, port={port}")
try:
server = ThreadingHTTPServer((host, port), ManagerHandler)
except Exception:
err = traceback.format_exc()
console(err)
error_log(err)
input("HTTPServer 시작 실패. Enter를 누르면 종료합니다...")
return
append_event("info", f"Dashboard listening http://127.0.0.1:{port}")
if bool(cfg.get("open_browser_on_start", False)):
try:
webbrowser.open(f"http://127.0.0.1:{port}")
except Exception:
pass
server.serve_forever()
except KeyboardInterrupt:
append_event("warning", "Server Manager stopped by Ctrl+C")
except Exception:
err = traceback.format_exc()
console(err)
error_log(err)
input("Server Manager 오류 발생. Enter를 누르면 종료합니다...")
if __name__ == "__main__":
main()