Spaces:
Running
Running
| """ | |
| 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() | |