「覚えたつもり」を卒業したい！LINE通知で記憶に刻み込むサーバーレスアプリ

1度で覚えられません！

私は知らないことがあったときに意味を調べますが、１度で覚えられずまた調べてしまうことが多いです。

そこで、今回は記憶に定着させるためのアプリをWeb・LINE・自動通知の3つで、復習サイクルを回せるようにサーバーレスで実装してみました。

Webでは単語の登録、登録日や単語での検索ができます。

また、lineから単語を登録することも可能で、登録した単語を送るとその意味が返ってくるようになっています。

その日登録した単語の一覧が自動でlineに通知されます。

アーキテクチャ

 [ブラウザ] index.html
       │ 
       ▼
 [API Gateway]──> [Lambda: api_handler.py]
       │                   │
       │                   ▼
       │               [DynamoDB]
       │
       ├──> [Lambda: line_webhook.py] <── LINE Bot
       │                   │
       │                   ▼
       │               [DynamoDB]
       │
       └──> [EventBridge]──> [Lambda: notifier.py]──> LINE Push通知

実装

↓ 全体のコードです。

api_handler.pyのコード

import os, json, boto3, datetime, zoneinfo
from boto3.dynamodb.conditions import Key

WORDS_TABLE = os.environ["WORDS_TABLE"]
dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(WORDS_TABLE)

def _res(status, body):
    return {"statusCode": status,
            "headers": {"Content-Type":"application/json","Access-Control-Allow-Origin":"*"},
            "body": json.dumps(body, ensure_ascii=False)}

def handler(event, context):
    path = event.get("path", "")
    method = event.get("httpMethod", "GET")
    body = event.get("body")
    qsp = event.get("queryStringParameters") or {}

    # 単語追加
    if path.endswith("/words") and method=="POST":
        data = json.loads(body or "{}")
        word, meaning = data.get("word"), data.get("meaning")
        now = datetime.datetime.now(zoneinfo.ZoneInfo("Asia/Tokyo"))
        date = now.strftime("%Y-%m-%d")
        created_at = now.isoformat()
        item = {
            "pk": date, "sk": created_at,
            "word": word, "meaning": meaning,
            "date": date, "createdAt": created_at,
            "GSI1PK": "words", "GSI1SK": word,
            "GSI2PK": "words", "GSI2SK": created_at
        }
        table.put_item(Item=item)
        return _res(200, {"ok":True})

    # 一覧取得（今日 or 最近）
    if path.endswith("/words") and method=="GET":
        date = qsp.get("date") if qsp else None
        if date:
            res = table.query(KeyConditionExpression=Key("pk").eq(date))
            return _res(200, res.get("Items", []))
        else:
            res = table.query(IndexName="GSI2",
                              KeyConditionExpression=Key("GSI2PK").eq("words"),
                              ScanIndexForward=False, Limit=50)
            return _res(200, res.get("Items", []))

    # 検索（前方一致）
    if path.endswith("/words/search") and method=="GET":
        q = qsp.get("word","")
        res = table.query(IndexName="GSI1",
                          KeyConditionExpression=Key("GSI1PK").eq("words") & Key("GSI1SK").begins_with(q))
        return _res(200, res.get("Items", []))

    return _res(404, {"error":"not found"})

line_webhook.pyのコード

import os, json, base64, hmac, hashlib, urllib.request, urllib.error, logging
import boto3
from boto3.dynamodb.conditions import Key

logger = logging.getLogger()
logger.setLevel(logging.INFO)

WORDS_TABLE = os.environ["WORDS_TABLE"]
LINE_CHANNEL_SECRET = os.environ["LINE_CHANNEL_SECRET"]
LINE_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]

ddb = boto3.resource("dynamodb")
table = ddb.Table(WORDS_TABLE)

