Python으로 간단한 ActivityPub 리마인더 봇 만들기

AmaseCocoa @cocoa@hackers.pub

이 튜토리얼에서는 Python을 사용하여 간단한 ActivityPub 봇을 만드는 방법을 안내합니다. 이 봇은 멘션을 수신하고, 특정 형식의 메시지를 받으면 지정된 시간 후에 사용자에게 리마인더를 예약하고 전송합니다.

예를 들어, 사용자가 "@reminder@your.host.com 10m check the oven"과 같은 메시지로 봇을 멘션하면, 봇은 10분 후에 "🔔 Reminder for @user: check the oven"과 같은 메시지로 응답합니다.

사전 요구사항

이 튜토리얼을 따라하려면 Python 3.10 이상과 다음 라이브러리가 필요합니다:

  • apkit[server]: Python에서 ActivityPub 애플리케이션을 구축하기 위한 강력한 툴킷. FastAPI 기반 컴포넌트를 포함하는 server 추가 기능을 사용합니다.
  • uvicorn: FastAPI 애플리케이션을 실행하기 위한 ASGI 서버.
  • cryptography: ActivityPub에 필요한 암호화 키를 생성하고 관리하는 데 사용됩니다.
  • uv: 선택 사항이지만 권장되는 빠른 패키지 관리자.

이러한 의존성은 uv 또는 pip를 사용하여 설치할 수 있습니다.

# uv로 새 프로젝트 초기화
uv init

# 의존성 설치
uv add "apkit[server]" uvicorn cryptography

프로젝트 구조

프로젝트 구조는 봇의 로직을 위한 단일 Python 파일로 구성된 최소한의 구조입니다.

.
├── main.py
└── private_key.pem
  • main.py: 봇을 위한 모든 코드가 포함되어 있습니다.
  • private_key.pem: 봇의 Actor를 위한 개인 키. 첫 실행 시 자동으로 생성됩니다.

코드 설명

우리의 애플리케이션 로직은 다음 단계로 나눌 수 있습니다:

  1. 임포트 및 구성: 필요한 임포트와 기본 구성 변수를 설정합니다.
  2. 키 생성: 활동 서명에 필요한 암호화 키를 준비합니다.
  3. Actor 정의: Fediverse에서 봇의 정체성을 정의합니다.
  4. 서버 초기화: apkit ActivityPub 서버를 설정합니다.
  5. 데이터 저장소: 생성된 활동을 위한 간단한 인메모리 저장소를 구현합니다.
  6. 리마인더 로직: 리마인더를 파싱하고 알림을 보내는 핵심 로직을 코딩합니다.
  7. 엔드포인트 정의: 필요한 웹 엔드포인트(/actor, /inbox 등)를 생성합니다.
  8. 활동 핸들러: 다른 서버에서 들어오는 활동을 처리합니다.
  9. 애플리케이션 시작: 서버를 실행합니다.

main.py 파일의 각 섹션을 자세히 살펴보겠습니다.

1. 임포트 및 구성

먼저, 필요한 모듈을 임포트하고 봇의 기본 구성을 정의합니다.

# main.py

import asyncio
import logging
import re
import uuid
import os
from datetime import timedelta, datetime

# Imports from FastAPI, cryptography, and apkit
from fastapi import Request, Response
from fastapi.responses import JSONResponse
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import serialization as crypto_serialization

from apkit.config import AppConfig
from apkit.server import ActivityPubServer
from apkit.server.types import Context, ActorKey
from apkit.server.responses import ActivityResponse
from apkit.models import (
    Actor, Application, CryptographicKey, Follow, Create, Note, Mention, Actor as APKitActor, OrderedCollection,
)
from apkit.client import WebfingerResource, WebfingerResult, WebfingerLink
from apkit.client.asyncio.client import ActivityPubClient

# --- Logging Setup ---
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# --- Basic Configuration ---
HOST = "your.host.com"  # Replace with your domain
USER_ID = "reminder"      # The bot's username

your.host.com을 봇이 호스팅될 실제 도메인으로 변경해야 합니다. 이 값들은 봇의 고유 식별자(예: @reminder@your.host.com)를 결정합니다.

2. 키 생성 및 지속성

ActivityPub는 서버 간 통신을 보호하기 위해 HTTP 서명을 사용합니다. 이를 위해 각 액터는 공개/개인 키 쌍이 필요합니다. 다음 코드는 개인 키를 생성하고 아직 존재하지 않는 경우 파일에 저장합니다.

# main.py (continued)

