first commit

config.toml

@@ -0,0 +1,11 @@
+safe_entry = "mcunc"
+tokenTTL = 3600 # token有效期 单位：s
+cacheTTL = 300 # token过期缓存，单位：s
+
+# 数据库配置
+#db_host = "sz.mcunc.site"
+db_host = "localhost"
+db_port = 3306
+db_user = "root"
+db_password = "ljh20080924"
+db_database = "emoji"

controllers/compress_image.py

@@ -0,0 +1,52 @@
+import os
+from PIL import Image
+from models.logger import setup_logger
+
+logger = setup_logger()
+
+def compress_image(input_path, output_folder: str = 'compressed_images', target_kb=200, quality_step=5) -> str:
+    """
+    将图片压缩到指定大小（以KB为单位）
+
+    Args:
+        input_path: 输入图片路径
+        output_folder: 输出文件夹
+        target_kb: 目标文件大小（默认200KB）
+        quality_step: 每次迭代中质量减少的步长（默认5）
+
+    Returns:
+         压缩后的图片路径
+    """
+    # 确保输出文件夹存在
+    os.makedirs(output_folder, exist_ok=True)
+
+    # 获取文件基本信息
+    file_size_kb = os.path.getsize(input_path) / 1024
+    base_name = os.path.basename(input_path)
+    output_path = os.path.join(output_folder, base_name.rsplit('.', 1)[0] + '.jpg')  # 默认转为jpg
+
+    if file_size_kb <= target_kb:
+        # 如果文件小于等于目标大小，直接复制
+        img = Image.open(input_path)
+        if img.mode in ("RGBA", "P"):
+            img = img.convert("RGB")
+        img.save(output_path, "JPEG", optimize=True, quality=95)
+        return str(output_path)
+
+    # 初始质量设置
+    quality = 85
+
+    while quality > 10:  # 设置最低质量限制避免过低质量的图片
+        with Image.open(input_path) as img:
+            if img.mode in ("RGBA", "P"):
+                img = img.convert("RGB")
+            img.save(output_path, "JPEG", optimize=True, quality=quality)
+
+        # 检查文件大小
+        if os.path.getsize(output_path) / 1024 < target_kb:
+            return str(output_path)
+
+        quality -= quality_step  # 减少质量设置
+
+    logger.error(f"图片 {input_path} 压缩失败!")
+    return ''

controllers/localfile_handles.py

@@ -0,0 +1,14 @@
+import os
+from models.logger import setup_logger
+
+logger = setup_logger()
+
+get_emoji_count = lambda folder_path: sum(
+    1 for file in os.listdir(folder_path)
+    if os.path.isfile(os.path.join(folder_path, file)) and file.lower().endswith(('.jpg', '.jpeg'))
+)
+
+def yield_emoji_path(folder_path: str):
+    for root, dirs, files in os.walk(folder_path):
+        for file in files:
+            yield os.path.join(root, file)

main.py

@@ -0,0 +1,17 @@
+import os
+from dashscope import MultiModalConversation
+import dashscope
+
+# 本地图像的绝对路径
+local_path = r"D:\Python\mcunc\EmojiTextGenerator\emojis\00e4f193-74af-40f7-8722-59d5a5931f92.jpg"
+image_path = f"file://{local_path}"
+messages = [
+                {'role':'user',
+                'content': [{'image': image_path},
+                            {'text': '使用简洁语言描述该表情包 ，例如 在吗 ，生气，？'}]}]
+response = MultiModalConversation.call(
+    api_key="sk-213f83213fce4e2ba41b8e67721f19cb",
+    model='qwen3-vl-plus',
+    messages=messages)
+
+print(response.output.choices[0].message.content[0]["text"])

models/connect_db.py

