短视频购物app

python操作AWS personalize 进行 推荐

"""
AKIATQSBE6U5RUQGCBST
s6gaP48EgRG9K9jAH6blzs/IIdMY4/69o3D3sDWq
"""
import boto3

# Let's use Amazon S3
s3 = boto3.resource('s3')

# print(s3)

# for bucket in s3.buckets.all():
#     print(bucket.name)

personalize = boto3.client('personalize')

# print(personalize)

# response = personalize.list_recipes()

# for recipe in response['recipes']:
#     print (recipe)   

# with open('schema.json') as f:
#     createSchemaResponse = personalize.create_schema(
#         name = 'Schema',
#         schema = f.read()
#     )

# schema_arn = createSchemaResponse['schemaArn']

# print('Schema ARN:' + schema_arn )   


# 创建数据集合

# response = personalize.create_dataset_group(name = 'YourDatasetGroup')
# dsg_arn = response['datasetGroupArn']

# description = personalize.describe_dataset_group(datasetGroupArn = dsg_arn)['datasetGroup']

# print('Name: ' + description['name'])
# print('ARN: ' + description['datasetGroupArn'])
# print('Status: ' + description['status'])


# response = personalize.create_dataset(
#     name = 'YourDataset',
#     schemaArn = "arn:aws:personalize:us-west-2:241728550203:schema/Schema",
#     datasetGroupArn = "arn:aws:personalize:us-west-2:241728550203:dataset-group/YourDatasetGroup",
#     datasetType = 'Interactions')

# print ('Dataset Arn: ' + response['datasetArn'])


response = personalize.create_dataset_import_job(
    jobName = 'YourImportJob',
    datasetArn = 'arn:aws:personalize:us-west-2:241728550203:dataset/YourDatasetGroup/INTERACTIONS',
    dataSource = {'dataLocation':'s3://meeshin/data.csv'},
    roleArn = "arn:aws:iam::241728550203:role/PersonalizeRole")

dsij_arn = response['datasetImportJobArn']

# print ('Dataset Import Job arn: ' + dsij_arn)

description = personalize.describe_dataset_import_job(
    datasetImportJobArn = dsij_arn)['datasetImportJob']

print('Name: ' + description['jobName'])
print('ARN: ' + description['datasetImportJobArn'])
print('Status: ' + description['status'])

wxpython_aws.py

import wx
import boto3
import random

s3 = boto3.resource('s3')
personalize = boto3.client('personalize')

