Spaces:
Running
Running
File size: 4,367 Bytes
c85625f 5ddfd1f c85625f | 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 | """
Server HTTP untuk Syntax Checker — http://127.0.0.1:8008
Endpoint:
GET /api/status — health-check
POST /api/syntax — {"text": "...", "language": "id"}
→ {"findings": [...], "finding_count": int}
Jalankan:
python -m src.syntax.syntax_server
python src/syntax/syntax_server.py --no-ml # nonaktifkan model (deteksi mati)
"""
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.parent))
from syntax.syntax_detector import SyntaxChecker
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-7s %(message)s",
datefmt="%H:%M:%S",
)
logger = logging.getLogger(__name__)
_checker: SyntaxChecker | None = None
_use_ml = True
def _get() -> SyntaxChecker:
global _checker
if _checker is None:
_checker = SyntaxChecker(use_ml=_use_ml)
_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": True,
"ml_active": chk.ml_active,
})
else:
self._json(404, {"error": "Not found"})
def do_POST(self):
if self.path != "/api/syntax":
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
language = str(body.get("language", "id"))
findings = _get().check(text, language=language)
self._json(200, {
"text": text,
"finding_count": len(findings),
"findings": [
{
"sentence": f.sentence,
"start": f.start,
"end": f.end,
"score": f.score,
"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=8008)
parser.add_argument("--host", default="127.0.0.1")
parser.add_argument("--no-ml", action="store_true",
help="Nonaktifkan model (deteksi mati, server tetap jalan).")
args = parser.parse_args()
_use_ml = not args.no_ml
logger.info("Memuat Syntax Checker (ML: %s)...", "aktif" if _use_ml else "nonaktif")
chk = _get()
logger.info("Syntax Checker siap (ML aktif: %s).", chk.ml_active)
server = HTTPServer((args.host, args.port), Handler)
logger.info("Syntax server berjalan di http://%s:%d", args.host, args.port)
try:
server.serve_forever()
except KeyboardInterrupt:
server.server_close()
|