体验华为开发者空间基于华为云开发环境桌面版的人脸颜值打分案例，反馈改进建议，请直接在评论区反馈即可体验指导：cid:link_0

体验华为开发者空间【案例共创】基于华为云主机的软件定义网络SDN实践案例，反馈改进建议，请直接在评论区反馈即可体验指导：cid:link_0

体验华为开发者空间【案例共创】基于华为云主机的剪刀手手势识别案例，反馈改进建议，请直接在评论区反馈即可体验指导：cid:link_0

1 引言在进行数据存储和管理时，我们常常需要一个轻量级而又功能强大的数据库系统。SQLite 是一个极受欢迎的轻量级数据库，它被广泛应用于各种大小的项目中。Python 自带了对 SQLite 的支持，使得在 Python 中使用 SQLite 变得非常简单和直观。本文将详细介绍 Python 中如何使用 SQLite3 库进行数据库操作。2 SQLite3 简介SQLite 是一个基于 C 语言的轻量级 SQL 数据库引擎。它不需要独立的服务器进程，数据库完整地存储在一个磁盘文件中。Python 标准库中的 sqlite3 模块提供了一个与 SQLite 数据库交互的接口。3 基本步骤使用 SQLite3 模块主要包括以下几个步骤：连接数据库创建表插入数据查询数据更新/删除数据关闭数据库连接4 示例代码4.1 连接数据库1234import sqlite3 # 连接到数据库（如果不存在，则创建）conn = sqlite3.connect('example.db')4.2 创建表123456# 创建一个 Cursor 对象cursor = conn.cursor() # 创建表cursor.execute('''CREATE TABLE IF NOT EXISTS users               (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)''')4.3 插入数据123456# 插入数据cursor.execute("INSERT INTO users (name, age) VALUES ('Alice', 30)")cursor.execute("INSERT INTO users (name, age) VALUES ('Bob', 25)") # 提交事务conn.commit()4.4 查询数据123# 查询数据cursor.execute("SELECT * FROM users")print(cursor.fetchall())4.5 更新/删除数据12345678# 更新数据cursor.execute("UPDATE users SET age = 31 WHERE name = 'Alice'") # 删除数据cursor.execute("DELETE FROM users WHERE name = 'Bob'") # 提交事务conn.commit()4.6 关闭数据库连接123# 关闭 Cursor 和 Connectioncursor.close()conn.close()注意事项异常处理：在进行数据库操作时，使用 try...except 语句处理异常是一个好习惯。事务管理：使用 commit() 来保存更改，使用 rollback() 来撤销未提交的事务。数据库安全：避免 SQL 注入攻击，最好使用参数化查询。5 实例演示以下是一个使用 Python 中的 sqlite3 模块来操作 SQLite 数据库的完整示例。这个例子将展示如何连接到数据库、创建表、插入、查询、更新、删除数据，并最终关闭数据库连接。我们将这些操作封装在一个名为 sqlite_example.py 的 Python 文件中。123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107# sqlite_example.pyimport sqlite3 def create_connection(db_file):    """ 创建数据库连接 """    conn = None    try:        conn = sqlite3.connect(db_file)        return conn    except sqlite3.Error as e:        print(e)    return conn def create_table(conn):    """ 创建表 """    try:        cursor = conn.cursor()        cursor.execute('''CREATE TABLE IF NOT EXISTS users                        (id INTEGER PRIMARY KEY, name TEXT, age INTEGER)''')    except sqlite3.Error as e:        print(e) def insert_user(conn, user):    """ 插入新用户 """    try:        cursor = conn.cursor()        cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", user)        conn.commit()    except sqlite3.Error as e:        print(e) def update_user(conn, user):    """ 更新用户信息 """    try:        cursor = conn.cursor()        cursor.execute("UPDATE users SET age = ? WHERE name = ?", user)        conn.commit()    except sqlite3.Error as e:        print(e) def delete_user(conn, name):    """ 删除用户 """    try:        cursor = conn.cursor()        cursor.execute("DELETE FROM users WHERE name = ?", (name,))        conn.commit()    except sqlite3.Error as e:        print(e) def select_all_users(conn):    """ 查询所有用户 """    try:        cursor = conn.cursor()        cursor.execute("SELECT * FROM users")        rows = cursor.fetchall()        for row in rows:            print(row)    except sqlite3.Error as e:        print(e) def main():    database = "example.db"     # 创建数据库连接    conn = create_connection(database)     # 创建表    if conn is not None:        create_table(conn)                 # 插入数据        insert_user(conn, ('Alice', 30))        insert_user(conn, ('Bob', 25))         # 更新数据        update_user(conn, (35, 'Alice'))         # 查询数据        print("查询到的所有用户:")        select_all_users(conn)         # 删除数据        delete_user(conn, 'Bob')         # 再次查询数据        print("\n删除后的用户数据:")        select_all_users(conn)         # 关闭数据库连接        conn.close()    else:        print("Error! 无法创建数据库连接。") if __name__ == '__main__':    main()     """ 控制台输出结果：查询到的所有用户: (1, 'Alice', 35) (2, 'Alice', 35) (3, 'Bob', 25)   删除后的用户数据: (1, 'Alice', 35) (2, 'Alice', 35) """在这个脚本中，我们定义了不同的函数来处理数据库的创建连接、创建表、插入、更新、删除和查询操作。main 函数中整合了这些操作的流程。运行这个脚本将在同目录下创建一个名为 example.db 的 SQLite 数据库文件，并在其中执行了一系列的数据库操作。TIPS可以通过数据库浏览器查询创建的数据库，

为什么需要个人笔记管理器在当今信息爆炸的数字时代，每天我们都会接触到海量的知识和信息。研究表明，一个普通上班族平均每天要处理超过100条各类信息。如何高效地组织和检索这些信息，成为现代人必须面对的挑战。Markdown作为一种轻量级标记语言，因其简洁的语法和出色的兼容性，已经成为技术文档撰写和日常笔记记录的事实标准。它的纯文本特性使其具备极佳的跨平台性，无论是Windows、macOS还是Linux系统都能完美支持，且文件大小通常只有几KB，非常适合作为个人知识管理的基础格式。本文将详细指导你使用Python构建一个功能完备的命令行笔记管理器。这个工具具有以下核心特点：完全基于命令行操作，无需图形界面使用纯文本文件存储数据，无需安装数据库支持笔记的增删改查基本功能采用模块化设计，便于后续功能扩展特别值得一提的是，这个项目非常适合编程新手作为第一个完整的Python应用来实践，你将会学习到：文件读写操作命令行参数解析正则表达式应用基本的异常处理整个项目代码量控制在200行以内，但涵盖了软件开发的核心流程，是理解程序如何从想法到实现的绝佳案例。核心功能设计我们的笔记管理器将实现以下基础功能：笔记创建功能快速新建Markdown文件：支持一键创建新的Markdown文档，自动生成基本文档结构（包含标题、日期等元信息）多种创建方式：快捷键操作（如Ctrl+N）命令行快速创建（note new 文件名）右键菜单选项智能命名：支持自动生成文件名（基于时间戳或首行标题）集成系统默认编辑器：自动调用用户设置的默认文本编辑器（如VS Code、Sublime Text等）编辑特性：语法高亮支持实时保存机制版本历史记录（可选）多编辑器支持：可配置使用不同编辑器打开特定类型的文档按日期自动分类存储：文件按"年/月/日"层级自动归档目录结构示例：notes/├── 2023/│   ├── 01/│   │   ├── 15_meeting-notes.md│   │   └── 20_project-ideas.md│   └── 02/│       └── 03_daily-journal.md└── 2024/    └── ...自定义分类：支持按项目、标签等附加分类方式基于关键词快速定位：支持以下搜索方式：单关键词搜索（如"会议"）多关键词组合搜索（如"项目 + 进度"）正则表达式搜索搜索范围：可限定特定日期范围或目录层级搜索结果展示：高亮显示匹配内容，并显示上下文片段终端渲染Markdown基础格式：在终端中直接预览渲染效果，支持：标题级别显示列表项缩进代码块高亮简单表格渲染预览模式：支持以下两种方式：实时预览（分屏显示）静态渲染（执行命令后输出渲染结果）会议记录：快速创建笔记 → 编辑内容 → 自动归档 → 后续可通过关键词搜索查找学习笔记：结构化存储学习资料 → 终端预览复习 → 按日期追踪学习进度项目管理：分类管理各项目文档 → 快速检索相关文件 → 团队成员共享笔记目录graph TD    A[用户输入命令] --> B[创建笔记]    A --> C[搜索笔记]    A --> D[编辑笔记]    A --> E[预览笔记]    B --> F[按日期存储]    C --> G[关键词匹配]环境准备确保安装Python 3.6+，无需额外库：12python --version  # 检查版本mkdir md_notes && cd md_notes  # 创建项目目录代码实现（分模块讲解）1. 文件管理模块1234567891011import osfrom datetime import datetime NOTE_DIR = "notes" def init_storage():    """创建按日期分类的存储目录"""    today = datetime.now().strftime("%Y-%m-%d")    daily_dir = os.path.join(NOTE_DIR, today)    os.makedirs(daily_dir, exist_ok=True)    return daily_dir2. 笔记创建功能1234567891011121314def create_note():    """创建新笔记并打开编辑器"""    note_dir = init_storage()    title = input("输入笔记标题: ").strip()    filename = f"{title.replace(' ', '_')}.md"    filepath = os.path.join(note_dir, filename)         with open(filepath, 'w') as f:        f.write(f"# {title}\n\n创建时间: {datetime.now()}\n\n")         # 调用系统默认编辑器    editor = os.getenv('EDITOR', 'vim')  # Windows可改为'notepad'    os.system(f"{editor} {filepath}")    print(f"笔记已保存至: {filepath}")3. 搜索功能实现1234567891011def search_notes(keyword):    """全文搜索关键词"""    matches = []    for root, _, files in os.walk(NOTE_DIR):        for file in files:            if file.endswith(".md"):                path = os.path.join(root, file)                with open(path, 'r') as f:                    if keyword.lower() in f.read().lower():                        matches.append(path)    return matches4. 终端预览功能12345678910def preview_note(filepath):    """在终端显示Markdown基础渲染"""    from rich.markdown import Markdown    from rich.console import Console         with open(filepath, 'r') as f:        content = f.read()         console = Console()    console.print(Markdown(content))主程序逻辑1234567891011121314151617181920212223import argparse def main():    parser = argparse.ArgumentParser(description="Markdown笔记管理器")    subparsers = parser.add_subparsers(dest='command')         # 创建子命令    create_parser = subparsers.add_parser('new', help='创建新笔记')    search_parser = subparsers.add_parser('search', help='搜索笔记')    search_parser.add_argument('keyword', help='搜索关键词')         args = parser.parse_args()         if args.command == 'new':        create_note()    elif args.command == 'search':        results = search_notes(args.keyword)        print(f"找到{len(results)}条匹配记录:")        for r in results:            print(f"- {r}") if __name__ == "__main__":    main()使用示例# 创建笔记python notes.py new输入笔记标题: Python学习笔记# 搜索笔记python notes.py search 装饰器找到3条匹配记录:- notes/2023-08-15/Python高级技巧.md- notes/2023-08-18/设计模式实践.md

