app_flask/app_flask.py

from flask import Flask, render_template, request, redirect, url_for, jsonify, make_response
from flask_cors import CORS
from flask_restx import Api, Resource, fields
#from WXBizJsonMsgCrypt import WXBizJsonMsgCrypt
import time
import os
import sys
import json

# 创建 Flask 应用
app = Flask(__name__)
app.config['JSON_AS_ASCII'] = False
app.config['JSON_SORT_KEYS'] = False

# 启用 CORS
CORS(app, resources={r"/*": {"origins": "*"}})

# 首页路由
@app.route("/")
def home():
    return render_template("index.html", message="欢迎来到 Flask Demo！")

# 带参数的路由
@app.route("/hello/<name>")
@app.route("/hello")  # 兼容无参数情况
def hello(name=None):
    if name is None:
        name = "访客"
    return f"<h1>你好，{name}！</h1><a href='./'>返回首页</a>"

# 表单页面
@app.route("/form", methods=["GET", "POST"])
def form():
    if request.method == "POST":
        username = request.form.get("username")
        return redirect(url_for("./hello", name=username))  # 提交后跳转到打招呼页面
    return """
        <form method="post">
            <label>输入你的名字：</label>
            <input type="text" name="username" required>
            <button type="submit">提交</button>
        </form>
        <a href='./'>返回首页</a>
    """

# 健康检查
@app.route('/health', methods=['GET'])
def health_check():
    return jsonify({
        'status': 'healthy',
        'service': 'flask',
        'version': '1.0.0',
        'timestamp': time.time(),
        'environment': os.getenv('ENVIRONMENT', 'development')
    })

