feature(wqj): add Chinese meme

backend/chinese_meme_generation_prompt.txt

@@ -0,0 +1,26 @@
+# **表情包分析顾问**
+我是一个表情包分析顾问，我致力于分析表情包，并帮助我的客户做到以下事情：(1)理解表情包所蕴含的情感 (2)分析这个表情包的发送者的意图 (3)例举一些这个表情包可能的使用场景。
+
+## 背景介绍
+我即将分析的表情包都是从发送者和接收者的线上聊天情景中产生的。我擅长从表情包的文字和图片中提取关键信息，用于分析(1)表情包所传达出来的情感 (2)表情包发送者的意图，比如发送者为什么要发送这个表情包给接收者？发送者发送这个表情包给接收者是想表达什么意思？(3)表情包的使用场景，比如这个表情包的发送者和接收者的身份，以及发送者发送这个表情包时发送者和接收者的聊天场景。
+
+## 目标
+- 详细并准确地分析表情包的意思并输出
+    (0)表情包的内容
+    (1)表情包所传达出来的情感
+    (2)表情包发送者的意图
+    (3)表情包的使用场景
+并且保证输出格式与下面的样例保持一致
+
+## 输出样例
+### 样例1
+(0)这个表情包描述了一个皱眉，叹气的小狗，表情包中的文字是“生活好累”。在这个表情包中，小狗实际上代表发送者。
+(1)这个表情包传达出一种失落，抑郁的情绪。
+(2)发送者的意图也许是为了表达他的难过，发送者正感觉处在低谷并且希望接收者可以安慰他或者给他一点帮助和建议。
+(3)发送者和接收者的身份也许是朋友/恋人。聊天发生的场景也许是日常聊天。
+
+### 样例2
+(0)这个表情包描述了一个敬礼的人，表情包中的文字是“收到”。
+(1)这个表情包传达出一种认同感或责任感。
+(2)发送者的意图也许是为了表达他对接收者之前所说话的认同，并且表达他会做好交代的事情，发送者感觉良好并且接受了接收者交代的任务。
+(3)发送者和接收者的身份也许是雇员（发送者）和经理（接收者） / 学生（发送者）和导师（接收者） / 丈夫（发送者）和妻子（接收者）。 聊天发生的场景也许是接收者之前给发送者交代了一个任务，发送者表示接受这个任务并保证会完成。

backend/chinese_meme_generator.py