Flask详解Flask简介Flask是一个用Python编写的轻量级Web应用框架。它被称为“微框架”，因为它不包含ORM（对象关系映射）、表单验证等开箱即用的功能。相反，Flask提供了核心功能，并允许开发者根据项目需求自由选择和集成各种扩展。这使得Flask非常灵活，适合构建小型应用、API服务以及作为大型项目中的微服务。安装和基础配置安装Flask非常简单，通常使用pip进行安装：1pip install Flask核心概念Werkzeug：一个WSGI（Web Server Gateway Interface）工具包，用于处理HTTP请求和响应。Jinja2：一个现代的、设计友好的Python模板引擎，用于渲染HTML页面。路由（Routing）：将URL映射到Python函数。视图函数（View Function）：处理请求并返回响应的Python函数。模板（Templates）：用于生成HTML的文本文件，通常使用Jinja2语法。路由和视图Flask使用@app.route()装饰器来定义路由和视图函数：1234567from flask import Flaskapp = Flask(__name__)@app.route('/')def hello_world():    return 'Hello, World!'if __name__ == '__main__':    app.run(debug=True)模板系统Flask使用Jinja2作为其模板引擎。你可以在templates文件夹中创建HTML文件，并在视图函数中渲染它们：1234567from flask import Flaskapp = Flask(__name__)@app.route('/')def hello_world():    return 'Hello, World!'if __name__ == '__main__':    app.run(debug=True)templates/user.html:123456789<!DOCTYPE html><html><head>    <title>User Page</title></head><body>    <h1>Hello, {{ name }}!</h1></body></html>数据库集成Flask本身不提供数据库集成，但可以通过各种扩展来支持。例如，可以使用Flask-SQLAlchemy来集成SQLAlchemy ORM：pip install Flask-SQLAlchemy1234567891011121314151617181920212223242526from flask import Flaskfrom flask_sqlalchemy import SQLAlchemyapp = Flask(__name__)app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///test.db'db = SQLAlchemy(app)class User(db.Model):    id = db.Column(db.Integer, primary_key=True)    username = db.Column(db.String(80), unique=True, nullable=False)    email = db.Column(db.String(120), unique=True, nullable=False)    def __repr__(self):        return '<User %r>' % self.usernamewith app.app_context():    db.create_all()# 示例：添加用户# with app.app_context():#     admin = User(username='admin', email='admin@example.com')#     db.session.add(admin)#     db.session.commit()​# 示例：查询用户# with app.app_context():#     user = User.query.filter_by(username='admin').first()#     print(user.email)​if __name__ == '__main__':    app.run(debug=True)实际示例下面是一个简单的Flask应用，演示了如何创建一个Web页面并显示一些数据：首先，创建一个名为app.py的文件：1234567891011from flask import Flask, render_templateapp = Flask(__name__)@app.route('/')def index():    data = {        'title': '我的Flask应用',        'items': ['项目A', '项目B', '项目C']    }    return render_template('index.html', data=data)if __name__ == '__main__':    app.run(debug=True)然后，在与app.py同级目录下创建一个名为templates的文件夹，并在其中创建一个名为index.html的文件：1234567891011121314<!DOCTYPE html><html><head>    <title>{{ data.title }}</title></head><body>    <h1>{{ data.title }}</h1>    <ul>        {% for item in data.items %}        <li>{{ item }}</li>        {% endfor %}    </ul></body></html>运行应用：python app.py在浏览器中访问http://127.0.0.1:5000/，你将看到一个包含标题和列表的页面。Streamlit详解Streamlit简介Streamlit是一个开源的Python库，它允许数据科学家和机器学习工程师快速创建交互式Web应用，而无需Web开发经验。Streamlit的特点是简单、快速，只需几行Python代码就能将数据脚本、模型和可视化转换为可分享的Web应用。安装和基础配置安装Streamlit非常简单，通常使用pip进行安装：1pip install streamlit运行Streamlit应用：1streamlit run your_app.py核心概念基于脚本：Streamlit应用是Python脚本，每次用户交互（如滑动、输入）都会重新运行整个脚本。数据流：Streamlit通过缓存机制优化数据处理，避免重复计算。Widget：Streamlit提供了丰富的UI组件（如滑块、按钮、文本输入框），用于用户交互。组件和布局Streamlit提供了多种组件和布局选项，可以轻松构建复杂的界面：1234567891011121314151617181920212223import streamlit as stst.title("我的Streamlit应用")st.header("这是一个标题")st.subheader("这是一个副标题")st.write("Hello, Streamlit!")name = st.text_input("请输入你的名字")if name:    st.write(f"你好, {name}!")number = st.slider("选择一个数字", 0, 100, 50)st.write(f"你选择的数字是: {number}")option = st.selectbox(    "你喜欢哪种水果?",    ("苹果", "香蕉", "橙子"))st.write(f"你选择的水果是: {option}")# 布局col1, col2 = st.columns(2)with col1:    st.header("第一列")    st.write("这是第一列的内容。")with col2:    st.header("第二列")    st.write("这是第二列的内容。")数据可视化Streamlit与流行的Python数据可视化库（如Matplotlib、Plotly、Altair）无缝集成，可以轻松展示数据：12345678910111213141516import streamlit as stimport pandas as pdimport numpy as npimport matplotlib.pyplot as pltst.title("数据可视化示例")# 生成一些随机数据data = pd.DataFrame(    np.random.randn(20, 3),    columns=["a", "b", "c"])st.line_chart(data)# 使用Matplotlib绘制图表st.subheader("Matplotlib图表")fig, ax = plt.subplots()ax.scatter(data["a"], data["b"])st.pyplot(fig)实际示例下面是一个简单的Streamlit应用，用于展示一个随机生成的数据表格和图表：首先，创建一个名为my_streamlit_app.py的文件：123456789101112131415161718import streamlit as stimport pandas as pdimport numpy as npst.title("随机数据生成器")# 用户输入生成数据的大小num_rows = st.slider("选择行数", 10, 100, 50)# 生成随机数据data = pd.DataFrame(    np.random.randn(num_rows, 5),    columns=["A", "B", "C", "D", "E"])st.subheader("原始数据")st.dataframe(data)st.subheader("数据折线图")st.line_chart(data)# 显示一些统计信息st.subheader("数据统计")st.write(data.describe())运行应用：streamlit run my_streamlit_app.py在浏览器中，你将看到一个交互式的数据应用，可以调整行数并实时查看数据表格和折线图的变化。FastAPI详解FastAPI简介FastAPI是一个现代、高性能的Web框架，用于使用Python 3.7+构建API。它基于标准的Python类型提示，并利用Starlette（用于Web部分）和Pydantic（用于数据部分）来提供出色的性能和开发体验。FastAPI的主要特点包括：高性能：与Node.js和Go相当，得益于Starlette和Pydantic。快速开发：减少了大约20%的开发时间。更少的Bug：减少了大约40%的人为错误。直观：强大的编辑器支持，代码补全。健壮：自动生成交互式API文档（Swagger UI和ReDoc）。基于标准：基于OpenAPI和JSON Schema。安装和基础配置安装FastAPI及其推荐的ASGI服务器Uvicorn：1pip install fastapi uvicorn核心概念Pydantic：用于数据验证、设置和文档的Python库。FastAPI使用它来定义请求体和响应模型。Starlette：一个轻量级的ASGI框架/工具包，FastAPI在其之上构建。类型提示（Type Hints）：Python 3.5+引入的特性，用于声明变量类型。FastAPI广泛使用它们来提供数据验证、序列化和自动文档。路径操作（Path Operations）：定义API端点和HTTP方法。路径操作FastAPI使用装饰器来定义路径操作，类似于Flask的路由：123456789from fastapi import FastAPIapp = FastAPI()@app.get("/")async def read_root():    return {"Hello": "World"}@app.get("/items/{item_id}")async def read_item(item_id: int, q: str | None = None):    return {"item_id": item_id, "q": q}​运行应用：uvicorn main:app --reload数据验证FastAPI利用Pydantic进行数据验证。你可以定义Pydantic模型来处理请求体：1234567891011from fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Item(BaseModel):    name: str    description: str | None = None    price: float    tax: float | None = None@app.post("/items/")async def create_item(item: Item):    return item异步支持FastAPI原生支持异步（async/await），这使得它在处理I/O密集型任务时表现出色：1234567from fastapi import FastAPIimport asyncioapp = FastAPI()@app.get("/async_data/")async def get_async_data():    await asyncio.sleep(1) # 模拟异步I/O操作    return {"data": "这是异步数据"}实际示例下面是一个简单的FastAPI应用，演示了如何创建一个带有数据验证的API：首先，创建一个名为main.py的文件：12345678910111213141516from fastapi import FastAPIfrom pydantic import BaseModelapp = FastAPI()class Product(BaseModel):    name: str    price: float    is_available: bool = True@app.post("/products/")async def create_product(product: Product):    return {"message": "产品创建成功", "product": product}@app.get("/products/{product_name}")async def get_product(product_name: str):    # 实际应用中会从数据库查询    if product_name == "apple":        return Product(name="apple", price=1.0, is_available=True)    return {"message": "产品未找到"}运行应用：uvicorn main:app --reload打开浏览器访问 http://127.0.0.1:8000/docs，你将看到自动生成的Swagger UI文档，可以测试API。框架对比特性FlaskStreamlitFastAPI类型微框架，Web应用和API数据应用和仪表板API框架，高性能Web服务学习曲线中等，需要了解Web开发基础低，Python脚本即可中等，需要了解异步编程和类型提示性能良好，但不如FastAPI适用于数据展示，非高并发Web服务极高，与Node.js和Go相当主要用途传统Web应用、小型API、微服务数据科学应用、机器学习演示、仪表板高性能API、微服务、数据服务异步支持不原生支持，需借助扩展不适用原生支持async/await数据验证需借助扩展（如WTForms、Marshmallow）内置UI组件提供简单验证基于Pydantic，自动数据验证和序列化自动文档无，需借助扩展（如Flask-RESTX）无内置Swagger UI和ReDoc生态系统成熟，大量扩展和社区支持活跃，专注于数据科学领域快速增长，得益于Starlette和Pydantic最佳实践选择合适的框架：如果你需要构建一个传统的Web应用，或者一个轻量级的API服务，并且希望有更大的灵活性和控制权，Flask是一个不错的选择。如果你是一名数据科学家或机器学习工程师，希望快速将数据分析、模型演示或交互式仪表板部署为Web应用，而不想深入Web开发细节，Streamlit是你的首选。如果你需要构建高性能的API服务，对性能有较高要求，并且希望利用Python的类型提示进行数据验证和自动文档生成，FastAPI是最佳选择。项目结构：对于Flask和FastAPI项目，建议采用模块化的项目结构，将路由、模型、服务等分离到不同的文件中，保持代码的清晰和可维护性。Streamlit应用通常是单个Python脚本，但对于复杂的应用，也可以考虑将逻辑拆分为多个函数或模块。依赖管理：始终使用pipenv或poetry等工具管理项目依赖，确保环境的隔离和可复现性。错误处理和日志：在所有框架中，都应该实现健壮的错误处理机制，并使用日志记录来跟踪应用的行为和问题。安全性：对于Web应用和API，安全性至关重要。始终注意输入验证、身份验证、授权、CORS等安全问题。测试：为你的应用编写单元测试和集成测试，确保代码的质量和功能的正确性。部署：根据框架和应用类型选择合适的部署方式。Flask和FastAPI应用通常部署在Gunicorn、Uvicorn等ASGI/WSGI服务器上，并通过Nginx或Apache进行反向代理。Streamlit应用可以直接通过streamlit run运行，也可以部署到Streamlit Cloud或其他云平台。

基本使用方法1. 安装1pip install pywebview2. 简单示例（Python与HTML交互）12345678910111213141516171819202122import webviewdef print_message(message):    print(f"收到前端消息: {message}")# 创建一个HTML字符串作为界面html = """<!DOCTYPE html><html><head>    <title>PyWebView示例</title></head><body>    <button onclick="python.print_message('你好，Python！')">发送消息到Python</button>    <script>        // 调用Python函数        window.python = pywebview.api;    </script></body></html>"""# 创建窗口并加载HTMLwindow = webview.create_window("PyWebView应用", html=html)webview.start(gui=None, debug=True)  # gui=None自动选择系统默认浏览器引擎结合Flask与waitress开发应用架构设计Flask：提供Web API，处理业务逻辑。pywebview：作为桌面应用的外壳，加载Flask的Web界面。waitress：在生产环境中替代Flask内置服务器，提供更好的性能。开发环境代码示例12345678910111213141516171819202122232425262728293031# app.py - Flask后端from flask import Flask, jsonifyapp = Flask(__name__)@app.route('/')def index():    return """    <!DOCTYPE html>    <html>    <head>        <title>Flask + PyWebView应用</title>    </head>    <body>        <h1>Hello from Flask!</h1>        <button onclick="fetchData()">获取数据</button>        <div id="result"></div>        <script>            async function fetchData() {                const response = await fetch('/api/data');                const data = await response.json();                document.getElementById('result').textContent = data.message;            }        </script>    </body>    </html>    """@app.route('/api/data')def get_data():    return jsonify({"message": "这是来自Flask后端的数据！"})if __name__ == '__main__':    # 开发环境：使用Flask内置服务器    app.run(debug=True, port=5000)1234567891011121314# desktop.py - PyWebView桌面应用包装import webviewimport threadingimport subprocessdef run_flask():    # 启动Flask应用（开发环境）    subprocess.run(['python', 'app.py'])if __name__ == '__main__':    # 在后台线程中启动Flask    flask_thread = threading.Thread(target=run_flask, daemon=True)    flask_thread.start()    # 创建PyWebView窗口，加载Flask应用    window = webview.create_window("桌面应用", "http://localhost:5000")    webview.start(gui=None, debug=True)生产环境配置（使用waitress）1. 修改Flask应用（app.py）123456789101112# app.py - Flask后端（生产环境）from flask import Flask, jsonifyapp = Flask(__name__)@app.route('/')def index():    return """    <!-- 同上，HTML内容 -->    """@app.route('/api/data')def get_data():    return jsonify({"message": "这是来自Flask后端的数据！"})# 移除开发环境的app.run()，改为导出app实例2. 使用waitress启动Flask12345# server.py - 使用waitress启动Flask（生产环境）from waitress import servefrom app import appif __name__ == '__main__':    serve(app, host='127.0.0.1', port=5000, threads=8)  # 生产环境使用waitress3. 修改PyWebView应用（desktop.py）1234567891011121314# desktop.py - PyWebView桌面应用（生产环境）import webviewimport threadingimport subprocessdef run_flask():    # 启动Flask应用（生产环境使用waitress）    subprocess.run(['python', 'server.py'])if __name__ == '__main__':    # 在后台线程中启动Flask    flask_thread = threading.Thread(target=run_flask, daemon=True)    flask_thread.start()    # 创建PyWebView窗口，加载Flask应用    window = webview.create_window("桌面应用", "http://localhost:5000")    webview.start(gui=None, debug=False)  # 生产环境关闭调试模式打包应用的建议为了将应用分发给最终用户，可以使用PyInstaller或cx_Freeze将Python代码打包成单个可执行文件：12# 使用PyInstaller打包pyinstaller --onefile --windowed desktop.py注意事项：打包时需确保包含所有依赖项（如Flask、waitress、pywebview）。在macOS/Linux上可能需要额外配置以确保WebView组件正确加载。总结通过结合pywebview、Flask和waitress，可以开发出兼具美观界面和强大功能的跨平台桌面应用：开发阶段：使用Flask内置服务器和调试模式，提高开发效率。生产阶段：使用waitress替代Flask服务器，提升性能和稳定性。部署阶段：使用PyInstaller打包为独立可执行文件，方便分发。完整代码以下是在Windows平台上结合PyWebView、Flask和Waitress的完整代码实现。代码分为三个主要文件，分别负责Flask后端、桌面应用包装和生产环境启动。1. Flask后端代码（app.py）123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475# app.py - Flask后端应用from flask import Flask, jsonify, render_template_stringimport osapp = Flask(__name__)# 确保中文正常显示app.config['JSON_AS_ASCII'] = False# 首页路由@app.route('/')def index():    return render_template_string("""    <!DOCTYPE html>    <html lang="zh-CN">    <head>        <meta charset="UTF-8">        <title>PyWebView + Flask应用</title>        <style>            body { font-family: 'Microsoft YaHei', sans-serif; max-width: 800px; margin: 0 auto; padding: 20px; }            .container { background-color: #f9f9f9; border-radius: 8px; padding: 20px; box-shadow: 0 2px 5px rgba(0,0,0,0.1); }            h1 { color: #333; }            button { background-color: #4CAF50; color: white; border: none; padding: 10px 20px;                     text-align: center; text-decoration: none; display: inline-block;                     font-size: 16px; margin: 4px 2px; cursor: pointer; border-radius: 4px; }            #result { margin-top: 20px; padding: 10px; background-color: #e8f5e9; border-radius: 4px; }        </style>    </head>    <body>        <div class="container">            <h1>PyWebView + Flask桌面应用</h1>            <p>这是一个基于Web技术的跨平台桌面应用示例。</p>            <button onclick="fetchData()">获取数据</button>            <button onclick="callPythonFunction()">调用Python函数</button>            <div id="result">点击按钮查看结果...</div>            <script>                // 从Python后端获取数据                async function fetchData() {                    try {                        const response = await fetch('/api/data');                        if (!response.ok) throw new Error('网络响应失败');                        const data = await response.json();                        document.getElementById('result').textContent = data.message;                    } catch (error) {                        document.getElementById('result').textContent = '错误: ' + error.message;                    }                }                // 调用Python函数（通过PyWebView的API）                async function callPythonFunction() {                    try {                        // 确保PyWebView API已加载                        if (window.pywebview && window.pywebview.api) {                            const result = await window.pywebview.api.multiply_numbers(5, 3);                            document.getElementById('result').textContent = `Python计算结果: 5 × 3 = ${result}`;                        } else {                            document.getElementById('result').textContent = 'PyWebView API未加载';                        }                    } catch (error) {                        document.getElementById('result').textContent = '调用Python函数时出错: ' + error.message;                    }                }            </script>        </div>    </body>    </html>    """)# API路由 - 返回JSON数据@app.route('/api/data')def get_data():    return jsonify({        "message": "这是来自Flask后端的数据！",        "timestamp": str(os.times()),        "platform": "Windows桌面应用"    })# 导出Flask应用实例供其他模块使用if __name__ == '__main__':    # 仅在直接运行此文件时使用Flask内置服务器（开发环境）    app.run(debug=True, port=5000)2. 开发环境启动脚本（desktop_dev.py）1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253# desktop_dev.py - 开发环境下的桌面应用启动脚本import webviewimport threadingimport subprocessimport timeimport sysfrom flask import Flaskdef run_flask():    """在子进程中启动Flask开发服务器"""    # 确定Python可执行文件路径    python_exe = sys.executable    # 启动Flask应用    server = subprocess.Popen(        [python_exe, 'app.py'],        stdout=subprocess.PIPE,        stderr=subprocess.STDOUT,        universal_newlines=True    )    # 等待服务器启动    for line in server.stdout:        print(line.strip())        if 'Running on' in line:            break    return serverclass Api:    """供前端JavaScript调用的Python API"""    def multiply_numbers(self, a, b):        """示例函数：计算两个数的乘积"""        return a * bif __name__ == '__main__':    # 启动Flask服务器    flask_server = run_flask()    try:        # 创建API实例        api = Api()        # 创建PyWebView窗口        window = webview.create_window(            title="PyWebView桌面应用",            url="http://localhost:5000",            width=800,            height=600,            resizable=True,            fullscreen=False,            js_api=api  # 暴露Python API给JavaScript        )        # 启动PyWebView主循环        webview.start(debug=True)    finally:        # 关闭Flask服务器        if flask_server:            flask_server.terminate()            flask_server.wait()            print("Flask服务器已关闭")3. 生产环境启动脚本（desktop_prod.py）12345678910111213141516171819202122232425262728293031323334353637383940414243# desktop_prod.py - 生产环境下的桌面应用启动脚本import webviewimport threadingimport subprocessimport timeimport sysfrom waitress import servefrom app import appclass Api:    """供前端JavaScript调用的Python API"""    def multiply_numbers(self, a, b):        """示例函数：计算两个数的乘积"""        return a * bdef run_waitress():    """使用waitress启动Flask应用（生产环境）"""    print("正在启动Waitress服务器...")    serve(app, host='127.0.0.1', port=5000, threads=8)if __name__ == '__main__':    # 在后台线程中启动waitress服务器    server_thread = threading.Thread(target=run_waitress, daemon=True)    server_thread.start()    # 等待服务器启动（给服务器一些时间初始化）    print("等待服务器启动...")    time.sleep(2)    try:        # 创建API实例        api = Api()        # 创建PyWebView窗口        window = webview.create_window(            title="PyWebView桌面应用",            url="http://localhost:5000",            width=800,            height=600,            resizable=True,            fullscreen=False,            js_api=api  # 暴露Python API给JavaScript        )        # 启动PyWebView主循环（关闭调试模式）        webview.start(debug=False)    except Exception as e:        print(f"应用启动失败: {e}")    finally:        print("应用已关闭")4. 运行说明开发环境安装依赖：1pip install flask pywebview waitress运行开发环境脚本：1python desktop_dev.py会自动启动Flask开发服务器和PyWebView窗口。修改代码后刷新页面即可看到变化。生产环境打包应用（可选）：1pyinstaller --onefile --windowed desktop_prod.py直接运行生产环境脚本：1python desktop_prod.py使用waitress服务器替代Flask内置服务器，性能更好。关闭了调试模式，更安全稳定。功能说明前后端交互：前端通过fetch调用Flask API（如/api/data）。前端通过window.pywebview.api调用Python函数（如multiply_numbers）。界面特点：使用Tailwind CSS风格的界面（内联样式确保打包时不丢失）。支持中文显示（使用Microsoft YaHei字体）。响应式设计，适配不同屏幕尺寸。这个实现可以作为Windows桌面应用的基础框架，你可以根据需要扩展Flask API或修改前端界面。另一个应用示例代码123456789101112131415161718192021222324252627282930313233343536373839404142434445464748# create Desktop App by  pywebview"""Replace the flask module with waitress module.To avoid the warning: WARNING: This is a development server. Do not use it in a production deployment. Use a production WSGI server instead."""import webviewfrom flask import Flask, render_template_stringimport threadingfrom waitress import serve# Create a Flask applicationapp = Flask(__name__)# Define the route@app.route('/')def index():    return render_template_string("""    <!DOCTYPE html>    <html lang="zh">    <head>        <meta charset="UTF-8">        <title>Pywebview + Flask 示例</title>    </head>    <body>        <h1>欢迎使用 Pywebview 和 Flask 构建的桌面应用！</h1>        <a href="http://cnliutz.ipyingshe.net" rel="external nofollow" >Blog Site</a>                            </body>    </html>    """)def create_webview():    # Define the URL of the Flask application to load    url = "http://127.0.0.1:5000"    # Create a window and load the specified URL    window = webview.create_window('Pywebview + Flask 应用', url)    # Run the application    webview.start()if __name__ == '__main__':    # Start the Waitress server in a separate thread    def run_waitress():        try:            # Start Waitress to run the Flask application            serve(app, host='127.0.0.1', port=5000)        except Exception as e:            print(f"Error starting Waitress: {e}")    waitress_thread = threading.Thread(target=run_waitress)    waitress_thread.daemon = True    waitress_thread.start()    # Start pywebview    create_webview()