@@ -0,0 +1,88 @@
+import toml
+import pymysql
+from pymysql.cursors import DictCursor
+from pymysql.connections import Connection
+from models.logger import setup_logger
+
+logger = setup_logger()
+
+def connect_db() -> Connection | None:
+    """连接数据库，如果没有则初始化数据库和表"""
+    try:
+        config = toml.load("../config.toml")
+        db_host = config["db_host"]
+        db_port = config["db_port"]
+        db_user = config["db_user"]
+        db_password = config["db_password"]
+        db_database = config["db_database"]
+    except Exception as e:
+        logger.error(f"读取数据库配置错误{e}")
+        return None
+
+    connection = pymysql.connect(
+        host=db_host,
+        port=int(db_port),
+        user=db_user,
+        password=db_password,
+        database=db_database,
+        charset='utf8mb4',
+        cursorclass=DictCursor
+    )
+
+    try:
+        with connection.cursor() as cursor:
+            # 创建 emoji 表
+            cursor.execute('''
+                        CREATE TABLE IF NOT EXISTS emoji (
+                            id INT AUTO_INCREMENT PRIMARY KEY,
+                            uuid VARCHAR(255) NOT NULL,
+                            description VARCHAR(255),
+                            url VARCHAR(255) NOT NULL,
+                            hash VARCHAR(255) NOT NULL
+                        )
+                ''')
+
+            # 创建 approved 表
+            cursor.execute('''
+                           CREATE TABLE IF NOT EXISTS approved
+                           (
+                               id          INT AUTO_INCREMENT PRIMARY KEY,
+                               uuid        VARCHAR(255) NOT NULL,
+                               approved    BOOL,
+                               approver    VARCHAR(255)
+                           )
+                           ''')
+
+            # 创建 token 表
+            cursor.execute('''
+                           CREATE TABLE IF NOT EXISTS token
+                           (
+                               id INT AUTO_INCREMENT PRIMARY KEY,
+                               token VARCHAR(255) NOT NULL ,
+                               userid INT NOT NULL ,
+                               expires TIMESTAMP NOT NULL
+                           )
+                           ''')
+
+            # 创建 user 表
+            cursor.execute('''
+                           CREATE TABLE IF NOT EXISTS user
+                           (
+                               id INT AUTO_INCREMENT PRIMARY KEY,
+                               username VARCHAR(255) NOT NULL,
+                               passwd VARCHAR(255) NOT NULL,
+                               email VARCHAR(255),
+                               lastlogin TIMESTAMP,
+                               submission_count INT DEFAULT 0
+                           )
+                           ''')
+
+        return connection
+    except Exception as e:
+        logger.error(f"数据库连接或表创建失败: {e}")
+        return None
+
+
+if __name__ == "__main__":
+    conn = connect_db()
+    conn.cursor()

models/emoji_db.py

@@ -0,0 +1,73 @@
+from models.connect_db import connect_db
+from models.logger import setup_logger
+
+logger = setup_logger()
+connection = connect_db()
+
+def update_description(emoji_uuid: str, description: str) -> bool:
+    """
+    更新表情包描述词
+
+    Args:
+        emoji_uuid: 表情包UUID
+        description: 新的描述词
+
+    Returns:
+        成功返回True，失败返回False
+    """
+    try:
+        sql = """
+            UPDATE emoji 
+            SET description = %s 
+            WHERE uuid = %s
+        """
+
+        params = (description, emoji_uuid)
+
+        with connection.cursor() as cursor:
+            affected_rows = cursor.execute(sql, params)
+
+            if affected_rows > 0:
+                logger.info(f"成功更新表情包描述，UUID: {emoji_uuid}")
+                return True
+            else:
+                logger.warning(f"未找到要更新的表情包，UUID: {emoji_uuid}")
+                return False
+
+    except Exception as e:
+        logger.error(f"更新表情包描述失败: {e}")
+        return False
+
+
+def check_emoji(uuid: str) -> bool:
+    """
+    检查UUID是否存在于emoji表中
+
+    Args:
+        uuid: 要检查的UUID字符串
+
+    Returns:
+        uuid 在数据库中存在则为 True，否则 False
+    """
+    # 参数验证
+    if not uuid or not isinstance(uuid, str):
+        logger.warning(f"无效的UUID参数: {uuid}")
+        return False
+
+    try:
+        with connection.cursor() as cursor:
+            # 执行查询
+            sql = "SELECT 1 FROM emoji WHERE uuid = %s LIMIT 1"
+            cursor.execute(sql, (uuid,))
+
+            # 如果查询到结果则返回True
+            result = cursor.fetchone()
+            return result is not None
+
+    except Exception as e:
+        logger.error(f"检查UUID失败: {uuid}, 错误: {e}")
+        return False
+
+if __name__ == "__main__":
+    record = check_emoji("ffe1663c-44e1-4719-a5ba-01485f70a87e")
+    print(record)

models/generator.py

