added agents

agents/calendar-api/README.md

@@ -0,0 +1,24 @@
+# calendar-api
+
+A simple calendar API powered by FastAPI.
+
+## Architecture
+
+```md
+- database
+	- build.py
+	- database_handler.py
+- server
+	- method.py
+	- server.py
+- client
+	- client.py
+```
+
+## Usage
+
+1. Clone this repo from GitHub
+2. Install Python packages: `fastapi` and `uvicorn`
+3. Run `python build.py` to create a SQLite database (only run for the first time)
+4. Run `uvicorn server:app --reload`
+5. Run test code in `client.py`, or try it out on `http://127.0.0.1:8000/docs`

agents/calendar-api/build.py

@@ -0,0 +1,19 @@
+import configparser
+import json
+import database_handler
+
+
+def build_bd():
+    config = configparser.ConfigParser()
+    config.read('db.conf')
+    info = config['DEFAULT']
+
+    dbh = database_handler.DatabaseHandler(db_name=info['db_name'])
+
+    dbh.create_table(
+        table_name=info['table_name'],
+        columns=json.loads(info['columns']))
+
+
+if __name__ == '__main__':
+    build_bd()

agents/calendar-api/client.py

@@ -0,0 +1,109 @@
+import requests
+import json
+
+
+class Interface:
+    """client interface"""
+
+    def __init__(self, url, app_name):
+        self.url = url
+        self.app_url = url + '/' + app_name
+
+    def get(self, url, sid):
+        """get schedule"""
+        url = url + '/' + sid
+        return requests.get(url)
+
+    def get_all(self, url, payload=None):
+        """get schedules"""
+        return requests.get(url, params=payload)
+
+    def post(self, url, data):
+        """post schedule"""
+        return requests.post(url, data=json.dumps(data))
+
+    def update(self, url, data):
+        """update schedule"""
+        sid = data['sid']
+        url = url + '/' + sid
+        return requests.put(url, data=json.dumps(data))
+
+    def delete(self, url, sid):
+        """delete schedule"""
+        url = url + '/' + sid
+        return requests.delete(url)
+
+
+if __name__ == '__main__':
+    url = "http://127.0.0.1:8000"
+    i = Interface(url, 'schedules')
+
+    # --------------------------------------
+    # get all items
+    # --------------------------------------
+
+    # r = i.get_all(i.app_url)
+
+    # --------------------------------------
+    # get an item
+    # --------------------------------------
+
+    # sid = '1'
+    # r = i.get(i.app_url, sid)
+
+    # --------------------------------------
+    # post an item
+    # --------------------------------------
+
+    data = {
+        "sid": "1",
+        "name": "John Doe",
+        "content": "Test",
+        "category": "Business",
+        "level": 1,
+        "status": 0,
+        "creation_time": "2024-11-12 01:23:45",
+        "start_time": "2024-11-14 03:20:00",
+        "end_time": "2024-11-14 05:30:00"
+    }
+
+    r = i.post(i.app_url, data=data)
+
+    # --------------------------------------
+    # update an item
+    # --------------------------------------
+
+    # data = {
+    #     "sid": "1",
+    #     "name": "Work",
+    #     "content": "Shopping",
+    #     "category": "Business",
+    #     "level": 1,
+    #     "status": 0,
+    #     "creation_time": "2024-11-11 01:23:45",
+    #     "start_time": "2024-11-12 03:20:00",
+    #     "end_time": "2024-11-12 05:30:00"
+    # }
+
+    # r = i.update(i.app_url, data=data)
+
+    # --------------------------------------
+    # delete an item
+    # --------------------------------------
+
+    # sid = "1"
+    # r = i.delete(i.app_url, sid)
+
+    # --------------------------------------
+    # log
+    # --------------------------------------
+
+    print(r)
+    print(r.url)
+    print(r.ok)
+    print(r.content, type(r.content))
+    print(r.text, type(r.text))
+
+    # import ast
+    # t = ast.literal_eval(r.text)
+    # print(t, type(t))

agents/calendar-api/database_handler.py