技术方案工具选择PyPDF2：轻量级PDF处理库，适合基本PDF操作os：处理文件和目录路径argparse：解析命令行参数安装依赖1pip install PyPDF2完整代码实现123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123import osimport argparsefrom PyPDF2 import PdfReader, PdfWriter def split_pdf(input_path, output_folder, ranges):    """    将PDF文件按指定页数范围分割         参数:        input_path (str): 输入PDF文件路径        output_folder (str): 输出文件夹路径        ranges (list): 分割范围列表，格式为[(起始页, 结束页), ...]    """    # 确保输出文件夹存在    if not os.path.exists(output_folder):        os.makedirs(output_folder)         # 获取文件名（不含扩展名）    file_name = os.path.splitext(os.path.basename(input_path))[0]         try:        with open(input_path, 'rb') as file:            reader = PdfReader(file)            total_pages = len(reader.pages)                         print(f"开始处理: {input_path}")            print(f"总页数: {total_pages}")                         # 处理每个分割范围            for i, (start, end) in enumerate(ranges):                # 处理结束页为-1（表示到文档末尾）的情况                if end == -1:                    end = total_pages                                 # 验证页码范围                if start < 1 or end > total_pages or start > end:                    print(f"⚠️ 警告: 无效的页码范围 [{start}, {end}]，跳过")                    continue                                 # 创建PDF写入对象                writer = PdfWriter()                                 # 添加指定范围的页面（PyPDF2使用0-based索引）                for page_num in range(start - 1, end):                    writer.add_page(reader.pages[page_num])                                 # 生成输出文件名                part_name = f"{file_name}_part{i+1}_p{start}-{end}.pdf"                output_path = os.path.join(output_folder, part_name)                                 # 写入文件                with open(output_path, 'wb') as output_file:                    writer.write(output_file)                                 print(f"✅ 已创建: {part_name} (页数: {start}-{end})")                             print(f"分割完成! 文件保存在: {output_folder}")         except Exception as e:        print(f"❌ 处理过程中发生错误: {str(e)}") def parse_ranges(range_str):    """    解析页数范围字符串         参数:        range_str (str): 页数范围字符串，格式如 "1,2-9,10-"         返回:        list: 解析后的页数范围列表    """    ranges = []    parts = range_str.split(',')         for part in parts:        if '-' in part:            start_end = part.split('-')            start = int(start_end[0].strip())                         # 处理结束页为空的特殊情况（表示到文档末尾）            if start_end[1].strip() == '':                end = -1  # 使用-1表示到文档末尾            else:                end = int(start_end[1].strip())        else:            # 单个页码            start = int(part.strip())            end = start                 ranges.append((start, end))         return ranges if __name__ == "__main__":    # 设置命令行参数解析    parser = argparse.ArgumentParser(        description="PDF文件分割工具 - 按指定页数范围分割PDF文档",        formatter_class=argparse.RawTextHelpFormatter    )         parser.add_argument('input', help='输入PDF文件路径')    parser.add_argument('ranges',                         help='页数范围，格式如 "1,2-9,10-"\n'                             '示例: "1" - 仅第一页\n'                             '       "1-5" - 1到5页\n'                             '       "10-" - 从第10页到文档末尾\n'                             '       "1,3-5,7-" - 多个范围')    parser.add_argument('-o', '--output', default='pdf_split_output',                        help='输出文件夹 (默认: pdf_split_output)')         args = parser.parse_args()         # 解析页数范围    try:        ranges = parse_ranges(args.ranges)        print(f"解析的分割范围: {ranges}")                 # 执行PDF分割        split_pdf(args.input, args.output, ranges)    except ValueError:        print("❌ 错误: 页数范围格式无效。请使用格式如 '1,2-9,10-'")    except Exception as e:        print(f"❌ 发生错误: {str(e)}")使用说明基本用法1python pdf_splitter.py 输入文件.pdf "页数范围" [-o 输出文件夹]示例命令分割为三个部分（第1页、第2-9页、第10页到最后）：1python pdf_splitter.py "吉林省镇赉县*.pdf" "1,2-9,10-" -o 分割结果仅提取特定页面：12345678# 提取第5页python pdf_splitter.py input.pdf "5" # 提取第10-15页python pdf_splitter.py input.pdf "10-15" # 提取从第20页到文档末尾python pdf_splitter.py input.pdf "20-"提取多个不连续范围：12# 提取封面(1)、目录(3-4)和正文(6-)python pdf_splitter.py book.pdf "1,3-4,6-"输出示例解析的分割范围: [(1, 1), (2, 9), (10, -1)]开始处理: 吉林省镇赉县*.pdf总页数: 75✅ 已创建: 吉林省镇赉县*_part1_p1-1.pdf (页数: 1-1)✅ 已创建: 吉林省镇赉县*_part2_p2-9.pdf (页数: 2-9)✅ 已创建: 吉林省镇赉县*_part3_p10-75.pdf (页数: 10-75)分割完成! 文件保存在: 分割结果技术亮点1.灵活的页数范围解析支持单页、连续页和到文档末尾的表示法支持多个不连续范围的组合智能处理页码边界情况2.健壮的错误处理无效页码检测文件不存在处理格式错误提示3.用户友好的输出清晰的处理进度显示有意义的文件名生成详细的操作结果反馈4.命令行友好详细的帮助信息直观的参数设计默认值简化操作实际应用场景1.文档预处理分离封面、目录和正文提取合同中的关键条款分割大型报告为多个章节2.工作流程优化仅分发相关人员需要的部分创建演示材料的子集提取扫描文档中的特定页面3.自动化处理与OCR工具集成批量处理多个文档定时任务自动分割新文档扩展建议添加GUI界面：使用PyQt或Tkinter创建图形界面集成OCR功能：结合pytesseract提取文本内容添加水印功能：在分割后的文件添加特定水印支持批量处理：处理整个文件夹的PDF文件添加PDF压缩：减小输出文件大小

、pymysql的基本定位pymysql 是一个纯Python实现的 MySQL 数据库连接驱动，用于在 Python 程序中操作 MySQL 数据库。核心功能：提供连接数据库、执行 SQL 语句、处理查询结果等接口，类似于 Python 2 时代的 MySQLdb 模块。版本兼容性Python 3 专属：pymysql 设计用于 Python 3.x 环境，解决了 MySQLdb 在 Python 3 中不兼容的问题（MySQLdb 仅支持 Python 2）。兼容原理：pymysql 在接口层面模拟了 MySQLdb 的 API，便于开发者从 Python 2 迁移到 Python 3 时无缝过渡。pymysql的典型用法import pymysql # 连接数据库 conn = pymysql.connect( host='localhost', user='root', password='123456', database='test_db', port=3306 ) # 执行查询 with conn.cursor() as cursor: sql = "SELECT * FROM users WHERE id = %s" cursor.execute(sql, (1,)) result = cursor.fetchone() print(result) # 关闭连接 conn.close() 总结pymysql 的核心定位是 Python 3 环境下操作 MySQL 数据库的标准模块，其设计目标是兼容 Python 3 并替代 Python 2 中的 MySQLdb。在实际开发中，它是 Python 与 MySQL 交互的首选工具之一。