# --- Key Persistence ---
KEY_FILE = "private_key.pem"

# Load the private key if it exists, otherwise generate a new one
if os.path.exists(KEY_FILE):
    logger.info(f"Loading existing private key from {KEY_FILE}.")
    with open(KEY_FILE, "rb") as f:
        private_key = crypto_serialization.load_pem_private_key(f.read(), password=None)
else:
    logger.info(f"No key file found. Generating new private key and saving to {KEY_FILE}.")
    private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
    with open(KEY_FILE, "wb") as f:
        f.write(private_key.private_bytes(
            encoding=crypto_serialization.Encoding.PEM,
            format=crypto_serialization.PrivateFormat.PKCS8,
            encryption_algorithm=crypto_serialization.NoEncryption()
        ))

# Generate the public key from the private key
public_key_pem = private_key.public_key().public_bytes(
    encoding=crypto_serialization.Encoding.PEM,
    format=crypto_serialization.PublicFormat.SubjectPublicKeyInfo
).decode('utf-8')

3. Actor 정의

다음으로, 봇의 Actor를 정의합니다. Actor는 ActivityPub 네트워크에서 봇의 정체성입니다. 이 엔티티는 자동화되어 있으므로 Application 타입을 사용합니다.

# main.py (continued)

# --- Actor Definition ---
actor = Application(
    id=f"https://{HOST}/actor",
    name="Reminder Bot",
    preferredUsername=USER_ID,
    summary="A bot that sends you reminders. Mention me like: @reminder 5m Check the oven",
    inbox=f"https://{HOST}/inbox",      # 활동을 수신하는 엔드포인트
    outbox=f"https://{HOST}/outbox",    # 활동을 전송하는 엔드포인트
    publicKey=CryptographicKey(
        id=f"https://{HOST}/actor#main-key",
        owner=f"https://{HOST}/actor",
        publicKeyPem=public_key_pem
    )
)

4. 서버 초기화

apkit에서 ActivityPubServer를 초기화하고, 발신 활동에 서명하기 위한 Actor의 키를 검색하는 함수를 제공합니다.

# main.py (continued)

# --- Key Retrieval Function ---
async def get_keys_for_actor(identifier: str) -> list[ActorKey]:
    """주어진 Actor ID에 대한 키를 반환합니다."""
    if identifier == actor.id:
        return [ActorKey(key_id=actor.publicKey.id, private_key=private_key)]
    return []

# --- Server Initialization ---
app = ActivityPubServer(apkit_config=AppConfig(
    actor_keys=get_keys_for_actor  # 키 검색 함수 등록
))

5. 인메모리 저장소 및 캐시

생성된 활동을 제공하기 위해 어딘가에 저장해야 합니다. 간단하게 하기 위해 이 예제에서는 기본 인메모리 딕셔너리를 저장소와 캐시로 사용합니다. 프로덕션 애플리케이션에서는 이를 영구적인 데이터베이스(SQLite나 PostgreSQL 같은)와 적절한 캐시(Redis 같은)로 대체해야 합니다.

# main.py (continued)

# --- In-memory Store and Cache ---
ACTIVITY_STORE = {} # 생성된 활동을 저장하는 간단한 딕셔너리
CACHE = {}          # 최근에 접근한 활동을 위한 캐시
CACHE_TTL = timedelta(minutes=5) # 캐시 만료 시간(5분)

6. 리마인더 파싱 및 전송 로직

이것은 우리 봇의 핵심 로직입니다. parse_reminder 함수는 정규 표현식을 사용하여 멘션에서 지연 시간과 메시지를 추출하고, send_reminder는 알림을 예약합니다.

# main.py (continued)

# --- Reminder Parsing Logic ---
def parse_reminder(text: str) -> tuple[timedelta | None, str | None, str | None]:
    """'5m do something'와 같은 리마인더 텍스트를 파싱합니다."""
    # ... (간결함을 위해 구현 생략)

# --- Reminder Sending Function ---
async def send_reminder(ctx: Context, delay: timedelta, message: str, target_actor: APKitActor, original_note: Note):
    """지정된 지연 시간 후에 리마인더를 보냅니다."""
    logger.info(f"Scheduling reminder for {target_actor.id} in {delay}: '{message}'")
    await asyncio.sleep(delay.total_seconds()) # 비동기적으로 대기
    
    logger.info(f"Sending reminder to {target_actor.id}")
    
    # 리마인더 Note 생성
    reminder_note = Note(...)
    # Create 활동으로 감싸기
    reminder_create = Create(...)
    
    # 생성된 활동 저장
    ACTIVITY_STORE[reminder_note.id] = reminder_note
    ACTIVITY_STORE[reminder_create.id] = reminder_create
    
    # 대상 액터의 inbox로 활동 전송
    keys = await get_keys_for_actor(f"https://{HOST}/actor")
    await ctx.send(keys, target_actor, reminder_create)
    logger.info(f"Reminder sent to {target_actor.id}")