'''
# 创建 RESTX API
api = Api(
    app,
    version='1.0',
    title='Flask REST API',
    description='高性能 Flask REST API 服务',
    doc='/docs'
)

# 数据模型
user_model = api.model('User', {
    'id': fields.Integer(readonly=True, description='用户ID'),
    'username': fields.String(required=True, description='用户名'),
    'email': fields.String(required=True, description='邮箱'),
    'created_at': fields.DateTime(readonly=True, description='创建时间')
})

task_model = api.model('Task', {
    'id': fields.Integer(readonly=True, description='任务ID'),
    'title': fields.String(required=True, description='任务标题'),
    'description': fields.String(description='任务描述'),
    'completed': fields.Boolean(default=False, description='是否完成'),
    'created_at': fields.DateTime(readonly=True, description='创建时间')
})

# 内存数据库
users_db = []
tasks_db = []
user_counter = 1
task_counter = 1

# 根路径
@app.route('/', methods=['GET'])
def root():
    return jsonify({
        'message': '欢迎使用 Flask 服务',
        'docs': '/docs',
        'endpoints': [
            '/health',
            '/api/users',
            '/api/tasks',
            '/api/info'
        ]
    })

# 获取服务信息
@app.route('/info', methods=['GET'])
def get_info():
    return jsonify({
        'service': 'flask',
        #'host': request.host,
        #'server_port':  request.environ.get('SERVER_PORT'),
        #'server_name': request.environ.get('SERVER_NAME'),
        #"remote_addr": request.remote_addr,
        'port': os.getenv('PORT', 5000),
        'python_version': os.sys.version,
        'environment': os.getenv('ENVIRONMENT', 'development'),
        'startup_time': app.config.get('STARTUP_TIME', time.time())
    })

# 用户资源
@api.route('/users')
class UserList(Resource):
    @api.doc('list_users')
    @api.marshal_list_with(user_model)
    def get(self):
        """获取所有用户"""
        return users_db
    
    @api.doc('create_user')
    @api.expect(user_model)
    @api.marshal_with(user_model, code=201)
    def post(self):
        """创建新用户"""
        global user_counter
        data = api.payload
        user = {
            'id': user_counter,
            'username': data['username'],
            'email': data['email'],
            'created_at': time.strftime('%Y-%m-%d %H:%M:%S')
        }
        users_db.append(user)
        user_counter += 1
        return user, 201

@api.route('/users/<int:user_id>')
@api.response(404, '用户未找到')
@api.param('user_id', '用户ID')
class User(Resource):
    @api.doc('get_user')
    @api.marshal_with(user_model)
    def get(self, user_id):
        """根据ID获取用户"""
        user = next((u for u in users_db if u['id'] == user_id), None)
        if user is None:
            api.abort(404, f"用户 {user_id} 未找到")
        return user
    
    @api.doc('update_user')
    @api.expect(user_model)
    @api.marshal_with(user_model)
    def put(self, user_id):
        """更新用户信息"""
        user = next((u for u in users_db if u['id'] == user_id), None)
        if user is None:
            api.abort(404, f"用户 {user_id} 未找到")
        
        data = api.payload
        user.update({
            'username': data.get('username', user['username']),
            'email': data.get('email', user['email'])
        })
        return user
    
    @api.doc('delete_user')
    @api.response(204, '用户已删除')
    def delete(self, user_id):
        """删除用户"""
        global users_db
        users_db = [u for u in users_db if u['id'] != user_id]
        return '', 204

# 任务资源
@api.route('/tasks')
class TaskList(Resource):
    @api.doc('list_tasks')
    @api.marshal_list_with(task_model)
    def get(self):
        """获取所有任务"""
        return tasks_db
    
    @api.doc('create_task')
    @api.expect(task_model)
    @api.marshal_with(task_model, code=201)
    def post(self):
        """创建新任务"""
        global task_counter
        data = api.payload
        task = {
            'id': task_counter,
            'title': data['title'],
            'description': data.get('description', ''),
            'completed': data.get('completed', False),
            'created_at': time.strftime('%Y-%m-%d %H:%M:%S')
        }
        tasks_db.append(task)
        task_counter += 1
        return task, 201

@api.route('/tasks/<int:task_id>')
@api.response(404, '任务未找到')
@api.param('task_id', '任务ID')
class Task(Resource):
    @api.doc('get_task')
    @api.marshal_with(task_model)
    def get(self, task_id):
        """根据ID获取任务"""
        task = next((t for t in tasks_db if t['id'] == task_id), None)
        if task is None:
            api.abort(404, f"任务 {task_id} 未找到")
        return task
    
    @api.doc('update_task')
    @api.expect(task_model)
    @api.marshal_with(task_model)
    def put(self, task_id):
        """更新任务信息"""
        task = next((t for t in tasks_db if t['id'] == task_id), None)
        if task is None:
            api.abort(404, f"任务 {task_id} 未找到")
        
        data = api.payload
        task.update({
            'title': data.get('title', task['title']),
            'description': data.get('description', task['description']),
            'completed': data.get('completed', task['completed'])
        })
        return task
    
    @api.doc('delete_task')
    @api.response(204, '任务已删除')
    def delete(self, task_id):
        """删除任务"""
        global tasks_db
        tasks_db = [t for t in tasks_db if t['id'] != task_id]
        return '', 204

# 计算功能
@app.route('/compute', methods=['POST'])
def compute():
    """执行计算操作"""
    try:
        data = request.get_json()
        numbers = data.get('numbers', [])
        operation = data.get('operation', 'sum').lower()
        
        if not numbers:
            return jsonify({'error': '数字列表不能为空'}), 400
        
        if operation == 'sum':
            result = sum(numbers)
        elif operation == 'product':
            result = 1
            for num in numbers:
                result *= num
        elif operation == 'average':
            result = sum(numbers) / len(numbers)
        elif operation == 'max':
            result = max(numbers)
        elif operation == 'min':
            result = min(numbers)
        else:
            return jsonify({'error': '不支持的操作类型'}), 400
        
        return jsonify({
            'result': result,
            'operation': operation,
            'timestamp': time.time()
        })
    
    except Exception as e:
        return jsonify({'error': str(e)}), 500

# 性能测试
@app.route('/benchmark/<int:iterations>', methods=['GET'])
def benchmark(iterations=1000000):
    """性能基准测试"""
    start_time = time.time()
    
    # 执行一些计算密集型操作
    total = 0
    for i in range(iterations):
        total += i * i
    
    end_time = time.time()
    
    return jsonify({
        'iterations': iterations,
        'total': total,
        'time_seconds': end_time - start_time,
        'operations_per_second': iterations / (end_time - start_time) if (end_time - start_time) > 0 else 0
    })

# 启动事件
@app.before_first_request
def before_first_request():
    app.config['STARTUP_TIME'] = time.time()
    print(f"Flask 服务启动在端口 {os.getenv('PORT', 5000)}")

# 错误处理
@app.errorhandler(404)
def not_found(error):
    return jsonify({'error': '资源未找到'}), 404

@app.errorhandler(500)
def internal_error(error):
    return jsonify({'error': '服务器内部错误'}), 500

if __name__ == '__main__':
    port = int(os.getenv('PORT', 5000))
    app.run(host='0.0.0.0', port=port, debug=False)

'''