def _reply(reply_token: str, text: str):
    url = "https://api.line.me/v2/bot/message/reply"
    body = {"replyToken": reply_token, "messages": [{"type": "text", "text": text}]}
    req = urllib.request.Request(url, method="POST")
    req.add_header("Content-Type", "application/json")
    req.add_header("Authorization", f"Bearer {LINE_TOKEN}")
    with urllib.request.urlopen(req, data=json.dumps(body).encode("utf-8")) as res:
        logger.info("LINE reply status: %s", res.status)

def _verify_signature(event_body: str, signature: str) -> bool:
    mac = hmac.new(LINE_CHANNEL_SECRET.encode("utf-8"),
                   event_body.encode("utf-8"),
                   hashlib.sha256).digest()
    expected = base64.b64encode(mac).decode("utf-8")
    return hmac.compare_digest(expected, signature)

def _lookup_meaning(word: str) -> str | None:
    # 完全一致
    res = table.query(
        IndexName="GSI1",
        KeyConditionExpression=Key("GSI1PK").eq("words") & Key("GSI1SK").eq(word)
    )
    items = res.get("Items", [])
    if items:
        items.sort(key=lambda x: x.get("createdAt", ""), reverse=True)
        return items[0].get("meaning")

    # 前方一致の候補（最大5件）
    res2 = table.query(
        IndexName="GSI1",
        KeyConditionExpression=Key("GSI1PK").eq("words") & Key("GSI1SK").begins_with(word),
        Limit=5
    )
    cand = [f"{x.get('word','?')}：{x.get('meaning','')}" for x in res2.get("Items",[])]
    if cand:
        return "候補:\n" + "\n".join(f"・{c}" for c in cand)

    return None

def _try_register(text: str) -> str | None:
    """
    「単語:意味」または「単語＝意味」で来たら登録する（任意の拡張）
    例: "apple:りんご", "apple＝りんご"
    """
    sep = ":" if ":" in text else "＝" if "＝" in text else None
    if not sep: 
        return None
    w, m = [s.strip() for s in text.split(sep, 1)]
    if not w or not m:
        return "登録形式は「単語:意味」です。例: apple:りんご"

    import datetime, zoneinfo
    now = datetime.datetime.now(zoneinfo.ZoneInfo("Asia/Tokyo"))
    pk = now.strftime("%Y-%m-%d")
    created_at = now.isoformat(timespec="milliseconds")

    # 例: 既存のスキーマに合わせて保存（必要に応じて項目名は調整）
    item = {
        "pk": pk,
        "sk": created_at,
        "word": w,
        "meaning": m,
        "createdAt": created_at,
        "date": pk,
        # GSI1 用キー
        "GSI1PK": "words",
        "GSI1SK": w,
        # GSI2 用キー
        "GSI2PK": "words",
        "GSI2SK": created_at,
    }
    table.put_item(Item=item)
    return f"登録しました：{w}＝{m}"

def handler(event, context):
    body = event.get("body") or ""
    headers = {k.lower(): v for k, v in (event.get("headers") or {}).items()}
    signature = headers.get("x-line-signature", "")

    if not _verify_signature(body, signature):
        logger.warning("invalid signature")
        return {"statusCode": 403, "body": "invalid signature"}

    j = json.loads(body)
    for ev in j.get("events", []):
        if ev.get("type") == "message" and ev.get("message", {}).get("type") == "text":
            reply_token = ev.get("replyToken")
            text = ev["message"]["text"].strip()

            # 1) 「単語:意味」形式なら登録
            reg = _try_register(text)
            if reg:
                _reply(reply_token, reg)
                continue

            # 2) 検索（完全一致→候補）
            meaning = _lookup_meaning(text)
            if meaning is None:
                _reply(reply_token, f"「{text}」は見つかりませんでした。")
            else:
                _reply(reply_token, f"{text}：{meaning}")

    return {"statusCode": 200, "body": "ok"}

notifier.pyのコード

import os, json, boto3, datetime, urllib.request, zoneinfo
from boto3.dynamodb.conditions import Key

