统一消息系统

# login.py

from flask import Flask, request, session

import pika

app = Flask(__name__)

app.secret_key = 'your_secret_key'

def send_message_to_queue(user_id, role):

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))

channel = connection.channel()

channel.queue_declare(queue='notification')

message = f"User {user_id} with role {role} has logged in."

channel.basic_publish(exchange='', routing_key='notification', body=message)

connection.close()

@app.route('/login', methods=['POST'])

def login():

username = request.json.get('username')

password = request.json.get('password')

# 简单的模拟登录逻辑

if username == 'admin' and password == '123456':

session['user_id'] = 1

session['role'] = 'admin'

send_message_to_queue(1, 'admin')

return {'status': 'success', 'message': 'Login successful'}

else:

return {'status': 'error', 'message': 'Invalid credentials'}

if __name__ == '__main__':

app.run(debug=True)

# file_api.py

from flask import Flask, jsonify, session

app = Flask(__name__)

# 模拟招标文件数据

tender_files = [

{"id": 1, "name": "项目A招标文件.pdf", "type": "public"},

{"id": 2, "name": "项目B招标文件.docx", "type": "private"},

{"id": 3, "name": "项目C招标文件.xlsx", "type": "private"}

]

@app.route('/tender_files', methods=['GET'])

def get_tender_files():

user_role = session.get('role')

if not user_role:

return jsonify({"error": "Not logged in"}), 401

filtered_files = [f for f in tender_files if f['type'] == 'public' or (f['type'] == 'private' and user_role == 'admin')]

return jsonify(filtered_files)

if __name__ == '__main__':

app.run(debug=True)