110 lines
3.2 KiB
Python
110 lines
3.2 KiB
Python
from models.connect_db import connect_db
|
||
from models.logger import setup_logger
|
||
|
||
logger = setup_logger()
|
||
|
||
def update_description(emoji_uuid: str, description: str, connection = connect_db()) -> bool:
|
||
"""
|
||
更新表情包描述词
|
||
|
||
Args:
|
||
emoji_uuid: 表情包UUID
|
||
description: 新的描述词
|
||
connection: 数据库socket
|
||
|
||
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}")
|
||
connection.commit()
|
||
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, connection = connect_db()) -> bool:
|
||
"""
|
||
检查UUID是否存在于emoji表中
|
||
|
||
Args:
|
||
uuid: 要检查的UUID字符串
|
||
connection: 数据库socket
|
||
|
||
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 approved FROM approved WHERE uuid = %s LIMIT 1"
|
||
cursor.execute(sql, (uuid,))
|
||
|
||
# 如果查询到结果为1
|
||
result = cursor.fetchone()
|
||
return result is not None
|
||
|
||
except Exception as e:
|
||
logger.error(f"检查UUID失败: {uuid}, 错误: {e}")
|
||
return False
|
||
|
||
def get_emoji_content(uuid: str, content_index: str = 'url', connection = connect_db()) -> tuple | None:
|
||
"""
|
||
获取图片信息，检查图片在 approved 库中检索 approved 是否为 1，为 1 则返回获取到的索引内容
|
||
|
||
Args:
|
||
uuid: emoji的UUID
|
||
content_index: 内容检索索引
|
||
connection: 数据库socket
|
||
|
||
Returns:
|
||
content: 获取到的内容，比如 url
|
||
"""
|
||
# 参数验证
|
||
if not uuid or not isinstance(uuid, str):
|
||
logger.warning(f"无效的UUID参数: {uuid}")
|
||
return None
|
||
|
||
try:
|
||
with connection.cursor() as cursor:
|
||
sql1 = "SELECT approved FROM approved WHERE uuid = %s LIMIT 1"
|
||
cursor.execute(sql1, (uuid,))
|
||
|
||
result = cursor.fetchone()
|
||
if result['approved'] == 1:
|
||
sql2 = f"SELECT {content_index} FROM emoji WHERE uuid = %s LIMIT 1"
|
||
cursor.execute(sql2, (uuid,))
|
||
contents = cursor.fetchone()
|
||
return contents[content_index]
|
||
|
||
return None
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取失败: {uuid}, 错误: {e}")
|
||
return None
|
||
|
||
if __name__ == "__main__":
|
||
record = check_emoji("ffe1663c-44e1-4719-a5ba-01485f70a87e")
|
||
print(record) |