프로젝트 1 — 사내 규정 질의 API

import hashlib, hmac, json, os, re, secrets, time, urllib.request
from collections import defaultdict
from datetime import datetime
import numpy as np
from openai import OpenAI
from sentence_transformers import SentenceTransformer

# Ollama 모델을 내려 깨끗하게 다시 올린다(10차시)
def unload():
    urllib.request.urlopen(urllib.request.Request("http://localhost:11434/api/generate", headers={"Content-Type": "application/json"},
                                                  data=json.dumps({"model": "qwen2.5:3b", "keep_alive": 0}).encode())).read()
unload()

articles = [l.strip() for l in open("company_rules.txt", encoding="utf-8") if re.match(r"^제\d+조\(", l)]
titles = [re.match(r"^제\d+조\((.+?)\)", a).group(1) for a in articles]
emb = SentenceTransformer("intfloat/multilingual-e5-small", device="cpu")
A = emb.encode(["passage: " + a for a in articles], normalize_embeddings=True)
llm = OpenAI(base_url="http://localhost:11434/v1", api_key="ollama")
SYSTEM = "당신은 솔내기술(가상 회사)의 규정 도우미입니다. [조항]에 있는 내용만으로 한두 문장으로 답하세요. 없으면 '규정에서 찾을 수 없습니다'라고 답하세요."


def answer(question):
    """조항 3개를 찾아 LLM 으로 답한다. 오류나 깨진 글자면 모델을 내렸다 두 번까지 다시 묻는다"""
    q = emb.encode("query: " + question, normalize_embeddings=True)
    top = np.argsort(-(A @ q))[:3]
    msgs = [{"role": "system", "content": SYSTEM},
            {"role": "user", "content": "[조항]\n" + "\n".join(articles[i] for i in top) + f"\n\n[질문]\n{question}"}]
    for _ in range(3):
        try:
            text = llm.chat.completions.create(model="qwen2.5:3b", messages=msgs, temperature=0, seed=42).choices[0].message.content.strip()
            if "@" not in text:
                return text, [titles[i] for i in top]
        except Exception as e:
            text = f"(오류: {e})"
        unload()
    return text, [titles[i] for i in top]


print(len(articles), "조항 ·", titles[:5])
print(answer("집에서 일하는 건 일주일에 며칠까지 돼요?"))

from fastapi import Depends, FastAPI, HTTPException, Request, Security
from fastapi.security import APIKeyHeader
from fastapi.testclient import TestClient
from pydantic import BaseModel, Field

# 시험용 키를 만들어 환경 변수에 넣는다. 실제로는 관리자가 발급해 비밀 저장소에 둔다
staff = {"김솔내": "직원", "이영업": "직원", "박관리": "관리자"}
issued = {name: secrets.token_urlsafe(24) for name in staff}
os.environ["RULES_API_KEYS"] = json.dumps({name: {"키": issued[name], "역할": role} for name, role in staff.items()}, ensure_ascii=False)

digest = lambda k: hashlib.sha256(k.encode()).hexdigest()
KEYS = {digest(v["키"]): {"이름": n, "역할": v["역할"]} for n, v in json.loads(os.environ["RULES_API_KEYS"]).items()}
LIMIT, WINDOW = 10, 60                                   # 1분에 10번까지
recent = defaultdict(list)                               # 사용자 → 최근 요청 시각들
LOG = []
header = APIKeyHeader(name="X-API-Key", auto_error=False)


def who(request: Request, key: str | None = Security(header)):
    """키 확인(401) → 횟수 제한(429) → 기록. 통과하면 사용자 정보를 돌려준다"""
    user = next((u for h, u in KEYS.items() if key and hmac.compare_digest(h, digest(key))), None)
    if user is None:
        LOG.append({"시각": datetime.now().isoformat(timespec="seconds"), "사용자": None, "주소": request.url.path, "상태": 401})
        raise HTTPException(status_code=401, detail="API 키가 없거나 맞지 않습니다")
    now = time.time()
    recent[user["이름"]] = [t for t in recent[user["이름"]] if now - t < WINDOW]     # 1분이 지난 기록은 버린다
    if len(recent[user["이름"]]) >= LIMIT:
        LOG.append({"시각": datetime.now().isoformat(timespec="seconds"), "사용자": user["이름"], "주소": request.url.path, "상태": 429})
        raise HTTPException(status_code=429, detail=f"1분에 {LIMIT}번까지 부를 수 있습니다")
    recent[user["이름"]].append(now)
    return user