@@ -0,0 +1,239 @@
+import csv
+import base64
+import os
+from openai import OpenAI
+import random
+import json
+import re
+import numpy as np
+from PIL import Image
+
+translation = {
+    'sentiment category': '情感类别',
+    'happiness': '开心',
+    'love': '爱',
+    'anger': '愤怒',
+    'sorrow': '悲伤',
+    'fear': '恐惧',
+    'hate': '恨',
+    'surprise': '惊讶',
+    'sentiment degree': '情感程度',
+    'slightly': '轻微',
+    'moderately': '中等',
+    'very': '强烈',
+    'intention detection': '情感意图',
+    'interactive': '交互',
+    'expressive': '表达',
+    'entertaining': '娱乐',
+    'offensive': '冒犯',
+    'other': '其他',
+    'offensiveness detection': '冒犯性检测',
+    'non-offensive': '非冒犯',
+}
+
+
+def get_total_data(data_path: str) -> dict:
+    # Create a dictionary to store the processed data
+    data_dict = {}
+
+    # Open the CSV file and read its contents
+    with open(data_path, mode='r', encoding='utf-8') as csv_file:
+        csv_reader = csv.DictReader(csv_file)
+
+        # Iterate over each row in the CSV
+        for row in csv_reader:
+            # Use 'images_name' as the key for the outer dictionary
+            image_name = row['images_name']
+            # Remove 'images_name' from the row to create the inner dictionary
+            row.pop('images_name')
+            # Store the inner dictionary in the outer dictionary
+            data_dict[image_name] = row
+    labels = ['sentiment category', 'sentiment degree', 'intention detection', 'offensiveness detection']
+    # polish content of the dict
+    for image_name, details in data_dict.items():
+        for key in labels:
+            content = details[key]
+            # original data: 4(sorrow)/2(moderately)
+            # using content[2:-1] to remove number & '()'
+            details[key] = content[2:-1]
+    #data_dict: image_path -> content dict
+    return data_dict
+
+
+def get_samples_random(data: dict, sample_num: int = None) -> list[int]:
+    #return the index of selected samples
+    if sample_num is None:
+        sample_num = 100
+    random_numbers = random.sample(range(0, len(data)), sample_num)
+    return random_numbers
+
+
+def encode_image(image_path: str) -> tuple:
+    # return base64 image and origin format of the image
+    with open(image_path, "rb") as image_file:
+        img_base64 = base64.b64encode(image_file.read()).decode('utf-8')
+    with Image.open(image_path) as img:
+        img_format = img.format.lower()
+    return img_base64, img_format
+
+
+def get_ans_chmeme_chinese(
+        data_dict: dict, image_list: list, client, end_point: str, cared_keys: list, sys_prompt: str
+) -> dict:
+    # main function of using api to generate data
+    # return: dict, image_num -> detail_dict
+    ch_dataset = {}
+    error_imgs = []
+    token_usage = 0
+    for image_path in image_list:
+        print(f"processing {image_path}...")
+        base64_image, image_type = encode_image(image_path)
+        img_name = image_path.split('/')[-1]
+        img_details = data_dict[img_name]
+        question = '\n下面我将根据上面的要求分析这个表情包，并且按照样例的格式输出我的分析结果。这个表情包的'
+        for element in cared_keys:
+            detail = f'{translation[element]} 是 {translation[img_details[element]]}, '
+            question += detail
+        if img_details['metaphor occurrence'] == '1':
+            add_prompt = '注意在这个表情包中存在隐喻，其中'
+            src = img_details['source domain']
+            tgt = img_details['target domain']
+            src_parts = re.split(r'[;；]', src)
+            tgt_parts = re.split(r'[;；]', tgt)
+            for src_item, tgt_item in zip(src_parts, tgt_parts):
+                add_prompt += f"， ‘{tgt_item}’ 有 ‘{src_item}’ 的意思"
+            add_prompt += '。'
+            question += add_prompt
+
+        try:
+            response = client.chat.completions.create(
+                # change accroding to your config
+                model=end_point,
+                messages=[
+                    {
+                        "role": "system",
+                        "content": sys_prompt
+                    }, {
+                        "role": "user",
+                        "content": [
+                            {
+                                "type": "text",
+                                "text": question,
+                            },
+                            {
+                                "type": "image_url",
+                                "image_url": {
+                                    "url": f"data:image/{image_type};base64,{base64_image}"
+                                },
+                            },
+                        ],
+                    }
+                ],
+            )
+            ret_text = response.choices[0].message.content
+            token_usage += response.usage.total_tokens
+            tmpdata = {}
+            tmpdata['description'] = ret_text
+            for key in cared_keys:
+                tmpdata[key] = img_details[key]
+            tmpdata['prompt'] = question
+            ch_dataset[image_path] = tmpdata
+            print(f"processing {image_path} finished")
+        except:
+            error_imgs.append(image_path)
+            print(f"Error processing image {image_path}")
+            continue
+    print(f'You have used {token_usage} tokens for the dataset generation')
+    print(f'The number of error images:{len(error_imgs)}')
+    print(f'error imgs:')
+    for img in error_imgs:
+        print(img)
+    return ch_dataset
+
+
+def save_to_json(data: dict, output_path: str) -> None:
+    with open(output_path, 'w', encoding='utf-8') as f:
+        json.dump(data, f, ensure_ascii=False, indent=4)
+
+
+def get_dataset(
+        data: dict, prompt_path: str, client_endpoint: str, img_prefix_dir: str, select_numbers: list = None
+) -> dict:
+    data_dict = data
+    if select_numbers is None:
+        random_numbers = get_smaples_random(data_dict, 10)
+    else:
+        random_numbers = select_numbers
+
+    random_numbers.sort()
+    image_list = []
+
+    #img_l=[656,1371,1558,1584,656,908,186,1037,4410,5554,6004,5976,5609,4299]
+    #can be modified
+    img_l = random_numbers
+
+    for num in img_l:
+        img_path = os.path.join(img_prefix_dir, f"Image_({num}).jpg")
+        image_list.append(img_path)
+
+    client = OpenAI(
+        # get API KEY from env
+        api_key=api_key,
+        base_url=model_url,
+        #base_url="https://ark.cn-beijing.volces.com/api/v3",
+    )
+    cared_keys = ['sentiment category', 'sentiment degree', 'intention detection', 'offensiveness detection']
+
+    with open(prompt_path, 'r', encoding='utf-8') as file:
+        PROMPT = file.read()
+    ch_dataset = get_ans_chmeme_chinese(data_dict, image_list, client, client_endpoint, cared_keys, PROMPT)
+
+    json_dataset = {}
+    for num in img_l:
+        img_path = f'{img_prefix_dir}({num}).jpg'
+        new_img_path = f'{num}'
+        cur_data = {}
+        cur_details = ch_dataset[img_path]
+        text_details = cur_details['description']
+        lines = text_details.strip().split('\n')
+        for line in lines:
+            if line.startswith("(0)"):
+                cur_data["content"] = line[3:].strip()
+
+            elif line.startswith("(1)"):
+                cur_data["emotion"] = line[3:].strip()
+
+            elif line.startswith("(2)"):
+                cur_data["intention"] = line[3:].strip()
+
+            elif line.startswith("(3)"):
+                cur_data["scene"] = line[3:].strip()
+        json_dataset[new_img_path] = cur_data
+
+    return json_dataset
+
+
+csv_file_path = os.environ.get("csv_file_path")
+# path to the image files
+image_prefix = os.environ.get("img_prefix")
+# path to the prompt
+prompt_path = './chinese_meme_generation_prompt.txt'
+# path to save the dataset
+dataset_path_json = os.environ.get("data_path_json")
+# need to set API_KEY
+api_key = os.environ.get("ARK_API_KEY")
+# set model_url
+model_url = "https://ark.cn-beijing.volces.com/api/v3"
+# first need to set model_endpoint in website
+model_endpoint = "ep-20250124115350-pwjxt"
+
+
+def main():
+    data = get_total_data(csv_file_path)
+    select_numbers = range(0, 6046)
+    json_dataset = get_dataset(data, prompt_path, model_endpoint, image_prefix, select_numbers)
+    save_to_json(json_dataset, dataset_path_json)
+
+
+if __name__ == "__main__":
+    main()