class Main(wx.Frame):

    def __init__(self):
        wx.Frame.__init__(self,None,-1,u'AWS',size=(500,800))

        # 初始配置文件
        self.config_view()
        self.config_str()
        self.config_arn()



    def config_view(self):
        self.Bind(wx.EVT_CLOSE, self.OnCloseWindow)
        self.panel = wx.Panel(self, wx.ID_ANY)

        # 按钮
        self.searchS3 = wx.Button(self.panel,-1,u'读取S3',pos=(25,20),size=(200,30))
        self.list_recipes = wx.Button(self.panel,-1,u'list_recipes',pos=(250,20),size=(200,30))

        self.create_dataset_group = wx.Button(self.panel,-1,u'create_dataset_group',pos=(25,60),size=(200,30))
        self.create_dataset_group_text = wx.TextCtrl(self.panel, -1, "",pos=(250,60), size=(200, 30))

        self.create_schema = wx.Button(self.panel,-1,u'create_schema',pos=(25,100),size=(200,30))
        self.schema_text = wx.TextCtrl(self.panel, -1, "",pos=(250,100), size=(200, 30))
        self.schema_users_text = wx.TextCtrl(self.panel, -1, "",pos=(250,140), size=(200, 30))
        self.schema_items_text = wx.TextCtrl(self.panel, -1, "",pos=(250,180), size=(200, 30))

        self.create_dataset = wx.Button(self.panel,-1,u'create_dataset',pos=(25,220),size=(200,30))
        self.create_dataset_text = wx.TextCtrl(self.panel, -1, "",pos=(250,220), size=(200, 30))
        self.create_dataset_items_text = wx.TextCtrl(self.panel, -1, "",pos=(250,260), size=(200, 30))
        self.create_dataset_users_text = wx.TextCtrl(self.panel, -1, "",pos=(250,300), size=(200, 30))

        self.create_dataset_import_job = wx.Button(self.panel,-1,u'create_dataset_import_job',pos=(25,340),size=(200,30))
        self.create_dataset_import_job_text = wx.TextCtrl(self.panel, -1, "",pos=(250,340), size=(200, 30))
        self.create_dataset_import_job_items_text = wx.TextCtrl(self.panel, -1, "",pos=(250,380), size=(200, 30))
        self.create_dataset_import_job_users_text = wx.TextCtrl(self.panel, -1, "",pos=(250,420), size=(200, 30))

        self.create_solution = wx.Button(self.panel,-1,u'create_solution',pos=(25,460),size=(200,30))
        self.create_solution_text = wx.TextCtrl(self.panel, -1, "",pos=(250,460), size=(200, 30))

        self.create_solution_version = wx.Button(self.panel,-1,u'create_solution_version',pos=(25,500),size=(200,30))
        self.create_solution_version_text = wx.TextCtrl(self.panel, -1, "",pos=(250,500), size=(200, 30))

        self.describe_solution_version = wx.Button(self.panel,-1,u'describe_solution_version',pos=(25,540),size=(200,30))
        self.describe_solution_version_text = wx.TextCtrl(self.panel, -1, "",pos=(250,540), size=(200, 30))

        self.create_campaign = wx.Button(self.panel,-1,u'create_campaign',pos=(25,580),size=(200,30))
        self.create_campaign_text = wx.TextCtrl(self.panel, -1, "",pos=(250,580), size=(200, 30))
        self.get_recommendations = wx.Button(self.panel,-1,u'get_recommendations',pos=(25,620),size=(200,30))
        # self.create_schema_text = wx.TextCtrl(self.panel, -1, "",pos=(250,380), size=(200, 30))


        # 绑定事件
        self.searchS3.Bind(wx.EVT_BUTTON,self.OnSearchS3)
        self.list_recipes.Bind(wx.EVT_BUTTON,self.Onlist_recipes)
        self.create_schema.Bind(wx.EVT_BUTTON,self.Oncreate_schema)
        self.create_dataset_group.Bind(wx.EVT_BUTTON,self.Oncreate_dataset_group)
        self.create_dataset.Bind(wx.EVT_BUTTON,self.Oncreate_dataset)
        self.create_dataset_import_job.Bind(wx.EVT_BUTTON,self.Oncreate_dataset_import_job)
        self.create_solution.Bind(wx.EVT_BUTTON,self.Oncreate_solution)
        self.create_solution_version.Bind(wx.EVT_BUTTON,self.Oncreate_solution_version)
        self.describe_solution_version.Bind(wx.EVT_BUTTON,self.Ondescribe_solution_version)
        self.create_campaign.Bind(wx.EVT_BUTTON,self.Oncreate_campaign)
        self.get_recommendations.Bind(wx.EVT_BUTTON,self.Onget_recommendations)

        self.messageText = wx.TextCtrl(self.panel,-1,pos=(20,680),size=(460,50),style=wx.TE_MULTILINE|wx.TE_RICH2| wx.TE_READONLY | wx.TE_MULTILINE | wx.BORDER_NONE)


    def config_arn(self):
        arn_list = []
        with open("file.txt", 'r+',encoding='UTF-8', errors='replace') as f:
            while True:
                line = f.readline()
                if line:
                    arn_list.append(line)
                else:
                    break

        len_arn_list = len(arn_list)
        if len_arn_list>0:self.create_dataset_group_text.SetValue(arn_list[0]) 
        if len_arn_list>1:self.schema_text.SetValue(arn_list[1]) 
        if len_arn_list>2:self.schema_users_text.SetValue(arn_list[2]) 
        if len_arn_list>3:self.schema_items_text.SetValue(arn_list[3]) 

        if len_arn_list>4:self.create_dataset_text.SetValue(arn_list[4]) 
        if len_arn_list>5:self.create_dataset_items_text.SetValue(arn_list[5]) 
        if len_arn_list>6:self.create_dataset_users_text.SetValue(arn_list[6]) 

        if len_arn_list>7:self.create_dataset_import_job_text.SetValue(arn_list[7]) 
        if len_arn_list>8:self.create_dataset_import_job_items_text.SetValue(arn_list[8]) 
        if len_arn_list>9:self.create_dataset_import_job_users_text.SetValue(arn_list[9]) 



    def Oncreate_dataset(self,evn):
        create_dataset_group_text = str(self.create_dataset_group_text.GetValue()).replace("\n",'')
        if not create_dataset_group_text:return false
        self.Interactions_response = personalize.create_dataset(
            name = 'PyInteractionDataset'+str(random.randint(1,1001)),
            schemaArn = str(self.schema_text.GetValue()).replace("\n",''),
            datasetGroupArn = create_dataset_group_text.replace("\n",''),
            datasetType = 'Interactions')
        self.Items_response = personalize.create_dataset(
            name = 'PyItemsDataset'+str(random.randint(1,1001)),
            schemaArn = str(self.schema_items_text.GetValue()).replace("\n",''),
            datasetGroupArn = create_dataset_group_text,
            datasetType = 'Items')
        self.Users_response = personalize.create_dataset(
            name = 'PyUsersDataset'+str(random.randint(1,1001)),
            schemaArn = str(self.schema_users_text.GetValue()).replace("\n",''),
            datasetGroupArn = create_dataset_group_text,
            datasetType = 'Users')

        self.setMessageText(self.Interactions_response['datasetArn'])
        self.setMessageText(self.Items_response['datasetArn'])
        self.setMessageText(self.Users_response['datasetArn'])

        self.create_dataset_text.SetValue(self.Interactions_response['datasetArn'])
        self.create_dataset_items_text.SetValue(self.Items_response['datasetArn'])
        self.create_dataset_users_text.SetValue(self.Users_response['datasetArn'])

    def Oncreate_dataset_import_job(self,evn):
        roleArn = "arn:aws:iam::241728550203:role/PersonalizeRole"
        response = personalize.create_dataset_import_job(
            jobName = 'pyJobInteractions'+str(random.randint(1,1001)),
            datasetArn = str(self.create_dataset_text.GetValue()),
            dataSource = {'dataLocation':'s3://mees1/data.csv'},
            roleArn = roleArn)
        dsij_arn = response['datasetImportJobArn']
        description = personalize.describe_dataset_import_job(
            datasetImportJobArn = dsij_arn)['datasetImportJob']

        self.setMessageText(description['datasetImportJobArn'])
        self.create_dataset_import_job_text.SetValue(description['datasetImportJobArn'])

        response = personalize.create_dataset_import_job(
            jobName = 'pyJobItems'+str(random.randint(1,1001)),
            datasetArn = str(self.create_dataset_items_text.GetValue()),
            dataSource = {'dataLocation':'s3://mees1/items.csv'},
            roleArn = roleArn)
        dsij_arn = response['datasetImportJobArn']
        description = personalize.describe_dataset_import_job(
            datasetImportJobArn = dsij_arn)['datasetImportJob']

        self.setMessageText(description['datasetImportJobArn'])
        self.create_dataset_import_job_items_text.SetValue(description['datasetImportJobArn'])

        response = personalize.create_dataset_import_job(
            jobName = 'pyJobUsers'+str(random.randint(1,1001)),
            datasetArn = str(self.create_dataset_users_text.GetValue()),
            dataSource = {'dataLocation':'s3://mees1/users.csv'},
            roleArn = roleArn)
        dsij_arn = response['datasetImportJobArn']
        description = personalize.describe_dataset_import_job(
            datasetImportJobArn = dsij_arn)['datasetImportJob']

        self.setMessageText(description['datasetImportJobArn'])
        self.create_dataset_import_job_users_text.SetValue(description['datasetImportJobArn'])


    def Oncreate_solution(self,evn):
        response = personalize.create_solution(
            datasetGroupArn = self.create_dataset_group_text.GetValue().replace("\n",''),
            name = "SolutionNamePyAwsAuto"+str(random.randint(1,1001)),
            # recipeArn  = "arn:aws:personalize:::recipe/aws-hrnn-metadata",
            performHPO = True,
            performAutoML = True,
            solutionConfig = {

            }

        )

        print(response)

        # # Get the solution ARN.
        solution_arn = response['solutionArn']
        self.setMessageText('Solution ARN: ' + solution_arn)


    def Oncreate_solution_version(self,evn):
        pass


    def Ondescribe_solution_version(self,evn):
        pass


    def Ondescribe_solution_version(self,evn):
        pass


    def Oncreate_campaign(self,evn):
        pass


    def Onget_recommendations(self,evn):
        pass



    def config_str(self):
        self.datagroup_arn = ''
        self.dataset_arn = ''
        self.Pyschema = ''
        self.PyUserschema = ''
        self.PyItemsschema = ''

    def OnCloseWindow(self, event):
        self.Destroy()

    # 获取S3名称
    def OnSearchS3(self,event):
        for bucket in s3.buckets.all():
            self.setMessageText(bucket.name)

    def Onlist_recipes(self,evt):
        response = personalize.list_recipes()

        for recipe in response['recipes']:
            self.setMessageText(str(recipe['recipeArn']))

    def Oncreate_schema(self,evn):
        index = random.randint(1000,100001)
        with open('schema.json') as f:
            createSchemaResponse = personalize.create_schema(
                name = 'Pyschema'+str(index),
                schema = f.read()
            )
        self.Pyschema = createSchemaResponse['schemaArn']


        self.setMessageText(self.Pyschema)

        with open('user.json') as f:
            createSchemaResponse = personalize.create_schema(
                name = 'PyUserschema'+str(index),
                schema = f.read()
            )
        self.PyUserschema = createSchemaResponse['schemaArn']
        self.setMessageText(self.PyUserschema)


        with open('items.json') as f:
            createSchemaResponse = personalize.create_schema(
                name = 'PyItemsschema'+str(index),
                schema = f.read()
            )
        self.PyItemsschema = createSchemaResponse['schemaArn']
        self.setMessageText(self.PyItemsschema)
        self.schema_users_text.SetValue(self.PyUserschema)
        self.schema_items_text.SetValue(self.PyItemsschema)
        self.schema_text.SetValue(self.Pyschema)



    def Oncreate_dataset_group(self,evn):
        response = personalize.create_dataset_group(name = 'PyDatasetGroup'+str(random.randint(1,1001)))
        dsg_arn = response['datasetGroupArn']
        description = personalize.describe_dataset_group(datasetGroupArn = dsg_arn)['datasetGroup']

        self.datagroup_arn = description['datasetGroupArn']
        self.create_dataset_group_text.SetValue(self.datagroup_arn)

        self.setMessageText(description['datasetGroupArn'])



    def setMessageText(self,text):
        with open("file.txt",'a+') as f:
            f.writelines(str(text)+"\n")
        self.messageText.AppendText(text+"\n")

if __name__ == '__main__':
    app = wx.PySimpleApp()
    frame = Main()
    frame.Show()
    app.MainLoop()        

转载请注明来源,欢迎对文章中的引用来源进行考证,欢迎指出任何有错误或不够清晰的表达。可以在下面评论区评论,也可以邮件至 anaf@163.com
目录