1
 2
 3
 4
 5
 6
 7
 8
 9
10

clean-bot/
├── lib/                      # Core modules
│   ├── clean_task.py         # Task state management
│   ├── message.py            # Command parser
│   ├── line.py               # LINE API
│   └── s3_client.py          # S3 operations
├── test/                     # Tests
├── line_clean_bot.py         # Lambda entry point
├── template.yaml             # SAM template
└── Makefile                  # Development commands

 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

# line_clean_bot.py

def process_user_message(event, context):
    """LINE message webhook handler"""
    body = json.loads(event.get('body'))

    # Get user ID or group ID
    if body['events'][0]['source']['type'] == 'user':
        line_id = body['events'][0]['source']['userId']
    else:
        line_id = body['events'][0]['source']['groupId']

    message = body['events'][0]['message']['text']

    # Fetch user data from S3
    s3client = S3client(BUCKET_NAME)
    obj_key = line_id + '.json'
    if not s3client.check_exist_object(obj_key):
        s3client.update_object(obj_key, '{"tasks": []}')

    # Process message and reply
    # ...

def push_message_periodically(event, context):
    """Send reminder notifications on schedule"""
    current_time = datetime.now() + timedelta(hours=9)  # JST

    s3client = S3client(BUCKET_NAME)
    for obj_list in s3client.list_objects():
        clean_task = CleanTask(s3client.get_object_body(obj_list.key))

        # Check if notification conditions are met
        if clean_task.should_notify(current_time):
            # Send message
            # ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12

# lib/clean_task.py

def __evaluate_cleanup_timing(self, task):
    """Determine if task is overdue"""
    task_time = datetime.strptime(task['updated_at'], self.date_format)
    return (self.now - task_time).days >= int(task['duration'])

def get_todo_tasks(self):
    """Get list of overdue tasks"""
    return [task for task in self.tasks
            if not task.get('paused', False)
            and self.__evaluate_cleanup_timing(task)]

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19

def should_notify(self, current_time):
    """Determine if notification should be sent"""
    if not self.notification['enabled']:
        return False

    # Check day of week
    weekday_map = {0: 'Mon', 1: 'Tue', 2: 'Wed', 3: 'Thu', 4: 'Fri', 5: 'Sat', 6: 'Sun'}
    current_weekday = weekday_map[current_time.weekday()]
    if current_weekday not in self.notification['days']:
        return False

    # Check time
    if current_time.hour < self.notification['hour']:
        return False

    # Check if already notified today
    # ...

    return True

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

# lib/message.py

def get_return_message(self, message, s3client):
    task_operation = self.__get_task_operation_name(message)  # First word
    task_name = self.__get_task_name(message)                  # Second word

    if task_operation == 'done':
        for task_name in self.__get_all_task_name(message):
            self.clean_task.update_task_updated_at(task_name)
        s3client.update_object(self.object_keyname, self.clean_task.get_json())
        return "Completed"

    if task_operation == 'add':
        duration = self.__get_duration(message)  # Third word
        self.clean_task.add_task(task_name, duration)
        # ...

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

{
  "tasks": [
    {
      "task_name": "vacuum",
      "updated_at": "2024-01-01 12:00:00",
      "duration": 7,
      "paused": false
    }
  ],
  "notification": {
    "enabled": true,
    "days": ["Mon", "Wed", "Fri"],
    "hour": 7,
    "last_notified_at": "2024-01-01 07:00:00"
  }
}

 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

# template.yaml
AWSTemplateFormatVersion: '2010-09-09'
Transform: AWS::Serverless-2016-10-31

Parameters:
  ChannelAccessToken:
    Type: String
    NoEcho: true
  BucketName:
    Type: String
    Default: your-bucket-name

Globals:
  Function:
    Runtime: python3.11
    Timeout: 300
    Environment:
      Variables:
        CHANNEL_ACCESS_TOKEN: !Ref ChannelAccessToken
        BUCKET_NAME: !Ref BucketName

Resources:
  # Message processing Lambda
  ProcessUserMessageFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: line_clean_bot.process_user_message
      CodeUri: .
      Policies:
        - S3CrudPolicy:
            BucketName: !Ref BucketName
      Events:
        ApiEvent:
          Type: HttpApi
          Properties:
            Path: /process_user_message
            Method: ANY

  # Scheduled notification Lambda
  PushMessagePeriodicallyFunction:
    Type: AWS::Serverless::Function
    Properties:
      Handler: line_clean_bot.push_message_periodically
      CodeUri: .
      Policies:
        - S3CrudPolicy:
            BucketName: !Ref BucketName
      Events:
        ScheduleEvent:
          Type: Schedule
          Properties:
            Schedule: cron(0 * * * ? *)  # Every hour at minute 0

1
2
3
4
5

# Build
sam build

# Deploy
sam deploy --parameter-overrides ChannelAccessToken=your_token

1
2
3
4
5
6
7

# Start local API server
sam local start-api --env-vars env.json

# Run tests
curl -X POST http://localhost:3000/process_user_message \
  -H "Content-Type: application/json" \
  -d @events/line_message.json