WORDS_TABLE = os.environ["WORDS_TABLE"]
LINE_TOKEN = os.environ["LINE_CHANNEL_ACCESS_TOKEN"]
LINE_USER_ID = os.environ["LINE_USER_ID"]

dynamodb = boto3.resource("dynamodb")
table = dynamodb.Table(WORDS_TABLE)

def push_line(text):
    body = {"to": LINE_USER_ID, "messages":[{"type":"text","text":text}]}
    req = urllib.request.Request("https://api.line.me/v2/bot/message/push", method="POST")
    req.add_header("Content-Type","application/json")
    req.add_header("Authorization",f"Bearer {LINE_TOKEN}")
    urllib.request.urlopen(req, data=json.dumps(body).encode("utf-8"))

def handler(event, context):
    now = datetime.datetime.now(zoneinfo.ZoneInfo("Asia/Tokyo"))
    today = now.strftime("%Y-%m-%d")

    res = table.query(KeyConditionExpression=Key("pk").eq(today))
    items = res.get("Items", [])
    if not items:
        msg = f"{today} の新しい単語はありません"
    else:
        lines = [f"【{today}の単語】"] + [f"・{x['word']}：{x['meaning']}" for x in items]
        msg = "\n".join(lines)

    push_line(msg)
    return {"ok":True}

index.htmlのコード

<!doctype html>
<html lang="ja">
<head>
  <meta charset="utf-8">
  <title>単語メモ</title>
  <style>
    body {
      font-family: 'Helvetica Neue', Arial, sans-serif;
      background: #f4f6f9;
      color: #333;
      display: flex;
      justify-content: center;
      align-items: flex-start;
      padding: 40px;
    }
    .container {
      background: #fff;
      padding: 30px;
      border-radius: 12px;
      box-shadow: 0 4px 20px rgba(0,0,0,0.1);
      width: 500px;
    }
    h1 {
      text-align: center;
      margin-bottom: 20px;
      color: #4a90e2;
    }
    input, textarea {
      padding: 10px;
      margin: 5px 0;
      width: calc(100% - 22px);
      border: 1px solid #ccc;
      border-radius: 6px;
      font-size: 14px;
    }
    textarea {
      min-height: 80px; /* ← 高さを指定 */
      resize: vertical; /* ← ユーザーが縦方向にリサイズできるように */
    }
    button {
      background: #4a90e2;
      color: white;
      border: none;
      padding: 10px 16px;
      border-radius: 6px;
      cursor: pointer;
      font-size: 14px;
      margin: 5px 5px 10px 0;
    }
    button:hover {
      background: #357ab8;
    }
    #list {
      list-style: none;
      padding: 0;
    }
    #list li {
      background: #f9f9f9;
      margin: 8px 0;
      padding: 10px;
      border-radius: 6px;
      border-left: 5px solid #4a90e2;
    }
  </style>
</head>
<body>
  <div class="container">
    <h1>単語メモ</h1>
    <input id="word" placeholder="単語">
    <textarea id="meaning" placeholder="意味（文章も入力可）"></textarea>
    <button id="add">追加</button>
    <br>
    <input id="date" placeholder="YYYY-MM-DD">
    <button id="load">読み込み</button>
    <br>
    <input id="q" placeholder="検索語">
    <button id="search">検索</button>
    <ul id="list"></ul>
  </div>
  <script>
    const API="https://xxxx.execute-api.ap-northeast-1.amazonaws.com/prod";
    add.onclick=async()=>{
      await fetch(API+"/words",{
        method:"POST",
        headers:{"Content-Type":"application/json"},
        body:JSON.stringify({word:word.value,meaning:meaning.value})
      });
      alert("追加しました");
    }
    load.onclick=async()=>{
      const d=date.value;
      const url=d?API+"/words?date="+d:API+"/words";
      const r=await fetch(url);
      render(await r.json());
    }
    search.onclick=async()=>{
      const r=await fetch(API+"/words/search?word="+q.value);
      render(await r.json());
    }
    function render(items){
      list.innerHTML="";
      items.forEach(x=>{
        const li=document.createElement("li");
        li.textContent=x.word+"："+x.meaning;
        list.appendChild(li)
      })
    }
  </script>