7. 엔드포인트 정의

필요한 ActivityPub 엔드포인트를 정의합니다. apkit는 FastAPI를 기반으로 하므로 표준 FastAPI 데코레이터를 사용할 수 있습니다. 주요 엔드포인트는 다음과 같습니다:

  • Webfinger: 다른 서버의 사용자가 @user@host와 같은 주소를 사용하여 봇을 발견할 수 있게 합니다. 이는 연합(federation)을 위한 중요한 첫 단계입니다.
  • /actor: 봇의 프로필 정보와 공개 키를 포함하는 봇의 Actor 객체를 제공합니다.
  • /inbox: 봇이 다른 서버로부터 활동을 수신하는 엔드포인트입니다. apkit는 이 라우트를 자동으로 처리하여 다음 단계에서 정의할 핸들러로 활동을 전달합니다.
  • /outbox: 봇이 생성한 활동의 컬렉션입니다. 하지만 이는 자리 표시자 컬렉션을 반환합니다.
  • /notes/{note_id} 및 /creates/{create_id}: 봇이 생성한 특정 객체를 제공하는 엔드포인트로, 다른 서버가 고유 ID로 이를 가져올 수 있게 합니다.

다음은 이러한 엔드포인트를 정의하는 코드입니다:

# main.py (continued)

# inbox 엔드포인트는 apkit에 의해 자동으로 처리됩니다.
app.inbox("/inbox")

@app.webfinger()
async def webfinger_endpoint(request: Request, acct: WebfingerResource) -> Response:
    """봇을 발견할 수 있도록 Webfinger 요청을 처리합니다."""
    if not acct.url:
        # acct:user@host와 같은 리소스 쿼리 처리
        if acct.username == USER_ID and acct.host == HOST:
            link = WebfingerLink(rel="self", type="application/activity+json", href=actor.id)
            wf_result = WebfingerResult(subject=acct, links=[link])
            return JSONResponse(wf_result.to_json(), media_type="application/jrd+json")
    else:
        # URL을 사용한 리소스 쿼리 처리
        if acct.url == f"https://{HOST}/actor":
            link = WebfingerLink(rel="self", type="application/activity+json", href=actor.id)
            wf_result = WebfingerResult(subject=acct, links=[link])
            return JSONResponse(wf_result.to_json(), media_type="application/jrd+json")
    return JSONResponse({"message": "Not Found"}, status_code=404)

@app.get("/actor")
async def get_actor_endpoint():
    """봇의 Actor 객체를 제공합니다."""
    return ActivityResponse(actor)

@app.get("/outbox")
async def get_outbox_endpoint():
    """봇이 보낸 활동의 컬렉션을 제공합니다."""
    items = sorted(ACTIVITY_STORE.values(), key=lambda x: x.id, reverse=True)
    outbox_collection = OrderedCollection(
        id=actor.outbox,
        totalItems=len(items),
        orderedItems=items
    )
    return ActivityResponse(outbox_collection)

@app.get("/notes/{note_id}")
async def get_note_endpoint(note_id: uuid.UUID):
    """특정 Note 객체를 캐싱과 함께 제공합니다."""
    note_uri = f"https://{HOST}/notes/{note_id}"
    
    # 먼저 캐시 확인
    if note_uri in CACHE and (datetime.now() - CACHE[note_uri]["timestamp"]) < CACHE_TTL:
        return ActivityResponse(CACHE[note_uri]["activity"])
        
    # 캐시에 없으면 저장소에서 가져오기
    if note_uri in ACTIVITY_STORE:
        activity = ACTIVITY_STORE[note_uri]
        # 반환하기 전에 캐시에 추가
        CACHE[note_uri] = {"activity": activity, "timestamp": datetime.now()}
        return ActivityResponse(activity)
        
    return Response(status_code=404) # Not Found

