#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# app.py - основной скрипт для Apache2 с mod_wsgi
import os
import sys
import logging
import re
import json
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
# Переменные для API доступа
API_USER = "event"
API_PASS = "Event2026"
# Файл для хранения статуса переданных событий
SENT_EVENTS_FILE = "/opt/api/sent_events.json"
# Файл для хранения подтвержденных событий
CONFIRMED_EVENTS_FILE = "/opt/api/confirmed_events.json"
# Убедимся, что директория для логов существует
if not os.path.exists(LOG_DIR):
os.makedirs(LOG_DIR, mode=0o755)
logger.info(f"Created log directory: {LOG_DIR}")
# Создаем директорию для файлов состояния если нужно
os.makedirs(os.path.dirname(SENT_EVENTS_FILE), exist_ok=True)
logger.info(f"Application started with LOG_DIR={LOG_DIR}")
logger.info(f"Configured username: {USERNAME}")
logger.info(f"Bot token: {BOT_TOKEN}")
logger.info(f"API credentials configured")
def load_sent_events():
"""Загружает список отправленных событий"""
if os.path.exists(SENT_EVENTS_FILE):
try:
with open(SENT_EVENTS_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except:
return []
return []
def save_sent_events(events):
"""Сохраняет список отправленных событий"""
try:
with open(SENT_EVENTS_FILE, 'w', encoding='utf-8') as f:
json.dump(events, f, ensure_ascii=False, indent=2)
return True
except Exception as e:
logger.error(f"Error saving sent events: {e}")
return False
def load_confirmed_events():
"""Загружает список подтвержденных событий"""
if os.path.exists(CONFIRMED_EVENTS_FILE):
try:
with open(CONFIRMED_EVENTS_FILE, 'r', encoding='utf-8') as f:
return json.load(f)
except:
return []
return []
def save_confirmed_events(events):
"""Сохраняет список подтвержденных событий"""
try:
with open(CONFIRMED_EVENTS_FILE, 'w', encoding='utf-8') as f:
json.dump(events, f, ensure_ascii=False, indent=2)
return True
except Exception as e:
logger.error(f"Error saving confirmed events: {e}")
return False
def get_current_time():
"""Возвращает текущее время с часовым поясом UTC"""
return datetime.now(timezone.utc)
def add_timestamp_to_event(message):
"""Добавляет временную метку к событию в формате YYYYMMDD_hhmmss"""
now = datetime.now()
timestamp = now.strftime("%Y%m%d_%H%M%S")
return f"[{timestamp}] {message}"
def parse_event_from_log(event_text):
"""Извлекает timestamp и текст события из строки лога"""
pattern = r'\[(\d{8}_\d{6})\] (.*)'
match = re.match(pattern, event_text)
if match:
return {
'event_date': match.group(1),
'event_text': match.group(2)
}
return None
def format_event_for_display(event_date_str, event_text):
"""Форматирует событие для отображения на странице"""
try:
# Парсим дату из формата YYYYMMDD_hhmmss
date_part = event_date_str.split('_')[0]
time_part = event_date_str.split('_')[1]
year = date_part[0:4]
month = date_part[4:6]
day = date_part[6:8]
hour = time_part[0:2]
minute = time_part[2:4]
second = time_part[4:6]
return f"Дата события: {day} {month} {year} Время события: {hour} час. {minute} мин. {second} сек.\n{event_text}"
except:
return event_text
def get_events_for_date_range(start_date, end_date):
"""Получает все события за указанный диапазон дат"""
events = []
event_id = 1
# Преобразуем даты
start = datetime.strptime(start_date, '%Y%m%d')
end = datetime.strptime(end_date, '%Y%m%d')
current = start
while current <= end:
date_str = current.strftime('%Y%m%d')
log_file = os.path.join(LOG_DIR, f'api_{date_str}.log')
if os.path.exists(log_file):
with open(log_file, 'r', encoding='utf-8') as f:
content = f.read()
blocks = re.split(r'\n\s*\n', content.strip())
for block in blocks:
if block.strip():
parsed = parse_event_from_log(block.strip())
if parsed:
events.append({
'event_id': str(event_id),
'event_date': parsed['event_date'],
'event_text': parsed['event_text']
})
event_id += 1
current += timedelta(days=1)
return events
def get_events_for_date(date_str):
"""Получает все события за конкретную дату"""
events = []
event_id = 1
log_file = os.path.join(LOG_DIR, f'api_{date_str}.log')
if os.path.exists(log_file):
with open(log_file, 'r', encoding='utf-8') as f:
content = f.read()
blocks = re.split(r'\n\s*\n', content.strip())
for block in blocks:
if block.strip():
parsed = parse_event_from_log(block.strip())
if parsed:
events.append({
'event_id': str(event_id),
'event_date': parsed['event_date'],
'event_text': parsed['event_text']
})
event_id += 1
return events
def get_all_events():
"""Получает все события из всех лог-файлов"""
events = []
event_id = 1
# Получаем все файлы логов
for filename in os.listdir(LOG_DIR):
if filename.startswith('api_') and filename.endswith('.log'):
log_file = os.path.join(LOG_DIR, filename)
with open(log_file, 'r', encoding='utf-8') as f:
content = f.read()
blocks = re.split(r'\n\s*\n', content.strip())
for block in blocks:
if block.strip():
parsed = parse_event_from_log(block.strip())
if parsed:
events.append({
'event_id': str(event_id),
'event_date': parsed['event_date'],
'event_text': parsed['event_text']
})
event_id += 1
return events
def get_new_events():
"""Получает новые события, которые еще не были отправлены"""
all_events = get_all_events()
sent_events = load_sent_events()
# Получаем ID отправленных событий
sent_ids = [event['event_id'] for event in sent_events]
# Фильтруем только новые события
new_events = [event for event in all_events if event['event_id'] not in sent_ids]
return new_events
def mark_events_as_sent(events):
"""Отмечает события как отправленные"""
sent_events = load_sent_events()
sent_ids = [event['event_id'] for event in sent_events]
for event in events:
if event['event_id'] not in sent_ids:
sent_events.append(event)
save_sent_events(sent_events)
def mark_events_as_confirmed(events):
"""Отмечает события как подтвержденные"""
confirmed_events = load_confirmed_events()
confirmed_ids = [event['event_id'] for event in confirmed_events]
for event in events:
if event['event_id'] not in confirmed_ids:
confirmed_events.append(event)
save_confirmed_events(confirmed_events)
# 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}")
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 jsonify({'logs': [], 'date': date_str})
with open(log_file, 'r', encoding='utf-8') as f:
content = f.read()
messages = []
blocks = re.split(r'\n\s*\n', content.strip())
for block in blocks:
if block.strip():
parsed = parse_event_from_log(block.strip())
if parsed:
formatted = format_event_for_display(parsed['event_date'], parsed['event_text'])
messages.append(formatted)
else:
messages.append(block.strip())
logger.info(f"Read {len(messages)} messages from {log_file}")
return jsonify({'logs': messages, 'date': date_str})
except Exception as e:
logger.error(f"Error reading logs for {date_str}: {e}")
return jsonify({'error': 'Ошибка чтения файла'}), 500
@app.route('/api/logs//count', methods=['GET'])
@login_required
def get_logs_count(date_str):
"""API для получения количества сообщений за дату"""
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):
return jsonify({'count': 0}), 200
with open(log_file, 'r', encoding='utf-8') as f:
content = f.read()
blocks = re.split(r'\n\s*\n', content.strip())
count = len([b for b in blocks if b.strip()])
return jsonify({'count': count}), 200
except Exception as e:
logger.error(f"Error counting logs for {date_str}: {e}")
return jsonify({'count': 0}), 200
@app.route('/getdata/', methods=['GET'])
def get_data():
"""API для получения событий с аутентификацией"""
# Получаем параметры аутентификации
user = request.args.get('user')
password = request.args.get('pass')
# Проверяем аутентификацию
if user != API_USER or password != API_PASS:
logger.warning(f"Unauthorized access attempt with user: {user}")
return jsonify({'error': 'Unauthorized'}), 401
# Проверяем параметр status=ok для подтверждения отправки
if request.args.get('status') == 'ok':
# Получаем последние отправленные события
sent_events = load_sent_events()
# Отмечаем их как подтвержденные
if sent_events:
mark_events_as_confirmed(sent_events)
logger.info(f"Confirmed {len(sent_events)} events as sent successfully")
return jsonify({'status': 'ok', 'message': 'Events confirmed'}), 200
# Проверяем параметр date
date_param = request.args.get('date')
if date_param:
# Проверяем формат даты
if len(date_param) == 8 and date_param.isdigit():
events = get_events_for_date(date_param)
if events:
logger.info(f"Returning {len(events)} events for date {date_param}")
return jsonify({'alex_events': events})
else:
logger.info(f"No events found for date {date_param}")
return jsonify({'alex_events': 'empty'})
else:
return jsonify({'error': 'Invalid date format. Use YYYYMMDD'}), 400
# Проверяем параметры date_start и date_end
date_start = request.args.get('date_start')
date_end = request.args.get('date_end')
if date_start and date_end:
# Проверяем формат дат
if (len(date_start) == 8 and date_start.isdigit() and
len(date_end) == 8 and date_end.isdigit()):
events = get_events_for_date_range(date_start, date_end)
if events:
logger.info(f"Returning {len(events)} events from {date_start} to {date_end}")
return jsonify({'alex_events': events})
else:
logger.info(f"No events found from {date_start} to {date_end}")
return jsonify({'alex_events': 'empty'})
else:
return jsonify({'error': 'Invalid date format. Use YYYYMMDD'}), 400
# Если нет параметров даты, возвращаем все новые события
new_events = get_new_events()
if new_events:
# Отмечаем события как отправленные
mark_events_as_sent(new_events)
logger.info(f"Returning {len(new_events)} new events")
return jsonify({'alex_events': new_events})
else:
logger.info("No new events available")
return jsonify({'alex_events': 'empty'})
@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
# Добавляем timestamp к сообщению перед записью
message_with_timestamp = add_timestamp_to_event(message)
if write_log(message_with_timestamp):
logger.info("Telegram API: Message successfully logged with timestamp")
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]}...")
# Добавляем timestamp к сообщению перед записью
message_with_timestamp = add_timestamp_to_event(message)
if write_log(message_with_timestamp):
logger.info("Message successfully logged with timestamp")
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\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:
content = f.read()
messages = []
blocks = re.split(r'\n\s*\n', content.strip())
for block in blocks:
if block.strip():
messages.append(block.strip())
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