[Feature] add core features

.github/workflows/lint.yaml

@@ -0,0 +1,18 @@
+name: Lint
+
+on:
+  push:
+    paths:
+      - '**.py'
+
+jobs:
+  run-linters:
+    name: Run linters
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Check out Git repository
+        uses: actions/checkout@v4
+
+      - name: Lint check ruff
+        uses: chartboost/ruff-action@v1

.pre-commit-config.yaml

@@ -0,0 +1,24 @@
+repos:
+- repo: https://github.com/pre-commit/pre-commit-hooks
+  rev: v5.0.0
+  hooks:
+    - id: trailing-whitespace
+    - id: end-of-file-fixer
+    - id: check-yaml
+    - id: check-json
+    - id: check-toml
+    - id: check-added-large-files
+    - id: check-merge-conflict
+
+- repo: https://github.com/astral-sh/ruff-pre-commit
+  rev: v0.11.8
+  hooks:
+    - id: ruff
+      args: [ --fix ]
+    - id: ruff-format
+
+- repo: https://github.com/pycqa/isort
+  rev: 6.0.1
+  hooks:
+    - id: isort
+      args: ["--profile", "black"]

README.md

@@ -1 +1,35 @@
-# NaverWorksKit
+# NaverWorksKit
+
+A simple toolkit to integrate with Naver Works Bot API — so you don’t have to suffer through the official documentation.
+
+## Features
+- Easy bot message sending (channel & user)
+- Automatic access token management
+- Simple FastAPI server to handle webhook events
+
+## Getting Started
+1. Clone the repository
+```bash
+git clone https://github.com/hanacardData/NaverWorksKit.git
+cd NaverWorksKit
+```
+
+2. Create a .env file in the project root and fill in your credentials:
+```
+WORKS_CLIENT_ID=your_client_id
+WORKS_CLIENT_SECRET=your_client_secret
+SERVICE_ACCOUNT=your_service_account
+PRIVATE_KEY_PATH=./secret/private.pem
+BOT_ID=your_bot_id
+```
+
+3. Install dependencies
+```python
+pip install -r requirements.txt
+```
+
+4. Run the FastAPI server with Uvicorn
+```bash
+uvicorn app.main:app --reload --host 0.0.0.0 --port 8001 # dev
+uvicorn app.main:app --host 0.0.0.0 --port 8001 --workers 4 # production
+```

app/config/settings.py

@@ -0,0 +1,16 @@
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    works_client_id: str
+    works_client_secret: str
+    bot_secret: str
+    bot_id: str
+    private_key_path: str
+    service_account: str
+
+    class Config:
+        env_file = ".env"
+
+
+settings = Settings()

app/handlers/event_handler.py

@@ -0,0 +1,44 @@
+from enum import Enum
+
+from fastapi.responses import JSONResponse
+
+from app.services.post_message import post_message_to_channel, post_message_to_user
+
+
+class BotStatus(str, Enum):
+    """Enum for bot status."""
+
+    IGNORED = "ignored"
+    OK = "ok"
+
+
+async def process_event(data: dict) -> JSONResponse:
+    event_type = data.get("type")
+    source = data.get("source", {})
+    channel_id = source.get("channelId")
+    user_id = source.get("userId")
+
+    if event_type == "join":
+        await post_message_to_channel(
+            channel_id=channel_id,
+            message="Hello! I am a bot.",
+        )
+        return JSONResponse(status_code=200, content={"status": BotStatus.OK})
+
+    if event_type == "message":
+        content = data.get("content", {})
+        text = content.get("text", "")
+        if not text.startswith("/"):
+            return JSONResponse(status_code=200, content={"status": BotStatus.IGNORED})
+
+        if channel_id:
+            await post_message_to_channel(
+                channel_id=channel_id, message=f"Pong! {text}"
+            )
+            return JSONResponse(status_code=200, content={"status": BotStatus.OK})
+
+        if user_id:
+            await post_message_to_user(user_id=user_id, message=f"Pong! {text}")
+            return JSONResponse(status_code=200, content={"status": BotStatus.OK})
+
+    return JSONResponse(status_code=200, content={"status": BotStatus.IGNORED})

app/logger.py

@@ -0,0 +1,33 @@
+import logging
+import sys
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+LOG_DIR = Path("logs")
+LOG_DIR.mkdir(exist_ok=True)
+
+
+def init_logger() -> logging.Logger:
+    logger = logging.getLogger("WorksBot")
+    logger.setLevel(logging.INFO)
+
+    formatter = logging.Formatter(
+        fmt="%(asctime)s - %(levelname)s - %(name)s - %(message)s",
+        datefmt="%Y-%m-%d %H:%M:%S",
+    )
+
+    console_handler = logging.StreamHandler(sys.stdout)
+    console_handler.setFormatter(formatter)
+    logger.addHandler(console_handler)
+
+    file_handler = RotatingFileHandler(
+        LOG_DIR / "app.log", maxBytes=5 * 1024 * 1024, backupCount=5
+    )
+    file_handler.setFormatter(formatter)
+    logger.addHandler(file_handler)
+
+    logger.propagate = False
+    return logger
+
+
+logger = init_logger()

app/main.py