@app.get("/creates/{create_id}")
async def get_create_endpoint(create_id: uuid.UUID):
    """특정 Create 활동을 캐싱과 함께 제공합니다."""
    create_uri = f"https://{HOST}/creates/{create_id}"
    
    if create_uri in CACHE and (datetime.now() - CACHE[create_uri]["timestamp"]) < CACHE_TTL:
        return ActivityResponse(CACHE[create_uri]["activity"])
        
    if create_uri in ACTIVITY_STORE:
        activity = ACTIVITY_STORE[create_uri]
        CACHE[create_uri] = {"activity": activity, "timestamp": datetime.now()}
        return ActivityResponse(activity)
        
    return Response(status_code=404)

8. 활동 핸들러

@app.on() 데코레이터를 사용하여 우리의 inbox에 게시된 특정 활동 유형에 대한 핸들러를 정의합니다.

  • on_follow_activity: Follow 요청을 자동으로 수락합니다.
  • on_create_activity: 들어오는 Create 활동(특히 Note 객체)을 파싱하여 리마인더를 예약합니다.
# main.py (continued)

# Follow 활동에 대한 핸들러
@app.on(Follow)
async def on_follow_activity(ctx: Context):
    """팔로우 요청을 자동으로 수락합니다."""
    # ... (간결함을 위해 구현 생략)

# Create 활동에 대한 핸들러
@app.on(Create)
async def on_create_activity(ctx: Context):
    """멘션을 파싱하여 리마인더를 예약합니다."""
    activity = ctx.activity
    # Note가 아니면 무시
    if not (isinstance(activity, Create) and isinstance(activity.object, Note)):
        return Response(status_code=202)

    note = activity.object
    
    # 봇이 멘션되었는지 확인
    is_mentioned = any(
        isinstance(tag, Mention) and tag.href == actor.id for tag in (note.tag or [])
    )
    
    if not is_mentioned:
        return Response(status_code=202)

    # ... (리마인더 텍스트 파싱)
    delay, message, time_str = parse_reminder(command_text)

    # 파싱이 성공하면 백그라운드 태스크로 리마인더 예약
    if delay and message and sender_actor:
        asyncio.create_task(send_reminder(ctx, delay, message, sender_actor, note))
        reply_content = f"<p>✅ OK! I will remind you in {time_str}.</p>"
    else:
        # 파싱이 실패하면 사용법 안내 전송
        reply_content = "<p>🤔 Sorry, I didn\'t understand. Please use the format: `@reminder [time] [message]`.</p><p>Example: `@reminder 10m Check the oven`</p>"

    # ... (응답 Note 생성 및 전송)

9. 애플리케이션 실행

마지막으로, uvicorn을 사용하여 애플리케이션을 실행합니다.

# main.py (continued)

if __name__ == "__main__":
    import uvicorn
    logger.info("Starting uvicorn server...")
    uvicorn.run(app, host="0.0.0.0", port=8000)

봇 실행 방법

  1. main.py에서 HOST와 USER_ID 변수를 환경에 맞게 설정합니다.

  2. 터미널에서 서버를 실행합니다:

    uvicorn main:app --host 0.0.0.0 --port 8000

  3. 봇이 http://0.0.0.0:8000에서 실행됩니다.

이제 Fediverse 어디에서나 봇을 멘션하여(예: @reminder@your.host.com) 리마인더를 설정할 수 있습니다.

다음 단계

이 튜토리얼은 간단한 ActivityPub 봇을 만드는 기본 사항을 다룹니다. 인메모리 저장소만 사용하기 때문에 서버가 재시작되면 모든 리마인더가 손실됩니다. 다음은 몇 가지 잠재적인 개선 사항입니다:

  • 영구 저장소: 인메모리 ACTIVITY_STORE를 SQLite나 PostgreSQL과 같은 데이터베이스로 대체합니다.
  • 강력한 태스크 큐잉: 서버가 재시작되어도 리마인더가 손실되지 않도록 Redis나 RabbitMQ 브로커와 함께 Celery와 같은 전용 태스크 큐를 사용합니다.
  • 고급 명령: 반복 리마인더와 같은 더 복잡한 명령에 대한 지원을 추가합니다.

이 가이드가 여러분만의 ActivityPub 애플리케이션을 구축하는 데 좋은 출발점이 되기를 바랍니다!

https://fedi-libs.github.io/apkit/

https://github.com/fedi-libs/apkit

https://github.com/AmaseCocoa/activitypub-reminder-bot

15
1
0

1 comment

If you have a fediverse account, you can comment on this article from your own instance. Search https://hackers.pub/ap/articles/01994967-5928-7a5d-b7bd-9eac2425e8a5 on your instance and reply to it.

0
0
0