day14-Trae之一键换脸APP开发04

今日内容

1 Python 对接Coze工作流

# 1 之前使用coze做了一个工作流-换脸-当时只能发布到，集成到智能体中---》发布到coze商店---》链接地址---》把链接地址发送给其他人使用-我想对这个功能收费？-我想做一款app/微信小程序---》在app，小程序上使用功能-拍照，相册中照片直接用-还想在app如收费功能。。。# 2 现在目标：使用Python代码，能调用这个工作流完成操作-后续任意工作流，都是这个流程############ python可以调用，其他语言也可以调用（我们以python为例）################ 3 Python调用coze工作流流程1 工作流发布2 根据api，使用不同语言写代码:参照这个地址：https://www.coze.cn/open/docs/developer_guides/workflow_run-1 https://api.coze.cn/v1/files/upload  上传两张图片-返回两个文件id号-2 https://api.coze.cn/v1/workflow/run  执行工作流-携带两个文件的id号-返回工作流执行的id号-3 https://api.coze.cn/v1/workflows/:workflow_id/run_histories/:execute_id -带着工作流执行的id号--查询结果-4 自己写代码把图片保存到本地3 获取api_key-https://www.coze.cn/open/oauth/pats-自己设定一个长一点的时间4 编写代码测试-api_key:我的：pat_2mYHUdcRixq90o4Nd2Jp0nPhpBcIZg4wFZkkmvxe3UHcztZdXi0glSX1rPqxHK9T-你们需要获取自己的-workflow_id：工作流id-我的是：7536960050292998154-两张图片：自己换5 测试完成图片放在-coze_results 文件夹下# 我们app后端使用Python写的---》这个代码可以直接放在我们后端项目中使用# 直接用Trae对接coze工作流，可能会有错--》所以我提前写好了一个测试通过的
# 使用提示词，让Trae它参照我的代码，再去写就不会有错了
参照： test_coze.py ，这个测试脚本我已经测试通过，按照这个修改后端coze换脸工作流代码，使其能顺利调用coze工作流

