云容笔谈·东方红颜影像生成系统MySQL数据库集成教程用户画像与生成记录管理你是不是也遇到过这样的问题自己搭建的AI影像生成系统用户今天生成的图片明天就找不到了或者想给老用户一些个性化的生成建议却发现根本没有他们的历史数据。数据就像流水一样用过就没了。我之前给“云容笔谈·东方红颜”这个影像生成系统做数据管理时就碰到了这个痛点。系统跑得挺好用户反馈也不错但所有生成记录都只是临时存在内存里服务器一重启或者用户换个设备登录之前的所有痕迹就都消失了。这显然不行尤其是当你想做用户行为分析、提供个性化服务或者只是简单地让用户能回顾自己的创作历史时。所以我决定给系统加上一个“记忆中枢”——MySQL数据库。今天这篇文章我就手把手带你走一遍整个集成过程。从零开始在部署环境里装上MySQL设计两张核心表来存用户画像和生成记录再用Python代码把它们和我们的AI生成系统连接起来。整个过程我会尽量讲得直白哪怕你之前没怎么碰过数据库跟着做也能搞定。1. 环境准备在服务器上安装和配置MySQL首先我们得有个地方放数据库。假设你的“云容笔谈”系统已经部署在了一台Linux服务器上比如Ubuntu 20.04/22.04我们现在就在这台服务器上安装MySQL。打开服务器的终端用SSH连上去然后我们开始操作。1.1 安装MySQL服务器在Ubuntu上安装MySQL非常方便用系统自带的包管理器就行。我们安装的是MySQL社区版。# 首先更新一下软件包列表确保获取到最新的版本信息 sudo apt update # 安装MySQL服务器 sudo apt install mysql-server -y这个-y参数是自动确认安装省得你再手动输入一次“Y”。安装过程可能会花几分钟取决于你的网络速度。安装完成后MySQL服务会自动启动。你可以用下面这个命令检查一下它是不是在正常运行sudo systemctl status mysql如果看到绿色的“active (running)”字样就说明MySQL已经成功安装并运行起来了。1.2 进行安全初始化配置刚安装好的MySQL默认配置不太安全比如root用户可能没有密码。我们需要运行一个安全脚本进行加固。sudo mysql_secure_installation运行这个命令后它会引导你完成一系列设置验证密码强度组件它会问你是否要设置密码验证策略。对于内部系统我们可以先选“否”输入n简化流程。设置root密码这是最重要的一步。它会提示你为MySQL的root用户设置一个密码。请务必设置一个强密码并牢记。输入密码时屏幕上不会有任何显示这是正常的输完回车即可。移除匿名用户选择“是”输入y删除默认存在的匿名用户提高安全性。禁止root远程登录选择“是”输入y。这样root用户只能从本机服务器自己登录防止从外网直接攻击。删除测试数据库选择“是”输入y移除名为test的默认数据库。重新加载权限表选择“是”输入y让刚才的所有安全设置立刻生效。完成这些步骤你的MySQL基础环境就算安全地搭建好了。1.3 创建一个专用的数据库和用户为了系统安全我们不应该直接用root用户来连接应用。最佳实践是创建一个专门给“云容笔谈”系统使用的数据库和一个拥有相应权限的普通用户。我们先以root身份登录MySQL命令行sudo mysql -u root -p输入你刚才设置的root密码。成功登录后你会看到提示符变成了mysql。然后执行以下SQL命令-- 创建一个名为 cloud_ink_talk 的数据库用来存放我们系统的所有数据 CREATE DATABASE cloud_ink_talk CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 创建一个新用户用户名是 ink_user并设置一个强密码 YourStrongPassword123! -- 请务必把 YourStrongPassword123! 换成你自己想的复杂密码 CREATE USER ink_userlocalhost IDENTIFIED BY YourStrongPassword123!; -- 把 cloud_ink_talk 数据库的所有操作权限授予 ink_user 用户 GRANT ALL PRIVILEGES ON cloud_ink_talk.* TO ink_userlocalhost; -- 让权限设置立即生效 FLUSH PRIVILEGES; -- 退出MySQL命令行 EXIT;这里有几个关键点utf8mb4字符集能支持存储Emoji等特殊字符避免乱码。ink_userlocalhost表示这个用户只能从本机连接这很安全因为我们的Python应用就和数据库在同一台服务器上。记住你为ink_user设置的密码等下写Python连接代码时要用到。好了数据库服务端的基础工作就完成了。接下来我们设计一下要存什么。2. 核心设计用户画像表与生成记录表数据库就像一个大仓库表就是里面一个个分类明确的货架。我们需要设计两个“货架”的结构一个放用户信息一个放生成记录。2.1 用户画像表 (user_profile)这张表记录用户的基本信息和偏好是实现个性化的基础。我们再次登录MySQL这次可以用新建的ink_user测试一下然后创建表。你可以用sudo mysql -u root -p登录后再切换到cloud_ink_talk数据库或者直接用mysql -u ink_user -p cloud_ink_talk输入ink_user的密码。然后在mysql提示符下执行以下SQLCREATE TABLE user_profile ( user_id VARCHAR(64) PRIMARY KEY COMMENT ‘系统分配的唯一用户ID例如UUID’, username VARCHAR(50) UNIQUE NOT NULL COMMENT ‘用户登录名’, nickname VARCHAR(50) COMMENT ‘用户昵称’, avatar_url TEXT COMMENT ‘用户头像图片链接’, preferred_style VARCHAR(100) COMMENT ‘偏好的影像风格如“古风”、“现代”、“唯美”’, default_model VARCHAR(50) DEFAULT ‘default’ COMMENT ‘默认使用的生成模型’, total_credits INT DEFAULT 0 COMMENT ‘用户总积分或剩余生成次数’, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT ‘账户创建时间’, last_login_at TIMESTAMP NULL COMMENT ‘最后登录时间’, profile_meta JSON COMMENT ‘扩展信息以JSON格式存储如标签、等级等’ ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT‘用户画像信息表’;设计思路解读user_id是主键绝对唯一用字符串存兼容各种ID生成方式。preferred_style和default_model是关键字段系统可以根据这些值在用户生成时提供默认参数提升体验。total_credits可以用来做简单的积分或次数管理。profile_meta字段类型是JSON这非常灵活。以后如果想增加“用户等级”、“喜欢的颜色”等属性可以直接往这个JSON里加而不用频繁修改表结构。2.2 生成历史记录表 (generation_history)这张表记录每一次图像生成的详细“档案”。CREATE TABLE generation_history ( record_id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT ‘记录ID自增长主键’, user_id VARCHAR(64) NOT NULL COMMENT ‘关联的用户ID’, task_id VARCHAR(128) NOT NULL COMMENT ‘生成任务的唯一标识可用于查询进度’, prompt_text TEXT NOT NULL COMMENT ‘用户输入的生成提示词’, negative_prompt TEXT COMMENT ‘负面提示词’, style_param VARCHAR(100) COMMENT ‘本次生成使用的风格参数’, model_name VARCHAR(50) COMMENT ‘本次生成使用的模型名称’, image_url TEXT NOT NULL COMMENT ‘生成结果图片的存储URL’, thumbnail_url TEXT COMMENT ‘缩略图URL用于列表快速展示’, image_size VARCHAR(20) COMMENT ‘图片尺寸如“1024x768”’, generation_time_ms INT COMMENT ‘生成耗时毫秒’, status VARCHAR(20) DEFAULT ‘success’ COMMENT ‘任务状态pending, processing, success, failed’, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT ‘记录创建时间’, INDEX idx_user_id (user_id), INDEX idx_created_at (created_at), INDEX idx_status (status), FOREIGN KEY (user_id) REFERENCES user_profile(user_id) ON DELETE CASCADE ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT‘影像生成历史记录表’;设计思路解读record_id是自增主键确保每条记录都有唯一编号。task_id很重要如果你的生成任务是异步的提交后等一会儿才出图可以用这个ID去轮询或回调查询结果。我们把关键的prompt_text提示词和image_url结果图都存了下来。这样用户就能完整回顾自己的创作历史和成果。建立了多个索引INDEX在user_id,created_at,status上建索引能极大提升“查询某个用户的所有记录”、“按时间排序”、“查找失败任务”等操作的速度。设置了外键约束FOREIGN KEYuser_id字段关联到user_profile表的user_id。ON DELETE CASCADE表示如果用户被删除他所有的生成记录也会自动级联删除保持数据干净。表设计好了仓库的货架就搭完了。接下来我们要让Python程序能跟这个仓库对话。3. 编写Python连接与操作数据库的代码我们的“云容笔谈”系统很可能是用Python写的比如用了FastAPI、Flask等框架。这里我以最通用的方式展示如何用Python连接MySQL并实现数据的增删改查。3.1 安装Python的MySQL驱动首先需要在你的Python项目环境里安装连接MySQL的库。最常用的是pymysql或mysql-connector-python。这里我用pymysql因为它纯Python实现兼容性好。在你的项目终端里运行pip install pymysql3.2 创建数据库连接工具类我们创建一个单独的Python文件比如叫database.py来管理所有数据库连接和操作。这样代码清晰也便于复用。# database.py import pymysql import json from typing import Optional, Dict, Any, List import logging # 设置日志方便出问题时排查 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class DatabaseManager: def __init__(self, host‘localhost’, user‘ink_user’, password‘YourStrongPassword123!’, database‘cloud_ink_talk’): 初始化数据库连接配置。 注意在生产环境中密码等敏感信息应从环境变量或配置文件中读取不要硬编码在代码里 self.host host self.user user self.password password self.database database self.connection None def get_connection(self): 获取数据库连接如果连接不存在或已关闭则创建新连接。 if self.connection is None or not self.connection.open: try: self.connection pymysql.connect( hostself.host, userself.user, passwordself.password, databaseself.database, charset‘utf8mb4’, cursorclasspymysql.cursors.DictCursor # 返回字典形式的结果更方便 ) logger.info(“成功连接到MySQL数据库”) except pymysql.Error as e: logger.error(f“连接数据库失败: {e}”) raise return self.connection def close_connection(self): 关闭数据库连接。 if self.connection and self.connection.open: self.connection.close() self.connection None logger.info(“数据库连接已关闭”) # 接下来我们将在这里添加具体的操作方法重要安全提醒上面的代码里密码是写死的这仅仅是为了演示。在实际项目中你必须通过环境变量、配置文件并加入.gitignore或密钥管理服务来管理数据库密码3.3 实现用户画像的增删改查方法在DatabaseManager类里继续添加操作user_profile表的方法。# 在 DatabaseManager 类中继续添加以下方法 def create_or_update_user(self, user_data: Dict[str, Any]) - bool: 创建或更新用户画像。 使用 INSERT ... ON DUPLICATE KEY UPDATE 语法如果用户ID已存在则更新。 sql “““ INSERT INTO user_profile ( user_id, username, nickname, avatar_url, preferred_style, default_model, total_credits, last_login_at, profile_meta ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s ) ON DUPLICATE KEY UPDATE nickname VALUES(nickname), avatar_url VALUES(avatar_url), preferred_style VALUES(preferred_style), default_model VALUES(default_model), total_credits VALUES(total_credits), last_login_at VALUES(last_login_at), profile_meta VALUES(profile_meta) ”““ try: with self.get_connection().cursor() as cursor: # 处理JSON字段 meta_json json.dumps(user_data.get(‘profile_meta’, {})) if user_data.get(‘profile_meta’) else None cursor.execute(sql, ( user_data[‘user_id’], user_data[‘username’], user_data.get(‘nickname’), user_data.get(‘avatar_url’), user_data.get(‘preferred_style’), user_data.get(‘default_model’, ‘default’), user_data.get(‘total_credits’, 0), user_data.get(‘last_login_at’), meta_json )) self.connection.commit() logger.info(f“用户 {user_data[‘user_id’]} 信息已保存/更新”) return True except pymysql.Error as e: logger.error(f“保存用户信息失败: {e}”) self.connection.rollback() return False def get_user_by_id(self, user_id: str) - Optional[Dict[str, Any]]: 根据用户ID查询用户信息。 sql “SELECT * FROM user_profile WHERE user_id %s” try: with self.get_connection().cursor() as cursor: cursor.execute(sql, (user_id,)) result cursor.fetchone() if result and result.get(‘profile_meta’): # 将JSON字符串解析回Python字典 result[‘profile_meta’] json.loads(result[‘profile_meta’]) return result except pymysql.Error as e: logger.error(f“查询用户信息失败: {e}”) return None3.4 实现生成记录的存储与查询方法同样在DatabaseManager类里添加操作generation_history表的方法。def insert_generation_record(self, record_data: Dict[str, Any]) - Optional[int]: 插入一条新的生成记录。 返回插入记录的 record_id。 sql “““ INSERT INTO generation_history ( user_id, task_id, prompt_text, negative_prompt, style_param, model_name, image_url, thumbnail_url, image_size, generation_time_ms, status ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ”““ try: with self.get_connection().cursor() as cursor: cursor.execute(sql, ( record_data[‘user_id’], record_data[‘task_id’], record_data[‘prompt_text’], record_data.get(‘negative_prompt’), record_data.get(‘style_param’), record_data.get(‘model_name’), record_data[‘image_url’], record_data.get(‘thumbnail_url’), record_data.get(‘image_size’), record_data.get(‘generation_time_ms’), record_data.get(‘status’, ‘success’) )) self.connection.commit() record_id cursor.lastrowid # 获取刚插入记录的自增ID logger.info(f“生成记录已保存record_id: {record_id}”) return record_id except pymysql.Error as e: logger.error(f“保存生成记录失败: {e}”) self.connection.rollback() return None def get_user_generation_history(self, user_id: str, limit: int 20, offset: int 0) - List[Dict[str, Any]]: 查询某个用户的生成历史记录按时间倒序排列。 sql “““ SELECT * FROM generation_history WHERE user_id %s ORDER BY created_at DESC LIMIT %s OFFSET %s ”““ try: with self.get_connection().cursor() as cursor: cursor.execute(sql, (user_id, limit, offset)) results cursor.fetchall() return results except pymysql.Error as e: logger.error(f“查询生成历史失败: {e}”) return []数据库操作的核心代码就准备好了。现在我们要考虑如何优雅地将这些操作融入到现有的、可能涉及耗时任务的AI生成流程中。4. 实现生成任务的异步存储与查询AI生成图片是个耗时操作如果让用户提交生成请求后一直等到图片生成完、数据存到数据库才返回响应那体验就太差了。我们应该采用异步的方式。4.1 在Web框架中集成异步存储假设你的“云容笔谈”系统使用FastAPI这是目前Python异步Web框架的主流选择。我们可以这样设计接口提交生成任务接口用户提交提示词系统立即返回一个task_id并异步启动生成任务。异步任务中存储记录在后台生成图片的异步函数里图片生成成功后调用我们的DatabaseManager.insert_generation_record方法将结果存入数据库。查询任务结果/历史接口用户可以用task_id查询单次结果或用user_id查询所有历史。下面是一个简化的FastAPI示例展示这个流程# main.py (FastAPI 应用示例) from fastapi import FastAPI, BackgroundTasks, HTTPException from pydantic import BaseModel from typing import Optional import uuid from your_ai_module import generate_image # 假设这是你的AI生成函数 from database import DatabaseManager # 导入我们刚才写的数据库管理类 app FastAPI(title“云容笔谈·东方红颜影像生成系统”) db_manager DatabaseManager() # 定义请求和响应模型 class GenerationRequest(BaseModel): user_id: str prompt: str style: Optional[str] None model: Optional[str] None class GenerationResponse(BaseModel): task_id: str status: str message: str “任务已提交正在处理中” # 全局字典或使用Redis等临时存储任务状态实际生产环境建议用消息队列结果后端 task_status_cache {} async def async_generation_task(task_id: str, request: GenerationRequest): 后台异步生成任务 try: # 1. 调用AI模型生成图片 (这是一个耗时的IO操作) image_url, thumbnail_url, gen_time await generate_image( promptrequest.prompt, stylerequest.style, modelrequest.model ) # 2. 图片生成成功后将记录存入数据库 record_data { ‘user_id’: request.user_id, ‘task_id’: task_id, ‘prompt_text’: request.prompt, ‘style_param’: request.style, ‘model_name’: request.model, ‘image_url’: image_url, ‘thumbnail_url’: thumbnail_url, ‘generation_time_ms’: gen_time, ‘status’: ‘success’ } db_manager.insert_generation_record(record_data) # 3. 更新任务缓存状态 task_status_cache[task_id] { ‘status’: ‘success’, ‘image_url’: image_url, ‘record_id’: record_id # 假设insert方法返回了ID } logger.info(f“异步任务 {task_id} 完成并已存储”) except Exception as e: logger.error(f“异步任务 {task_id} 失败: {e}”) # 即使失败也可以记录一条状态为failed的记录到数据库 db_manager.insert_generation_record({ ‘user_id’: request.user_id, ‘task_id’: task_id, ‘prompt_text’: request.prompt, ‘status’: ‘failed’ }) task_status_cache[task_id] {‘status’: ‘failed’, ‘error’: str(e)} app.post(“/generate”, response_modelGenerationResponse) async def submit_generation_task(request: GenerationRequest, background_tasks: BackgroundTasks): 提交影像生成任务 task_id str(uuid.uuid4()) # 生成唯一任务ID # 将耗时任务放入后台执行 background_tasks.add_task(async_generation_task, task_id, request) # 立即返回任务ID return GenerationResponse(task_idtask_id, status“pending”) app.get(“/task/{task_id}”) async def get_task_result(task_id: str): 根据task_id查询任务结果 result task_status_cache.get(task_id) if not result: # 如果缓存没有尝试从数据库查询适用于服务重启后 # 这里需要你实现一个根据task_id查数据库的方法 # result db_manager.get_record_by_task_id(task_id) raise HTTPException(status_code404, detail“任务不存在或已过期”) return result app.get(“/user/{user_id}/history”) async def get_user_history(user_id: str, page: int 1, size: int 10): 查询用户的生成历史 offset (page - 1) * size history db_manager.get_user_generation_history(user_id, limitsize, offsetoffset) return {“user_id”: user_id, “page”: page, “history”: history}4.2 处理高并发与数据一致性当用户量变大时你需要考虑更健壮的方案使用消息队列用Celery Redis/RabbitMQ来管理异步任务替代简单的BackgroundTasks。这样可以更好地处理任务重试、结果持久化和分布式 worker。数据库连接池上面的DatabaseManager每次操作都可能创建新连接高并发下不好。可以使用DBUtils或SQLAlchemy配合asyncpg等异步驱动来管理连接池。事务处理确保“更新用户积分”和“插入生成记录”这样的关联操作在一个事务里完成避免数据不一致。定期清理与归档generation_history表会增长很快。需要制定策略比如将超过一年的记录迁移到归档表或者只保留图片URL而将图片本身转移到对象存储如S3、MinIO的生命周期管理中。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
云容笔谈·东方红颜影像生成系统MySQL数据库集成教程：用户画像与生成记录管理
云容笔谈·东方红颜影像生成系统MySQL数据库集成教程用户画像与生成记录管理你是不是也遇到过这样的问题自己搭建的AI影像生成系统用户今天生成的图片明天就找不到了或者想给老用户一些个性化的生成建议却发现根本没有他们的历史数据。数据就像流水一样用过就没了。我之前给“云容笔谈·东方红颜”这个影像生成系统做数据管理时就碰到了这个痛点。系统跑得挺好用户反馈也不错但所有生成记录都只是临时存在内存里服务器一重启或者用户换个设备登录之前的所有痕迹就都消失了。这显然不行尤其是当你想做用户行为分析、提供个性化服务或者只是简单地让用户能回顾自己的创作历史时。所以我决定给系统加上一个“记忆中枢”——MySQL数据库。今天这篇文章我就手把手带你走一遍整个集成过程。从零开始在部署环境里装上MySQL设计两张核心表来存用户画像和生成记录再用Python代码把它们和我们的AI生成系统连接起来。整个过程我会尽量讲得直白哪怕你之前没怎么碰过数据库跟着做也能搞定。1. 环境准备在服务器上安装和配置MySQL首先我们得有个地方放数据库。假设你的“云容笔谈”系统已经部署在了一台Linux服务器上比如Ubuntu 20.04/22.04我们现在就在这台服务器上安装MySQL。打开服务器的终端用SSH连上去然后我们开始操作。1.1 安装MySQL服务器在Ubuntu上安装MySQL非常方便用系统自带的包管理器就行。我们安装的是MySQL社区版。# 首先更新一下软件包列表确保获取到最新的版本信息 sudo apt update # 安装MySQL服务器 sudo apt install mysql-server -y这个-y参数是自动确认安装省得你再手动输入一次“Y”。安装过程可能会花几分钟取决于你的网络速度。安装完成后MySQL服务会自动启动。你可以用下面这个命令检查一下它是不是在正常运行sudo systemctl status mysql如果看到绿色的“active (running)”字样就说明MySQL已经成功安装并运行起来了。1.2 进行安全初始化配置刚安装好的MySQL默认配置不太安全比如root用户可能没有密码。我们需要运行一个安全脚本进行加固。sudo mysql_secure_installation运行这个命令后它会引导你完成一系列设置验证密码强度组件它会问你是否要设置密码验证策略。对于内部系统我们可以先选“否”输入n简化流程。设置root密码这是最重要的一步。它会提示你为MySQL的root用户设置一个密码。请务必设置一个强密码并牢记。输入密码时屏幕上不会有任何显示这是正常的输完回车即可。移除匿名用户选择“是”输入y删除默认存在的匿名用户提高安全性。禁止root远程登录选择“是”输入y。这样root用户只能从本机服务器自己登录防止从外网直接攻击。删除测试数据库选择“是”输入y移除名为test的默认数据库。重新加载权限表选择“是”输入y让刚才的所有安全设置立刻生效。完成这些步骤你的MySQL基础环境就算安全地搭建好了。1.3 创建一个专用的数据库和用户为了系统安全我们不应该直接用root用户来连接应用。最佳实践是创建一个专门给“云容笔谈”系统使用的数据库和一个拥有相应权限的普通用户。我们先以root身份登录MySQL命令行sudo mysql -u root -p输入你刚才设置的root密码。成功登录后你会看到提示符变成了mysql。然后执行以下SQL命令-- 创建一个名为 cloud_ink_talk 的数据库用来存放我们系统的所有数据 CREATE DATABASE cloud_ink_talk CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci; -- 创建一个新用户用户名是 ink_user并设置一个强密码 YourStrongPassword123! -- 请务必把 YourStrongPassword123! 换成你自己想的复杂密码 CREATE USER ink_userlocalhost IDENTIFIED BY YourStrongPassword123!; -- 把 cloud_ink_talk 数据库的所有操作权限授予 ink_user 用户 GRANT ALL PRIVILEGES ON cloud_ink_talk.* TO ink_userlocalhost; -- 让权限设置立即生效 FLUSH PRIVILEGES; -- 退出MySQL命令行 EXIT;这里有几个关键点utf8mb4字符集能支持存储Emoji等特殊字符避免乱码。ink_userlocalhost表示这个用户只能从本机连接这很安全因为我们的Python应用就和数据库在同一台服务器上。记住你为ink_user设置的密码等下写Python连接代码时要用到。好了数据库服务端的基础工作就完成了。接下来我们设计一下要存什么。2. 核心设计用户画像表与生成记录表数据库就像一个大仓库表就是里面一个个分类明确的货架。我们需要设计两个“货架”的结构一个放用户信息一个放生成记录。2.1 用户画像表 (user_profile)这张表记录用户的基本信息和偏好是实现个性化的基础。我们再次登录MySQL这次可以用新建的ink_user测试一下然后创建表。你可以用sudo mysql -u root -p登录后再切换到cloud_ink_talk数据库或者直接用mysql -u ink_user -p cloud_ink_talk输入ink_user的密码。然后在mysql提示符下执行以下SQLCREATE TABLE user_profile ( user_id VARCHAR(64) PRIMARY KEY COMMENT ‘系统分配的唯一用户ID例如UUID’, username VARCHAR(50) UNIQUE NOT NULL COMMENT ‘用户登录名’, nickname VARCHAR(50) COMMENT ‘用户昵称’, avatar_url TEXT COMMENT ‘用户头像图片链接’, preferred_style VARCHAR(100) COMMENT ‘偏好的影像风格如“古风”、“现代”、“唯美”’, default_model VARCHAR(50) DEFAULT ‘default’ COMMENT ‘默认使用的生成模型’, total_credits INT DEFAULT 0 COMMENT ‘用户总积分或剩余生成次数’, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT ‘账户创建时间’, last_login_at TIMESTAMP NULL COMMENT ‘最后登录时间’, profile_meta JSON COMMENT ‘扩展信息以JSON格式存储如标签、等级等’ ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT‘用户画像信息表’;设计思路解读user_id是主键绝对唯一用字符串存兼容各种ID生成方式。preferred_style和default_model是关键字段系统可以根据这些值在用户生成时提供默认参数提升体验。total_credits可以用来做简单的积分或次数管理。profile_meta字段类型是JSON这非常灵活。以后如果想增加“用户等级”、“喜欢的颜色”等属性可以直接往这个JSON里加而不用频繁修改表结构。2.2 生成历史记录表 (generation_history)这张表记录每一次图像生成的详细“档案”。CREATE TABLE generation_history ( record_id BIGINT AUTO_INCREMENT PRIMARY KEY COMMENT ‘记录ID自增长主键’, user_id VARCHAR(64) NOT NULL COMMENT ‘关联的用户ID’, task_id VARCHAR(128) NOT NULL COMMENT ‘生成任务的唯一标识可用于查询进度’, prompt_text TEXT NOT NULL COMMENT ‘用户输入的生成提示词’, negative_prompt TEXT COMMENT ‘负面提示词’, style_param VARCHAR(100) COMMENT ‘本次生成使用的风格参数’, model_name VARCHAR(50) COMMENT ‘本次生成使用的模型名称’, image_url TEXT NOT NULL COMMENT ‘生成结果图片的存储URL’, thumbnail_url TEXT COMMENT ‘缩略图URL用于列表快速展示’, image_size VARCHAR(20) COMMENT ‘图片尺寸如“1024x768”’, generation_time_ms INT COMMENT ‘生成耗时毫秒’, status VARCHAR(20) DEFAULT ‘success’ COMMENT ‘任务状态pending, processing, success, failed’, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP COMMENT ‘记录创建时间’, INDEX idx_user_id (user_id), INDEX idx_created_at (created_at), INDEX idx_status (status), FOREIGN KEY (user_id) REFERENCES user_profile(user_id) ON DELETE CASCADE ) ENGINEInnoDB DEFAULT CHARSETutf8mb4 COMMENT‘影像生成历史记录表’;设计思路解读record_id是自增主键确保每条记录都有唯一编号。task_id很重要如果你的生成任务是异步的提交后等一会儿才出图可以用这个ID去轮询或回调查询结果。我们把关键的prompt_text提示词和image_url结果图都存了下来。这样用户就能完整回顾自己的创作历史和成果。建立了多个索引INDEX在user_id,created_at,status上建索引能极大提升“查询某个用户的所有记录”、“按时间排序”、“查找失败任务”等操作的速度。设置了外键约束FOREIGN KEYuser_id字段关联到user_profile表的user_id。ON DELETE CASCADE表示如果用户被删除他所有的生成记录也会自动级联删除保持数据干净。表设计好了仓库的货架就搭完了。接下来我们要让Python程序能跟这个仓库对话。3. 编写Python连接与操作数据库的代码我们的“云容笔谈”系统很可能是用Python写的比如用了FastAPI、Flask等框架。这里我以最通用的方式展示如何用Python连接MySQL并实现数据的增删改查。3.1 安装Python的MySQL驱动首先需要在你的Python项目环境里安装连接MySQL的库。最常用的是pymysql或mysql-connector-python。这里我用pymysql因为它纯Python实现兼容性好。在你的项目终端里运行pip install pymysql3.2 创建数据库连接工具类我们创建一个单独的Python文件比如叫database.py来管理所有数据库连接和操作。这样代码清晰也便于复用。# database.py import pymysql import json from typing import Optional, Dict, Any, List import logging # 设置日志方便出问题时排查 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) class DatabaseManager: def __init__(self, host‘localhost’, user‘ink_user’, password‘YourStrongPassword123!’, database‘cloud_ink_talk’): 初始化数据库连接配置。 注意在生产环境中密码等敏感信息应从环境变量或配置文件中读取不要硬编码在代码里 self.host host self.user user self.password password self.database database self.connection None def get_connection(self): 获取数据库连接如果连接不存在或已关闭则创建新连接。 if self.connection is None or not self.connection.open: try: self.connection pymysql.connect( hostself.host, userself.user, passwordself.password, databaseself.database, charset‘utf8mb4’, cursorclasspymysql.cursors.DictCursor # 返回字典形式的结果更方便 ) logger.info(“成功连接到MySQL数据库”) except pymysql.Error as e: logger.error(f“连接数据库失败: {e}”) raise return self.connection def close_connection(self): 关闭数据库连接。 if self.connection and self.connection.open: self.connection.close() self.connection None logger.info(“数据库连接已关闭”) # 接下来我们将在这里添加具体的操作方法重要安全提醒上面的代码里密码是写死的这仅仅是为了演示。在实际项目中你必须通过环境变量、配置文件并加入.gitignore或密钥管理服务来管理数据库密码3.3 实现用户画像的增删改查方法在DatabaseManager类里继续添加操作user_profile表的方法。# 在 DatabaseManager 类中继续添加以下方法 def create_or_update_user(self, user_data: Dict[str, Any]) - bool: 创建或更新用户画像。 使用 INSERT ... ON DUPLICATE KEY UPDATE 语法如果用户ID已存在则更新。 sql “““ INSERT INTO user_profile ( user_id, username, nickname, avatar_url, preferred_style, default_model, total_credits, last_login_at, profile_meta ) VALUES ( %s, %s, %s, %s, %s, %s, %s, %s, %s ) ON DUPLICATE KEY UPDATE nickname VALUES(nickname), avatar_url VALUES(avatar_url), preferred_style VALUES(preferred_style), default_model VALUES(default_model), total_credits VALUES(total_credits), last_login_at VALUES(last_login_at), profile_meta VALUES(profile_meta) ”““ try: with self.get_connection().cursor() as cursor: # 处理JSON字段 meta_json json.dumps(user_data.get(‘profile_meta’, {})) if user_data.get(‘profile_meta’) else None cursor.execute(sql, ( user_data[‘user_id’], user_data[‘username’], user_data.get(‘nickname’), user_data.get(‘avatar_url’), user_data.get(‘preferred_style’), user_data.get(‘default_model’, ‘default’), user_data.get(‘total_credits’, 0), user_data.get(‘last_login_at’), meta_json )) self.connection.commit() logger.info(f“用户 {user_data[‘user_id’]} 信息已保存/更新”) return True except pymysql.Error as e: logger.error(f“保存用户信息失败: {e}”) self.connection.rollback() return False def get_user_by_id(self, user_id: str) - Optional[Dict[str, Any]]: 根据用户ID查询用户信息。 sql “SELECT * FROM user_profile WHERE user_id %s” try: with self.get_connection().cursor() as cursor: cursor.execute(sql, (user_id,)) result cursor.fetchone() if result and result.get(‘profile_meta’): # 将JSON字符串解析回Python字典 result[‘profile_meta’] json.loads(result[‘profile_meta’]) return result except pymysql.Error as e: logger.error(f“查询用户信息失败: {e}”) return None3.4 实现生成记录的存储与查询方法同样在DatabaseManager类里添加操作generation_history表的方法。def insert_generation_record(self, record_data: Dict[str, Any]) - Optional[int]: 插入一条新的生成记录。 返回插入记录的 record_id。 sql “““ INSERT INTO generation_history ( user_id, task_id, prompt_text, negative_prompt, style_param, model_name, image_url, thumbnail_url, image_size, generation_time_ms, status ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ”““ try: with self.get_connection().cursor() as cursor: cursor.execute(sql, ( record_data[‘user_id’], record_data[‘task_id’], record_data[‘prompt_text’], record_data.get(‘negative_prompt’), record_data.get(‘style_param’), record_data.get(‘model_name’), record_data[‘image_url’], record_data.get(‘thumbnail_url’), record_data.get(‘image_size’), record_data.get(‘generation_time_ms’), record_data.get(‘status’, ‘success’) )) self.connection.commit() record_id cursor.lastrowid # 获取刚插入记录的自增ID logger.info(f“生成记录已保存record_id: {record_id}”) return record_id except pymysql.Error as e: logger.error(f“保存生成记录失败: {e}”) self.connection.rollback() return None def get_user_generation_history(self, user_id: str, limit: int 20, offset: int 0) - List[Dict[str, Any]]: 查询某个用户的生成历史记录按时间倒序排列。 sql “““ SELECT * FROM generation_history WHERE user_id %s ORDER BY created_at DESC LIMIT %s OFFSET %s ”““ try: with self.get_connection().cursor() as cursor: cursor.execute(sql, (user_id, limit, offset)) results cursor.fetchall() return results except pymysql.Error as e: logger.error(f“查询生成历史失败: {e}”) return []数据库操作的核心代码就准备好了。现在我们要考虑如何优雅地将这些操作融入到现有的、可能涉及耗时任务的AI生成流程中。4. 实现生成任务的异步存储与查询AI生成图片是个耗时操作如果让用户提交生成请求后一直等到图片生成完、数据存到数据库才返回响应那体验就太差了。我们应该采用异步的方式。4.1 在Web框架中集成异步存储假设你的“云容笔谈”系统使用FastAPI这是目前Python异步Web框架的主流选择。我们可以这样设计接口提交生成任务接口用户提交提示词系统立即返回一个task_id并异步启动生成任务。异步任务中存储记录在后台生成图片的异步函数里图片生成成功后调用我们的DatabaseManager.insert_generation_record方法将结果存入数据库。查询任务结果/历史接口用户可以用task_id查询单次结果或用user_id查询所有历史。下面是一个简化的FastAPI示例展示这个流程# main.py (FastAPI 应用示例) from fastapi import FastAPI, BackgroundTasks, HTTPException from pydantic import BaseModel from typing import Optional import uuid from your_ai_module import generate_image # 假设这是你的AI生成函数 from database import DatabaseManager # 导入我们刚才写的数据库管理类 app FastAPI(title“云容笔谈·东方红颜影像生成系统”) db_manager DatabaseManager() # 定义请求和响应模型 class GenerationRequest(BaseModel): user_id: str prompt: str style: Optional[str] None model: Optional[str] None class GenerationResponse(BaseModel): task_id: str status: str message: str “任务已提交正在处理中” # 全局字典或使用Redis等临时存储任务状态实际生产环境建议用消息队列结果后端 task_status_cache {} async def async_generation_task(task_id: str, request: GenerationRequest): 后台异步生成任务 try: # 1. 调用AI模型生成图片 (这是一个耗时的IO操作) image_url, thumbnail_url, gen_time await generate_image( promptrequest.prompt, stylerequest.style, modelrequest.model ) # 2. 图片生成成功后将记录存入数据库 record_data { ‘user_id’: request.user_id, ‘task_id’: task_id, ‘prompt_text’: request.prompt, ‘style_param’: request.style, ‘model_name’: request.model, ‘image_url’: image_url, ‘thumbnail_url’: thumbnail_url, ‘generation_time_ms’: gen_time, ‘status’: ‘success’ } db_manager.insert_generation_record(record_data) # 3. 更新任务缓存状态 task_status_cache[task_id] { ‘status’: ‘success’, ‘image_url’: image_url, ‘record_id’: record_id # 假设insert方法返回了ID } logger.info(f“异步任务 {task_id} 完成并已存储”) except Exception as e: logger.error(f“异步任务 {task_id} 失败: {e}”) # 即使失败也可以记录一条状态为failed的记录到数据库 db_manager.insert_generation_record({ ‘user_id’: request.user_id, ‘task_id’: task_id, ‘prompt_text’: request.prompt, ‘status’: ‘failed’ }) task_status_cache[task_id] {‘status’: ‘failed’, ‘error’: str(e)} app.post(“/generate”, response_modelGenerationResponse) async def submit_generation_task(request: GenerationRequest, background_tasks: BackgroundTasks): 提交影像生成任务 task_id str(uuid.uuid4()) # 生成唯一任务ID # 将耗时任务放入后台执行 background_tasks.add_task(async_generation_task, task_id, request) # 立即返回任务ID return GenerationResponse(task_idtask_id, status“pending”) app.get(“/task/{task_id}”) async def get_task_result(task_id: str): 根据task_id查询任务结果 result task_status_cache.get(task_id) if not result: # 如果缓存没有尝试从数据库查询适用于服务重启后 # 这里需要你实现一个根据task_id查数据库的方法 # result db_manager.get_record_by_task_id(task_id) raise HTTPException(status_code404, detail“任务不存在或已过期”) return result app.get(“/user/{user_id}/history”) async def get_user_history(user_id: str, page: int 1, size: int 10): 查询用户的生成历史 offset (page - 1) * size history db_manager.get_user_generation_history(user_id, limitsize, offsetoffset) return {“user_id”: user_id, “page”: page, “history”: history}4.2 处理高并发与数据一致性当用户量变大时你需要考虑更健壮的方案使用消息队列用Celery Redis/RabbitMQ来管理异步任务替代简单的BackgroundTasks。这样可以更好地处理任务重试、结果持久化和分布式 worker。数据库连接池上面的DatabaseManager每次操作都可能创建新连接高并发下不好。可以使用DBUtils或SQLAlchemy配合asyncpg等异步驱动来管理连接池。事务处理确保“更新用户积分”和“插入生成记录”这样的关联操作在一个事务里完成避免数据不一致。定期清理与归档generation_history表会增长很快。需要制定策略比如将超过一年的记录迁移到归档表或者只保留图片URL而将图片本身转移到对象存储如S3、MinIO的生命周期管理中。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