@@ -0,0 +1,44 @@
+import os
+from typing import Optional
+from dashscope import MultiModalConversation
+from models.logger import setup_logger
+
+logger = setup_logger()
+
+class Generator:
+    def __init__(self, api_key: str, model: str = 'qwen3-vl-plus'):
+        self.api_key = api_key
+        self.model = model
+
+    def process_single_image(self, image_path: str) -> Optional[str]:
+        """处理单张图片，生成描述词"""
+        try:
+            image_url = f"file://{image_path}"
+            messages = [
+                {
+                    'role': 'user',
+                    'content': [
+                        {'image': image_url},
+                        {'text': '使用简洁语言描述该表情包 ，例如 在吗 ，生气，？'}
+                    ]
+                }
+            ]
+
+            response = MultiModalConversation.call(
+                model=self.model,
+                messages=messages
+            )
+
+            if response and hasattr(response, 'output'):
+                if hasattr(response.output.choices[0].message.content[0], 'text'):
+                    return response.output.choices[0].message.content[0]["text"]
+                elif isinstance(response.output.choices[0].message.content[0], dict):
+                    return response.output.choices[0].message.content[0].get("text", "")
+
+            return None
+
+        except Exception as e:
+            logger.error(f"API调用失败: {e}")
+            return None
+
+    

models/logger.py

@@ -0,0 +1,61 @@
+import logging
+from logging.handlers import RotatingFileHandler
+from datetime import datetime
+from pathlib import Path
+
+def setup_logger(log_file=None, log_level=logging.INFO, max_bytes=10485760, backup_count=5):
+    """
+    初始化日志记录器，将日志按指定格式输出并记录到文件
+
+    Args:
+        log_file (str): 日志文件路径，默认为None，会自动生成以当天日期命名的日志文件
+        log_level: 日志级别
+        max_bytes (int): 单个日志文件最大字节数
+        backup_count (int): 保留的备份日志文件数量
+
+    Returns:
+        logging.Logger: 配置好的日志记录器
+    """
+    # 创建logger
+    logger = logging.getLogger('emoji-text-generator')
+    logger.setLevel(log_level)
+    logger.propagate = False
+
+    # 避免重复添加handler
+    if logger.handlers:
+        return logger
+
+    # 创建logs目录（如果不存在）
+    log_dir = Path('logs')
+    log_dir.mkdir(exist_ok=True)
+
+    # 如果未指定日志文件名，则使用当天日期命名
+    if log_file is None:
+        today = datetime.now().strftime('%Y-%m-%d')
+        log_file = log_dir / f'{today}.log'
+
+    # 创建格式化器
+    formatter = logging.Formatter(
+        '[%(asctime)s] [%(levelname)s] [%(filename)s:%(lineno)d] - %(message)s'
+    )
+
+    # 创建文件处理器（带轮转）
+    file_handler = RotatingFileHandler(
+        log_file,
+        maxBytes=max_bytes,
+        backupCount=backup_count,
+        encoding='utf-8'
+    )
+    file_handler.setLevel(log_level)
+    file_handler.setFormatter(formatter)
+
+    # 创建控制台处理器
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(log_level)
+    console_handler.setFormatter(formatter)
+
+    # 添加处理器到logger
+    logger.addHandler(file_handler)
+    logger.addHandler(console_handler)
+
+    return logger

models/logs/2025-12-07.log

@@ -0,0 +1,2 @@
+[2025-12-07 13:44:14,574] [ERROR] [connect_db.py:19] - 读取数据库配置错误[Errno 2] No such file or directory: 'config.toml'
+[2025-12-07 13:44:14,574] [ERROR] [emoji_db.py:68] - 检查表情包是否存在失败: 'NoneType' object has no attribute 'cursor'

test.py

@@ -0,0 +1,36 @@
+import os
+from pathlib import Path
+from models.generator import Generator
+from models.emoji_db import check_emoji, update_description
+from controllers.localfile_handles import get_emoji_count, yield_emoji_path
+from controllers.compress_image import compress_image
+from models.logger import setup_logger
+
+logger = setup_logger()
+
+emoji_folders = "D:\\Python\\mcunc\\EmojiTextGenerator\\emojis\\"
+emoji = yield_emoji_path(emoji_folders)
+
+api_key = os.getenv('DASHSCOPE_API_KEY')
+
+def main():
+    gen = Generator(api_key)
+    for f in range(get_emoji_count(emoji_folders)):
+        emoji_file = next(emoji)
+        emoji_uuid = ''
+        p = Path(emoji_file)
+        if p.suffix == ".jpg" or p.suffix == ".jpeg":
+            emoji_uuid = p.stem
+        # print(emoji_uuid)
+        image = compress_image(emoji_file)
+
+        # if not check_emoji(emoji_uuid):
+        #     logger.error(f"图片 {emoji_uuid} 不在数据库中！")
+        # else:
+        description = gen.process_single_image(image)
+        if update_description(emoji_uuid, description)
+            logger.info(f"图片 {emoji_file} 的描述词生成完毕， 其 description 为: {description}")
+
+
+if __name__ == "__main__":
+    main()