import os
import requestsimport json
import timeclass CozeWorkflowTester:def __init__(self, api_key, workflow_id, base_url='https://api.coze.cn'):self.api_key = api_keyself.workflow_id = workflow_idself.base_url = base_urlself.headers = {'Authorization': f'Bearer {self.api_key}','Content-Type': 'application/json'}self.file_upload_endpoint = f"{self.base_url}/v1/files/upload"self.workflow_endpoint = f"{self.base_url}/v1/workflow/run"# 根据用户提供的正确接口，使用参数化URL格式self.task_status_endpoint_template = "{base_url}/v1/workflows/{workflow_id}/run_histories/{execute_id}"def upload_file(self, file_path):"""上传文件到Coze并获取文件ID"""if not os.path.exists(file_path):raise FileNotFoundError(f"文件不存在: {file_path}")try:# 构建请求头，不包含Content-Type，让requests自动处理upload_headers = {'Authorization': f'Bearer {self.api_key}'}# 读取文件并准备上传with open(file_path, 'rb') as f:files = {'file': (os.path.basename(file_path), f)}# 发送请求response = requests.post(url=self.file_upload_endpoint,headers=upload_headers,files=files)# 解析响应result = response.json()# 检查是否成功if response.status_code == 200 and 'data' in result and result.get('code') == 0:file_id = result['data'].get('id')if file_id:return file_idelse:raise Exception(f"无法从响应中提取文件ID: {result}")else:raise Exception(f"文件上传失败: {result}")except Exception as e:raise Exception(f"文件上传失败")def run_workflow(self, source_image_path, target_image_path):"""运行Coze工作流"""try:# 先上传图片获取文件IDsource_file_id = self.upload_file(source_image_path)target_file_id = self.upload_file(target_image_path)payload = {"workflow_id": self.workflow_id,"parameters": {"face": json.dumps({"file_id": source_file_id}),"backed": json.dumps({"file_id": target_file_id}),"text": "执行换脸操作"},"app_id": "","is_async": True}# 发送请求response = requests.post(self.workflow_endpoint,headers=self.headers,json=payload)# 检查响应if response.status_code != 200:error_msg = f"工作流调用失败，状态码: {response.status_code}, 响应: {response.text}"raise Exception(error_msg)# 解析响应try:result = response.json()except json.JSONDecodeError:error_msg = f"无法解析响应JSON: {response.text}"raise Exception(error_msg)# 根据用户提示，正确的任务ID提取方式应该是检查msg为Success和提取execute_id# 检查响应消息是否成功if result.get('code') == 0 or result.get('msg') == '':# 提取execute_id作为任务IDtask_id = result.get('execute_id')if task_id:return task_idelse:error_msg = f"msg为Success但未找到execute_id: {result}"raise Exception(error_msg)else:error_msg = f"工作流调用失败，返回消息: {result.get('msg')}，详情: {result}"raise Exception(error_msg)except Exception as e:raise Exception(f"工作流调用异常: {str(e)}")def get_workflow_result(self, task_id):"""获取工作流执行结果"""try:# 使用用户提供的正确接口URL格式: https://api.coze.cn/v1/workflows/:workflow_id/run_histories/:execute_id# 构建完整的任务状态查询URLtask_status_endpoint = self.task_status_endpoint_template.format(base_url=self.base_url,workflow_id=self.workflow_id,execute_id=task_id)# 发送GET请求获取任务状态，新接口使用GET而不是POSTresponse = requests.get(task_status_endpoint,headers=self.headers)# 检查响应if response.status_code != 200:raise Exception(f"查询结果失败: {response.status_code} - {response.text}")# 解析响应result = response.json()# 根据用户要求，只要code为0表示成功，返回的数据在data中# 检查任务状态code = result.get('code')if code == 0:# 任务成功完成# 获取data中的结果数据data = result.get('data', {})[0]execute_status = data.get('execute_status', '')if execute_status == 'Success':result_image_url = json.loads(json.loads(data.get('output')).get('Output')).get('output')if result_image_url:# 下载结果图片到本地local_path = self.download_image(result_image_url, task_id)return {'status': 'success','result_image_url': result_image_url,'local_image_path': local_path,'processing_time': result.get('processing_time', 0),'task_id': task_id}else:return {'status': 'success','message': '任务成功但未返回图片','task_id': task_id}elif execute_status == 'Fail':return {'status': 'failed','error_message': '任务执行失败','task_id': task_id}else:return {'status': 'processing','task_id': task_id}else:error_msg = f"查询结果失败，返回消息: {result.get('msg')}，详情: {result}"raise Exception(error_msg)except Exception as e:raise Exception(f"查询结果异常: {str(e)}")def download_image(self, image_url, task_id):"""下载图片到本地"""try:# 创建保存目录result_dir = os.path.join(os.getcwd(), 'coze_results')os.makedirs(result_dir, exist_ok=True)# 生成保存路径image_filename = f"result_{task_id}_{int(time.time())}.jpg"save_path = os.path.join(result_dir, image_filename)response = requests.get(image_url, stream=True)response.raise_for_status()# 保存图片with open(save_path, 'wb') as f:for chunk in response.iter_content(chunk_size=8192):f.write(chunk)return save_pathexcept Exception as e:return Noneif __name__ == '__main__':coze = CozeWorkflowTester(api_key='pat_2mYHUdcRixq90o4Nd2Jp0nPhpBcIZg4wFZkkmvxe3UHcztZdXi0glSX1rPqxHK9T',workflow_id='7536960050292998154')try:# 运行工作流task_id = coze.run_workflow('./source.jpg', './target.jpg')# 轮询结果，最多等待120秒max_wait_time = 120wait_interval = 5elapsed_time = 0while elapsed_time < max_wait_time:result = coze.get_workflow_result(task_id)if result['status'] == 'success':print(f"\n测试成功！")print(f"任务ID: {result['task_id']}")print(f"处理时间: {result.get('processing_time', '未知')}秒")print(f"结果图片URL: {result.get('result_image_url', '无')}")print(f"本地保存路径: {result.get('local_image_path', '无')}")breakelif result['status'] == 'failed':print(f"\n测试失败！")print(f"任务ID: {result['task_id']}")print(f"错误信息: {result.get('error_message', '未知错误')}")break# 等待一段时间后重试print(f"任务处理中，{wait_interval}秒后再次查询...")time.sleep(wait_interval)elapsed_time += wait_intervalprint(f"\n测试超时！任务仍在处理中，已等待{max_wait_time}秒")print(f"任务ID: {task_id}")print("请稍后手动查询结果")except Exception as e:print(f"\n测试异常: {str(e)}")

重要