backend/chinese_meme_recommender.py

@@ -0,0 +1,90 @@
+import json
+import os
+import torch
+from transformers import AutoModel
+import requests
+from sklearn.metrics.pairwise import cosine_similarity
+from PIL import Image
+
+
+class ChineseMemeRecommender:
+
+    def __init__(self, data: dict, embedding_model_name: str = 'online'):
+        self.data = data
+        self.embedding_model_name = embedding_model_name
+        # ids of chmeme
+        self.rec_nums = list(data.keys())
+        keys = list(data[self.rec_nums[0]].keys())
+        self.key_len = len(keys)
+        if embedding_model_name == 'online':
+            self.embedding_model_url = os.getenv("EMBEDDING_URL")
+            if self.embedding_model_url is None:
+                self.embedding_model_url = os.environ.get("embedding_url")
+        elif embedding_model_name == 'jina':
+            self.embedding_model = AutoModel.from_pretrained("jinaai/jina-embeddings-v3", trust_remote_code=True)
+        else:
+            raise KeyError(f'Incorrect arg: {self.embedding_model_name}')
+        device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
+        if self.embedding_model_name != 'online':
+            self.embedding_model = self.embedding_model.to(device)
+
+    def _get_embed_sim(self, query: str):
+        data_sentence = []
+        for num in self.rec_nums:
+            item = self.data[num]
+            for key in list(item.keys()):
+                data_sentence.append(item[key])
+        if self.embedding_model_name == 'bert':
+            similarities = self.embedding_model.similarity(query, data_sentence)
+        else:
+            if self.embedding_model_name == 'online':
+                payload = {"text": [query] + data_sentence, "model": "bge-large-zh-v1.5"}
+                headers = {"Content-Type": "application/json"}
+                response = requests.post(self.embedding_model_url, json=payload, headers=headers)
+                embeddings = response.json()['embedding']
+            else:
+                embeddings = self.embedding_model.encode([query] + data_sentence)
+            embeddings_query = torch.tensor(embeddings[0]).reshape(1, -1)
+            embeddings_sentences = torch.tensor(embeddings[1:])
+            similarities = cosine_similarity(embeddings_query, embeddings_sentences)
+        return similarities[0]
+
+    def _get_sim_score(self, query: str):
+        sims = self._get_embed_sim(query)
+        sims = torch.from_numpy(sims)
+        key_len = self.key_len
+        sims = sims.view(-1, key_len).float()
+        scores = sims.mean(dim=1).tolist()
+        return scores
+
+    def get_topk_meme(self, query: str, k: int = 2) -> list:
+        scores = self._get_sim_score(query)
+        sorted_with_index = sorted(enumerate(scores), key=lambda x: x[1], reverse=True)
+
+        top_k_indexes = [index for index, _ in sorted_with_index[:k]]
+
+        res = []
+        for ind in top_k_indexes:
+            res.append(self.rec_nums[ind])
+        return res
+
+
+def test_rec_chmeme(file_path: str, query: str = '耗子尾汁', k: int = 4):
+    # json file path
+    with open(file_path, 'r', encoding='utf-8') as file:
+        ch_dataset = json.load(file)
+    # as for ChineseMemeRecommender, only support online embedding model
+    # other models can be added later
+    my_chmeme = ChineseMemeRecommender(data=ch_dataset)
+    return_id = my_chmeme.get_topk_meme(query, k)
+    prefix = os.environ.get('prefix')
+    image_path = [f'{prefix}({x}).jpg' for x in return_id]
+    for path in image_path:
+        print(path)
+        img = Image.open(path)
+        img.show()
+
+
+if __name__ == "__main__":
+    file_path = ''  #your file path
+    test_rec_chmeme(file_path)