@@ -0,0 +1,35 @@
+import logging
+
+from fastapi import FastAPI, Header, HTTPException, Request
+from fastapi.responses import JSONResponse
+from uvicorn.config import LOGGING_CONFIG
+
+from app.config.settings import settings
+from app.handlers.event_handler import process_event
+from app.utils.signature import verify_signature
+
+app = FastAPI()
+logger = logging.getLogger(__name__)
+LOGGING_CONFIG["formatters"]["default"]["fmt"] = (
+    "%(asctime)s [%(name)s] %(levelprefix)s %(message)s"
+)
+LOGGING_CONFIG["formatters"]["access"]["fmt"] = (
+    "%(asctime)s [%(name)s] %(levelprefix)s %(message)s"
+)
+
+
+@app.post("/")
+async def callback(
+    request: Request, x_works_signature: str = Header(None)
+) -> JSONResponse:
+    raw_body = await request.body()
+    raw_text = raw_body.decode()
+
+    if not x_works_signature or not verify_signature(
+        raw_text, x_works_signature, settings.bot_secret
+    ):
+        logger.warning("Invalid or missing signature.")
+        raise HTTPException(status_code=403, detail="Invalid or missing signature")
+
+    data = await request.json()
+    return await process_event(data)

app/services/access_token.py

@@ -0,0 +1,82 @@
+import time
+from logging import Logger
+
+import jwt
+import requests
+from cryptography.hazmat.primitives import serialization
+from logger import logger
+
+from app.config.settings import settings
+
+
+class TokenManager:
+    def __init__(self, logger: Logger):
+        self._access_token = None
+        self._token_expiry = 0
+        self.logger = logger
+
+    def get_token(self) -> str:
+        """get access token."""
+        if self._access_token and not self.__is_token_expired:
+            self.logger.debug("Returning cached access token.")
+            return self._access_token
+
+        self.logger.debug("Token expired or not available, requesting new token.")
+        return self._request_new_token()
+
+    def _request_new_token(self) -> str:
+        """request new access token when expired."""
+        now = int(time.time())
+        exp = now + 3600
+        payload = {
+            "iss": settings.works_client_id,
+            "sub": settings.service_account,
+            "iat": now,
+            "exp": exp,
+        }
+        try:
+            with open(settings.private_key_path, "r") as key_file:
+                private_key = serialization.load_pem_private_key(
+                    key_file.read().encode(), password=None
+                )
+            encoded_jwt = jwt.encode(payload, private_key, algorithm="RS256")
+        except Exception as e:
+            self.logger.error(f"Error loading private key: {e}")
+            raise
+
+        token_url = "https://auth.worksmobile.com/oauth2/v2.0/token"
+        headers = {"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"}
+        data = {
+            "assertion": encoded_jwt,
+            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
+            "client_id": settings.works_client_id,
+            "client_secret": settings.works_client_secret,
+            "scope": "bot bot.message bot.read",
+        }
+
+        try:
+            response = requests.post(token_url, headers=headers, data=data)
+            response.raise_for_status()
+        except requests.RequestException as e:
+            self.logger.error(f"Token request failed: {e}")
+            raise
+
+        if response.status_code == 200:
+            token_data = response.json()
+            self._access_token = token_data["access_token"]
+            self._token_expiry = exp
+            self.logger.info("Access token successfully obtained and cached.")
+            return self._access_token
+        else:
+            self.logger.error(
+                f"Failed to get token, status code: {response.status_code}"
+            )
+            self.logger.error(f"Response: {response.text}")
+            raise Exception("Token request failed")
+
+    @property
+    def __is_token_expired(self) -> bool:
+        return int(time.time()) >= self._token_expiry - 60
+
+
+token_manager = TokenManager(logger)

app/services/post_message.py

@@ -0,0 +1,51 @@
+### app/services/post_message.py
+import httpx
+from retry import retry
+
+from app.config.settings import settings
+from app.logger import logger
+from app.services.access_token import token_manager
+
+CHANNEL_MESSAGE_URL = (
+    "https://www.worksapis.com/v1.0/bots/{bot_id}/channels/{channel_id}/messages"
+)
+USER_MESSAGE_URL = (
+    "https://www.worksapis.com/v1.0/bots/{bot_id}/users/{user_id}/messages"
+)
+
+
+def _set_headers() -> dict[str, str]:
+    token = token_manager.get_token()
+    return {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
+
+
+def _set_text_payload(message: str) -> dict[str, dict[str, str]]:
+    return {"content": {"type": "text", "text": message}}
+
+
+@retry(tries=3, delay=1, backoff=2, exceptions=(httpx.RequestError, httpx.HTTPError))
+async def post_message_to_channel(channel_id: str, message: str) -> None:
+    url = CHANNEL_MESSAGE_URL.format(bot_id=settings.bot_id, channel_id=channel_id)
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.post(
+                url, headers=_set_headers(), json=_set_text_payload(message)
+            )
+            response.raise_for_status()
+        except (httpx.RequestError, httpx.HTTPStatusError) as e:
+            logger.error(f"Async post to channel failed: {e}")
+            raise
+
+
+@retry(tries=3, delay=1, backoff=2, exceptions=(httpx.RequestError, httpx.HTTPError))
+async def post_message_to_user(user_id: str, message: str) -> None:
+    url = USER_MESSAGE_URL.format(bot_id=settings.bot_id, user_id=user_id)
+    async with httpx.AsyncClient() as client:
+        try:
+            response = await client.post(
+                url, headers=_set_headers(), json=_set_text_payload(message)
+            )
+            response.raise_for_status()
+        except (httpx.RequestError, httpx.HTTPStatusError) as e:
+            logger.error(f"Async post to user failed: {e}")
+            raise

app/utils/signature.py

@@ -0,0 +1,14 @@
+import base64
+import hashlib
+import hmac
+
+
+def verify_signature(body: str, received_signature: str, secret: str) -> bool:
+    """Verify the HMAC signature of the request body."""
+    hash_digest = hmac.new(
+        key=secret.encode("utf-8"),
+        msg=body.encode("utf-8"),
+        digestmod=hashlib.sha256,
+    ).digest()
+    expected_signature = base64.b64encode(hash_digest).decode("utf-8")
+    return hmac.compare_digest(expected_signature, received_signature)