# 我们水平其实目前不太具备---》前后端都开发完成，然后一点错误都没有的能力--》即便借助了Trae-需要我们有点水平# 基础不是特别好的---》老师建议--》是在老师这个项目的基础上改-一会我讲如何在我基础上改
# 基础特别好---》可以按照我上课讲的，从零生成--》最终完成-这里面错误特别多：Trae这个ai编辑器，写android不是特别好-这里面还涉及了很多前后端联调的东西，所以有些难度# 大家如果从零生成有问题，按照一会讲的---》是在老师这个项目的基础上改# 我们学Trae的目的---》想让大家做一些小型项目---》后续慢慢成长

项目在本地运行

# 1 运行后端:方式一cd change_face_apipython manage.py runserver 0.0.0.0:8000
# 2 运行后端:方式二Trae的对话中提示词：这个文件夹：change_face_api是一个后端django项目，请帮我运行在 0.0.0.0:8000 这个地址

# 方式一：运行app端# 1 前端app运行---》使用Androidstudio打开项目
# 2 插上手机--》打开手的usb调试---》点绿色监听运行-vivo手机--》不允许安装未知来源app--》搜一下--》解开--》可以解开--》就可以装了
# 方式二：运行app端-在trae中使用提示词：这个文件夹：是一个app项目，请帮我运行在手机上# 本地跑的项目，手机无法访问必须保证：手机[无线]和电脑[有线]连同一个路由器# 老师我前两天在调试的时候，trae能打包安装成功，但是android里就有报错-原因：Androidstudio默认给使用了高版本的java导致的

2 云服务器

# 1 如果大家想上线项目，互联网用户都能使用【你的朋友都能用这个app，而不是只有我们自己玩】，必须有公网ip---》购买云服务器---》dify的云服务器部署--》讲过-如果大家只是做练习---》本地玩就够了--》互联网用户无法使用-你朋友真的要用：本地运行后端，你朋友连你家路由器，安装app，他就可以用了# 2 购买云服务-地址：106.15.79.178-使用finalshell链接

2.1 上线架构图

3 安装python3.11

# 1 阿里云的centos上有python环境:linux，mac 的很多服务使用了python开发的-我们开发项目，使用python11，这个解释器版本，我们不用，这个有点老了- python3.9.23     pip-->python/python3  pip/pip3 都被占了-咱们项目开发，在3.11上开发的，需要使用3.11的解释器来运行-我们目标【多版本共存在同一个机器上，不要用乱了】：有多个python解释器python  python3---》代指 python3.9    pip  pip3--》代指 python3.9的pippython3.11---》是我们自己装的           pip3.11 --》代指 python3.11的pip- 以后：这个云服务器上有两个python解释器python/python3/pip/pip3----->默认的 python3.9.23 ---》我们不用他python3.11/pip3.11------------>我们自己装的，运行我们的后端项目pip不是linux的一种命令嘛？为啥pip还有版本呢？
python解释器装完后会释放两个命令python pip  成对出现python执行代码  pip 下载第三方模块的# 2 我们使用源码编译安装######### 安装步骤 ######
#1  源码安装python，依赖一些第三方zlib* libffi-devel
dnf install openssl-devel bzip2-devel expat-devel readline-devel sqlite-devel psmisc libffi-devel zlib* libffi-devel  -y# 2 前往用户根目录
cd #2 下载  3.11.9 源码 服务器终端
# https://registry.npmmirror.com/binary.html?path=python/
wget https://registry.npmmirror.com/-/binary/python/3.11.9/Python-3.11.9.tgz#3  解压安装包
tar -xf Python-3.11.9.tgz#4 进入目标文件
cd Python-3.11.9#5  配置安装路径：/usr/local/python3
# 把3.11.9 编译安装到/usr/local/python311路径下
./configure --prefix=/usr/local/python311#6  编译并安装,如果报错，说明缺依赖
# yum install openssl-devel bzip2-devel expat-devel gdbm-devel readline-devel sqlite-devel psmisc libffi-devel zlib* libffi-devel  -y
# make只是编译----》可执行文件，没有安装
# 类似于在win上下载了安装包，但是没安装
# make install 安装---》类似于在win上下了安装包，一路下一步安装了，指定安装位置---》/usr/local/python39
make &&  make install#7  建立软连接：/usr/local/python311路径不在环境变量，终端命令 python3，pip3
ln -s /usr/local/python311/bin/python3 /usr/bin/python3.11
ln -s /usr/local/python311/bin/pip3 /usr/bin/pip3.11# 机器上有多个python和pip命令，对应关系如下
python       3.9      pip 
python3      3.9      pip3
python3.11   3.11      pip3.11#8  删除安装包与文件： 不删留着也可以
cd 
rm -rf Python-3.11.9
rm -rf Python-3.11.9.tgz

