tv_show_v1/api/main.py

import os
import time
import uuid

from flask import Blueprint, render_template, send_from_directory, current_app, redirect, url_for, request, jsonify, session
from werkzeug.utils import secure_filename

from api.utils import login_required, load_led_config, get_server_ip
from application.scheduler_service import scheduler_service
from application.self_check_service import run_all_checks
from application.uap_message_service import send_self_check_notification
from utils.logger_config import logger

main_bp = Blueprint('main', __name__)

ATTACHMENT_TEST_SESSION_KEY = 'attachment_test_filename'
# 附件测试页允许的类型（单附件，新上传会替换）
ATTACHMENT_TEST_EXTENSIONS = {
    'png', 'jpg', 'jpeg', 'gif', 'bmp', 'webp', 'heic', 'heif',
    'mp4', 'mov', 'webm', 'mkv', 'avi',
    'pdf', 'doc', 'docx', 'xls', 'xlsx', 'ppt', 'pptx',
    'txt', 'csv', 'zip', 'rar', '7z',
    'mp3', 'wav', 'm4a',
}


def _attachment_test_public_url(filename):
    if not filename:
        return None
    host = request.host
    if 'localhost' in host or '127.0.0.1' in host:
        server_ip = get_server_ip()
        port = request.environ.get('SERVER_PORT', '5000')
        host = f"{server_ip}:{port}"
    scheme = request.headers.get('X-Forwarded-Proto', request.scheme)
    return f"{scheme}://{host}/uploads/{filename}"


def _attachment_test_remove_stored():
    """删除 session 中记录的文件（若存在）。"""
    name = session.pop(ATTACHMENT_TEST_SESSION_KEY, None)
    if not name:
        return
    path = os.path.join(current_app.config['UPLOAD_FOLDER'], name)
    try:
        if os.path.isfile(path):
            os.remove(path)
    except OSError as e:
        logger.warning(f"删除附件测试文件失败 {path}: {e}")


def _guess_ext_from_content_type(content_type):
    if not content_type:
        return None
    ct = content_type.lower()
    mapping = {
        'image/png': 'png', 'image/jpeg': 'jpg', 'image/jpg': 'jpg', 'image/gif': 'gif',
        'image/webp': 'webp', 'image/bmp': 'bmp', 'image/heic': 'heic', 'image/heif': 'heif',
        'video/mp4': 'mp4', 'video/quicktime': 'mov', 'video/webm': 'webm',
        'application/pdf': 'pdf',
    }
    return mapping.get(ct.split(';')[0].strip())

@main_bp.route('/')
@login_required
def index():
    """重定向到 Kodi 控制页面"""
    return redirect(url_for('main.kodi_page'))

@main_bp.route('/kodi')
@login_required
def kodi_page():
    """Kodi 控制页面"""
    return render_template('kodi/index.html', active_page='kodi')

@main_bp.route('/door')
@login_required
def door_page():
    """门禁控制页面"""
    return render_template('door/index.html', active_page='door')

@main_bp.route('/led')
@login_required
def led_page():
    """LED 控制页面"""
    led_segments = load_led_config()
    return render_template('led/index.html', active_page='led', led_segments=led_segments)

@main_bp.route('/ha')
@login_required
def ha_page():
    """HA 灯光控制页面"""
    return render_template('ha/index.html', active_page='ha')

@main_bp.route('/self_check')
@login_required
def self_check_page():
    """设备自检页面"""
    return render_template('self_check.html', active_page='self_check')


@main_bp.route('/attachment_upload_test')
@login_required
def attachment_test_page():
    """附件上传测试页面（单附件，会话内替换）"""
    return render_template('attachment_test/index.html', active_page='attachment_test')


@main_bp.route('/api/attachment_test/status', methods=['GET'])
@login_required
def attachment_test_status():
    """当前会话中的附件信息"""
    fn = session.get(ATTACHMENT_TEST_SESSION_KEY)
    if not fn:
        return jsonify({'success': True, 'has_file': False})
    folder = current_app.config['UPLOAD_FOLDER']
    path = os.path.join(folder, fn)
    if not os.path.isfile(path):
        session.pop(ATTACHMENT_TEST_SESSION_KEY, None)
        return jsonify({'success': True, 'has_file': False})
    return jsonify({
        'success': True,
        'has_file': True,
        'filename': fn,
        'url': _attachment_test_public_url(fn),
    })


@main_bp.route('/api/attachment_test/upload', methods=['POST'])
@login_required
def attachment_test_upload():
    """上传单个附件，覆盖会话中已有文件"""
    if 'file' not in request.files:
        return jsonify({'success': False, 'message': '缺少文件字段 file'}), 400
    file = request.files['file']
    if not file or file.filename == '':
        return jsonify({'success': False, 'message': '未选择文件'}), 400

    original = file.filename
    base = secure_filename(original)
    ext = None
    if base and '.' in base:
        ext = base.rsplit('.', 1)[1].lower()
    if not ext or ext not in ATTACHMENT_TEST_EXTENSIONS:
        ext = _guess_ext_from_content_type(file.content_type or '')
    if not ext or ext not in ATTACHMENT_TEST_EXTENSIONS:
        return jsonify({
            'success': False,
            'message': f'不支持的文件类型，允许: {", ".join(sorted(ATTACHMENT_TEST_EXTENSIONS))}',
        }), 400

    _attachment_test_remove_stored()

    new_name = f"att_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}.{ext}"
    folder = current_app.config['UPLOAD_FOLDER']
    os.makedirs(folder, exist_ok=True)
    filepath = os.path.join(folder, new_name)
    file.save(filepath)
    session[ATTACHMENT_TEST_SESSION_KEY] = new_name

    return jsonify({
        'success': True,
        'message': '上传成功',
        'filename': new_name,
        'url': _attachment_test_public_url(new_name),
    })


@main_bp.route('/api/attachment_test', methods=['DELETE'])
@login_required
def attachment_test_delete():
    """删除当前会话中的附件"""
    if not session.get(ATTACHMENT_TEST_SESSION_KEY):
        return jsonify({'success': True, 'message': '没有可删除的附件'})
    _attachment_test_remove_stored()
    return jsonify({'success': True, 'message': '已删除'})

@main_bp.route('/api/send_report', methods=['POST'])
@login_required
def send_report_api():
    """手动发送自检报告"""
    try:
        data = request.get_json()
        target_email = data.get('email')
        
        # 执行全量自检
        results = run_all_checks()
        
        # 生成 HTML 报告
        html_report = scheduler_service.format_report_html(results)
        
        # 发送邮件
        subject = "手动触发设备自检报告"
        scheduler_service.send_email(subject, html_report, receivers=target_email)
        
        # 向统一登录平台运维人员推送通知
        summary = scheduler_service._format_summary_text(results)
        send_self_check_notification(summary)
        
        return jsonify({
            "success": True,
            "message": f"自检报告已发送至 {target_email if target_email else '默认邮箱'}，并已通知运维人员"
        })
    except Exception as e:
        logger.error(f"发送报告失败: {e}")
        return jsonify({
            "success": False,
            "message": f"发送报告失败: {str(e)}"
        }), 500

# 提供上传文件的访问接口
@main_bp.route('/uploads/<filename>')
def uploaded_file(filename):
    """提供上传文件的访问接口"""
    return send_from_directory(current_app.config['UPLOAD_FOLDER'], filename)