print({n: v["역할"] for n, v in json.loads(os.environ["RULES_API_KEYS"]).items()}, "· 키 해시", len(KEYS), "개")

app = FastAPI(title="솔내기술 사내 규정 질의 API")


class Ask(BaseModel):
    질문: str = Field(min_length=2, max_length=200)


@app.post("/ask")
def ask(body: Ask, user=Depends(who)):
    start = time.perf_counter()
    text, refs = answer(body.질문)
    ms = round((time.perf_counter() - start) * 1000)
    LOG.append({"시각": datetime.now().isoformat(timespec="seconds"), "사용자": user["이름"], "주소": "/ask", "상태": 200, "걸린ms": ms})
    return {"답": text, "근거조항": refs, "걸린ms": ms}


@app.get("/me")
def me(user=Depends(who)):
    return user


@app.get("/stats")
def stats(user=Depends(who)):
    if user["역할"] != "관리자":
        raise HTTPException(status_code=403, detail="관리자만 볼 수 있습니다")
    done = [r for r in LOG if r["상태"] == 200]
    return {"요청": len(LOG), "성공": len(done), "평균ms": round(np.mean([r["걸린ms"] for r in done])) if done else None,
            "사용자별": {u: sum(r["사용자"] == u for r in LOG) for u in staff},
            "상태별": {s: sum(r["상태"] == s for r in LOG) for s in sorted({r["상태"] for r in LOG})}}


client = TestClient(app)
print(client.get("/me", headers={"X-API-Key": issued["김솔내"]}).json())

qa = [json.loads(line) for line in open("company_rules_qa.jsonl", encoding="utf-8")][:8]
right = 0
for i, item in enumerate(qa):
    name = ["김솔내", "이영업"][i % 2]                     # 두 직원이 번갈아 묻는다
    r = client.post("/ask", json={"질문": item["질문"]}, headers={"X-API-Key": issued[name]})
    body = r.json()
    ok = any(k in body["답"].replace(" ", "") for k in item["정답열쇠"].replace(" ", "").split("|"))
    right += ok
    print("O" if ok else "X", r.status_code, item["질문"], "→", body["답"][:50], "|", body["근거조항"][0], f"{body['걸린ms']}ms")
print(f"맞힌 수: {right} / {len(qa)}")

checks = {
    "키 없음": client.post("/ask", json={"질문": "연차는 며칠이에요?"}).status_code,
    "틀린 키": client.post("/ask", json={"질문": "연차는 며칠이에요?"}, headers={"X-API-Key": "wrong-key-0000"}).status_code,
    "너무 짧은 질문": client.post("/ask", json={"질문": "?"}, headers={"X-API-Key": issued["이영업"]}).status_code,
    "직원이 통계": client.get("/stats", headers={"X-API-Key": issued["김솔내"]}).status_code,
}
# 김솔내가 /me 를 빠르게 여러 번 부른다. 앞서 /me 1번, /ask 4번, /stats 1번(모두 6번)을 불렀으니 곧 1분에 10번을 넘는다
codes = [client.get("/me", headers={"X-API-Key": issued["김솔내"]}).status_code for _ in range(8)]
checks["빠르게 8번"] = codes
print(checks)
r = client.get("/stats", headers={"X-API-Key": issued["박관리"]})
print("관리자 통계:", r.status_code, json.dumps(r.json(), ensure_ascii=False))
with open("out_rules_api_log.jsonl", "w", encoding="utf-8") as fp:
    fp.writelines(json.dumps(x, ensure_ascii=False) + "\n" for x in LOG)