#  python解释器、虚拟环境、conda、各种版本兼容、环境等等，这些蒙蒙的，怎么统一管理？python解释器是运行python代码的软件--》必须要有虚拟环境：多个项目，为了防止第三方模块冲突---》所有每个项目使用一个环境换脸项目---》第三方模块---》单独放在一个小房子里---》用自己的换肾项目---》第三方模块---》单独放在一个小房子里---》用自己的一个操作系统内上装了多个python解释器3.11   3.12 -conda 就是python虚拟环境、-第三模块有版本：不同项目用的版本又不一样# 服务器上有Py安装包，为什么不卸载然后装？linux 服务上有很多内置服务--》内置软件这个内置软件使用python写的---》使用的版本是python3.9.23--》如果把它写了--》很多系统服务就运行不了了---》系统就进不来不要动别人的，我们再这个基础上加我们自己的

4 安装nginx

# 软件：反向代理服务器   反向带代理服务器- 做请求转发    （前端来了个请求---》打在了80端口上---》转到本地8000端口，或者其他机器的某个端口）- 静态资源代理    前端项目直接放在服务器上某个位置----》请求来了，使用nginx拿到访问的内容，直接返回- 负载均衡       假设来了1000个请求--》打在nginx上，nginx性能很高，能顶住---》只转发到某个django项目，可能顶不住---》集群化的不是3台django---》均匀的打在3台机器上# 前往用户根目录
cd ~#下载nginx 1.28.0
wget https://nginx.org/download/nginx-1.28.0.tar.gz
#解压安装包
tar -xf nginx-1.28.0.tar.gz#进入目标文件
cd nginx-1.28.0# 配置安装路径：/usr/local/nginx
# 安装 PCRE 开发包的名称是 pcre-devel，支持https访问
#zlib-devel：用于 gzip 压缩模块。openssl-devel：用于 SSL/TLS 模块（如启用 HTTPS） gcc 和 make：编译工具链
dnf install -y pcre-devel gcc make zlib-devel openssl-devel
./configure --prefix=/usr/local/nginx --with-http_ssl_module
#编译并安装make &&  make install# 建立软连接：终端命令 nginx
ln -s /usr/local/nginx/sbin/nginx /usr/bin/nginx #删除安装包与文件：
cd ~
rm -rf nginx-1.28.0
rm -rf nginx-1.28.0.tar.xz# 测试Nginx环境，服务器运行nginx，本地访问服务器ip
nginx   # 启动nginx服务，监听80端口----》公网ip 80 端口就能看到页面了
服务器绑定的域名 或 ip:80http://106.15.79.178/ #就能看到下图# 部署一个飞机大战：http://106.15.79.178/玩飞机大战---》讲完课就释放了-trae生成一个js的飞机大战项目-压缩-上传到：/usr/local/nginx/html- cd /usr/local/nginx/html/-解压缩：unzip plan.zip -cd plan-执行：mv * ../ -停止：nginx -s stop -启动：nginx-访问：http://106.15.79.178/# 静态文件放的路径
/usr/local/nginx/html# 查看进程
ps aux | grep nginx# 关闭和启动
关闭：nginx -s stop 
启动： nginx# 它有配置文件---》配置监听那些地址，配置代理那些静态文件---》还没讲

5 安装mysql8

### 1 官方yum源
https://dev.mysql.com/downloads/repo/yum/### 2 下载对应版本mysql源到本地，如果系统是centos9，这里选择el9版本
# no architecture的缩写，说明这个包可以在各个不同的cpu上使用
我们选择  mysql84-community-release-el9-1.noarch.rpm### 3 或者直接来到：https://repo.mysql.com/
找到相应版本下载，我们下载
https://dev.mysql.com/downloads/file/?id=528548# 从这开始-----### 4 下载rpm包
wget https://dev.mysql.com/get/mysql84-community-release-el9-1.noarch.rpm### 5 安装rpm包
dnf install -y mysql84-community-release-el9-1.noarch.rpm### 6 开始安装
dnf install -y mysql-community-server --nogpgcheck  # 会自动把客户端装上
### 7 启动，查看状态
systemctl start mysqld
systemctl status mysqld### 8 查看默认密码并登录
grep "password" /var/log/mysqld.log  #     RDSfU%&)W9U2### 9 修改密码
mysql -uroot -p
ALTER USER 'root'@'localhost' IDENTIFIED BY 'Lqz12345?';#创建用户
CREATE USER 'root'@'%' IDENTIFIED BY 'Lqz12345?';
GRANT ALL PRIVILEGES ON *.* TO 'root'@'%' WITH GRANT OPTION;### 10 查看mysql版本
mysql -V######## 后面上线项目时应该做的## 现在直接做了#### 使用navicat 链接###
如果链接不上，就是安全组没开## 创建库：change_face- navicat 图形化解密创建，change_face# 12 安装mysqlclient
dnf install python3-devel mysql-devel --nogpgcheck -y
pip3.11 install mysqlclient

