#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# app.py - основной скрипт для Apache2 с mod_wsgi
import os
import sys
import logging
import re
from datetime import datetime, timedelta, timezone
from functools import wraps
from flask import Flask, request, Response, jsonify, render_template_string, session, redirect, url_for
import secrets
app = Flask(__name__)
# Секретный ключ для сессий
app.secret_key = secrets.token_hex(32)
# Настройка логирования
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler('/opt/api/logs/app.log'),
logging.StreamHandler(sys.stdout)
]
)
logger = logging.getLogger(__name__)
# Конфигурация
LOG_DIR = "/opt/api/apilog"
USERNAME = "admin"
PASSWORD = "Админ@@@2026"
BOT_TOKEN = "7957630117:AAHxO2DEdpR7Y8t4d55gDPyaKuO0Fd9M7rY"
SESSION_TIMEOUT_MINUTES = 0
# Убедимся, что директория для логов существует
if not os.path.exists(LOG_DIR):
os.makedirs(LOG_DIR, mode=0o755)
logger.info(f"Created log directory: {LOG_DIR}")
logger.info(f"Application started with LOG_DIR={LOG_DIR}")
logger.info(f"Configured username: {USERNAME}")
logger.info(f"Bot token: {BOT_TOKEN}")
def get_current_time():
"""Возвращает текущее время с часовым поясом UTC"""
return datetime.now(timezone.utc)
def is_new_message_start(line):
"""Определяет, является ли строка началом нового сообщения"""
# Паттерн для даты и времени в начале строки
date_pattern = r'^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}'
# Если строка начинается с даты - это новое сообщение
if re.match(date_pattern, line):
return True
return False
# HTML шаблон страницы логина
LOGIN_TEMPLATE = '''
Вход в систему
🔐 Вход в систему
{% if error %}
{{ error }}
{% endif %}
{% if message %}
{{ message }}
{% endif %}
'''
# HTML шаблон для страницы просмотра логов
HTML_TEMPLATE = '''
API Logs Viewer
📋 API Logs Viewer
👤 {{ username }}
Текущий день: {{ current_date }}
{% if session_timeout %}
Сессия истекает через:
{% endif %}
🔄 Автообновление: ВКЛ
{% if session_warning %}
⚠️ Внимание! Сессия истекает через {{ session_warning }} минут. Пожалуйста, сохраните данные.
{% endif %}
Загрузка логов...
'''
def login_required(f):
"""Декоратор для проверки авторизации"""
@wraps(f)
def decorated_function(*args, **kwargs):
if 'user_id' not in session:
logger.info("User not authenticated, redirecting to login")
return redirect(url_for('login'))
if SESSION_TIMEOUT_MINUTES:
login_time = session.get('login_time')
if login_time:
elapsed = (get_current_time() - login_time).total_seconds() / 60
if elapsed > SESSION_TIMEOUT_MINUTES:
logger.info(f"Session expired for user {session.get('username')}")
session.clear()
return redirect(url_for('login', message='Сессия истекла. Пожалуйста, войдите снова.'))
return f(*args, **kwargs)
return decorated_function
@app.route('/login', methods=['GET', 'POST'])
def login():
"""Страница входа"""
if 'user_id' in session:
if SESSION_TIMEOUT_MINUTES:
login_time = session.get('login_time')
if login_time:
elapsed = (get_current_time() - login_time).total_seconds() / 60
if elapsed <= SESSION_TIMEOUT_MINUTES:
return redirect(url_for('index'))
error = None
message = request.args.get('message')
if request.method == 'POST':
username = request.form.get('username')
password = request.form.get('password')
logger.info(f"Login attempt for user: {username}")
if username == USERNAME and password == PASSWORD:
logger.info(f"Successful login for user: {username}")
session['user_id'] = username
session['username'] = username
session['login_time'] = get_current_time()
session.permanent = True
if SESSION_TIMEOUT_MINUTES:
app.permanent_session_lifetime = timedelta(minutes=SESSION_TIMEOUT_MINUTES)
return redirect(url_for('index'))
else:
logger.warning(f"Failed login attempt for user: {username}")
error = 'Неверное имя пользователя или пароль'
return render_template_string(LOGIN_TEMPLATE, error=error, message=message)
@app.route('/logout', methods=['POST'])
def logout():
"""Выход из системы"""
logger.info(f"User {session.get('username')} logged out")
session.clear()
return redirect(url_for('login', message='Вы успешно вышли из системы'))
@app.route('/')
@login_required
def index():
"""Главная страница"""
current_date = datetime.now().strftime('%Y-%m-%d')
username = session.get('username', 'User')
session_timeout = None
session_warning = None
if SESSION_TIMEOUT_MINUTES:
login_time = session.get('login_time')
if login_time:
elapsed = (get_current_time() - login_time).total_seconds() / 60
remaining = SESSION_TIMEOUT_MINUTES - elapsed
if remaining > 0:
session_timeout = int(remaining)
if remaining <= 5:
session_warning = int(remaining)
return render_template_string(
HTML_TEMPLATE,
current_date=current_date,
username=username,
session_timeout=session_timeout,
session_warning=session_warning
)
@app.route('/api/logs/', methods=['GET'])
@login_required
def get_logs(date_str):
"""API для получения логов за определенную дату"""
logger.info(f"API request for logs date: {date_str}")
logs = read_logs_for_date(date_str)
if logs is None:
logger.error(f"Failed to read logs for {date_str}")
return jsonify({'error': 'Ошибка чтения файла'}), 500
return jsonify({'logs': logs, 'date': date_str})
@app.route('/api/logs//count', methods=['GET'])
@login_required
def get_logs_count(date_str):
"""API для получения количества сообщений за дату"""
logs = read_logs_for_date(date_str)
if logs is None:
return jsonify({'count': 0}), 200
return jsonify({'count': len(logs)})
@app.route('/bot/sendMessage', methods=['POST', 'GET'])
def telegram_send_message(token):
"""Эмуляция Telegram API: /bot/sendMessage"""
if token != BOT_TOKEN:
logger.warning(f"Invalid bot token attempt: {token}")
return jsonify({
'ok': False,
'error_code': 401,
'description': 'Unauthorized'
}), 401
if request.method == 'POST':
message = request.form.get('text') or request.args.get('text')
if not message and request.is_json:
data = request.get_json()
message = data.get('text') if data else None
else:
message = request.args.get('text')
chat_id = request.form.get('chat_id') or request.args.get('chat_id')
logger.info(f"Telegram API request - Token: {token[:10]}..., Chat ID: {chat_id}, Message length: {len(message) if message else 0}")
if not message:
logger.warning("Telegram API: No message provided")
return jsonify({
'ok': False,
'error_code': 400,
'description': 'Bad Request: message text is empty'
}), 400
if write_log(message):
logger.info("Telegram API: Message successfully logged")
response = {
'ok': True,
'result': {
'message_id': int(datetime.now().timestamp()),
'from': {
'id': 123456789,
'is_bot': True,
'first_name': 'API Logs Bot',
'username': 'api_logs_bot'
},
'chat': {
'id': int(chat_id) if chat_id and chat_id.isdigit() else 123456789,
'type': 'private',
'first_name': 'User',
'username': 'user'
},
'date': int(datetime.now().timestamp()),
'text': message
}
}
return jsonify(response), 200
else:
logger.error("Telegram API: Failed to write log")
return jsonify({
'ok': False,
'error_code': 500,
'description': 'Internal Server Error'
}), 500
@app.route('/send', methods=['POST'])
def send_alert():
"""Эндпоинт для получения сообщений от send.sh (без авторизации)"""
message = request.form.get('text') or request.args.get('text')
if not message:
logger.warning("Send request without message")
return jsonify({'error': 'No message provided'}), 400
logger.info(f"Received message via /send: {message[:100]}...")
if write_log(message):
logger.info("Message successfully logged")
return jsonify({'status': 'success', 'message': 'Message logged'}), 200
else:
logger.error("Failed to write log")
return jsonify({'error': 'Failed to write log'}), 500
@app.route('/health', methods=['GET'])
def health_check():
"""Эндпоинт для проверки работоспособности"""
return jsonify({'status': 'ok', 'timestamp': datetime.now().isoformat()})
def write_log(message):
"""Запись сообщения в лог-файл"""
today = datetime.now().strftime('%Y%m%d')
log_file = os.path.join(LOG_DIR, f'api_{today}.log')
try:
with open(log_file, 'a', encoding='utf-8') as f:
f.write(message + '\n')
return True
except Exception as e:
logger.error(f"Error writing log: {e}")
return False
def read_logs_for_date(date_str):
"""Чтение логов за определенную дату - объединяем многострочные сообщения"""
try:
datetime.strptime(date_str, '%Y-%m-%d')
log_file = os.path.join(LOG_DIR, f'api_{date_str.replace("-", "")}.log')
if not os.path.exists(log_file):
logger.info(f"Log file not found: {log_file}")
return []
with open(log_file, 'r', encoding='utf-8') as f:
lines = f.readlines()
# Объединяем многострочные сообщения
messages = []
current_message = []
for line in lines:
line = line.rstrip('\n')
if not line:
continue
# Если это начало нового сообщения
if current_message and is_new_message_start(line):
messages.append('\n'.join(current_message))
current_message = [line]
else:
current_message.append(line)
# Добавляем последнее сообщение
if current_message:
messages.append('\n'.join(current_message))
logger.info(f"Read {len(messages)} messages from {log_file}")
return messages
except Exception as e:
logger.error(f"Error reading logs for {date_str}: {e}")
return None
# Для WSGI
application = app