ساخت چت‌باتی برای تعامل با Amazon S3

1AWS_ACCESS_KEY_ID=<your-key>
2AWS_SECRET_ACCESS_KEY=<your-key>
3GILAS_API_KEY=<your-key>

1! pip install openai
2! pip install boto3
3! pip install tenacity
4! pip install python-dotenv

 1from openai import OpenAI
 2import json
 3import boto3
 4import os
 5import datetime
 6from urllib.request import urlretrieve
 7
 8# load environment variables
 9from dotenv import load_dotenv
10load_dotenv() 
11
12# Create openai client
13client = OpenAI(
14   api_key=os.environ.get(("GILAS_API_KEY", "<کلید API خود را اینجا بسازید https://dashboard.gilas.io/apiKey>")), 
15   base_url="https://api.gilas.io/v1/" # Gilas APIs
16)
17
18GPT_MODEL = "gpt-4o-mini"
19
20# Optional - if you had issues loading the environment file, you can set the AWS values using the below code
21# os.environ['AWS_ACCESS_KEY_ID'] = ''
22# os.environ['AWS_SECRET_ACCESS_KEY'] = ''
23
24# Create S3 client
25s3_client = boto3.client('s3')

 1# Functions dict to pass S3 operations details for the GPT model
 2functions = [
 3    {   
 4        "type": "function",
 5        "function":{
 6            "name": "list_buckets",
 7            "description": "List all available S3 buckets",
 8            "parameters": {
 9                "type": "object",
10                "properties": {}
11            }
12        }
13    },
14    {
15        "type": "function",
16        "function":{
17            "name": "list_objects",
18            "description": "List the objects or files inside a given S3 bucket",
19            "parameters": {
20                "type": "object",
21                "properties": {
22                    "bucket": {"type": "string", "description": "The name of the S3 bucket"},
23                    "prefix": {"type": "string", "description": "The folder path in the S3 bucket"},
24                },
25                "required": ["bucket"],
26            },
27        }
28    },
29    {   
30        "type": "function",
31        "function":{
32            "name": "download_file",
33            "description": "Download a specific file from an S3 bucket to a local distribution folder.",
34            "parameters": {
35                "type": "object",
36                "properties": {
37                    "bucket": {"type": "string", "description": "The name of the S3 bucket"},
38                    "key": {"type": "string", "description": "The path to the file inside the bucket"},
39                    "directory": {"type": "string", "description": "The local destination directory to download the file, should be specificed by the user."},
40                },
41                "required": ["bucket", "key", "directory"],
42            }
43        }
44    },
45    {
46        "type": "function",
47        "function":{
48            "name": "upload_file",
49            "description": "Upload a file to an S3 bucket",
50            "parameters": {
51                "type": "object",
52                "properties": {
53                    "source": {"type": "string", "description": "The local source path or remote URL"},
54                    "bucket": {"type": "string", "description": "The name of the S3 bucket"},
55                    "key": {"type": "string", "description": "The path to the file inside the bucket"},
56                    "is_remote_url": {"type": "boolean", "description": "Is the provided source a URL (True) or local path (False)"},
57                },
58                "required": ["source", "bucket", "key", "is_remote_url"],
59            }
60        }
61    },
62    {
63        "type": "function",
64        "function":{
65            "name": "search_s3_objects",
66            "description": "Search for a specific file name inside an S3 bucket",
67            "parameters": {
68                "type": "object",
69                "properties": {
70                    "search_name": {"type": "string", "description": "The name of the file you want to search for"},
71                    "bucket": {"type": "string", "description": "The name of the S3 bucket"},
72                    "prefix": {"type": "string", "description": "The folder path in the S3 bucket"},
73                    "exact_match": {"type": "boolean", "description": "Set exact_match to True if the search should match the exact file name. Set exact_match to False to compare part of the file name string (the file contains)"}
74                },
75                "required": ["search_name"],
76            },
77        }
78    }
79]

 1def datetime_converter(obj):
 2    if isinstance(obj, datetime.datetime):
 3        return obj.isoformat()
 4    raise TypeError(f"Object of type {obj.__class__.__name__} is not JSON serializable")
 5
 6def list_buckets():
 7    response = s3_client.list_buckets()
 8    return json.dumps(response['Buckets'], default=datetime_converter)
 9