简介uuid 模块是Python标准库中用于生成通用唯一标识符（Universally Unique Identifier，UUID）的模块。UUID是一个128位的标识符，可以保证在全球范围内的唯一性。这个模块在分布式系统、数据库主键、会话标识、文件命名等场景中非常有用。核心功能1. UUID版本uuid1(): 基于时间戳和MAC地址uuid3(): 基于命名空间和名称的MD5哈希uuid4(): 基于随机数uuid5(): 基于命名空间和名称的SHA-1哈希2. UUID属性hex: 32个字符的十六进制字符串int: 128位整数bytes: 16字节的字节串urn: URN格式的字符串version: UUID版本号variant: UUID变体3. 命名空间NAMESPACE_DNS: DNS命名空间NAMESPACE_URL: URL命名空间NAMESPACE_OID: ISO OID命名空间NAMESPACE_X500: X.500 DN命名空间使用场景1. 生成唯一标识符import uuiddef generate_identifiers():    # 基于时间戳的UUID    time_based_uuid = uuid.uuid1()    print(f"基于时间的UUID: {time_based_uuid}")        # 随机UUID    random_uuid = uuid.uuid4()    print(f"随机UUID: {random_uuid}")        # 基于名称的UUID    name_based_uuid = uuid.uuid5(uuid.NAMESPACE_DNS, 'example.com')    print(f"基于名称的UUID: {name_based_uuid}")2. 数据库主键import uuidfrom datetime import datetimeclass User:    def __init__(self, username, email):        self.id = uuid.uuid4()        self.username = username        self.email = email        self.created_at = datetime.now()        def __str__(self):        return f"User(id={self.id}, username={self.username})"def create_user():    user = User("张三", "zhangsan@example.com")    print(f"创建用户: {user}")    return user3. 文件命名import uuidimport osdef save_file_with_uuid(content, extension='.txt'):    # 生成唯一文件名    filename = f"{uuid.uuid4()}{extension}"        # 保存文件    with open(filename, 'w') as f:        f.write(content)        print(f"文件已保存: {filename}")4. 会话管理import uuidfrom datetime import datetime, timedeltaclass SessionManager:    def __init__(self):        self.sessions = {}        def create_session(self, user_id):        session_id = uuid.uuid4()        self.sessions[session_id] = {            'user_id': user_id,            'created_at': datetime.now(),            'expires_at': datetime.now() + timedelta(hours=24)        }        return session_id        def validate_session(self, session_id):        if session_id in self.sessions:            session = self.sessions[session_id]            if datetime.now() < session['expires_at']:                return True        return False使用注意事项UUID版本选择uuid1: 需要MAC地址，可能泄露隐私uuid3/5: 相同输入产生相同输出uuid4: 完全随机，最常用性能考虑uuid1 性能最好uuid4 随机性最好uuid3/5 适合需要确定性存储优化使用二进制存储考虑使用整数存储注意索引效率安全性避免使用 uuid1 泄露MAC地址注意UUID碰撞概率考虑使用加密UUID性能优化建议生成优化批量生成UUID缓存常用UUID使用合适的版本存储优化使用二进制格式压缩存储空间优化索引结构比较优化使用整数比较避免字符串比较缓存比较结果错误处理最佳实践UUID生成错误处理import uuiddef safe_generate_uuid():    try:        return uuid.uuid4()    except Exception as e:        print(f"UUID生成失败: {e}")        # 使用备用方案        return uuid.uuid1()UUID验证处理import uuiddef validate_uuid(uuid_str):    try:        # 尝试解析UUID        uuid_obj = uuid.UUID(uuid_str)        # 验证版本        if uuid_obj.version != 4:            print(f"警告: 使用了非随机UUID版本 {uuid_obj.version}")        return uuid_obj    except ValueError as e:        print(f"无效的UUID格式: {e}")实际应用案例1. 分布式ID生成器import uuidimport timeclass DistributedIdGenerator:    def __init__(self, node_id):        self.node_id = node_id        self.sequence = 0        def generate_id(self):        # 时间戳（41位）        timestamp = int(time.time() * 1000)        # 节点ID（10位）        node = self.node_id & 0x3FF        # 序列号（12位）        self.sequence = (self.sequence + 1) & 0xFFF                # 组合ID        unique_id = (timestamp << 22) | (node << 12) | self.sequence        return unique_id2. 文件版本控制import uuidimport osfrom datetime import datetimeclass FileVersionControl:    def __init__(self, base_dir):        self.base_dir = base_dir        os.makedirs(base_dir, exist_ok=True)        def save_version(self, file_content):        # 生成版本ID        version_id = uuid.uuid4()        timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')                # 保存文件        filename = f"v{timestamp}_{version_id}.txt"        filepath = os.path.join(self.base_dir, filename)                with open(filepath, 'w') as f:            f.write(file_content)                return version_id总结uuid 模块是Python中生成唯一标识符的强大工具。通过合理使用 uuid 模块，我们可以：生成全局唯一标识符实现分布式ID生成管理会话和缓存处理文件命名优化数据库设计————————————————                            版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。                        原文链接：https://blog.csdn.net/Aerkui/article/details/147775487

概述在 Windows 上使用 Waitress 部署 Flask 应用是一个不错的选择，因为 Waitress 是一个纯 Python 编写的 WSGI 服务器，它兼容 Windows 系统。以下是具体步骤来帮助你在 Windows 上通过 Waitress 部署 Flask 应用。windows 部署准备工作确保你已经安装了 Python：Waitress 需要 Python 环境，请确保你的系统上已经安装了 Python，并且版本适合你的项目需求。创建虚拟环境（可选但推荐）：为了管理项目的依赖关系，建议为你的 Flask 项目创建一个虚拟环境。python -m venv venv激活虚拟环境：   .\venv\Scripts\activate安装 Flask 和 Waitress：如果你还没有安装 Flask 和 Waitress，可以通过 pip 安装它们。pip install flask waitress使用 Waitress 部署 Flask 应用假设你已经有了一个简单的 Flask 应用，文件名为 app.py。接下来，我们将使用 Waitress 来运行这个应用。修改你的 Flask 应用：确保你的 Flask 应用可以被外部访问。通常情况下，你需要指定主机和端口。但是，Waitress 在 Windows 上运行时，你可以直接使用以下命令而不需要特别修改代码。使用 Waitress 运行 Flask 应用：打开命令提示符或 PowerShell，激活你的虚拟环境（如果使用的话），然后运行下面的命令来启动你的 Flask 应用。waitress-serve --port=8080 --call "app:create_app"这里有几个关键点需要注意：--port=8080：指定服务监听的端口号，可以根据需要进行调整。"app:create_app"：这是告诉 Waitress 如何加载你的 Flask 应用。这里的 app 是包含 Flask 应用实例的 Python 文件名（不包括 .py 扩展名），而 create_app 是该文件中返回 Flask 应用实例的函数名称。如果你的 Flask 应用实例是直接定义的（如 app = Flask(__name__)），可以直接引用它，而无需使用 --call 参数。waitress-serve --port=8080 app:app如果你希望使用 --call 参数，可以通过工厂函数来创建 Flask 应用实例。修改 app.py 文件如下：# app.pyfrom flask import Flaskdef create_app():    app = Flask(__name__)    @app.route('/')    def home():        return "Hello, World!"    return app然后运行以下命令：waitress-serve --port=8080 --call "app:create_app"这里 --call "app:create_app" 表示：app 是包含工厂函数的 Python 文件名。create_app 是返回 Flask 应用实例的函数。linux 部署2. 使用 WSGI 服务器Flask 内置的开发服务器不适合生产环境。推荐使用以下 WSGI 服务器：Gunicorn（适合 Linux/Unix 系统）Waitress（适合 Windows 系统）示例：使用 Gunicornpip install gunicorngunicorn -w 4 -b 0.0.0.0:8000 app:app-w 4：启动 4 个工作进程（根据 CPU 核心数调整）。-b 0.0.0.0:8000：绑定到所有网络接口的 8000 端口。app:app：第一个 app 是文件名（如 app.py），第二个 app 是 Flask 实例名称。nginx反向代理ubuntu 配置安装 Nginx：sudo apt updatesudo apt install nginx创建 Nginx 配置文件（例如 /etc/nginx/sites-available/flask_app）：server {    listen 80;    server_name your_domain_or_ip;    location / {        proxy_pass http://127.0.0.1:8000;        proxy_set_header Host $host;        proxy_set_header X-Real-IP $remote_addr;        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;        proxy_set_header X-Forwarded-Proto $scheme;    }}启用配置并重启 Nginx：sudo ln -s /etc/nginx/sites-available/flask_app /etc/nginx/sites-enabled/sudo nginx -t  # 测试配置是否正确sudo systemctl restart nginx5. 使用进程管理工具为了确保 Flask 应用在后台持续运行并能够自动重启，可以使用进程管理工具，如：SupervisorSystemdPM2（适合 Node.js，但也可以用于 Python）示例：使用 Systemd创建服务文件（例如 /etc/systemd/system/flask_app.service）：[Unit]Description=Gunicorn instance to serve Flask appAfter=network.target[Service]User=your_userGroup=www-dataWorkingDirectory=/path/to/your/appExecStart=/path/to/venv/bin/gunicorn -w 4 -b 127.0.0.1:8000 app:app[Install]WantedBy=multi-user.target启动并启用服务：sudo systemctl start flask_appsudo systemctl enable flask_app————————————————                            版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。                        原文链接：https://blog.csdn.net/ling_zhi_xin/article/details/147324917

一、os 库概述​功能：提供与操作系统交互的接口，包括文件管理、环境变量、进程控制、路径操作等。​跨平台特性：大部分函数支持 Windows、Linux、macOS，但部分功能存在系统差异。​核心模块：os：基础操作系统接口os.path：路径处理工具​导入方式：import osAI写代码二、常用功能分类详解​1. 路径操作（os.path 模块）​函数    描述    示例    输出示例os.path.abspath(path)    返回绝对路径    os.path.abspath("test.txt")    /home/user/test.txtos.path.dirname(path)    返回路径的目录部分    os.path.dirname("/a/b/c.txt")    /a/bos.path.basename(path)    返回路径的文件名部分    os.path.basename("/a/b/c.txt")    c.txtos.path.join(path1, path2)    智能拼接路径（跨平台安全）    os.path.join("dir", "file.txt")    dir/file.txtos.path.exists(path)    检查路径是否存在    os.path.exists("test.txt")    True/Falseos.path.isfile(path)    检查是否为文件    os.path.isfile("test.txt")    Trueos.path.isdir(path)    检查是否为目录    os.path.isdir("docs")    Trueos.path.getsize(path)    获取文件大小（字节）    os.path.getsize("test.txt")    1024os.path.splitext(path)    分割文件名和扩展名    os.path.splitext("file.txt")    ('file', '.txt') 2. 目录管理函数    描述    示例    注意os.getcwd()    获取当前工作目录    print(os.getcwd())    返回绝对路径os.chdir(path)    切换当前工作目录    os.chdir("..")    失败抛出 FileNotFoundErroros.listdir(path=".")    列出目录内容    os.listdir("/tmp")    返回文件名列表（不递归）os.mkdir(path)    创建单层目录    os.mkdir("new_dir")    目录已存在时抛出 FileExistsErroros.makedirs(path)    递归创建多层目录    os.makedirs("a/b/c")    类似 mkdir -pos.rmdir(path)    删除空目录    os.rmdir("empty_dir")    目录非空时抛出 OSErroros.removedirs(path)    递归删除空目录    os.removedirs("a/b/c")    从最深层开始向上删除 3. 文件操作函数    描述    示例    注意os.remove(path)    删除文件    os.remove("temp.txt")    文件不存在时抛出 FileNotFoundErroros.rename(src, dst)    重命名/移动文件或目录    os.rename("old.txt", "new.txt")    跨磁盘移动可能失败os.stat(path)    获取文件状态信息    st = os.stat("file.txt")st.st_size    返回包含大小、时间等的对象os.utime(path, times)    修改文件访问/修改时间    os.utime("file.txt", (atime, mtime))    时间戳为秒数 4. 进程管理函数    描述    示例    注意os.system(command)    执行系统命令    os.system("ls -l")    返回命令退出状态码os.popen(command)    执行命令并获取输出（已废弃）    output = os.popen("date").read()    建议改用 subprocessos.exec*() 系列    替换当前进程（如 os.execlp）    os.execlp("python", "python", "script.py")    执行后原进程终止os.kill(pid, signal)    向进程发送信号    os.kill(pid, signal.SIGTERM)    Windows 支持有限 5. 环境变量管理函数/属性    描述    示例    注意os.environ    字典形式的环境变量    os.environ["PATH"]    修改后仅影响当前进程os.getenv(key, default)    获取环境变量值    os.getenv("HOME", "/tmp")    键不存在时返回 defaultos.putenv(key, value)    设置环境变量（不推荐）    os.putenv("DEBUG", "1")    建议直接操作 os.environ 6. 系统信息函数/属性    描述    示例    输出示例os.name    操作系统名称    print(os.name)    posix（Linux/macOS）nt（Windows）os.sep    路径分隔符    os.sep    / 或 \os.linesep    行终止符    os.linesep    \n（Linux）\r\n（Windows）os.cpu_count()    返回 CPU 核心数    os.cpu_count()    8三、实际应用示例​1. 递归遍历目录def list_files(startpath):    for root, dirs, files in os.walk(startpath):        level = root.replace(startpath, '').count(os.sep)        indent = ' ' * 4 * level        print(f"{indent}{os.path.basename(root)}/")        subindent = ' ' * 4 * (level + 1)        for f in files:            print(f"{subindent}{f}") list_files("/tmp")AI写代码 os.walk(path) 是 Python 中用于递归遍历目录结构的核心函数，可深度遍历指定路径下的所有子目录和文件。以下从功能、参数、返回值到应用场景的全面解析：①、核心功能与运行机制​目录树遍历以 path 为根目录，逐层遍历所有子目录，生成目录树的完整结构。支持 ​深度优先 或 ​广度优先 遍历（通过 topdown 参数控制）。​数据生成方式返回一个 ​生成器​（generator），按需生成三元组 (root, dirs, files)，避免一次性加载所有路径导致内存压力。②、参数详解pythonos.walk(top, topdown=True, onerror=None, followlinks=False)AI写代码参数    类型    作用top    str    要遍历的根目录路径topdown    bool    默认 True：先返回父目录再子目录；False 则逆序（从叶子节点向上）onerror    callable    遍历出错时的回调函数（接收异常对象作为参数）followlinks    bool    是否追踪符号链接指向的真实目录（默认 False）③、返回值解析每次迭代返回的三元组包含：​**root**当前遍历的目录绝对路径（如 "/data/images/cats"）。​**dirs**当前目录下的 ​直接子目录名列表​（不含路径，如 ["2023", "2024"]）。​**files**当前目录下的 ​文件名列表​（不含路径，如 ["cat1.jpg", "cat2.png"]）。示例目录结构：markdownroot/├── dir1/│   ├── file1.txt│   └── subdir/└── dir2/    └── file2.logAI写代码对应的遍历输出：python# 第一次迭代('root', ['dir1', 'dir2'], [])  # 第二次迭代（进入 dir1）('root/dir1', ['subdir'], ['file1.txt'])  # 第三次迭代（进入 subdir）('root/dir1/subdir', [], [])  # 第四次迭代（进入 dir2）('root/dir2', [], ['file2.log'])AI写代码 遍历所有文件并获取完整路径pythonimport os for root, dirs, files in os.walk("data"):    for file in files:        full_path = os.path.join(root, file)  # 拼接完整路径[3,6](@ref)        print(f"发现文件：{full_path}")AI写代码 2. 批量重命名文件folder = "docs"for filename in os.listdir(folder):    if filename.endswith(".txt"):        new_name = filename.replace(" ", "_")        os.rename(            os.path.join(folder, filename),            os.path.join(folder, new_name)        )AI写代码 3. 安全创建临时目录import tempfile def create_temp_dir():    temp_dir = tempfile.mkdtemp()    try:        print(f"临时目录已创建: {temp_dir}")        # 执行操作...    finally:        os.rmdir(temp_dir)        print("临时目录已删除")AI写代码四、注意事项​1. 路径处理​不要手动拼接路径：使用 os.path.join() 替代字符串拼接，避免跨平台问题：# 错误方式path = "dir" + "/" + "file.txt"  # 在Windows下失效 # 正确方式path = os.path.join("dir", "file.txt")AI写代码2. 异常处理文件操作需捕获常见异常：try:    os.remove("non_exist.txt")except FileNotFoundError as e:    print(f"文件不存在: {e}")AI写代码3. 安全性​谨慎使用 os.system()：可能引发命令注入漏洞，优先使用 subprocess.run()：# 危险！用户输入可能包含恶意命令user_input = "some; rm -rf /"os.system(f"echo {user_input}") # 安全方式import subprocesssubprocess.run(["echo", user_input], check=True)AI写代码4. 跨平台兼容性检查系统差异：if os.name == 'posix':    print("运行在 Unix-like 系统")elif os.name == 'nt':    print("运行在 Windows 系统")AI写代码五、与 shutil 库的对比功能    os 库    shutil 库​文件复制​    ❌ 无直接支持    ✅ shutil.copy()​目录删除​    ✅ 只能删空目录    ✅ shutil.rmtree() 递归删除​压缩文件​    ❌ 无支持    ✅ shutil.make_archive()​元数据保留​    ❌ 需手动处理    ✅ copystat() 保留权限和时间————————————————                            版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。                        原文链接：https://blog.csdn.net/wanglaqqqq/article/details/146064391

