Spaces:
Running
Running
File size: 3,999 Bytes
85020b5 ea3fcbe 5ddfd1f ea3fcbe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 | """
Server HTTP untuk Profanity Checker — http://127.0.0.1:8005
Endpoint:
GET /api/status — health-check
POST /api/profanity — {"text": "..."}
→ {"findings": [...], "finding_count": int}
Jalankan:
python src/profanity_server.py
"""
from __future__ import annotations
import json
import logging
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent))
from profanity_detector import ProfanityChecker
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-7s %(message)s",
datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)
_checker: ProfanityChecker | None = None
def _get() -> ProfanityChecker:
global _checker
if _checker is None:
_checker = ProfanityChecker()
_checker.load()
return _checker
class Handler(BaseHTTPRequestHandler):
def _cors(self):
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
def do_OPTIONS(self):
self.send_response(204); self._cors(); self.end_headers()
def _json(self, status: int, body: object):
payload = json.dumps(body, ensure_ascii=False).encode()
try:
self.send_response(status)
self.send_header("Content-Type", "application/json; charset=utf-8")
self.send_header("Content-Length", str(len(payload)))
self._cors()
self.end_headers()
self.wfile.write(payload)
except (ConnectionAbortedError, BrokenPipeError, ConnectionResetError):
pass
def _body(self) -> dict | None:
n = int(self.headers.get("Content-Length", 0))
if not n:
return {}
try:
return json.loads(self.rfile.read(n))
except (json.JSONDecodeError, ValueError):
return None
def log_message(self, fmt, *args):
logger.info("%-6s %s", args[0] if args else "", args[1] if len(args) > 1 else "")
def do_GET(self):
if self.path == "/api/status":
chk = _get()
self._json(200, {"ready": chk.is_loaded, "lexicon_size": chk.lexicon_size})
else:
self._json(404, {"error": "Not found"})
def do_POST(self):
if self.path != "/api/profanity":
self._json(404, {"error": "Not found"}); return
body = self._body()
if body is None:
self._json(400, {"error": "JSON tidak valid."}); return
text = str(body.get("text", "")).strip()
if not text:
self._json(400, {"error": "Field 'text' kosong."}); return
findings = _get().check(text)
self._json(200, {
"text": text,
"finding_count": len(findings),
"findings": [
{
"word": f.word,
"normalized": f.normalized,
"start": f.start,
"end": f.end,
"severity": f.severity,
"reason": f.reason,
"confidence": f.confidence,
}
for f in findings
],
})
if __name__ == "__main__":
import argparse
parser = argparse.ArgumentParser()
parser.add_argument("--port", type=int, default=8005)
parser.add_argument("--host", default="127.0.0.1")
args = parser.parse_args()
logger.info("Memuat lexicon profanity…")
_get()
server = HTTPServer((args.host, args.port), Handler)
logger.info("Profanity server berjalan di http://%s:%d", args.host, args.port)
logger.info("Buka web/profanity-test.html untuk pengujian.")
try:
server.serve_forever()
except KeyboardInterrupt:
server.server_close()
|