</body>
</html>

DynamoDBテーブル作成

単語データを保存・検索するためのデータベースを作成します。
検索効率を高めるために日付＋GSIを設計し、学習履歴を自在に取得できます。

1. AWSコンソール → DynamoDB → 「テーブルの作成」
2. テーブル名: WordsTable
3. パーティションキー: pk(文字列)
4. ソートキー: sk(文字列)
5. 「追加のインデックス」からGSI1、GSI2を作成
   1. GSI1: GSI1PK(PK)、GSI1SK(SK) → 検索用
   2. GSI2: GSI2PK(PK)、GSI2SK(SK) → 最新取得用

Lambda関数作成

アプリの主要な処理ロジックを担うサーバーレス関数を用意します。
Web API処理、LINE Webhook処理、定期通知処理を分けて管理します。

APIリクエスト処理(Lambda: api_handler.py)

• 用途: WEBフロント用API
• 環境変数
  • WORDS_TABLE = WordsTable

Line Bot応答・登録処理(Lambda: line_webhook)

• 用途: LINE BotのWebhookを処理
• 環境変数
  • WORDS_TABLE = WordsTable
  • LINE_CHANNEL_SECRET = <LINEチャンネルシークレット>
  • LINE_CHANNEL_ACCESS_TOKEN = <LINEアクセストークン>

自動通知(Lambda: notifier.py)

• 用途: 毎朝その日の単語をLINEへ通知
• 環境変数
  • WORDS_TABLE = WordsTable
  • LINE_CHANNEL_ACCESS_TOKEN = <LINEアクセストークン>
  • LINE_USER_ID = <通知したいユーザーID>

API Gateway

WebフロントからLambdaを呼び出すためのREST APIを構築します。
エンドポイントを通して「単語追加・検索・取得」が可能になります。

1. API Gateway → REST API → 新規作成
2. リソース/wordsを作成
   1. GET: Lambda(api_handler)
   2. POST: Lambda(api_handler)
3. リソース/words/searchを作成
   1. GET: Lambda(api_handler)
4. ステージを作成
   1. エンドポイントURLを控えておく(https://xxxx.execute-api.ap-northeast-1.amazonaws.com/prod)

EventBridgeルール設定

毎日にLambdaを自動実行するスケジューラを作成します。
これにより毎朝自動で「その日の単語リスト」がLINEに届きます。

1. EventBridge → 「ルールを作成」
2. スケジュールパターン
   1. 頻度: 定期的なスケジュール
   2. タイムゾーン: (UTF+09:00)Asia/Tokyo
   3. スケジュールの種類: rateベースのスケジュール
   4. rate式: rate(1, days)

S3 + CloudFront(任意)

このアプリは単一のHTMLファイルなので、ローカルで開いても動作します。
他の人に配布したい場合や常時公開したい場合にのみ、S3+CloudFrontでのホスティングが有効です。

LINE Developers設定

LINE公式アカウントを作成し、WebhookとLambdaを接続します。
これにより、ユーザーのメッセージをLambdaが受け取って処理できます。

1. Line Developers → Messaging APIチャンネル作成
2. Webhook URLにLambda(line_webhook)で作成した関数URLを入れる
3. 「応答メッセージ」OFF、「Webhook送信」ON
4. アクセストークンとシークレットをLambdaに設定済みか確認

まとめ

今回は自分用で作成してみましたが、DynamoDBを変更すればユーザーごとの管理もできますし、クイズ形式で問題を出して正答率から苦手の単語を見極めたりもできると思います！