6 uwsgi

# 测试阶段使用 python manage.py runserver 0.0.0.0:8000-并发量低--》测试用可以-两三个人用# 上线阶段使用--》给互联网用户用--》需要高并发-成百上千-需要再项目中编写配置文件

#1 安装uwsgi dnf install -y python3-devel gcc libxml2-devel
pip3.11 install uwsgi
ln -s /usr/local/python311/bin/uwsgi /usr/bin/uwsgi  # 以后在任意路径下敲uwsgi都能找到-uwsgi 装好，没在环境变量中--》在其他位置敲这个命令，找不到-做了软连接--》在任意位置敲这个软件名字，都能找到# 2 敲 uwsgi  有反应，就是装好了

7 上传后端项目

7.1 编写uwsgi配置文件--change_face.ini

[uwsgi]
socket = 127.0.0.1:8000
chdir = /root/change_face_api/     # 这个是项目名，可能改，如果是自己项目
wsgi-file = change_face_project.wsgi # 这个需要改
processes = 4
threads = 2
master = true
daemonize = uwsgi.log

7.2 修改配置文件-settings.py

DEBUG = False
ALLOWED_HOSTS = ['*']
DATABASES = {'default': {'ENGINE': 'django.db.backends.mysql','NAME': 'change_face','USER': 'root','PASSWORD': 'Lqz12345?','HOST': '127.0.0.1','PORT': '3306','OPTIONS': {'charset': 'utf8mb4',},}
}

7.3 在项目目录下新建 requirements.txt 如果有就不需要创建了

Django==4.2
requests==2.31.0
djangorestframework==3.14.0
django-simpleui==2023.12.12
pillow ==11.2.1
djangorestframework-simplejwt==5.3.1

7.4 压缩上传

# 1 把后端项目压缩成zip--》放到桌面上
# 2 上传到服务器
# 3 解压unzip change_face_api.zip 
# 4 进入到文件中cd change_face_api# 5 安装依赖pip3.11 install -r requirements.txt# 6 使用manage.py 运行看看有没有错，如果每错，再使用uwsgi运行# 查看依赖是否装好 ：   pip3.11 listpython3.11 manage.py runserver 0.0.0.0:8000# 7 使用uwsgi运行#7. 1 启动uwsgi:注意目录，项目目录下uwsgi change_face.ini  # django项目跑在 8000端口了# 2 查看ps aux |grep uwsgi# 3 停止pkill -9 uwsgi

8 nginx配置

8.1 请求打在nginx上--转发给uwsgi

# 1 配置nginx转发
cd /usr/local/nginx/conf
mv nginx.conf nginx.conf.bak  # 把原来的配置文件备份一下
vi nginx.conf  # 创建一个新的
# 新增的server
events {worker_connections  1024;
}
http {include       mime.types;default_type  application/octet-stream;sendfile        on;server {listen 8080;server_name  127.0.0.1;charset utf-8;location / {include uwsgi_params;uwsgi_pass 127.0.0.1:8000;uwsgi_param UWSGI_SCRIPT change_face_project.wsgi; uwsgi_param UWSGI_CHDIR /root/change_face_api/;}location /static {alias /home/static;}}
}# 2 重启nginx 
nginx -s reload

8.2 导入假数据

# 1 本地导出-在数据库上右键--》转储sql文件--》数据和文件-你们可以直接用我导出的
# 2 云服务的数据库中导入-在数据库上右键---》运行sql文件--》选择导出的sql

8.3 浏览器访问-看到下面表示成功

http://106.15.79.178:8080/admin/login/?next=/admin/

9 media访问

10 app打包上线