python_ai/main.py

from flask import Flask, request, jsonify,Response
import os
from checkPlaceName import checkPlaceName
from checkRepeatText import checkRepeatText
from checkCompanyName import checkCompanyName
from checkDocumentError import getDocumentError
from checkTitleName import checkTitleName
from flask_cors import CORS
import qwen_agenttext
app = Flask(__name__)
cros = CORS(app)
UPLOAD_FOLDER = 'uploads'
usableTag=[0,0,0,0,0,0,0,0]
if not os.path.exists(UPLOAD_FOLDER):
    os.makedirs(UPLOAD_FOLDER)
@app.route('/upload', methods=['POST'])
def upload_file():
    if 'file' not in request.files:
        return jsonify({"error": "No file part"}), 400
    file = request.files['file']
    if file.filename == '':
        return jsonify({"error": "No selected file"}), 400
    if file:
        filename = file.filename
        file.save(os.path.join(UPLOAD_FOLDER,filename))
        return jsonify({"message": "File uploaded successfully"}), 200
@app.route('/stream' ,methods=["GET", "POST"])
def stream_numbers():
    context= request.args.get('context')
    # def generate_numbers():
    #     event_id=0
    #     for number in range(1, 10):
    #         json_data = json.dumps({"number": number})
    #         print(json_data)
    #         event_id += 1
    #         yield f"id: {event_id}\n"
    #         yield f"event: time-update\n"
    #         yield f"data: {json_data}\n\n"  # 每次生成一个数字就发送
    #         time.sleep(0.5)  # 为了演示，加入短暂延迟
    #     json_data = json.dumps({"number": "done"})
    #     yield f"id: {1}\n"
    #     yield f"event: time-update\n"
    #     yield f"data: {json_data}\n\n"  # 发送完成信号

    headers = {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET,POST",
        "Access-Control-Allow-Headers": "x-requested-with,content-type",
    }
    return Response(qwen_agenttext.getxinx(context),headers=headers)
@app.route('/sse/checkRepeatText', methods=['GET'])
def checkRepeatTextWeb():
    filename = request.args.get('filename')

    def generate_checkRepeatText(filename):
        id=0
        try:
            for i in checkRepeatText(filename):
                yield f"id: {id+1}\n"
                yield f"event: checkRepeatText\n"
                yield f"data: {i}\n\n"  # 发送完成信号
        except Exception as e:
            yield f"id: {id+1}\n"
            yield f"event: checkRepeatText\n"
            yield f"data: **程序出现异常**\n\n"  # 发送完成信号
    headers = {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET,POST",
        "Access-Control-Allow-Headers": "x-requested-with,content-type",
    }
    return Response(generate_checkRepeatText(filename), headers=headers)


@app.route('/sse/checkPlaceName', methods=['GET'])
def checkPlaceNameWebSse():
    filename = request.args.get('filename')

    def generate_checkPlaceName(filename):
        id=0
        for i in checkPlaceName(filename):
            yield f"id: {id+1}\n"
            yield f"event: checkPlaceName\n"
            yield f"data: {i}\n\n"  # 发送完成信号
    headers = {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET,POST",
        "Access-Control-Allow-Headers": "x-requested-with,content-type",
    }
    return Response(generate_checkPlaceName(filename), headers=headers)
@app.route('/sse/checkCompanyName', methods=['GET'])
def checkCompanyNameWebSse():
    filename = request.args.get('filename')

    def generate_checkCompanyName(filename):
        id = 0
        for i in checkCompanyName(filename):
            yield f"id: {id + 1}\n"
            yield f"event: checkCompanyName\n"
            yield f"data: {i}\n\n"  # 发送完成信号

    headers = {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET,POST",
        "Access-Control-Allow-Headers": "x-requested-with,content-type",
    }
    return Response(generate_checkCompanyName(filename), headers=headers)

@app.route('/sse/checkDocumentErrorWeb', methods=['GET'])
def checkDocumentErrorWebSse():
    filename = request.args.get('filename')

    def generate_checkDocumentError(filename):
        id = 0
        for i in getDocumentError(filename):
            yield f"id: {id + 1}\n"
            yield f"event: getDocumentError\n"
            yield f"data: {i}\n\n"  # 发送完成信号

    headers = {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET,POST",
        "Access-Control-Allow-Headers": "x-requested-with,content-type",
    }
    return Response(generate_checkDocumentError(filename), headers=headers)
@app.route('/sse/checkTitleName', methods=['GET'])
def checkTitleNameWebSse():
    filename = request.args.get('filename')

    def generate_checkTitleName(filename):
        id = 0
        for i in checkTitleName(filename):
            yield f"id: {id + 1}\n"
            yield f"event: checkTitleName\n"
            yield f"data: {i}\n\n"  # 发送完成信号

    headers = {
        "Content-Type": "text/event-stream",
        "Cache-Control": "no-cache",
        "X-Accel-Buffering": "no",
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET,POST",
        "Access-Control-Allow-Headers": "x-requested-with,content-type",
    }
    return Response(generate_checkTitleName(filename), headers=headers)
if __name__ == '__main__':
    app.run(host="0.0.0.0",port=80)