一、PyAutoGUI 库简介PyAutoGUI 是 Python 生态中 ​功能最全面的图形界面自动化库，通过模拟鼠标、键盘操作和屏幕图像识别，实现跨平台（Windows/macOS/Linux）的自动化任务。其核心价值在于：​解放重复劳动：自动化数据录入、文件整理、报表生成等耗时操作，效率提升 90% 以上。​精准控制能力：支持像素级坐标定位（误差 ±0.5 像素）、亚秒级操作延迟（可配置 0.01 秒精度）。​智能图像识别：通过截图匹配动态定位界面元素（如按钮、文本框），解决因窗口位置变化导致的脚本失效问题。​安全可靠：内置防误触机制（如失效安全点）、操作日志记录和多线程安全锁，保障自动化流程的稳定性。二、安装与环境配置​安装命令：bashpip install pyautogui  # 核心库pip install Pillow     # 图像处理依赖（截图功能必需）AI写代码​验证安装：pythonimport pyautoguiprint(pyautogui.__version__)  # 输出版本号（如 0.9.54）AI写代码​全局安全配置​（防止脚本失控）：pythonpyautogui.FAILSAFE = True     # 鼠标移至左上角时自动终止程序[2](@ref)pyautogui.PAUSE = 0.5         # 每次操作后暂停 0.5 秒[2](@ref)AI写代码 三、核心功能与常用函数​1. 鼠标控制函数/方法    功能描述    参数说明​**pyautogui.moveTo(x, y, duration)**    移动鼠标到绝对坐标 (x, y)    duration：移动耗时（秒），设为 0 则瞬间完成​**pyautogui.moveRel(dx, dy, duration)**    相对当前位置偏移 (dx, dy)    常用于动态调整位置（如拖拽文件）​**pyautogui.click(x, y, clicks, button)**    点击操作（左键/右键/双击）    button：'left'（默认）、'right'、'middle'​**pyautogui.dragTo(x, y, duration)**    拖拽到目标位置    需先按住鼠标，常用于文件拖放或绘图软件操作​**pyautogui.scroll(units)**    滚动鼠标滚轮    units：正数向上滚动，负数向下 示例：自动化文件拖放pythonpyautogui.moveTo(100, 200, duration=1)  # 移动至文件位置pyautogui.dragTo(500, 300, duration=2)  # 拖拽至目标文件夹AI写代码 2. 键盘操作函数/方法    功能描述    参数说明​**pyautogui.write(text, interval)**    模拟键盘输入文本    interval：字符输入间隔时间（秒）​**pyautogui.press(keys)**    按下并释放单个按键（如 'enter'、'tab'）    支持 200+ 种按键（见 pyautogui.KEYBOARD_KEYS）​**pyautogui.hotkey(*keys)**    组合快捷键（如 'ctrl+c'）    自动处理按键顺序（按下 ctrl → 按下 c → 释放 c → 释放 ctrl）​**pyautogui.keyDown(key) / keyUp(key)**    分别处理按键的按下和释放    用于长按操作（如游戏角色移动） 示例：自动填写表单pythonpyautogui.write('张三', interval=0.1)  # 输入姓名pyautogui.press('tab')                # 切换至下一输入框pyautogui.write('13812345678')        # 输入手机号pyautogui.hotkey('ctrl', 's')         # 保存表单[6](@ref)AI写代码 3. 屏幕与图像识别函数/方法    功能描述    参数说明​**pyautogui.screenshot(region)**    截取屏幕区域图像    region：指定区域 (x, y, width, height)，默认全屏​**pyautogui.locateOnScreen(image, confidence)**    在屏幕上匹配目标图像位置    confidence：匹配精度（0~1），推荐 0.8 以上​**pyautogui.locateCenterOnScreen(image)**    返回匹配图像的中心坐标    常用于点击动态位置的按钮​**pyautogui.pixel(x, y)**    获取指定像素点的 RGB 颜色值    用于判断界面状态（如按钮是否高亮）示例：智能点击动态按钮pythonbutton_pos = pyautogui.locateOnScreen('submit_button.png', confidence=0.9)if button_pos:    x, y = pyautogui.center(button_pos)    pyautogui.click(x, y)  # 点击按钮中心[1,4](@ref)else:    raise Exception("未找到提交按钮") 4. 窗口管理与高级控制函数/方法    功能描述    参数说明​**pyautogui.getWindowsWithTitle(title)**    获取指定标题的窗口对象    需配合 pygetwindow 库使用​**pyautogui.alert(text)**    弹出确认对话框    阻塞程序直至用户点击​**pyautogui.size()**    返回屏幕分辨率 (width, height)    用于计算相对坐标示例：多窗口协同操作python# 激活记事本窗口并输入内容notepad = pyautogui.getWindowsWithTitle('记事本')[0]notepad.activate()pyautogui.write('自动录入文本', interval=0.1)四、应用场景与实战案例​自动化办公​Excel 报表生成：通过热键操作自动打开文件、填充数据并保存，耗时从 2 小时缩短至 2 分钟。​邮件批量发送：自动填写收件人、主题和附件，结合 Outlook 实现无人值守处理。​数据采集与测试​动态网页截图：定时截取网页数据并保存，用于监控价格或库存变化。​UI 自动化测试：验证软件界面响应，替代人工点击操作。​游戏与娱乐​自动连招脚本：在游戏中模拟按键序列实现连续技能释放。​视频自动播放：通过图像识别跳过广告，自动切换剧集。​五、注意事项与优化技巧​兼容性优化不同操作系统的坐标系统可能差异（如 macOS 的菜单栏高度），需动态计算偏移量。高分辨率屏幕需调整图像识别参数（如 confidence=0.95）。​性能提升使用 pyautogui.locateAllOnScreen() 批量识别多个目标，减少截图次数。启用多线程时需加锁（with pyautogui.lock()），防止操作冲突。​异常处理捕获 pyautogui.ImageNotFoundException 以处理图像识别失败。记录操作日志（pyautogui.logScreenshot()）便于调试。————————————————                            版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。                        原文链接：https://blog.csdn.net/wanglaqqqq/article/details/146359956

本文基于当前仓颉通用版 (0.53.18) 进行开发演示，该版本对Linux平台提供Python FFI支持。文中的所有操作均在华为开发者空间的云主机中进行。关于Python FFI的更多说明，请参阅仓颉官方文档使用指南 -  ://docs.cangjie-lang.cn/docs/0.53.18/user_manual/source_zh_cn/FFI/cangjie-python.htmlAPI 文档  -  ://docs.cangjie-lang.cn/docs/0.53.18/libs/std/ffi_python/ffi_python_package_overview.html需要额外说明的是，截至2025年6月，仓颉尚未发布LTS版本，鸿蒙仓颉尚未提供Python FFI支持。。因此，Python FFI功能在0.53.x后续Canary版本中的变动不在本文讨论范围内。补充更新：2025年7月1日发布的仓颉（通用版）LTS1.0移除了Python FFI。0. 初始化开发者空间的云主机进入开发者空间 (cid:link_0)，系统选择Ubuntu，这样可以确保预装了Python、Cangjie、以及对应的CodeArts IDE。华为云空间提供完整的开发环境。打开云主机，初始化后进入云桌面，可以通过终端命令行指令检查系统当前的 Cangjie 和 Python 版本。问🤔：什么是云主机？答🤓：云主机是开发者空间众多功能的其中之一，而云桌面是云主机默认的登录方式，此外还有无图形界面的云终端形式。1. 创建项目虽然云主机自身提供一定程度的数据持久化功能，但为了更好管理自己的代码，我们需要一个完善的流程来管理项目代码，避免意外情况造成的数据丢失。云桌面上的CodeArts IDE非常贴心地提供了完整的GitCode服务支持，此处我们可以通过点击云桌面上的GitCode快捷方式，在云主机的Firefox浏览器中访问GitCode网站。💡tips: 考虑到云主机的计算资源有限，不建议在复杂场景下高强度使用云主机内置的浏览器，同样的需求可以在外部浏览器中得到满足。GitCode的注册非常简单，此处过程省略。打开CodeArts IDE for Cangjie，其界面应如下所示：1.3 导入工程为方便后续操作，这里选择 导入GitCode 工程 。接下来我们需要获取登录令牌，在登录GitCode之后，可以根据链接指示进行操作。下图是手动创建登录令牌的界面参考，主要需要关注的是有效期和权限设置。注：熟悉云主机功能的开发者可能有注意到一个叫作同步项目的功能。这个功能的原理和上面的手动配置操作类似，区别在于同步项目需要自己手动关联华为云账号和GitCode账号；其次它目前有一个缺点，每次同步都会创建新的项目目录，此处手动配置token正是为了避免这个问题。1.1 创建代码仓库因为本文的预设前提是从0开始，故此处选择创建一个空项目，勾选初始化README。创建时需要留意 .gitignore文件 和 LICENSE 这两个文件的选择。初始化创建新项目后，切回 CodeArts IDE，选择刚创建的项目。1.1 初始化仓颉项目新建空仓库后可通过终端命令 cjpm new 完成项目初始化，通过 cjpm run 测试初始化项目运行结果。看到hello world即说明仓颉开发环境能够正常工作，在开始Python FFI代码编写前，我们通过Git提交代码完成项目的初始化。2. 在仓颉中调用Python函数2.2 指定libpython动态库位置仓颉的Python FFI默认只提供动态链接库的位置检测支持，动态库加载策略如下图所示：由于云主机预装的Python 3.12是不满足第二种查找策略的两点要求，若直接执行以下示例代码，则会报错退出：import std.ffi.python.*import std.log.*main(): Int64 { Python.load(loglevel: LogLevel.INFO) print("${Python.getVersion()}\n") Python.unload() return 0}为了解决这个问题，我们可以通过指定 PYTHON_DYNLIB 环境变量，强制仓颉运行时使用策略1进行动态库加载：export PYTHON_DYNLIB=$(find $(python -c 'import sysconfig as s;print(s.get_path("stdlib"))') -name 'libpython*' 2>/dev/null)也可以通过仓颉的子进程调用完成这个操作：// import std.io.* // import std.os.* // import std.os.process.* func pylib_path(config!: Bool = false): String { const PYEK = "PYTHON_DYNLIB" if (let Some(ev) <- getEnv(PYEK)) { return ev } let p = Process.start("bash", "-c", ##"find $( python3 -c \ 'import sysconfig as s; print(s.get_path("stdlib"))' ) \ -name 'libpython*' 2> /dev/null "##, stdOut: ProcessRedirect.Pipe) let rv = StringReader(p.stdOut).readToEnd() if (config) { setEnv(PYEK, rv) } return rv }2.1 调整仓颉运行时栈大小由于 Python 互操作使用到大量 Python 库的 native 代码，这部分代码在仓颉侧无法对其进行相应的栈保护。仓颉栈保护默认大小为 64KB，在对 Python C API 进行调用过程中，容易造成 native 代码超出默认栈大小，发生溢出，会触发不可预期的结果。建议用户在执行 Python 互操作相关代码前，配置仓颉默认栈大小至少为 1MB：export cjStackSize=1MB 3. 在仓颉中调用Python3.1 导入模块3.2 执行代码

