웹 채팅(chat.hyowons.net) 추가 — 클로·레이·골디와 대화
- workspace/scripts/chat_web.py: 표준 라이브러리 단일 서버(포트 18793, launchd ai.openclaw.chat-web) - 로그인: iMessage OTP 단독 + 30일 세션, 코드 발송 1분 1회·1시간 5회 제한, 실패 시 잠금+텔레그램 알림 - 에이전트 호출은 openclaw agent --session-key web… (텔레그램과 별도 맥락, 답은 웹에만) - /clear(웹 전용), 선택지 버튼([[버튼: A | B]] 지정 + 문장 추출 폴백), 홈 화면 아이콘 - CLAUDE.md·TASKS.md(헬스체크 daemon 등록)·클로 MEMORY.md 문서화 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
hyowons
@@ -26,6 +26,7 @@
|
||||
| 이름 | 일정 (Asia/Seoul) | plist / 스크립트 |
|
||||
|---|---|---|
|
||||
| Gmail 라벨 분류/정리 | 매일 01:00 | `ai.openclaw.gmail-label-classify` → `scripts/gmail_label_classify.py` (`[오전/오후 브리핑]` → `브리핑`, `[비하이브 종목분석]` → `종목분석`, `[주식 리포트]` → `주식브리핑`, 24시간 지난 테스트메일 휴지통). `gog` CLI 직접 호출. 2026-06-26 cron→launchd 이관(본문 생성 없는 단순 스크립트). 보고 없음 — 로그 `logs/gmail-label-classify.{log,err.log}` + `state/gmail_label_classify.log` |
|
||||
| 웹 채팅 서버 | 상시 | `ai.openclaw.chat-web` → `scripts/chat_web.py serve` (2026-10-07 도입). `chat.hyowons.net` 에서 관리자님이 클로·레이·골디와 대화하는 자체 웹. 내게 오는 웹 메시지는 세션 키 `web…`(텔레그램과 별도 맥락)이고 답은 웹에만 남는다(텔레그램 발송 X). 기록 `state/chat_web/history/`, 로그 `logs/chat-web.{log,err.log}` |
|
||||
|
||||
### 원칙
|
||||
|
||||
|
||||
@@ -0,0 +1,673 @@
|
||||
#!/usr/bin/env python3
|
||||
"""클로·레이·골디 웹 채팅 (chat.hyowons.net).
|
||||
|
||||
텔레그램처럼 세 에이전트와 대화하는 자체 웹. Tailscale 없이 NAS 리버스 프록시로 공개되므로
|
||||
로그인은 본인 iMessage OTP(iOS 자동입력) → 30일 세션 쿠키. 비밀번호는 없다(관리자님 결정) — 대신 누구나
|
||||
'코드 받기'를 누를 수 있으므로 발송 횟수를 제한한다(1분 1회, 1시간 5회).
|
||||
|
||||
에이전트 호출은 `openclaw agent --agent <id> --session-key web… --json` (게이트웨이 경유, 게이트웨이 자체는
|
||||
외부에 열지 않는다). 세션 키가 텔레그램과 달라 웹 대화는 웹 전용 맥락이다. 응답은 텔레그램으로 나가지 않는다
|
||||
(`--deliver` 미지정).
|
||||
|
||||
CLI:
|
||||
python3 chat_web.py serve [--port 18793]
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import hmac
|
||||
import html
|
||||
import json
|
||||
import os
|
||||
import secrets
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
import urllib.parse
|
||||
from datetime import datetime
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
from pathlib import Path
|
||||
from zoneinfo import ZoneInfo
|
||||
|
||||
KST = ZoneInfo('Asia/Seoul')
|
||||
ROOT = Path.home() / '.openclaw'
|
||||
STATE = ROOT / 'workspace' / 'state' / 'chat_web'
|
||||
HIST_DIR = STATE / 'history'
|
||||
AUTH_FILE = STATE / 'auth.json'
|
||||
KEYS_FILE = STATE / 'session_keys.json'
|
||||
IMSG_CRED = ROOT / 'credentials' / 'admin_imessage.json'
|
||||
ICON_FILE = Path(__file__).with_name('chat_web_icon.png')
|
||||
OPENCLAW = str(Path.home() / '.local' / 'bin' / 'openclaw')
|
||||
|
||||
DOMAIN = 'chat.hyowons.net'
|
||||
DEFAULT_PORT = 18793
|
||||
|
||||
AGENTS = [('main', '클로', '🦞'), ('stock', '레이', ''), ('budget', '골디', '📒')]
|
||||
AGENT_IDS = {a[0] for a in AGENTS}
|
||||
|
||||
SESSION_TTL = 30 * 86400
|
||||
OTP_TTL = 300
|
||||
OTP_MAX_TRIES = 5
|
||||
FAIL_WINDOW = 900 # 15분 안에
|
||||
FAIL_LIMIT = 5 # 5번 틀리면
|
||||
LOCK_SEC = 900 # 15분 잠금
|
||||
SEND_GAP = 60 # 코드 발송 최소 간격
|
||||
SEND_HOURLY = 5 # 1시간 발송 상한 (넘으면 잠금 — 폰으로 코드 폭탄 방지)
|
||||
MAX_MSG = 8000
|
||||
RUN_TIMEOUT = 600
|
||||
HISTORY_LIMIT = 300
|
||||
# 웹 메시지에만 덧붙이는 형식 안내 — 에이전트가 선택지를 지정하면 웹이 버튼으로 그린다.
|
||||
# AGENTS.md 에 넣지 않는 이유: 텔레그램엔 버튼이 없어 마커가 날것으로 보인다. 화면 기록엔 저장하지 않는다.
|
||||
WEB_HINT = ('\n\n(웹 채팅 안내: 관리자님이 골라서 답할 질문이면 답변 맨 마지막 줄에 '
|
||||
'[[버튼: 선택1 | 선택2]] 형식으로 선택지를 적어주세요. 고를 것이 없으면 쓰지 마세요.)')
|
||||
|
||||
_auth_lock = threading.Lock()
|
||||
_otp: dict[str, dict] = {} # pre 토큰 → {otp, exp, tries} (메모리 — 재시작 시 재로그인)
|
||||
_pending: dict[str, float] = {} # agent → 시작 시각
|
||||
_run_locks = {a: threading.Lock() for a in AGENT_IDS}
|
||||
_hist_lock = threading.Lock()
|
||||
|
||||
|
||||
def now_kst() -> datetime:
|
||||
return datetime.now(KST)
|
||||
|
||||
|
||||
def log(msg: str) -> None:
|
||||
print(f'[{now_kst():%Y-%m-%d %H:%M:%S}] {msg}', flush=True)
|
||||
|
||||
|
||||
def _read_json(p: Path, default):
|
||||
try:
|
||||
return json.loads(p.read_text(encoding='utf-8'))
|
||||
except Exception:
|
||||
return default
|
||||
|
||||
|
||||
def _write_json(p: Path, data) -> None:
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = p.with_suffix(p.suffix + '.tmp')
|
||||
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding='utf-8')
|
||||
os.replace(tmp, p)
|
||||
|
||||
|
||||
# ---------------- 인증 상태 ----------------
|
||||
|
||||
def _sha(s: str) -> str:
|
||||
return hashlib.sha256(s.encode()).hexdigest()
|
||||
|
||||
|
||||
def _load_auth() -> dict:
|
||||
a = _read_json(AUTH_FILE, {})
|
||||
a.setdefault('sessions', {})
|
||||
a.setdefault('fails', [])
|
||||
a.setdefault('locked_until', 0)
|
||||
a.setdefault('sends', [])
|
||||
return a
|
||||
|
||||
|
||||
def _take_send_slot() -> str | None:
|
||||
"""코드 발송 가능하면 기록하고 None, 아니면 사유. 상한 초과는 잠금으로 이어진다."""
|
||||
t = time.time()
|
||||
alert = False
|
||||
with _auth_lock:
|
||||
a = _load_auth()
|
||||
a['sends'] = [s for s in a['sends'] if t - s < 3600]
|
||||
if a['sends'] and t - a['sends'][-1] < SEND_GAP:
|
||||
return f'방금 코드를 보냈습니다. {int(SEND_GAP - (t - a["sends"][-1])) + 1}초 뒤 다시 받을 수 있습니다.'
|
||||
if len(a['sends']) >= SEND_HOURLY:
|
||||
a['locked_until'] = t + LOCK_SEC
|
||||
a['sends'] = []
|
||||
_write_json(AUTH_FILE, a)
|
||||
alert = True
|
||||
else:
|
||||
a['sends'].append(t)
|
||||
_write_json(AUTH_FILE, a)
|
||||
if alert:
|
||||
log('LOCKED — 코드 요청 과다')
|
||||
_alert(f'🔒 chat.hyowons.net 인증 코드 요청이 1시간에 {SEND_HOURLY}회를 넘어 {LOCK_SEC // 60}분 잠갔습니다.')
|
||||
return f'요청이 너무 많아 {LOCK_SEC // 60}분 잠갔습니다.'
|
||||
return None
|
||||
|
||||
|
||||
def _locked() -> float:
|
||||
"""잠겨 있으면 남은 초, 아니면 0."""
|
||||
with _auth_lock:
|
||||
left = _load_auth()['locked_until'] - time.time()
|
||||
return max(0.0, left)
|
||||
|
||||
|
||||
def _record_fail(ip: str, why: str) -> None:
|
||||
t = time.time()
|
||||
alert = False
|
||||
with _auth_lock:
|
||||
a = _load_auth()
|
||||
a['fails'] = [f for f in a['fails'] if t - f < FAIL_WINDOW] + [t]
|
||||
if len(a['fails']) >= FAIL_LIMIT and a['locked_until'] < t:
|
||||
a['locked_until'] = t + LOCK_SEC
|
||||
a['fails'] = []
|
||||
alert = True
|
||||
_write_json(AUTH_FILE, a)
|
||||
log(f'login fail ({why}) ip={ip}')
|
||||
if alert:
|
||||
log(f'LOCKED {LOCK_SEC}s')
|
||||
_alert(f'🔒 chat.hyowons.net 로그인 {FAIL_LIMIT}회 실패로 {LOCK_SEC // 60}분 잠갔습니다.\n'
|
||||
f'마지막 시도 IP: {ip} ({why})')
|
||||
|
||||
|
||||
def _alert(text: str) -> None:
|
||||
try:
|
||||
sys.path.insert(0, str(ROOT / 'workspace' / 'scripts'))
|
||||
from codex_fallback_monitor import _send_telegram
|
||||
_send_telegram(text)
|
||||
except Exception as e:
|
||||
log(f'alert 실패: {e}')
|
||||
|
||||
|
||||
def _new_session(ip: str, ua: str) -> str:
|
||||
tok = secrets.token_urlsafe(32)
|
||||
t = time.time()
|
||||
with _auth_lock:
|
||||
a = _load_auth()
|
||||
a['sessions'] = {k: v for k, v in a['sessions'].items() if v.get('exp', 0) > t}
|
||||
a['sessions'][_sha(tok)] = {'exp': t + SESSION_TTL, 'created': now_kst().isoformat(timespec='seconds'),
|
||||
'ip': ip, 'ua': ua[:120]}
|
||||
a['fails'] = []
|
||||
_write_json(AUTH_FILE, a)
|
||||
return tok
|
||||
|
||||
|
||||
def _valid_session(tok: str | None) -> bool:
|
||||
if not tok:
|
||||
return False
|
||||
with _auth_lock:
|
||||
s = _load_auth()['sessions'].get(_sha(tok))
|
||||
return bool(s) and s.get('exp', 0) > time.time()
|
||||
|
||||
|
||||
def _drop_session(tok: str | None) -> None:
|
||||
if not tok:
|
||||
return
|
||||
with _auth_lock:
|
||||
a = _load_auth()
|
||||
a['sessions'].pop(_sha(tok), None)
|
||||
_write_json(AUTH_FILE, a)
|
||||
|
||||
|
||||
def _send_otp(code: str) -> bool:
|
||||
handle = (_read_json(IMSG_CRED, {}) or {}).get('handle')
|
||||
if not handle:
|
||||
log('admin_imessage.json 없음 — OTP 발송 불가')
|
||||
return False
|
||||
text = f'[{DOMAIN}] 채팅 로그인 인증\n코드: {code}\n\n@{DOMAIN} #{code}'
|
||||
try:
|
||||
# 나→나 iMessage 는 항상 imessage 서비스 강제(SMS 셀프발송은 자동입력이 깨진다)
|
||||
subprocess.Popen(['imsg', 'send', '--to', handle, '--text', text, '--service', 'imessage'],
|
||||
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, start_new_session=True)
|
||||
return True
|
||||
except Exception as e:
|
||||
log(f'imsg 발송 실패: {e}')
|
||||
return False
|
||||
|
||||
|
||||
# ---------------- 대화 기록·에이전트 호출 ----------------
|
||||
|
||||
def _session_key(agent: str) -> str:
|
||||
return (_read_json(KEYS_FILE, {}) or {}).get(agent, 'web')
|
||||
|
||||
|
||||
def _rotate_session(agent: str) -> str:
|
||||
keys = _read_json(KEYS_FILE, {}) or {}
|
||||
key = f'web-{now_kst():%Y%m%d%H%M%S}'
|
||||
keys[agent] = key
|
||||
_write_json(KEYS_FILE, keys)
|
||||
_append(agent, 'system', '새 대화를 시작했습니다')
|
||||
return key
|
||||
|
||||
|
||||
def _clear(agent: str) -> None:
|
||||
"""/clear — 화면 기록을 archive/ 로 옮기고(삭제 X) 새 세션으로 시작."""
|
||||
p = HIST_DIR / f'{agent}.jsonl'
|
||||
if p.exists():
|
||||
arc = HIST_DIR / 'archive'
|
||||
arc.mkdir(parents=True, exist_ok=True)
|
||||
with _hist_lock:
|
||||
os.replace(p, arc / f'{agent}-{now_kst():%Y%m%d%H%M%S}.jsonl')
|
||||
_rotate_session(agent)
|
||||
|
||||
|
||||
def _append(agent: str, role: str, text: str) -> dict:
|
||||
item = {'id': f'{time.time_ns()}', 'role': role, 'text': text,
|
||||
'ts': now_kst().isoformat(timespec='seconds')}
|
||||
with _hist_lock:
|
||||
HIST_DIR.mkdir(parents=True, exist_ok=True)
|
||||
with open(HIST_DIR / f'{agent}.jsonl', 'a', encoding='utf-8') as f:
|
||||
f.write(json.dumps(item, ensure_ascii=False) + '\n')
|
||||
return item
|
||||
|
||||
|
||||
def _history(agent: str) -> list[dict]:
|
||||
p = HIST_DIR / f'{agent}.jsonl'
|
||||
if not p.exists():
|
||||
return []
|
||||
with _hist_lock:
|
||||
lines = p.read_text(encoding='utf-8').splitlines()[-HISTORY_LIMIT:]
|
||||
out = []
|
||||
for ln in lines:
|
||||
try:
|
||||
out.append(json.loads(ln))
|
||||
except Exception:
|
||||
pass
|
||||
return out
|
||||
|
||||
|
||||
def _parse_reply(stdout: str) -> tuple[str, str | None]:
|
||||
"""(본문, 오류). JSON 앞에 다른 출력이 섞여도 첫 '{' 부터 읽는다."""
|
||||
i = stdout.find('{')
|
||||
if i < 0:
|
||||
return '', '응답을 해석할 수 없습니다'
|
||||
try:
|
||||
d = json.loads(stdout[i:])
|
||||
except Exception:
|
||||
return '', '응답을 해석할 수 없습니다'
|
||||
if d.get('status') != 'ok':
|
||||
return '', f"실행 실패: {d.get('summary') or d.get('status')}"
|
||||
parts = []
|
||||
for p in (d.get('result') or {}).get('payloads') or []:
|
||||
if p.get('text'):
|
||||
parts.append(p['text'])
|
||||
if p.get('mediaUrl'):
|
||||
parts.append(p['mediaUrl'])
|
||||
return '\n\n'.join(parts) or '(빈 응답)', None
|
||||
|
||||
|
||||
def _run_agent(agent: str, text: str) -> None:
|
||||
with _run_locks[agent]:
|
||||
_pending[agent] = time.time()
|
||||
msg_file = STATE / 'tmp' / f'{agent}-{time.time_ns()}.txt'
|
||||
try:
|
||||
msg_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
msg_file.write_text(text if text.startswith('/') else text + WEB_HINT, encoding='utf-8') # 슬래시 명령은 본체가 처리하므로 그대로
|
||||
r = subprocess.run(
|
||||
[OPENCLAW, 'agent', '--agent', agent, '--session-key', _session_key(agent),
|
||||
'--message-file', str(msg_file), '--json', '--timeout', str(RUN_TIMEOUT)],
|
||||
capture_output=True, text=True, timeout=RUN_TIMEOUT + 60, cwd=str(ROOT))
|
||||
body, err = _parse_reply(r.stdout)
|
||||
if err:
|
||||
log(f'{agent} 실패 rc={r.returncode} err={r.stderr[-500:]!r}')
|
||||
_append(agent, 'error', err)
|
||||
else:
|
||||
_append(agent, 'agent', body)
|
||||
except subprocess.TimeoutExpired:
|
||||
_append(agent, 'error', f'{RUN_TIMEOUT // 60}분 안에 응답이 없어 중단했습니다')
|
||||
except Exception as e:
|
||||
log(f'{agent} 예외: {e}')
|
||||
_append(agent, 'error', f'호출 오류: {e}')
|
||||
finally:
|
||||
msg_file.unlink(missing_ok=True)
|
||||
_pending.pop(agent, None)
|
||||
|
||||
|
||||
# ---------------- HTML ----------------
|
||||
|
||||
CSS = """
|
||||
:root{--bg:#f4f5f7;--panel:#fff;--fg:#1c1e21;--muted:#7a7f87;--line:#e3e5e8;--me:#d9f3c9;--them:#fff;
|
||||
--accent:#2a7ae2;--err:#c62828;--chip:#eef1f5}
|
||||
@media (prefers-color-scheme:dark){:root{--bg:#0f1115;--panel:#181b21;--fg:#e6e8eb;--muted:#8a9099;
|
||||
--line:#2a2e36;--me:#2b4a2a;--them:#22262e;--accent:#5b9cf0;--err:#ef6b6b;--chip:#232831}}
|
||||
*{box-sizing:border-box}html,body{margin:0;height:100%}
|
||||
body{background:var(--bg);color:var(--fg);font:15px/1.5 -apple-system,BlinkMacSystemFont,"Apple SD Gothic Neo",sans-serif}
|
||||
button,input,textarea{font:inherit;color:inherit}
|
||||
"""
|
||||
|
||||
LOGIN_CSS = CSS + """
|
||||
.box{max-width:340px;margin:12vh auto 0;padding:24px 20px;background:var(--panel);border:1px solid var(--line);border-radius:14px}
|
||||
h1{font-size:18px;margin:0 0 4px}p{color:var(--muted);font-size:13px;margin:0 0 16px}
|
||||
input{width:100%;font-size:17px;padding:11px 12px;border:1px solid var(--line);border-radius:10px;background:var(--bg);margin-bottom:12px}
|
||||
button{width:100%;padding:11px;border:0;border-radius:10px;background:var(--accent);color:#fff;font-weight:600}
|
||||
.err{color:var(--err);font-size:13px;margin:-4px 0 12px}
|
||||
"""
|
||||
|
||||
|
||||
def _page(title: str, body: str, css: str = LOGIN_CSS, extra_head: str = '') -> bytes:
|
||||
return (f'<!doctype html><html lang="ko"><head><meta charset="utf-8">'
|
||||
f'<meta name="viewport" content="width=device-width,initial-scale=1,viewport-fit=cover">'
|
||||
f'<title>{html.escape(title)}</title>'
|
||||
# 홈 화면 아이콘 — 없으면 iOS 가 제목 첫 글자('채')로 아이콘을 만든다
|
||||
f'<link rel="apple-touch-icon" href="/apple-touch-icon.png">'
|
||||
f'<meta name="apple-mobile-web-app-capable" content="yes">'
|
||||
f'<meta name="apple-mobile-web-app-title" content="채팅">'
|
||||
f'<style>{css}</style>{extra_head}</head>'
|
||||
f'<body>{body}</body></html>').encode()
|
||||
|
||||
|
||||
def _login_page(err: str = '') -> bytes:
|
||||
e = f'<div class="err">{html.escape(err)}</div>' if err else ''
|
||||
return _page('채팅 로그인', f'''<form class="box" method="post" action="/login/request">
|
||||
<h1>💬 채팅 로그인</h1><p>클로 · 레이 · 골디 — 관리자님 iMessage로 인증 코드를 보냅니다</p>{e}
|
||||
<button>인증 코드 받기</button></form>''')
|
||||
|
||||
|
||||
def _otp_page(err: str = '') -> bytes:
|
||||
e = f'<div class="err">{html.escape(err)}</div>' if err else ''
|
||||
return _page('인증 코드', f'''<form class="box" method="post" action="/login/otp">
|
||||
<h1>📱 인증 코드</h1><p>iMessage로 보낸 6자리 코드를 입력하세요 (5분 유효)</p>{e}
|
||||
<input name="otp" inputmode="numeric" pattern="[0-9]*" maxlength="6" autocomplete="one-time-code" autofocus required>
|
||||
<button>로그인</button></form>
|
||||
<form method="post" action="/login/request" style="text-align:center;margin-top:14px"><button style="width:auto;background:none;color:var(--muted);font-size:13px">코드 다시 받기</button></form>''')
|
||||
|
||||
|
||||
APP_CSS = CSS + """
|
||||
body{display:flex;flex-direction:column;height:100dvh}
|
||||
header{display:flex;align-items:center;gap:6px;padding:8px 10px;padding-top:max(8px,env(safe-area-inset-top));background:var(--panel);border-bottom:1px solid var(--line)}
|
||||
.tabs{display:flex;gap:6px;flex:1;min-width:0}
|
||||
.tab{position:relative;border:0;background:var(--chip);border-radius:18px;padding:6px 12px;white-space:nowrap;cursor:pointer}
|
||||
.tab.on{background:var(--accent);color:#fff}
|
||||
.tab .dot{position:absolute;top:2px;right:2px;width:8px;height:8px;border-radius:50%;background:var(--err);display:none}
|
||||
.tab.unread .dot{display:block}
|
||||
.menu-btn{border:0;background:none;font-size:20px;padding:4px 8px;cursor:pointer}
|
||||
.menu{position:absolute;right:10px;top:48px;background:var(--panel);border:1px solid var(--line);border-radius:10px;box-shadow:0 6px 20px rgba(0,0,0,.15);display:none;z-index:5}
|
||||
.menu.open{display:block}.menu button{display:block;width:100%;text-align:left;border:0;background:none;padding:10px 16px;cursor:pointer}
|
||||
main{flex:1;overflow-y:auto;padding:12px 10px}
|
||||
.msg{max-width:min(82%,640px);margin:6px 0;padding:8px 11px;border-radius:14px;word-wrap:break-word;overflow-wrap:anywhere}
|
||||
.msg.user{margin-left:auto;background:var(--me);border-bottom-right-radius:4px}
|
||||
.msg.agent{background:var(--them);border:1px solid var(--line);border-bottom-left-radius:4px}
|
||||
.msg.error{background:none;border:1px dashed var(--err);color:var(--err);font-size:13px}
|
||||
.msg .t{display:block;font-size:11px;color:var(--muted);margin-top:2px;text-align:right}
|
||||
.sys{text-align:center;color:var(--muted);font-size:12px;margin:14px 0}
|
||||
.msg pre{background:var(--chip);padding:8px;border-radius:8px;overflow-x:auto;font-size:13px;margin:6px 0}
|
||||
.msg code{background:var(--chip);padding:1px 4px;border-radius:4px;font-size:13px}.msg pre code{background:none;padding:0}
|
||||
.msg a{color:var(--accent)}
|
||||
.typing{color:var(--muted);font-size:13px;margin:8px 4px;display:none}.typing.on{display:block}
|
||||
footer{display:flex;gap:8px;align-items:flex-end;padding:8px 10px;padding-bottom:max(8px,env(safe-area-inset-bottom));background:var(--panel);border-top:1px solid var(--line)}
|
||||
textarea{flex:1;min-width:0;resize:none;max-height:40vh;font-size:16px;padding:9px 12px;border:1px solid var(--line);border-radius:20px;background:var(--bg)}
|
||||
#send{border:0;background:var(--accent);color:#fff;border-radius:50%;width:40px;height:40px;flex:none;font-size:18px;cursor:pointer}
|
||||
#send:disabled{opacity:.4}
|
||||
.empty{color:var(--muted);text-align:center;margin-top:30vh;font-size:14px}
|
||||
.choices{display:flex;flex-wrap:wrap;gap:6px;margin:2px 0 8px}
|
||||
.choice{border:1px solid var(--accent);background:var(--panel);color:var(--accent);border-radius:16px;padding:6px 14px;font-size:14px;cursor:pointer}
|
||||
.choice:disabled{opacity:.4}
|
||||
"""
|
||||
|
||||
|
||||
def _app_page() -> bytes:
|
||||
agents = json.dumps([{'id': a, 'name': n, 'emoji': e} for a, n, e in AGENTS], ensure_ascii=False)
|
||||
body = f'''<header><div class="tabs" id="tabs"></div>
|
||||
<button class="menu-btn" id="menuBtn" aria-label="메뉴">⋯</button>
|
||||
<div class="menu" id="menu"><button id="resetBtn">새 대화 시작</button>
|
||||
<form method="post" action="/logout" style="margin:0"><button>로그아웃</button></form></div></header>
|
||||
<main id="log"></main>
|
||||
<div class="typing" id="typing"></div>
|
||||
<footer><textarea id="input" rows="1" placeholder="메시지"></textarea>
|
||||
<button id="send" aria-label="전송">➤</button></footer>
|
||||
<script>window.AGENTS={agents};</script><script src="/app.js"></script>'''
|
||||
return _page('채팅', body, APP_CSS)
|
||||
|
||||
|
||||
APP_JS = r"""
|
||||
(function(){
|
||||
const A=window.AGENTS, $=s=>document.querySelector(s);
|
||||
const log=$('#log'), input=$('#input'), send=$('#send'), typing=$('#typing');
|
||||
let cur=localStorage.getItem('cw_agent')||'main', status={}, lastSeen={}, timer=null;
|
||||
try{lastSeen=JSON.parse(localStorage.getItem('cw_seen')||'{}')}catch(e){}
|
||||
if(!A.some(a=>a.id===cur))cur='main';
|
||||
|
||||
function api(path,opt){opt=opt||{};opt.headers=Object.assign({'X-CW':'1'},opt.headers||{});
|
||||
return fetch(path,opt).then(r=>{if(r.status===401){location.href='/';throw new Error('auth')}return r.json()})}
|
||||
function esc(s){return s.replace(/[&<>"]/g,c=>({'&':'&','<':'<','>':'>','"':'"'}[c]))}
|
||||
function md(s){
|
||||
const blocks=[];s=esc(s).replace(/\n?```[\w-]*\n?([\s\S]*?)\n?```\n?/g,(m,c)=>{blocks.push(c);return '\u0000'+(blocks.length-1)+'\u0000'});
|
||||
s=s.replace(/`([^`\n]+)`/g,'<code>$1</code>').replace(/\*\*([^*\n]+)\*\*/g,'<b>$1</b>')
|
||||
.replace(/(https?:\/\/[^\s<]+)/g,'<a href="$1" target="_blank" rel="noopener">$1</a>').replace(/\n/g,'<br>');
|
||||
return s.replace(/\u0000(\d+)\u0000/g,(m,i)=>'<pre><code>'+blocks[i]+'</code></pre>');
|
||||
}
|
||||
// 답변 속 "수정해줘"라고 / '네'라고 같은 '따옴표 + (이)라고' 표현을 선택지 버튼으로 뽑는다
|
||||
const BTN=/\[\[\s*버튼\s*:\s*([^\]\n]+)\]\]/;
|
||||
function strip(t){return t.replace(new RegExp(BTN.source,'g'),'').replace(/\n{3,}/g,'\n\n').trim()}
|
||||
// 1순위: 에이전트가 지정한 [[버튼: A | B]] / 없으면 답변 속 "수정해줘"라고 같은 표현에서 추출
|
||||
function choices(t){const g=BTN.exec(t);if(g)return [...new Set(g[1].split('|').map(x=>x.trim()).filter(x=>x&&x.length<=30))].slice(0,6);
|
||||
const out=[],re=/\*{0,2}[“"‘'「]([^”"’'」\n]{1,20})[”"’'」]\*{0,2}\s*(?:이)?라고/g;let m;
|
||||
while((m=re.exec(t))){const c=m[1].trim();if(c&&!out.includes(c))out.push(c)}return out.slice(0,6)}
|
||||
function hm(ts){return ts?ts.slice(11,16):''}
|
||||
function renderTabs(){
|
||||
$('#tabs').innerHTML=A.map(a=>{const st=status[a.id]||{};const un=a.id!==cur&&st.last&&st.last!==lastSeen[a.id];
|
||||
return '<button class="tab'+(a.id===cur?' on':'')+(un?' unread':'')+'" data-id="'+a.id+'">'+a.emoji+' '+a.name+(st.pending?' …':'')+'<span class="dot"></span></button>'}).join('');
|
||||
}
|
||||
function renderLog(items){
|
||||
const a=A.find(x=>x.id===cur);
|
||||
if(!items.length){log.innerHTML='<div class="empty">'+a.emoji+' '+a.name+'에게 말을 걸어보세요</div>';return}
|
||||
log.innerHTML=items.map(m=>m.role==='system'?'<div class="sys">— '+esc(m.text)+' · '+m.ts.slice(5,16).replace('T',' ')+' —</div>'
|
||||
:'<div class="msg '+m.role+'">'+(m.role==='user'?esc(m.text).replace(/\n/g,'<br>'):md(strip(m.text)))+'<span class="t">'+hm(m.ts)+'</span></div>').join('');
|
||||
const last=items[items.length-1], ch=last.role==='agent'?choices(last.text):[];
|
||||
if(ch.length)log.insertAdjacentHTML('beforeend','<div class="choices">'+ch.map(c=>'<button class="choice" data-choice="'+esc(c)+'"'+(send.disabled?' disabled':'')+'>'+esc(c)+'</button>').join('')+'</div>');
|
||||
log.scrollTop=log.scrollHeight;
|
||||
}
|
||||
function load(){return api('/api/history?agent='+cur).then(d=>{renderLog(d.items);
|
||||
if(d.items.length){lastSeen[cur]=d.items[d.items.length-1].id;localStorage.setItem('cw_seen',JSON.stringify(lastSeen))}
|
||||
setTyping(d.pending);renderTabs()})}
|
||||
function setTyping(p){const a=A.find(x=>x.id===cur);typing.textContent=a.emoji+' '+a.name+' 입력 중…';typing.classList.toggle('on',!!p);send.disabled=!!p;log.querySelectorAll('.choice').forEach(b=>b.disabled=!!p)}
|
||||
function poll(){
|
||||
api('/api/status').then(d=>{const prev=status;status=d;
|
||||
if(prev[cur]&&prev[cur].last!==d[cur].last)load();else{setTyping(d[cur].pending);renderTabs()}
|
||||
const any=Object.values(d).some(s=>s.pending);clearTimeout(timer);timer=setTimeout(poll,any?2000:20000)}).catch(()=>{clearTimeout(timer);timer=setTimeout(poll,5000)});
|
||||
}
|
||||
function doSend(choice){
|
||||
const fromInput=typeof choice!=='string',t=fromInput?input.value.trim():choice;if(!t||send.disabled)return;send.disabled=true;
|
||||
log.querySelectorAll('.choice').forEach(b=>b.disabled=true);
|
||||
api('/api/send',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({agent:cur,text:t})})
|
||||
.then(d=>{if(d.error){alert(d.error);send.disabled=false;return}if(fromInput){input.value='';autosize()}load().then(()=>{clearTimeout(timer);timer=setTimeout(poll,1500)})})
|
||||
.catch(()=>{send.disabled=false});
|
||||
}
|
||||
function autosize(){input.style.height='auto';input.style.height=Math.min(input.scrollHeight,window.innerHeight*0.4)+'px'}
|
||||
$('#tabs').addEventListener('click',e=>{const b=e.target.closest('.tab');if(!b)return;cur=b.dataset.id;localStorage.setItem('cw_agent',cur);load()});
|
||||
send.addEventListener('click',()=>doSend());
|
||||
log.addEventListener('click',e=>{const b=e.target.closest('.choice');if(b&&!b.disabled)doSend(b.dataset.choice)});
|
||||
input.addEventListener('input',autosize);
|
||||
input.addEventListener('keydown',e=>{if(e.key==='Enter'&&(e.metaKey||e.ctrlKey)){e.preventDefault();doSend()}});
|
||||
$('#menuBtn').addEventListener('click',e=>{e.stopPropagation();$('#menu').classList.toggle('open')});
|
||||
document.addEventListener('click',()=>$('#menu').classList.remove('open'));
|
||||
$('#resetBtn').addEventListener('click',()=>{const a=A.find(x=>x.id===cur);
|
||||
if(!confirm(a.name+'와의 대화 맥락을 새로 시작할까요? (기록은 화면에 남습니다)'))return;
|
||||
api('/api/reset',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({agent:cur})}).then(d=>{if(d.error)alert(d.error);load()})});
|
||||
document.addEventListener('visibilitychange',()=>{if(!document.hidden){load();poll()}});
|
||||
load();poll();
|
||||
})();
|
||||
"""
|
||||
|
||||
|
||||
# ---------------- HTTP ----------------
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
server_version = 'chat-web'
|
||||
|
||||
def log_message(self, *a):
|
||||
pass
|
||||
|
||||
# --- 공용 ---
|
||||
def _ip(self) -> str:
|
||||
xf = self.headers.get('X-Forwarded-For', '')
|
||||
return xf.split(',')[0].strip() if xf else self.client_address[0]
|
||||
|
||||
def _cookie(self, name: str) -> str | None:
|
||||
for part in (self.headers.get('Cookie') or '').split(';'):
|
||||
k, _, v = part.strip().partition('=')
|
||||
if k == name:
|
||||
return v
|
||||
return None
|
||||
|
||||
def _send(self, code: int, body: bytes, ctype: str = 'text/html; charset=utf-8',
|
||||
cookies: list[str] | None = None, location: str | None = None) -> None:
|
||||
self.send_response(code)
|
||||
self.send_header('Content-Type', ctype)
|
||||
self.send_header('Content-Length', str(len(body)))
|
||||
self.send_header('Cache-Control', 'no-store')
|
||||
self.send_header('X-Frame-Options', 'DENY')
|
||||
self.send_header('X-Content-Type-Options', 'nosniff')
|
||||
self.send_header('Referrer-Policy', 'no-referrer')
|
||||
self.send_header('Content-Security-Policy',
|
||||
"default-src 'self'; script-src 'self' 'unsafe-inline'; style-src 'unsafe-inline'; "
|
||||
"img-src 'self' data:; frame-ancestors 'none'; form-action 'self'")
|
||||
for c in cookies or []:
|
||||
self.send_header('Set-Cookie', c)
|
||||
if location:
|
||||
self.send_header('Location', location)
|
||||
self.end_headers()
|
||||
if self.command != 'HEAD':
|
||||
self.wfile.write(body)
|
||||
|
||||
def _json(self, data, code: int = 200) -> None:
|
||||
self._send(code, json.dumps(data, ensure_ascii=False).encode(), 'application/json; charset=utf-8')
|
||||
|
||||
def _redirect(self, to: str, cookies: list[str] | None = None) -> None:
|
||||
self._send(303, b'', cookies=cookies, location=to)
|
||||
|
||||
@staticmethod
|
||||
def _ck(name: str, val: str, max_age: int, samesite: str = 'Lax') -> str:
|
||||
return f'{name}={val}; Path=/; Max-Age={max_age}; HttpOnly; Secure; SameSite={samesite}'
|
||||
|
||||
def _authed(self) -> bool:
|
||||
return _valid_session(self._cookie('cw_sess'))
|
||||
|
||||
def _body(self) -> bytes:
|
||||
n = int(self.headers.get('Content-Length') or 0)
|
||||
return self.rfile.read(min(n, 64 * 1024)) if n > 0 else b''
|
||||
|
||||
def _form(self) -> dict:
|
||||
return {k: v[0] for k, v in urllib.parse.parse_qs(self._body().decode('utf-8', 'replace')).items()}
|
||||
|
||||
# --- GET ---
|
||||
def do_GET(self):
|
||||
path, _, qs = self.path.partition('?')
|
||||
q = {k: v[0] for k, v in urllib.parse.parse_qs(qs).items()}
|
||||
if path == '/':
|
||||
if self._authed():
|
||||
return self._send(200, _app_page())
|
||||
if self._cookie('cw_pre') in _otp:
|
||||
return self._send(200, _otp_page())
|
||||
return self._send(200, _login_page())
|
||||
if path in ('/apple-touch-icon.png', '/apple-touch-icon-precomposed.png'):
|
||||
# 인증 밖 — iOS 가 홈 화면 추가 시 쿠키 없이 가져갈 수 있다
|
||||
return self._send(200, ICON_FILE.read_bytes(), 'image/png')
|
||||
if not self._authed():
|
||||
return self._json({'error': 'auth'}, 401) if path.startswith('/api/') else self._redirect('/')
|
||||
if path == '/app.js':
|
||||
return self._send(200, APP_JS.encode(), 'application/javascript; charset=utf-8')
|
||||
if path == '/api/history':
|
||||
ag = q.get('agent', '')
|
||||
if ag not in AGENT_IDS:
|
||||
return self._json({'error': 'agent'}, 400)
|
||||
return self._json({'items': _history(ag), 'pending': ag in _pending})
|
||||
if path == '/api/status':
|
||||
out = {}
|
||||
for ag in AGENT_IDS:
|
||||
h = _history(ag)
|
||||
out[ag] = {'pending': ag in _pending, 'last': h[-1]['id'] if h else None}
|
||||
return self._json(out)
|
||||
self._send(404, b'not found', 'text/plain')
|
||||
|
||||
do_HEAD = do_GET
|
||||
|
||||
# --- POST ---
|
||||
def do_POST(self):
|
||||
path = self.path.partition('?')[0]
|
||||
ip = self._ip()
|
||||
|
||||
if path == '/login/request':
|
||||
has_pre = self._cookie('cw_pre') in _otp
|
||||
page = _otp_page if has_pre else _login_page
|
||||
left = _locked()
|
||||
if left:
|
||||
return self._send(429, _login_page(f'잠겨 있습니다. {int(left // 60) + 1}분 뒤 다시 시도하세요.'))
|
||||
why = _take_send_slot()
|
||||
if why:
|
||||
log(f'code request refused ip={ip}')
|
||||
return self._send(429, page(why))
|
||||
code = f'{secrets.randbelow(10**6):06d}'
|
||||
pre = secrets.token_urlsafe(24)
|
||||
t = time.time()
|
||||
for k in [k for k, v in _otp.items() if v['exp'] < t]:
|
||||
_otp.pop(k, None)
|
||||
_otp.pop(self._cookie('cw_pre') or '', None) # 다시 받기 — 이전 코드는 무효
|
||||
_otp[pre] = {'otp': code, 'exp': t + OTP_TTL, 'tries': 0}
|
||||
if not _send_otp(code):
|
||||
_otp.pop(pre, None)
|
||||
return self._send(503, _login_page('인증 코드를 보내지 못했습니다.'))
|
||||
log(f'otp sent ip={ip}')
|
||||
return self._redirect('/', [self._ck('cw_pre', pre, OTP_TTL, 'Strict')])
|
||||
|
||||
if path == '/login/otp':
|
||||
f = self._form()
|
||||
pre = self._cookie('cw_pre')
|
||||
ent = _otp.get(pre or '')
|
||||
clear = [self._ck('cw_pre', '', 0, 'Strict')]
|
||||
if _locked():
|
||||
_otp.pop(pre or '', None)
|
||||
return self._redirect('/', clear)
|
||||
if not ent or ent['exp'] < time.time():
|
||||
_otp.pop(pre or '', None)
|
||||
return self._send(401, _login_page('코드가 만료됐습니다. 다시 로그인하세요.'), cookies=clear)
|
||||
if not hmac.compare_digest(f.get('otp', '').strip(), ent['otp']):
|
||||
ent['tries'] += 1
|
||||
_record_fail(ip, 'otp')
|
||||
if ent['tries'] >= OTP_MAX_TRIES:
|
||||
_otp.pop(pre, None)
|
||||
return self._send(401, _login_page('코드를 여러 번 틀렸습니다. 다시 로그인하세요.'), cookies=clear)
|
||||
return self._send(401, _otp_page('코드가 맞지 않습니다.'))
|
||||
_otp.pop(pre, None)
|
||||
tok = _new_session(ip, self.headers.get('User-Agent', ''))
|
||||
log(f'login ok ip={ip}')
|
||||
return self._redirect('/', clear + [self._ck('cw_sess', tok, SESSION_TTL)])
|
||||
|
||||
if path == '/logout':
|
||||
_drop_session(self._cookie('cw_sess'))
|
||||
return self._redirect('/', [self._ck('cw_sess', '', 0)])
|
||||
|
||||
# 이하 API — 세션 + 커스텀 헤더(교차 출처 요청은 이 헤더를 붙이려면 CORS 사전요청을 거쳐야 해 차단된다)
|
||||
if not self._authed() or self.headers.get('X-CW') != '1':
|
||||
return self._json({'error': 'auth'}, 401)
|
||||
try:
|
||||
data = json.loads(self._body() or b'{}')
|
||||
except Exception:
|
||||
return self._json({'error': 'bad json'}, 400)
|
||||
ag = data.get('agent')
|
||||
if ag not in AGENT_IDS:
|
||||
return self._json({'error': 'agent'}, 400)
|
||||
|
||||
if path == '/api/send':
|
||||
text = (data.get('text') or '').strip()
|
||||
if not text:
|
||||
return self._json({'error': '빈 메시지'}, 400)
|
||||
if len(text) > MAX_MSG:
|
||||
return self._json({'error': f'{MAX_MSG}자 이하로 보내주세요'}, 400)
|
||||
if ag in _pending:
|
||||
return self._json({'error': '이전 답을 기다리는 중입니다'}, 409)
|
||||
if text.lower() == '/clear': # 웹 전용 명령 — 에이전트로 보내지 않는다
|
||||
_clear(ag)
|
||||
return self._json({'ok': True, 'cleared': True})
|
||||
item = _append(ag, 'user', text)
|
||||
_pending[ag] = time.time() # 스레드 시작 전 표시 — 연속 클릭 이중 발송 방지
|
||||
threading.Thread(target=_run_agent, args=(ag, text), daemon=True).start()
|
||||
return self._json({'ok': True, 'item': item})
|
||||
|
||||
if path == '/api/reset':
|
||||
if ag in _pending:
|
||||
return self._json({'error': '답을 기다리는 중에는 새로 시작할 수 없습니다'}, 409)
|
||||
return self._json({'ok': True, 'session': _rotate_session(ag)})
|
||||
|
||||
self._json({'error': 'not found'}, 404)
|
||||
|
||||
|
||||
def cmd_serve(port: int) -> None:
|
||||
STATE.mkdir(parents=True, exist_ok=True)
|
||||
srv = ThreadingHTTPServer(('', port), Handler)
|
||||
log(f'chat_web serve :{port}')
|
||||
srv.serve_forever()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
args = sys.argv[1:]
|
||||
if args[:1] == ['serve']:
|
||||
port = int(args[args.index('--port') + 1]) if '--port' in args else DEFAULT_PORT
|
||||
cmd_serve(port)
|
||||
else:
|
||||
print(__doc__)
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 20 KiB |
Reference in New Issue
Block a user