#!/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"
# Убедимся, что директория для логов существует
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 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}]\n{message}"
def parse_events_from_file(content):
"""Парсит файл логов, извлекая события с timestamp и телом сообщения"""
events = []
# Ищем все блоки, начинающиеся с [timestamp]
pattern = r'\[(\d{8}_\d{6})\]\n?(.*?)(?=\n\[|$)'
matches = re.findall(pattern, content, re.DOTALL)
for match in matches:
timestamp = match[0]
event_body = match[1].strip()
# Удаляем пустые строки в начале тела сообщения
event_body = re.sub(r'^\n+', '', event_body)
if event_body:
events.append({
'timestamp': timestamp,
'body': event_body
})
return events
def parse_event_from_log(event_text):
"""Извлекает timestamp и текст события из многострочного лога"""
lines = event_text.strip().split('\n')
if not lines:
return None
for i, line in enumerate(lines):
line = line.strip()
pattern = r'\[(\d{8}_\d{6})\]'
match = re.match(pattern, line)
if match:
event_date = match.group(1)
event_body = '\n'.join(lines[i+1:]).strip()
event_body = re.sub(r'^\n+', '', event_body)
return {
'event_date': event_date,
'event_text': event_body
}
return None
def format_event_for_display(event_date_str, event_text):
"""Форматирует событие для отображения на странице"""
try:
if '_' in event_date_str:
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]
header = f"Дата события: {day} {month} {year} Время события: {hour} час. {minute} мин. {second} сек."
return f"{header}\n{event_text}"
except:
pass
return event_text
def get_events_for_date_range(start_date, end_date):
"""Получает все события за указанный диапазон дат"""
events = []
start = datetime.strptime(start_date, '%Y%m%d')
end = datetime.strptime(end_date, '%Y%m%d')
current = start
event_counter = 1
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()
parsed_events = parse_events_from_file(content)
for parsed in parsed_events:
events.append({
'event_id': str(event_counter),
'event_date': parsed['timestamp'],
'event_text': parsed['body']
})
event_counter += 1
current += timedelta(days=1)
return events
def get_events_for_date(date_str):
"""Получает все события за конкретную дату"""
events = []
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()
parsed_events = parse_events_from_file(content)
for idx, parsed in enumerate(parsed_events, 1):
events.append({
'event_id': str(idx),
'event_date': parsed['timestamp'],
'event_text': parsed['body']
})
return events
def get_all_events():
"""Получает все события из всех лог-файлов"""
events = []
for filename in sorted(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()
parsed_events = parse_events_from_file(content)
for parsed in parsed_events:
events.append({
'event_id': f"{filename}_{parsed['timestamp']}",
'event_date': parsed['timestamp'],
'event_text': parsed['body']
})
return events
def get_new_events():
"""Получает новые события, которые еще не были отправлены"""
all_events = get_all_events()
sent_events = load_sent_events()
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)
# Декоратор для проверки API ключей (для мобильного приложения)
def api_auth_required(f):
"""Декоратор для проверки API аутентификации через параметры user/pass"""
@wraps(f)
def decorated_function(*args, **kwargs):
user = request.args.get('user') or request.headers.get('X-API-User')
password = request.args.get('pass') or request.headers.get('X-API-Key')
if user == API_USER and password == API_PASS:
return f(*args, **kwargs)
else:
logger.warning(f"API auth failed - User: {user}")
return jsonify({'error': 'Unauthorized', 'message': 'Invalid API credentials'}), 401
return decorated_function
# Декоратор для проверки веб-аутентификации (для браузера)
def login_required(f):
"""Декоратор для проверки авторизации в веб-интерфейсе"""
@wraps(f)
def decorated_function(*args, **kwargs):
# Проверяем, что запрос хочет JSON (мобильное приложение)
if request.headers.get('Accept') == 'application/json' or \
request.headers.get('User-Agent', '').startswith('AlexEventsList'):
# Для мобильного приложения используем API аутентификацию
user = request.args.get('user') or request.headers.get('X-API-User')
password = request.args.get('pass') or request.headers.get('X-API-Key')
if user == API_USER and password == API_PASS:
return f(*args, **kwargs)
else:
return jsonify({'error': 'Unauthorized', 'message': 'Invalid API credentials'}), 401
# Для браузера используем сессионную аутентификацию
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
# HTML шаблон страницы логина
LOGIN_TEMPLATE = '''
Вход в систему
🔐 Вход в систему
{% if error %}
{{ error }}
{% endif %}
{% if message %}
{{ message }}
{% endif %}
'''
# HTML шаблон для страницы просмотра логов с календарем
HTML_TEMPLATE = '''
API Logs Viewer
📋 API Logs Viewer
👤 {{ username }}
🔄 Автообновление: ВКЛ
Загрузка логов...
'''
@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():
"""Главная страница"""
username = session.get('username', 'User')
return render_template_string(HTML_TEMPLATE, username=username)
# API endpoints для веб-интерфейса (с поддержкой мобильного приложения)
@app.route('/api/events/today', methods=['GET'])
@login_required
def get_today_events():
"""Получение событий за сегодня"""
today = datetime.now().strftime('%Y%m%d')
events = get_events_for_date(today)
formatted_events = []
for event in events:
formatted_text = format_event_for_display(event['event_date'], event['event_text'])
formatted_events.append({
'event_id': event['event_id'],
'event_date': event['event_date'],
'event_text': formatted_text
})
return jsonify({'events': formatted_events})
@app.route('/api/events/today/count', methods=['GET'])
@login_required
def get_today_events_count():
"""Получение количества событий за сегодня"""
today = datetime.now().strftime('%Y%m%d')
events = get_events_for_date(today)
return jsonify({'count': len(events)})
@app.route('/api/events/date/', methods=['GET'])
@login_required
def get_events_by_date(date_str):
"""Получение событий за конкретную дату"""
date_clean = date_str.replace('-', '')
events = get_events_for_date(date_clean)
formatted_events = []
for event in events:
formatted_text = format_event_for_display(event['event_date'], event['event_text'])
formatted_events.append({
'event_id': event['event_id'],
'event_date': event['event_date'],
'event_text': formatted_text
})
return jsonify({'events': formatted_events})
@app.route('/api/events/range//', methods=['GET'])
@login_required
def get_events_by_range(start_date, end_date):
"""Получение событий за диапазон дат"""
start_clean = start_date.replace('-', '')
end_clean = end_date.replace('-', '')
events = get_events_for_date_range(start_clean, end_clean)
formatted_events = []
for event in events:
formatted_text = format_event_for_display(event['event_date'], event['event_text'])
formatted_events.append({
'event_id': event['event_id'],
'event_date': event['event_date'],
'event_text': formatted_text
})
return jsonify({'events': formatted_events})
# API endpoint для мобильного приложения (без сессий, только API ключи)
@app.route('/getdata/', methods=['GET'])
@api_auth_required
def get_data():
"""API для получения событий с аутентификацией через API ключи"""
logger.info(f"API request from mobile app - Headers: {dict(request.headers)}")
# Проверяем параметр status=ok для подтверждения отправки
if request.args.get('status') == 'ok':
sent_events = load_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({'events_list_items': events})
else:
logger.info(f"No events found for date {date_param}")
return jsonify({'events_list_items': '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({'events_list_items': events})
else:
logger.info(f"No events found from {date_start} to {date_end}")
return jsonify({'events_list_items': '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({'events_list_items': new_events})
else:
logger.info("No new events available")
return jsonify({'events_list_items': 'empty'})
# Альтернативный endpoint для мобильного приложения (без count и success)
@app.route('/api/v1/events', methods=['GET'])
@api_auth_required
def get_events_api_v1():
"""API v1 для получения событий (без полей count и success)"""
logger.info(f"API v1 request from mobile app - Params: {request.args}")
# Получаем параметры
date_param = request.args.get('date')
date_start = request.args.get('date_start')
date_end = request.args.get('date_end')
status = request.args.get('status')
# Подтверждение отправки
if status == 'ok':
sent_events = load_sent_events()
logger.info(f"Confirmed {len(sent_events)} events")
return jsonify({'status': 'ok', 'message': 'Events confirmed'}), 200
# Поиск по конкретной дате
if date_param:
if len(date_param) == 8 and date_param.isdigit():
events = get_events_for_date(date_param)
if events:
return jsonify({'events_list_items': events})
else:
return jsonify({'events_list_items': 'empty'})
else:
return jsonify({'error': 'Invalid date format. Use YYYYMMDD'}), 400
# Поиск по диапазону дат
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:
return jsonify({'events_list_items': events})
else:
return jsonify({'events_list_items': '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)
return jsonify({'events_list_items': new_events})
else:
return jsonify({'events_list_items': '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
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]}...")
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
# Для WSGI
application = app