基于仓颉 + DeepSeek + MCP 的智能膳食分析助手一、项目背景及需求分析1.1 项目背景随着人们对健康饮食关注度的提升，越来越多用户希望借助 AI 助手实现个性化的饮食分析与管理。然而，目前市面上的饮食类应用普遍存在如下痛点：1️⃣. 缺乏智能分析：大多数仅记录卡路里，无法提供专业点评与优化建议；2️⃣. 知识更新不及时：难以结合最新营养研究进行推荐与判断；3️⃣. 缺乏可扩展性：难以适配特定人群（如高血压、糖尿病、健身人群等）的差异需求。本项目旨在构建一个基于大语言模型（DeepSeek）和结构化协议（MCP）的智能饮食健康助手，通过自然语言交互，帮助用户实现饮食数据结构化、健康风险识别与个性化建议生成。1.2 核心需求1️⃣ 食品实体识别能力系统应具备从自然语言中准确识别出饮食相关的食品实体与关键词的能力。例如，用户输入：“我今天吃了煎饼果子、卤牛肉、奶茶”，系统应能够自动识别出食品列表：["煎饼果子", "卤牛肉", "奶茶"] 该能力是后续营养分析与健康判断的基础，应支持中文短语、多食物组合、模糊词等多种表达方式。2️⃣ 饮食结构化分析能力（MCP 中间层处理）系统应通过 MCP 模块完成对用户输入饮食内容的语义解析与结构化处理，主要包括：应提取食品清单；应识别饮食行为时段（如早餐、晚餐、夜宵等）；应对识别出的食物打上健康标签（如高糖、高油脂、高钠、低纤维等）；应构造符合 LLM 输入规范的 Prompt，以支撑后续模型理解与生成。结构化分析结果应作为 DeepSeek 模型调用的核心中间数据。3️⃣ 合理饮食建议生成能力（基于 DeepSeek）系统应利用 DeepSeek 大模型，根据 MCP 构造的 Prompt 生成具有科学性、专业性与可操作性的饮食建议，具体应包括：对当前饮食存在的健康风险进行分析（如糖分摄入超标、缺乏蔬菜等）；提供可替代食材建议（如炸鸡 → 烤鸡，奶茶 → 绿茶）；给出搭配优化方案（如餐前饮水、搭配高纤维蔬菜等）；提出个性化提示（如适合三高人群、健身人群、孕期饮食等）。模型输出应以自然语言形式呈现，逻辑清晰、结构明确，风格应贴近真实营养师表达方式。二、项目概述本项目旨在构建一套基于多模块协作的 智能膳食分析助手，通过自然语言输入，实现对个人饮食行为的结构化分析与健康建议生成，服务于日常健康管理、饮食优化与疾病预防等场景。系统采用模块化设计，主要由三大核心组成部分构成：2.1 技术选型模块技术栈职责🧑‍💻 客户交互仓颉语言提供终端或界面交互入口，接收用户自然语言输入并展示模型生成结果🔧 中间处理MCP 模块（Python + FastAPI）实现饮食语义解析、食物识别、健康标签打标、Prompt 生成与请求路由🤖 模型服务DeepSeek 大语言模型接收经过标准化构造的请求 Prompt，返回结构化营养建议、健康分析与饮食优化建议2.2 适用对象企业健康服务平台产品经理个人开发者高校学生营养师/健身爱好者病患群体2.3 案例时间环境配置：约30 分钟构建项目：约25分钟测试阶段：约10 分钟本案例总时长预计 65 分钟。2.4 项目流程整个系统的数据处理与响应流程如下所示：2.5 资源总览本案例预计花费 0 元。资源名称规格免费额度/价格开发者空间–云主机2CPU4g X86 Ubuntu 22.04 Server定制版0元CodeArts IDE for CangjieIDE 仓颉 云主机预装0元CodeArts IDE for PythonIDE Python 云主机预装0元Maas DeepSeek 模型服务试用200W Token0元2.6 架构亮点✅ 使用 MCP 解耦输入处理与模型调用，提升可扩展性与灵活性；✅ DeepSeek 专注自然语言生成任务，MCP 预处理提高 prompt 精度；✅ 仓颉终端可后续拓展为图形界面、Web 应用或移动端入口；✅ 本地配置与食物字典可按需调整，适配多种用户类型（如减脂、控糖、健身等）；✅ 架构支持多模型/多来源接入，可拓展至 Claude、SparkDesk、ChatGPT 等后端模型。三、环境准备3.1 申请云主机面向广大开发者群体，华为开发者空间提供一个随时访问的“开发桌面云主机”、丰富的“预配置工具集合”和灵活使用的“场景化资源池”，开发者开箱即用，快速体验华为根技术和资源。如果还没有领取开发者空间云主机，可以参考免费领取云主机文档领取。领取云主机后可以直接进入华为开发者空间工作台界面，点击进入桌面连接云主机。3.2 免费领取DeepSeek R1满血版华为云提供了单模型200万免费Tokens，包含DeepSeek-R1&V3满血版，我们可以登录华为云ModelArts Studio(MaaS)控制台领取免费额度，这里我们选择DeepSeek-R1满血版来搭建我们的专属AI聊天机器人。在云主机桌面底部菜单栏，点击打开火狐浏览器。用火狐浏览器访问ModelArts Studio首页：cid:link_2，点击ModelArts Studio控制台跳转到登录界面，按照登录界面提示登录，即可进入ModelArts Studio控制台。签署免责声明。进入ModelArts Studio控制台首页，区域选择西南-贵阳一，在左侧菜单栏，选择在线推理 > 预置服务 > 免费服务，选择DeepSeek-R1-32K模型，点击领取额度，领取200万免费token。领取后点击调用说明，可以获取到对应的API地址、模型名称。红色方块为API 地址 红色箭头为 API KEY四、项目构建本案例采用 本地MCP 服务+ 本地仓颉AI机器人对话 (参考来自于 (https://gitcode.com/CaseDeveloper/Cangjie-Examples)4.1 MCP 服务构建4.1.1 打开 CodeArts IDE for Python4.1.2 新建项目项目名称 构建为： food_mcp 然后点击 创建4.1.3 代码实施mcp_server.py# -*- coding: utf-8 -*- # mcp_server.py """基于仓颉 + DeepSeek + MCP 的智能膳食分析助手""" from fastapi import FastAPI from pydantic import BaseModel import json from typing import List, Dict, Any from fastapi.responses import StreamingResponse from deepseek_client import stream_deepseek, async_stream_deepseek from text_match import extract_food_simple, extract_food from loguru import logger from cachetools import TTLCache import sys # 加载食物字典 with open("food_tags.json", encoding="utf-8") as f: FOOD_TAGS: Dict[str, Dict[str, Any]] = json.load(f) app = FastAPI() class Input(BaseModel): input: str # 全局缓存：prompt -> full_reply_text CACHE = TTLCache(maxsize=500, ttl=1800) # 30分钟自动过期 logger.remove() logger.add(sys.stdout, level="INFO", enqueue=False, backtrace=False) # ---- 基础角色词 ---- BASE_SYSTEM_PROMPT = ( "你是一名资深注册营养师，擅长以简洁的 Markdown 格式给出科学、可执行的饮食建议。" "所有回答需包含以下小节：\n" "1. 总体评价\n" "2. 推荐摄入（量/食材）\n" "3. 过量风险\n" "4. 不宜同食\n" "5. 行动建议\n" "回答请使用中文，并尽量在 300 字以内。" ) # ---- 场景化模板 ---- PROMPTS = { "nutrition_review": ( BASE_SYSTEM_PROMPT + "\n\n【场景】饮食点评。请先总体评价，再按上表 1~5 小节输出：{input}" ), "diet_plan": ( BASE_SYSTEM_PROMPT + "\n\n【场景】减脂期餐单。请输出 3 日食谱 (表格形式)，并在每餐注明热量估计：{input}" ), "effects_inquiry": ( BASE_SYSTEM_PROMPT + "\n\n【场景】过量影响。请列出已知副作用及参考文献：{input}" ), "avoid_inquiry": ( BASE_SYSTEM_PROMPT + "\n\n【场景】相克查询。请说明不宜同食原因及替代方案：{input}" ), } def _process_input(user_input: str): """内部共用逻辑，返回分析结果和 prompt""" # === 1. 食物关键字匹配 === food_list, exact_hits, fuzzy_hits = extract_food(user_input, FOOD_TAGS.keys()) # 记录匹配详情到日志 logger.info(f"[匹配] 精确={exact_hits} 模糊={fuzzy_hits}") # === 2. 聚合静态信息 === tags = list({tag for food in food_list for tag in FOOD_TAGS[food].get("tags", [])}) effects = {food: FOOD_TAGS[food].get("effects") for food in food_list if FOOD_TAGS[food].get("effects")} avoid_with = list({aw for food in food_list for aw in FOOD_TAGS[food].get("avoid_with", [])}) diet_type = list({dt for food in food_list for dt in FOOD_TAGS[food].get("diet_type", [])}) # === 3. 根据关键词判定 mode === mode = "nutrition_review" if any(k in user_input for k in ["减肥", "减脂", "低卡", "少油", "瘦身", "个人食谱"]): mode = "diet_plan" elif any(k in user_input for k in ["吃多了", "过量", "上火", "副作用", "影响"]): mode = "effects_inquiry" elif any(k in user_input for k in ["不能一起", "相克", "不宜同食", "一起吃"]): mode = "avoid_inquiry" new_prompt_base = PROMPTS[mode].format(input=user_input) # === 4. 拼接静态附加信息 === notes = [] if effects: notes.append("【过量风险】" + "；".join(f"{food}:{desc}" for food, desc in effects.items())) if avoid_with: notes.append("【不宜同食】" + "、".join(avoid_with)) extra_info = "；".join(notes) if extra_info: new_prompt_base += ( "\n\n以下为已知静态信息（请务必先列出【过量风险】与【不宜同食】两个小节，并完整引用下列内容，否则视为回答不完整）：" f"{extra_info}" ) # 要求模型以 JSON 输出，配合 response_format new_prompt = new_prompt_base + "\n请使用 markdown bullet list 输出建议。" return { "food_list": food_list, "tags": tags, "effects": effects, "avoid_with": avoid_with, "diet_type": diet_type, "mode": mode, "routed_input": new_prompt, } @app.post("/mcp") async def route_prompt(data: Input): user_input = data.input result = _process_input(user_input) return result @app.post("/mcp_stream") async def route_prompt_stream(data: Input): user_input = data.input result = _process_input(user_input) async def event_generator(): meta_json = json.dumps({k: v for k, v in result.items() if k != 'routed_input'}, ensure_ascii=False) # 发送 Meta 事件 yield f"event: meta\ndata: {meta_json}\n\n" prompt_key = result["routed_input"] # 若缓存命中，直接按20字符切片发送缓存内容 if prompt_key in CACHE: logger.info("[缓存] 命中") cached_text = CACHE[prompt_key] # SSE data 行不能包含裸换行，需逐行加前缀 payload_lines = [f"data: {line}" for line in cached_text.split("\n")] payload_block = "\n".join(payload_lines) yield f"event: token\n{payload_block}\n\n" yield "event: done\ndata: [DONE]\n\n" return logger.info("[缓存] 未命中") collected_chunks = [] async for chunk in async_stream_deepseek(prompt_key): collected_chunks.append(chunk) yield f"event: token\ndata: {chunk}\n\n" # 缓存完整内容 full_text = "".join(collected_chunks) CACHE[prompt_key] = full_text # 结束事件 yield "event: done\ndata: [DONE]\n\n" return StreamingResponse(event_generator(), media_type="text/event-stream") deepseek_client.py# deepseek_client.py # 基于 DeepSeek API 的客户端 import requests import json from typing import Generator import httpx # 从 config.json 读取配置 with open("config.json", encoding="utf-8") as f: config = json.load(f) DEEPSEEK_API_KEY = config["DEEPSEEK_API_KEY"] DEEPSEEK_URL = config["DEEPSEEK_URL"] MODEL_NAME = config["MODEL_NAME"] def call_deepseek(prompt): headers = { "Content-Type": "application/json", "Authorization": f"Bearer {DEEPSEEK_API_KEY}" } payload = { "model": MODEL_NAME, "messages": [ {"role": "user", "content": prompt} ] } response = requests.post(DEEPSEEK_URL, headers=headers, json=payload) result = response.json() return result["choices"][0]["message"]["content"] # 新增：流式推理生成器 def stream_deepseek(prompt) -> Generator[str, None, None]: """Yield generated content chunks from DeepSeek API using SSE.""" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {DEEPSEEK_API_KEY}" } payload = { "model": MODEL_NAME, "messages": [ {"role": "user", "content": prompt} ], "stream": True, "max_tokens": 2048, "stream_options": {"include_usage": True}, } # 使用 stream=True 触发增量输出 response = requests.post(DEEPSEEK_URL, headers=headers, json=payload, stream=True) # iter_lines 将保持连接并逐行读取 for line in response.iter_lines(decode_unicode=True): if not line: continue # DeepSeek/SSE 行以 "data: " 开头 if line.startswith("data: "): data_str = line[6:] # 结束标志 if data_str.strip() == "[DONE]": break # 解析 JSON，获取增量内容 try: data_json = json.loads(data_str) # usage 块 choices 为空，跳过 choices = data_json.get("choices", []) if not choices: continue delta = choices[0]["delta"].get("content", "") if delta: yield delta except json.JSONDecodeError: # 忽略无法解析的片段 continue # 异步流式生成器 async def async_stream_deepseek(prompt): """异步生成器，从 DeepSeek API 生成内容（SSE）。""" headers = { "Content-Type": "application/json", "Authorization": f"Bearer {DEEPSEEK_API_KEY}" } payload = { "model": MODEL_NAME, "messages": [ {"role": "user", "content": prompt} ], "stream": True, "max_tokens": 2048, "stream_options": {"include_usage": True}, } async with httpx.AsyncClient(timeout=None) as client: async with client.stream("POST", DEEPSEEK_URL, headers=headers, json=payload) as resp: async for line in resp.aiter_lines(): if not line: continue if line.startswith("data: "): data_str = line[6:] if data_str.strip() == "[DONE]": break try: data_json = json.loads(data_str) choices = data_json.get("choices", []) if not choices: continue delta = choices[0]["delta"].get("content", "") if delta: yield delta except json.JSONDecodeError: continue test_pipeline.py# -*- coding: utf-8 -*- # test_pipeline.py import requests from deepseek_client import call_deepseek import json from requests.exceptions import ChunkedEncodingError def test_diet_analysis(): user_input = "中午吃了炸鸡和奶茶，晚上吃了牛肉火锅" # 调用 MCP mcp_resp = requests.post("http://localhost:8001/mcp", json={"input": user_input}) mcp_data = mcp_resp.json() print("MCP 返回：", mcp_data) # 调用 DeepSeek reply = call_deepseek(mcp_data["routed_input"]) print("DeepSeek 回复：", reply) def test_diet_plan(): user_input = "我想要一个一周减肥计划，主要食物包括香菇、鸡胸肉和燕麦" mcp_resp = requests.post("http://localhost:8001/mcp", json={"input": user_input}) mcp_data = mcp_resp.json() print("MCP 返回：", mcp_data) reply = call_deepseek(mcp_data["routed_input"]) print("DeepSeek 回复：", reply) def test_effects_inquiry(): user_input = "香菇吃多了会怎样？" mcp_resp = requests.post("http://localhost:8001/mcp", json={"input": user_input}) mcp_data = mcp_resp.json() print("MCP 返回：", mcp_data) reply = call_deepseek(mcp_data["routed_input"]) print("DeepSeek 回复：", reply) def test_avoid_inquiry(): user_input = "牛奶和虾能一起吃吗？" mcp_resp = requests.post("http://localhost:8001/mcp", json={"input": user_input}) mcp_data = mcp_resp.json() print("MCP 返回：", mcp_data) reply = call_deepseek(mcp_data["routed_input"]) print("DeepSeek 回复：", reply) def test_stream(): user_input = "我晚上吃什么比较好呢？ 中午吃了炸鸡和奶茶" try: resp = requests.post( "http://localhost:8001/mcp_stream", json={"input": user_input}, stream=True, timeout=None, ) print("Stream start →", resp.status_code) meta_printed = False for line in resp.iter_lines(decode_unicode=True): if not line: continue # SSE: 可能含 event: xxx if line.startswith("event: "): current_event = line[7:] # 下一行应该是 data: continue if not line.startswith("data: "): continue data = line[6:] # current_event 默认为 token current_event = locals().get("current_event", "token") if current_event == "meta": try: meta_obj = json.loads(data) pretty_meta = json.dumps(meta_obj, ensure_ascii=False) print("Meta:", pretty_meta) except json.JSONDecodeError: print("Meta:", data) meta_printed = True print("\n--- AI 回复 ---\n", end="") continue if current_event == "done": print("\n✅ Stream done") break # 其他 token 输出 print(data, end="", flush=True) except ChunkedEncodingError: # 服务器已发送完毕但提前关闭连接，可忽略 print("\n⚠️ 连接提前关闭，已接收全部内容。") finally: try: resp.close() except Exception: pass if __name__ == "__main__": # test_diet_analysis() # test_diet_plan() # test_effects_inquiry() # test_avoid_inquiry() test_stream() text_match.py# -*- coding: utf-8 -*- # text_match.py import jieba import difflib from typing import Iterable, List, Set, Tuple def extract_food( user_text: str, food_vocab: Iterable[str], fuzzy_threshold: float = 0.8, ) -> Tuple[List[str], List[str], List[str]]: """根据用户输入提取食物关键词。 1. 先使用 jieba 分词命中精确词。 2. 对分词结果做模糊匹配，解决同义词/花式写法。 Args: user_text: 用户原始输入字符串。 food_vocab: 食物静态数据库的键集合。 fuzzy_threshold: SequenceMatcher 相似度阈值 (0-1)。 Returns: 去重后的食物列表，按出现顺序返回。 """ vocab_set: Set[str] = set(food_vocab) tokens = jieba.lcut(user_text, cut_all=False) hits: List[str] = [] exact_hits: List[str] = [] fuzzy_hits: List[str] = [] added: Set[str] = set() for token in tokens: # 精确匹配 if token in vocab_set and token not in added: hits.append(token) exact_hits.append(token) added.add(token) continue # 模糊匹配 for food in vocab_set: if food in added: continue if difflib.SequenceMatcher(None, token, food).ratio() >= fuzzy_threshold: hits.append(food) fuzzy_hits.append(food) added.add(food) break return hits, exact_hits, fuzzy_hits # 向后兼容的简化接口 def extract_food_simple(user_text: str, food_vocab: Iterable[str], fuzzy_threshold: float = 0.8) -> List[str]: """返回仅 food list 的旧版接口，供其他模块调用。""" return extract_food(user_text, food_vocab, fuzzy_threshold)[0] food_tags.json{ "炸鸡": { "tags": ["高油脂", "高热量"], "effects": "经常食用可能导致能量过剩、血脂升高", "avoid_with": ["啤酒"], "diet_type": ["增重期", "周末放纵"], "recipe_hint": "可使用空气炸锅，减少50%油脂摄入" }, "奶茶": { "tags": ["高糖", "高热量"], "effects": "过量摄入易导致胰岛素抵抗、体重增加", "avoid_with": [], "diet_type": ["偶尔犒劳"], "recipe_hint": "选择低糖或无糖版本，减少珍珠" }, "可乐": { "tags": ["高糖"], "effects": "长期高糖饮料摄入可增加蛀牙及肥胖风险", "avoid_with": ["咖啡"], "diet_type": ["偶尔犒劳"], "recipe_hint": "选择零度可乐或气泡水替代" }, "卤牛肉": { "tags": ["高钠", "高蛋白"], "effects": "高钠摄入可能升高血压", "avoid_with": [], "diet_type": ["增肌期"], "recipe_hint": "配合蔬菜食用平衡营养" }, "牛肉火锅": { "tags": ["高油脂", "高钠"], "effects": "高盐高油易导致水肿", "avoid_with": ["冰啤酒"], "diet_type": ["增重期"], "recipe_hint": "控制汤底油脂和盐分，搭配蔬菜" }, "蔬菜汤": { "tags": ["低热量", "健康推荐"], "effects": "一般安全", "avoid_with": [], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "少盐少油" }, "白灼虾": { "tags": ["高蛋白", "健康推荐"], "effects": "嘌呤偏高，痛风患者需控制", "avoid_with": ["维生素C高的水果"], "diet_type": ["增肌期", "减脂期"], "recipe_hint": "控制蘸料用量" }, "香菇": { "tags": ["高纤维", "低脂"], "effects": "过量可能导致胀气", "avoid_with": ["寒性食物"], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "烹饪前泡发充分" }, "燕麦": { "tags": ["高纤维", "低GI"], "effects": "摄入过多可能引起腹胀", "avoid_with": [], "diet_type": ["减脂期", "增肌期"], "recipe_hint": "配合蛋白质食物提高饱腹" }, "糙米": { "tags": ["高纤维", "低GI"], "effects": "富含植酸，影响矿物质吸收，建议与其他谷物搭配", "avoid_with": [], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "提前浸泡12小时再煮" }, "炸薯条": { "tags": ["高油脂", "高热量"], "effects": "反式脂肪酸摄入风险", "avoid_with": ["汽水"], "diet_type": ["偶尔犒劳"], "recipe_hint": "可使用空气炸锅替代油炸" } , "牛油果": { "tags": ["高脂肪", "高纤维", "健康推荐"], "effects": "脂肪含量高，减脂期注意总热量", "avoid_with": [], "diet_type": ["增肌期", "日常维稳"], "recipe_hint": "搭配全麦面包或沙拉" }, "酸奶": { "tags": ["高蛋白", "益生菌"], "effects": "乳糖不耐人群可能腹胀", "avoid_with": ["高糖谷物"], "diet_type": ["减脂期", "增肌期", "日常维稳"], "recipe_hint": "选择无糖或低糖版本" }, "披萨": { "tags": ["高油脂", "高热量"], "effects": "高钠高脂，注意摄入频次", "avoid_with": ["可乐"], "diet_type": ["周末放纵"], "recipe_hint": "选择薄底、加多蔬菜版本" }, "红薯": { "tags": ["低GI", "高纤维", "健康推荐"], "effects": "过量可致腹胀", "avoid_with": [], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "蒸煮可保留营养" }, "鲑鱼": { "tags": ["高蛋白", "Omega-3"], "effects": "富含EPA/DHA，有助心血管健康", "avoid_with": ["富含草酸蔬菜"], "diet_type": ["增肌期", "日常维稳"], "recipe_hint": "推荐清蒸或空气炸锅" }, "绿茶": { "tags": ["低热量", "抗氧化"], "effects": "空腹或过量可致胃刺激", "avoid_with": ["牛奶"], "diet_type": ["减脂期", "日常维稳"], "recipe_hint": "餐后一小时饮用更佳" }, "巧克力": { "tags": ["高糖", "高脂"], "effects": "摄入过多易长痘和增重", "avoid_with": [], "diet_type": ["偶尔犒劳"], "recipe_hint": "优选 70% 以上黑巧" }, "能量饮料": { "tags": ["高糖", "咖啡因"], "effects": "过量可能导致心悸、睡眠障碍", "avoid_with": ["咖啡"], "diet_type": ["比赛备战", "偶尔提神"], "recipe_hint": "控制每日总咖啡因 < 400mg" } } config.json{ "DEEPSEEK_API_KEY": "刚刚申请的Key", "DEEPSEEK_URL": "申请得到的Url", "MCP_PORT": 8001, "MODEL_NAME": "DeepSeek-R1" } requirements.txtfastapi uvicorn requests httpx jieba cachetools loguru4.1.4 安装依赖点击 终端pip install -r requirements.txt -i https://mirrors.huaweicloud.com/repository/pypi/simple安装成功 示例图 如果显示仍然缺少依赖 请关闭ide 重新打开即可4.2 调试 MCP终端上执行命令uvicorn mcp_server:app --port 8001 运行成功示例图下一步 新建终端 点加号执行 测试程序python test_pipeline.py执行测试成功示例图 这里我们就可以看到 流式响应的一个效果然后我们最小化 这个 CodeArts IDE for Python4.3 仓颉交互AIChat(魔改MCP)打开 仓颉编辑器4.3.1 新建项目我们这里名称就叫 food_bot 然后点击 创建4.3.2 代码实施config.json{ "model": "", "api_key": "", "base_url": "http://localhost:8001/mcp_stream", "system_prompt": "" } cjpm.toml[dependencies] [package] cjc-version = "0.53.13" compile-option = "" description = "nothing here" link-option = "" name = "food_bot" output-type = "executable" src-dir = "" target-dir = "" version = "1.0.0" package-configuration = {} src/main.cjpackage food_bot import std.console.Console import std.collection.ArrayList func cli_chat(env_info: EnvInfo, stream!: Bool) { println("\n欢迎使用智能膳食分析助手，输入exit退出") var history: ArrayList<(String, String)> = ArrayList<(String, String)>() while (true) { print("Input: ") var prompt: String = "" match(Console.stdIn.readln()) { case Some(str1: String) => prompt=str1 case None => continue } if (prompt == "exit" || prompt == "exit()") { break } if (prompt == "clear") { history.clear() println("Output: 已清理历史对话信息。") continue } // 根据配置或命令行参数判断是否启用流式模式 var use_stream: Bool = stream if (!use_stream && env_info.base_url.endsWith("/mcp_stream")) { use_stream = true } print("ChatBox: ") let response_option: Option<String> = if (use_stream) { stream_chat(prompt, env_info, history) } else { chat(prompt, env_info, history) } match (response_option) { case Some(response: String) => println("${response}") history.append((prompt, response)) case None => println("发生错误，正在重试一次...") let retry_option = if (use_stream) { stream_chat(prompt, env_info, history) } else { chat(prompt, env_info, history) } match(retry_option) { case Some(resp2: String) => println("${resp2}") history.append((prompt, resp2)) case None => println("重试失败，即将退出") break } } } } main(args: Array<String>): Int64 { let env_info = load_env_info() var stream: Bool = false if (args.size == 1) { if (args[0] == "--stream") { stream = true } else { println("参数无效，仅支持 --stream 以开启流式输出") } } cli_chat(env_info, stream: stream) return 0 } env_info.cjpackage food_bot import encoding.json.stream.* import std.fs.File import std.fs.Path import std.io.ByteArrayStream public class EnvInfo <: JsonDeserializable<EnvInfo> & JsonSerializable { public let model: String // 模型名称 public let api_key: String // api密钥 public let base_url: String // 调用接口路径 public let system_prompt: String // 预置系统提示词 public init(model: String, api_key: String, base_url: String, system_prompt: String) { this.model = model this.api_key = api_key this.base_url = base_url this.system_prompt = system_prompt } public static func fromJson(r: JsonReader): EnvInfo { var temp_model: String = "" var temp_api_key: String = "sk-xxx" var temp_base_url: String = "http://xxx.xxx.xxx/v1" var temp_system_prompt: String = "You are a helpful assistant." while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match (n) { case "model" => temp_model = r.readValue<String>() case "api_key" => temp_api_key = r.readValue<String>() case "base_url" => temp_base_url = r.readValue<String>() case "system_prompt" => temp_system_prompt = r.readValue<String>() case _ => () } } r.endObject() break case _ => throw Exception() } } return EnvInfo(temp_model, temp_api_key, temp_base_url, temp_system_prompt) } public func toJson(w: JsonWriter): Unit { w.startObject() w.writeName("model").writeValue(this.model) w.writeName("api_key").writeValue(this.api_key) w.writeName("base_url").writeValue(this.base_url) w.writeName("system_prompt").writeValue(this.system_prompt) w.endObject() w.flush() } } public func save_env_info(): Unit { // 该函数用于测试EnvInfo类的序列化为json的能力，顺便生成一个env_sample.json样本做为参考 let env_path = Path("env_sample.json") if (File.exists(env_path)) { File.delete(env_path) } let file = File.create(env_path) let env_info = EnvInfo( "xxxx", "sk-xxxxxx", "http://xxx.xxx.xxx/v1/chat/completions", "You are a helpful assistant." ) var byte_stream = ByteArrayStream() var json_writer = JsonWriter(byte_stream) let write_config = WriteConfig.pretty json_writer.writeConfig = write_config env_info.toJson(json_writer) file.write(byte_stream.readToEnd()) println("`env_sample.json` save ok") file.close() } public func load_env_info(): EnvInfo { // 用于加载配置文件 let env_path = Path("config.json") if (!File.exists(env_path)) { throw Exception("The config file not exists, please check again") } let file = File.openRead(env_path) let file_str: Array<UInt8> = file.readToEnd() var byte_stream = ByteArrayStream() byte_stream.write(file_str) let json_reader = JsonReader(byte_stream) let env_info: EnvInfo = EnvInfo.fromJson(json_reader) file.close() // println("model: ${env_info.model}") // println("api_key: ${env_info.api_key}") // println("base_url: ${env_info.base_url}") // println("system_prompt: ${env_info.system_prompt}") return env_info } chat.cjpackage food_bot import encoding.json.stream.* import net.http.ClientBuilder import net.http.HttpHeaders import net.http.HttpRequestBuilder import net.tls.TlsClientConfig import net.tls.CertificateVerifyMode import std.collection.ArrayList import std.io.ByteArrayStream import std.time.Duration import std.unicode.UnicodeExtension // for String.trim() // ===== 可配置常量 ===== public let READ_TIMEOUT_SECONDS: Int64 = 300 // 长轮询 SSE 建议 300 秒 // =========================== public enum RoleType { User | Assistant | System } public func role_type_to_str(role: RoleType): Option<String> { return match(role) { case RoleType.User => Some("user") case RoleType.Assistant => Some("assistant") case RoleType.System => Some("system") } } public func str_to_role_type(role_option_str: Option<String>): RoleType { return match(role_option_str) { case Some(str) => match(str) { case "user" => RoleType.User case "assistant" => RoleType.Assistant case "system" => RoleType.System case _ => RoleType.Assistant } case None => RoleType.Assistant } } public struct Message<: JsonDeserializable<Message> & JsonSerializable { public let role: RoleType public var content: String public init(role: RoleType, content: String) { this.role = role this.content = content } public static func fromJson(r: JsonReader): Message { var temp_role: Option<String> = None var temp_content: String = "" while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match(n) { case "role" => temp_role = r.readValue<Option<String>>() case "content" => temp_content = r.readValue<String>() case _ => r.skip() } } r.endObject() break case _ => throw Exception("can't deserialize for Message") } } let role_type: RoleType = str_to_role_type(temp_role) return Message(role_type, temp_content) } public func toJson(w: JsonWriter) { w.startObject() w.writeName("role").writeValue<Option<String>>(role_type_to_str(this.role)) w.writeName("content").writeValue<String>(this.content) w.endObject() w.flush() } } public struct ChatRequest <: JsonSerializable { private let model: String private let messages: ArrayList<Message> private let max_tokens: Int64 private let stream: Bool public init( model: String, messages: ArrayList<Message>, max_tokens: Int64, stream: Bool ) { // construction function with messages this.model = model this.messages = messages this.max_tokens = max_tokens this.stream = stream } public init( model: String, prompt: String, history: ArrayList<(String, String)>, system_prompt: String, max_tokens: Int64, stream: Bool ){ // construction function with prompt and system_prompt this.model = model this.messages = ArrayList<Message>([ Message(RoleType.System, system_prompt) ]) for ((use_msg, bot_msg) in history) { this.messages.append(Message(RoleType.User, use_msg)) this.messages.append(Message(RoleType.Assistant, bot_msg)) } this.messages.append(Message(RoleType.User, prompt)) this.max_tokens = max_tokens this.stream = stream } public init( model: String, prompt: String, history: ArrayList<(String, String)>, system_prompt: String, stream: Bool ){ // construction function with prompt and default arguments this.model = model this.messages = ArrayList<Message>([ Message(RoleType.System, system_prompt) ]) for ((use_msg, bot_msg) in history) { this.messages.append(Message(RoleType.User, use_msg)) this.messages.append(Message(RoleType.Assistant, bot_msg)) } this.messages.append(Message(RoleType.User, prompt)) this.max_tokens = 2000 this.stream = stream } public func toJson(w: JsonWriter) { w.startObject() w.writeName("model").writeValue<String>(this.model) w.writeName("messages").writeValue<ArrayList<Message>>(this.messages) w.writeName("max_tokens").writeValue<Int64>(this.max_tokens) w.writeName("stream").writeValue<Bool>(this.stream) w.endObject() w.flush() } } public struct Choice <: JsonDeserializable<Choice> & JsonSerializable { public let index: Int32 public let message: Option<Message> public let delta: Option<Message> public let finish_reason: Option<String> public let logprobs: Option<Float64> // dashscope for qwen need public init( index: Int32, message: Option<Message>, delta: Option<Message>, finish_reason: Option<String>, logprobs: Option<Float64> ) { this.index = index this.message = message this.delta = delta this.finish_reason = finish_reason this.logprobs = logprobs } public static func fromJson(r: JsonReader): Choice { var temp_index: Int32 = -1 var temp_message: Option<Message> = None var temp_delta: Option<Message> = None var temp_finish_reason: Option<String> = None var temp_logprobs: Option<Float64> = None while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match (n) { case "index" => temp_index = r.readValue<Int32>() case "message" => temp_message = r.readValue<Option<Message>>() case "delta" => temp_delta = r.readValue<Option<Message>>() case "finish_reason" => temp_finish_reason = r.readValue<Option<String>>() case "logprobs" => temp_logprobs = r.readValue<Option<Float64>>() case _ => r.skip() } } r.endObject() break case _ => throw Exception("can't deserialize for Choice") } } return Choice(temp_index, temp_message, temp_delta, temp_finish_reason, temp_logprobs) } public func toJson(w: JsonWriter) { w.startObject() w.writeName("index").writeValue<Int32>(this.index) w.writeName("message").writeValue<Option<Message>>(this.message) w.writeName("delta").writeValue<Option<Message>>(this.delta) w.writeName("finish_reason").writeValue<Option<String>>(this.finish_reason) w.writeName("logprobs").writeValue<Option<Float64>>(this.logprobs) w.endObject() w.flush() } } public struct Usage <: JsonDeserializable<Usage> & JsonSerializable { public let prompt_tokens: UInt64 public let completion_tokens: UInt64 public let total_tokens: UInt64 public init(prompt_tokens: UInt64, completion_tokens: UInt64, total_tokens: UInt64) { this.prompt_tokens = prompt_tokens this.completion_tokens = completion_tokens this.total_tokens = total_tokens } public static func fromJson(r: JsonReader): Usage { var temp_prompt_tokens: UInt64 = 0 var temp_completion_tokens: UInt64 = 0 var temp_total_tokens: UInt64 = 0 while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match (n) { case "prompt_tokens" => temp_prompt_tokens = r.readValue<UInt64>() case "completion_tokens" => temp_completion_tokens = r.readValue<UInt64>() case "total_tokens" => temp_total_tokens = r.readValue<UInt64>() case _ => r.skip() } } r.endObject() break case _ => throw Exception("can't deserialize for Usage") } } return Usage(temp_prompt_tokens, temp_completion_tokens, temp_total_tokens) } public func toJson(w: JsonWriter) { w.startObject() w.writeName("prompt_tokens").writeValue<UInt64>(this.prompt_tokens) w.writeName("completion_tokens").writeValue<UInt64>(this.completion_tokens) w.writeName("total_tokens").writeValue<UInt64>(this.total_tokens) w.endObject() w.flush() } } public struct ChatResponse <: JsonDeserializable<ChatResponse> { // some api names `id`, and some names `request_id` public let id: Option<String> public let request_id: Option<String> public let system_fingerprint: Option<String> public let model: String public let object: String public let created: UInt64 public let choices: ArrayList<Choice> public let usage: Option<Usage> public init( id: Option<String>, request_id: Option<String>, system_fingerprint: Option<String>, model: String, object: String, created: UInt64, choices: ArrayList<Choice>, usage: Option<Usage> ) { this.id = id this.request_id = request_id this.system_fingerprint = system_fingerprint this.model = model this.object = object this.created = created this.choices = choices this.usage = usage } public static func fromJson(r: JsonReader): ChatResponse { var temp_id: Option<String> = None var temp_request_id: Option<String> = None var temp_system_fingerprint: Option<String> = None var temp_model: String = "" var temp_object: String = "" var temp_created: UInt64 = 0 var temp_choices: ArrayList<Choice> = ArrayList<Choice>([]) var temp_usage: Option<Usage> = None while (let Some(v) <- r.peek()) { match(v) { case BeginObject => r.startObject() while(r.peek() != EndObject) { let n = r.readName() match (n) { case "id" => temp_id = r.readValue<Option<String>>() case "request_id" => temp_request_id = r.readValue<Option<String>>() case "system_fingerprint" => temp_system_fingerprint = r.readValue<Option<String>>() case "model" => temp_model = r.readValue<String>() case "object" => temp_object = r.readValue<String>() case "created" => temp_created = r.readValue<UInt64>() case "choices" => temp_choices = r.readValue<ArrayList<Choice>>() case "usage" => temp_usage = r.readValue<Option<Usage>>() case _ => r.skip() } } r.endObject() break case _ => throw Exception("can't deserialize for ChatResponse") } } return ChatResponse( temp_id, temp_request_id, temp_system_fingerprint, temp_model, temp_object, temp_created, temp_choices, temp_usage ) } } public func get_domain( url: String ): String { var temp_url = url if (temp_url.startsWith("https://")) { temp_url = temp_url["https://".size..] } else if (temp_url.startsWith("http://")) { temp_url = temp_url["http://".size..] } let domain: String = temp_url.split("?")[0].split("/")[0] return domain } public func build_http_client( prompt: String, env_info: EnvInfo, history: ArrayList<(String, String)>, stream!: Bool ){ // prepare input data // If we are targeting the custom `/mcp_stream` backend we must send // a very simple JSON body `{ "input": "<prompt>" }` instead of the // OpenAI-style `ChatRequest`. Detect this via URL suffix. let is_mcp_stream = env_info.base_url.endsWith("/mcp_stream") var post_data: Array<UInt8> if (is_mcp_stream) { var local_stream = ByteArrayStream() let local_writer = JsonWriter(local_stream) // { "input": "..." } local_writer.startObject() local_writer.writeName("input").writeValue(prompt) local_writer.endObject() local_writer.flush() post_data = local_stream.readToEnd() } else { var array_stream = ByteArrayStream() let json_writer = JsonWriter(array_stream) let chat_res = ChatRequest( env_info.model, prompt, history, env_info.system_prompt, stream ) chat_res.toJson(json_writer) post_data = array_stream.readToEnd() } // build headers var headers: HttpHeaders = HttpHeaders() if (!is_mcp_stream) { // local backend doesn't require auth header headers.add("Authorization", "Bearer ${env_info.api_key}") } headers.add("Content-Type", "application/json") if (stream) { headers.add("Accept", "text/event-stream") } let request = HttpRequestBuilder() .url(env_info.base_url) .method("POST") .body(post_data) .readTimeout(Duration.second * READ_TIMEOUT_SECONDS) .addHeaders(headers) .build() let client = if (env_info.base_url.startsWith("https")) { var tls_client_config = TlsClientConfig() tls_client_config.verifyMode = CertificateVerifyMode.TrustAll tls_client_config.domain = get_domain(env_info.base_url) ClientBuilder() .tlsConfig(tls_client_config) .build() } else { ClientBuilder().build() } return (request, client) } public func chat( prompt: String, env_info: EnvInfo, history: ArrayList<(String, String)> ): Option<String> { let (request, client) = build_http_client( prompt, env_info, history, stream: false ) var result_message: Option<String> = None var res_text = "" try { // call api let response = client.send( request ) // read result (support max revice 100k data) let buffer = Array<Byte>(102400, item: 0) let length = response.body.read(buffer) res_text = String.fromUtf8(buffer[..length]) // println("res_text: ${res_text}") var input_stream = ByteArrayStream() input_stream.write(res_text.toArray()) // convert text to ChatResponse object let json_reader = JsonReader(input_stream) let res_object = ChatResponse.fromJson(json_reader) let choices: ArrayList<Choice> = res_object.choices if (choices.size > 0) { let message = choices[0].message.getOrThrow() // println("message: ${message.content}") result_message = Some(message.content) } else { println("can't found any response") } } catch (e: Exception) { println("ERROR: ${e.message}, reviced text is ${res_text}") } client.close() return result_message } public func stream_chat( prompt: String, env_info: EnvInfo, history: ArrayList<(String, String)> ): Option<String> { let (request, client) = build_http_client( prompt, env_info, history, stream: true ) let is_mcp_stream = env_info.base_url.endsWith("/mcp_stream") var result_response: String = "" var temp_text2 = "" try { // call api let response = client.send( request ) // read result let buffer = Array<Byte>(10240, item: 0) if (is_mcp_stream) { var done = false var current_event = "" while(!done) { let length = response.body.read(buffer) if (length == 0) { break } let res_text = String.fromUtf8(buffer[..length]) for (line in res_text.split("\n")) { let trimmed = line.trim() if (trimmed.size == 0) { continue } if (trimmed.startsWith("event: ")) { current_event = trimmed["event: ".size..] continue } if (trimmed.startsWith("data: ")) { let data = trimmed["data: ".size..] if (current_event == "token") { // 累积 token，暂不直接打印 result_response = result_response + data } else if (current_event == "ping") { // 服务器心跳，忽略即可 () } else if (current_event == "error") { println("服务器错误: ${data}") done = true result_response = "" // force empty so caller sees None break } else if (current_event == "done") { done = true break } } } if (done) { break } } } else { var finish_reason: Option<String> = None while(finish_reason.isNone() && temp_text2 != "[DONE]") { let length = response.body.read(buffer) let res_text = String.fromUtf8(buffer[..length]) for (temp_text in res_text.split("\n")) { temp_text2 = if (temp_text.startsWith("data: ")) { temp_text["data: ".size..] } else { temp_text } if (temp_text2.size == 0) { continue } if (temp_text2 == "[DONE]") { break } var input_stream = ByteArrayStream() input_stream.write(temp_text2.toArray()) // convert text to ChatResponse object let json_reader = JsonReader(input_stream) let res_object = ChatResponse.fromJson(json_reader) let choices: ArrayList<Choice> = res_object.choices if (choices.size > 0) { finish_reason = choices[0].finish_reason if (finish_reason.isNone()) { let delta = choices[0].delta.getOrThrow() result_response = result_response + delta.content } } else { println("can't found any response") } } } } } catch (e: Exception) { println("ERROR: ${e.message}, reviced text is ${temp_text2}") } client.close() if (result_response.size > 0) { return Some(result_response) } else { return None } } 4.4 调试AIChat点击 终端执行 命令cjpm run运行成功示例图输入测试语句例如：我中午吃了炸鸡和奶茶还有香菇 我晚上吃点什么比较好呢？切回 CodeArts IDE for Python 切换 终端我们就可以看到 MCP 服务端 接受到请求的日志然后我们切回 仓颉编辑器 查看 这是响应正常 那么我们来测试 缓存 的效果 再次发出这个命令我中午吃了炸鸡和奶茶还有香菇 我晚上吃点什么比较好呢？你会看到它 立刻返回了结果然后我们可以 切换到 Python编辑器 看一下 命中缓存的日志💡这是两次请求1️⃣请求未命中则去请求DeepSeek2️⃣请求相同数据 则命中缓存五、项目总结5.1 MCPMCP，全称 Model Context Protocol，中文叫“模型上下文协议”。你可以把它想象成 AI 的“USB 接口” --让不同的 AI 模型、工具和应用程序能用统一的方式交流。那么我的理解是：它更像是一个适配器来调节各种AI不同的接口 达到一致的效果，让AI的交流更加简单，即使没有身份预设，走MCP是完美的让AI成为你的最佳助手。5.2 项目tag本次案例参考：https://devstation.connect.huaweicloud.com/space/devportal/casecenter/9a5b50615ac44a21af30f8699c81efc6/1https://devstation.connect.huaweicloud.com/space/devportal/casecenter/5a7144a66d1b43e088592ae447d518ed/1https://gitcode.com/CaseDeveloper/Cangjie-Examples简单的实现了 MCP并发(案例教程没有提及 uvicorn mcp_server:app --port 8001 --workers 4 )后续可拓展例如：food_bot 🔜 vue等各种UI 前端 各种适配机器人food_mcp🔜 独立开发为自己更加需要 模式更加契合的"USB" 让AI更加懂你关于缓存方面：缓存可以采用本地文件、redis等更加高性能的方案来替换这样就可以让实现由AI回答 到AI理解到了所表达和呈现的每一个需求点六、活动地址​ 如果对该项目有疑问，可以评论区留言交流~如果喜欢可以点点赞~​ 我正在参加【案例共创】第4期 基于华为开发者空间+仓颉/DeepSeek/MCP完成应用构建开发实践 https://bbs.huaweicloud.com/forum/thread-02127182415062274055-1-1.html