10def list_objects(bucket, prefix=''):
11    response = s3_client.list_objects_v2(Bucket=bucket, Prefix=prefix)
12    return json.dumps(response.get('Contents', []), default=datetime_converter)
13
14def download_file(bucket, key, directory):
15    
16    filename = os.path.basename(key)
17    
18    # Resolve destination to the correct file path
19    destination = os.path.join(directory, filename)
20    
21    s3_client.download_file(bucket, key, destination)
22    return json.dumps({"status": "success", "bucket": bucket, "key": key, "destination": destination})
23
24def upload_file(source, bucket, key, is_remote_url=False):
25    if is_remote_url:
26        file_name = os.path.basename(source)
27        urlretrieve(source, file_name)
28        source = file_name
29       
30    s3_client.upload_file(source, bucket, key)
31    return json.dumps({"status": "success", "source": source, "bucket": bucket, "key": key})
32
33def search_s3_objects(search_name, bucket=None, prefix='', exact_match=True):
34    search_name = search_name.lower()
35    
36    if bucket is None:
37        buckets_response = json.loads(list_buckets())
38        buckets = [bucket_info["Name"] for bucket_info in buckets_response]
39    else:
40        buckets = [bucket]
41
42    results = []
43
44    for bucket_name in buckets:
45        objects_response = json.loads(list_objects(bucket_name, prefix))
46        if exact_match:
47            bucket_results = [obj for obj in objects_response if search_name == obj['Key'].lower()]
48        else:
49            bucket_results = [obj for obj in objects_response if search_name in obj['Key'].lower()]
50
51        if bucket_results:
52            results.extend([{"Bucket": bucket_name, "Object": obj} for obj in bucket_results])
53
54    return json.dumps(results)

1available_functions = {
2    "list_buckets": list_buckets,
3    "list_objects": list_objects,
4    "download_file": download_file,
5    "upload_file": upload_file,
6    "search_s3_objects": search_s3_objects
7}

 1def chat_completion_request(messages, functions=None, function_call='auto', 
 2                            model_name=GPT_MODEL):
 3    
 4    if functions is not None:
 5        return client.chat.completions.create(
 6            model=model_name,
 7            messages=messages,
 8            tools=functions,
 9            tool_choice=function_call)
10    else:
11        return client.chat.completions.create(
12            model=model_name,
13            messages=messages)

 1def run_conversation(user_input, topic="S3 bucket functions.", is_log=False):
 2
 3    system_message=f"Don't make assumptions about what values to plug into functions. Ask for clarification if a user request is ambiguous. If the user ask question not related to {topic} response your scope is {topic} only."
 4    
 5    messages = [{"role": "system", "content": system_message},
 6                {"role": "user", "content": user_input}]
 7    
 8    # Call the model to get a response
 9    response = chat_completion_request(messages
10
11, functions=functions)
12    response_message = response.choices[0].message
13    
14    if is_log:
15        print(response.choices)
16    
17    # check if GPT wanted to call a function
18    if response_message.tool_calls:
19        function_name = response_message.tool_calls[0].function.name
20        function_args = json.loads(response_message.tool_calls[0].function.arguments)
21        
22        # Call the function
23        function_response = available_functions[function_name](**function_args)
24        
25        # Add the response to the conversation
26        messages.append(response_message)
27        messages.append({
28            "role": "tool",
29            "content": function_response,
30            "tool_call_id": response_message.tool_calls[0].id,
31        })
32        
33        # Call the model again to summarize the results
34        second_response = chat_completion_request(messages)
35        final_message = second_response.choices[0].message.content
36    else:
37        final_message = response_message.content
38
39    return final_message

1print(run_conversation('لطفا همه S3 Bucket های را لیست کن'))

1search_file = '<file_name>'
2print(run_conversation(f'دنبال فایلی با نام {search_file} در تمام سطل ها بگرد'))
3
4search_word = '<file_name_part>'
5bucket_name = '<bucket_name>'
6print(run_conversation(f'دنبال فایلی که اسمش شامل {search_word} در باکت {bucket_name} بگرد'))

1print(run_conversation('یک فایل را جستجو کن'))
2
3# Output:
4# مطمئناً، برای اینکه بتوانم به شما کمک کنم آنچه را که می‌خواهید پیدا کنید، لطفاً نام فایل و نام سطل S3 را ارائه دهید. همچنین، آیا جستجو باید دقیقاً با نام فایل مطابقت داشته باشد یا باید به مطابقت‌های جزئی نیز توجه کند؟

1# the model should not answer details not related to the scope
2print(run_conversation('هوا امروز چطوره؟'))
3
4# Output:
5# پوزش می‌خواهم برای سوءتفاهم، اما من فقط می‌توانم در زمینه عملکردهای سطل S3 کمک کنم. لطفاً یک سوال مرتبط با عملکردهای سطل S3 بپرسید؟

1search_file = '<file_name>'
2bucket_name = '<bucket_name>'
3local_directory = '<directory_path>'
4print(run_conversation(f'فایل {search_file} را از {bucket_name} به یک {local_directory} دیکشنری دانلود کن'))

1local_file = '<file_name>'
2bucket_name = '<bucket_name>'
3print(run_conversation(f'فایل {local_file} را به سبد {bucket_name} آپلود کن'))