PyBot/web/app.py

from flask import Flask, jsonify, render_template
from collections import deque
import os
import logging

app = Flask(__name__)

# 配置文件路径
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
PARENT_DIR = os.path.dirname(BASE_DIR)  # 上一个文件夹
SEC_NEWS_PATH = os.path.join(PARENT_DIR, 'resources', 'history', 'sec_news.md')
TECH_PASSAGE_PATH = os.path.join(PARENT_DIR, 'resources', 'history', 'tech_passage.md')
CORE_LOG_PATH = os.path.join(PARENT_DIR, 'resources', 'log', 'core.log')
WEB_LOG_PATH = os.path.join(PARENT_DIR, 'resources', 'log', 'app.log')

# 配置日志记录器
logging.basicConfig(
    filename=WEB_LOG_PATH,
    level=logging.INFO,
    format= '%(asctime)s - %(levelname)s - %(message)s'
)

def replace_content(content):
    content = content.replace('####', '###')
    content = content.replace(r"e:\Self-Tool-Code\PyBot", '.')  # 修改: 使用原始字符串避免转义问题
    return content

@app.route('/')
def index():
    logging.info("访问主页")
    return render_template('index.html')

def get_records(content, num_records=20, delimiter='-'*40):
    records = content.split(delimiter)
    # 去除空字符串，去除之后MD格式会混乱
    # records = [record.strip() for record in records if record.strip()]
    selected_records = records[:num_records]
    return delimiter.join(selected_records)

def read_last_lines(file_path, num_lines=100):
    with open(file_path, 'r', encoding='utf-8') as file:
        return deque(file, maxlen=num_lines)

@app.route('/get-sec-news')
def get_sec_news():
    logging.info(f"尝试打开安全新闻历史推送文件: {SEC_NEWS_PATH}")
    try:
        with open(SEC_NEWS_PATH, 'r', encoding='utf-8') as file:
            content = file.read()
        content = replace_content(content)
        content = get_records(content)
        return jsonify({'content': content}), 200
    except FileNotFoundError:
        logging.error(f"文件缺失: {SEC_NEWS_PATH}")
        return jsonify({'error': '安全新闻历史推送文件缺失！'}), 404
    except Exception as e:
        logging.error(f"读取时出错: {SEC_NEWS_PATH}, 原因: {str(e)}")
        return jsonify({'error': str(e)}), 500

@app.route('/get-tech-passage')
def get_tech_passage():
    logging.info(f"尝试打开技术文章历史推送文件: {TECH_PASSAGE_PATH}")
    try:
        with open(TECH_PASSAGE_PATH, 'r', encoding='utf-8') as file:
            content = file.read()
        content = replace_content(content)
        content = get_records(content)
        return jsonify({'content': content}), 200
    except FileNotFoundError:
        logging.error(f"文件缺失: {TECH_PASSAGE_PATH}")
        return jsonify({'error': '技术文章历史推送文件缺失！'}), 404
    except Exception as e:
        logging.error(f"读取时出错: {TECH_PASSAGE_PATH}, 原因: {str(e)}")
        return jsonify({'error': str(e)}), 500

@app.route('/log')
def get_log():
    logging.info(f"尝试打开核心日志文件: {CORE_LOG_PATH}")
    # 读取日志文件内容
    log_content = read_last_lines(CORE_LOG_PATH)
    log_content = '\n'.join(log_content)
    # 将日志内容传递给模板
    return render_template('log.html', log_content=log_content)

@app.route('/weblog')
def get_weblog():
    logging.info(f"尝试打开Web应用日志文件: {WEB_LOG_PATH}")
    log_content = read_last_lines(WEB_LOG_PATH)
    log_content = '\n'.join(log_content)
    log_content = replace_content(log_content)
    return render_template('log.html', log_content=log_content)

def run_server():
    app.run(host='0.0.0.0', port=5000)

if __name__ == '__main__':
    app.run(debug=True)  # 在生产环境中应设置为 False