@@ -0,0 +1,116 @@
+import sqlite3
+
+
+class DatabaseHandler:
+    """database handler"""
+
+    def __init__(self, db_name: str, check_same_thread: bool = True):
+        """init"""
+        self.db_name = db_name
+        self.conn = sqlite3.connect(
+            '{}.db'.format(db_name), check_same_thread=check_same_thread)
+        self.c = self.conn.cursor()
+
+    def execute(self, cmd: str):
+        """Execute command"""
+        self.c.execute(cmd)
+        self.conn.commit()
+
+    def create_table(self, table_name: str, columns: dict):
+        """Create table"""
+        lst = [str(k) + ' ' + str(v) for k, v in columns.items()]
+        columns_str = ','.join(lst)
+
+        cmd = 'CREATE TABLE {table_name}({columns_str})'
+
+        self.execute(cmd.format(
+            table_name=table_name,
+            columns_str=columns_str))
+
+    def insert_data(self, table_name: str, columns: dict, data: dict):
+        """Insert a row of data"""
+        lst = ["'" + str(v) + "'" if columns[k] == 'TEXT' else str(v)
+               for k, v in data.items()]
+        data_str = ','.join(lst)
+
+        cmd = 'INSERT INTO {table_name} VALUES ({data_str})'
+
+        self.execute(cmd.format(
+            table_name=table_name,
+            data_str=data_str))
+
+    def update_data(self, table_name: str, columns: dict, data: dict, condition: dict):
+        """Update data"""
+        lst1 = [str(k) + '=' + "'" + str(v) + "'" if columns[k] == 'TEXT'
+                else str(k) + '=' + str(v)
+                for k, v in data.items()]
+        value_str = ','.join(lst1)
+
+        lst2 = [str(k) + '=' + "'" + str(v) + "'" if columns[k] == 'TEXT'
+                else str(k) + '=' + str(v)
+                for k, v in condition.items()]
+        condition_str = ' AND '.join(lst2)
+
+        cmd = 'UPDATE {table_name} SET {value_str} WHERE {condition_str}'
+
+        self.execute(cmd.format(
+            table_name=table_name,
+            value_str=value_str,
+            condition_str=condition_str))
+
+    def delete_data(self, table_name: str, columns: dict, condition: dict):
+        """Delete data"""
+        lst = [str(k) + '=' + "'" + str(v) + "'" if columns[k] == 'TEXT'
+               else str(k) + '=' + str(v)
+               for k, v in condition.items()]
+        condition_str = ' AND '.join(lst)
+
+        cmd = 'DELETE FROM {table_name} WHERE {condition_str}'
+
+        self.execute(cmd.format(
+            table_name=table_name,
+            condition_str=condition_str))
+
+    def fetch_data(self, table_name: str, columns: dict, condition: dict):
+        """Fetch data"""
+        lst = [str(k) + '=' + "'" + str(v) + "'" if columns[k] == 'TEXT'
+               else str(k) + '=' + str(v)
+               for k, v in condition.items()]
+        condition_str = ' AND '.join(lst)
+
+        cmd = 'SELECT * FROM {table_name} WHERE {condition_str}'
+
+        self.execute(cmd.format(
+            table_name=table_name,
+            condition_str=condition_str))
+
+        return self.c.fetchall()
+
+    def fetch_all(self, table_name: str):
+        """Fetch data"""
+        cmd = 'SELECT * FROM {table_name}'
+
+        self.execute(cmd.format(
+            table_name=table_name))
+
+        return self.c.fetchall()
+
+    def check_existence(self, table_name: str, columns: dict, condition: dict):
+        """check the existence of item"""
+        try:
+            res = self.fetch_data(table_name, columns, condition)
+            if len(res) == 0:
+                return False
+        except Exception:
+            return False
+        return True
+
+
+if __name__ == '__main__':
+    dbh = DatabaseHandler(db_name="CalendarDB")
+    print(dbh.check_existence(
+        'calendar',
+        {"sid": "TEXT", "name": "TEXT", "content": "TEXT", "category": "TEXT", "level": "INTEGER",
+            "status": "REAL", "creation_time": "TEXT", "start_time": "TEXT", "end_time": "TEXT"},
+        {"sid": "22"}
+    ))

agents/calendar-api/db.conf

@@ -0,0 +1,4 @@
+[DEFAULT]
+db_name = CalendarDB
+table_name = calendar
+columns = {"sid": "TEXT", "name": "TEXT", "content": "TEXT", "category": "TEXT", "level": "INTEGER", "status": "REAL", "creation_time": "TEXT", "start_time": "TEXT", "end_time": "TEXT"}

agents/calendar-api/method.py

@@ -0,0 +1,88 @@
+import configparser
+import json
+import datetime
+
+
+class Method:
+    """Method"""
+    def __init__(self, conf_file):
+        """init"""
+        self.config = configparser.ConfigParser()
+        self.config.read(conf_file)  # 'db.conf'
+
+        self.info = self.config['DEFAULT']
+        self.columns = json.loads(self.info['columns'])
+
+    def check_params(self, jsn):
+        """检查参数值"""
+        if jsn['level'] not in [0, 1, 2, 3]:
+            return False
+
+        if jsn['status'] < 0 or jsn['status'] > 1:
+            return False
+
+        try:
+            lst = [
+                jsn['creation_time'],
+                jsn['start_time'],
+                jsn['end_time']
+            ]
+
+            for t in lst:
+                _ = datetime.datetime.strptime(t, '%Y-%m-%d %H:%M:%S')
+
+        except Exception:
+            return False
+
+        return True
+
+    def get(self, dbh, schedule_id):
+        return dbh.fetch_data(
+            table_name=self.info['table_name'],
+            columns=self.columns,
+            condition={'sid': schedule_id})
+
+    def post(self, dbh, schedule):
+        schedule_id = schedule.dict()['sid']
+        if dbh.check_existence(self.info['table_name'], self.columns, {'sid': schedule_id}):
+            return False
+
+        if not self.check_params(schedule.dict()):
+            return False
+
+        dbh.insert_data(
+            table_name=self.info['table_name'],
+            columns=self.columns,
+            data=schedule.dict())
+
+        return True
+
+    def update(self, dbh, schedule_id, schedule):
+        if not dbh.check_existence(self.info['table_name'], self.columns, {'sid': schedule_id}):
+            return False
+
+        if not self.check_params(schedule.dict()):
+            return False
+
+        dbh.update_data(
+            table_name=self.info['table_name'],
+            columns=self.columns,
+            data=schedule.dict(),
+            condition={'sid': schedule_id})
+
+        return True
+
+    def delete(self, dbh, schedule_id):
+        if not dbh.check_existence(self.info['table_name'], self.columns, {'sid': schedule_id}):
+            return False
+
+        dbh.delete_data(
+            table_name=self.info['table_name'],
+            columns=self.columns,
+            condition={'sid': schedule_id})
+
+        return True
+
+
+if __name__ == '__main__':
+    m = Method(conf_file='db.conf')