# app.py - Основной сервер TG Hosting
from flask import Flask, request, jsonify, send_file
from telegram import Bot
from telegram.error import TelegramError
import json
import os
import re
import logging
from datetime import datetime
import sqlite3
import hashlib
# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = Flask(__name__)
# Конфигурация
BOT_TOKEN = os.getenv('BOT_TOKEN', 'YOUR_BOT_TOKEN_HERE')
ADMIN_PASSWORD = os.getenv('ADMIN_PASSWORD', 'admin123')
# Инициализация бота
bot = Bot(token=BOT_TOKEN)
# База данных
def init_db():
conn = sqlite3.connect('tg_hosting.db')
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS projects (
id INTEGER PRIMARY KEY AUTOINCREMENT,
subdomain TEXT UNIQUE,
project_name TEXT,
telegram_channel_id TEXT,
project_type TEXT,
config TEXT,
created_at TEXT,
status TEXT DEFAULT 'active'
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS files (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER,
file_path TEXT,
telegram_message_id INTEGER,
file_type TEXT,
created_at TEXT,
FOREIGN KEY (project_id) REFERENCES projects (id)
)
''')
conn.commit()
conn.close()
init_db()
class TelegramStorage:
def __init__(self, bot):
self.bot = bot
self.cache = {}
async def create_project_channel(self, project_name, project_type):
"""Создает канал для проекта"""
try:
channel = await self.bot.create_channel(
title=f"Project_{project_name}",
description=f"TG Hosting: {project_name} | Type: {project_type}"
)
# Сохраняем метаданные проекта в закрепленное сообщение
meta = {
'name': project_name,
'type': project_type,
'created_at': str(datetime.now()),
'version': '1.0'
}
meta_message = await self.bot.send_message(
chat_id=channel.id,
text=f"PROJECT_META: {json.dumps(meta, ensure_ascii=False)}"
)
await self.bot.pin_chat_message(channel.id, meta_message.message_id)
return {
'channel_id': channel.id,
'meta_message_id': meta_message.message_id
}
except TelegramError as e:
logger.error(f"Error creating channel: {e}")
return None
async def upload_file(self, channel_id, file_path, content):
"""Загружает файл в канал проекта"""
try:
if isinstance(content, str) and any(file_path.endswith(ext) for ext in ['.txt', '.html', '.css', '.js', '.json', '.xml']):
# Текстовые файлы
message = await self.bot.send_message(
chat_id=channel_id,
text=f"FILE: {file_path}\n\n{content}"
)
else:
# Бинарные файлы (упрощенная версия)
message = await self.bot.send_message(
chat_id=channel_id,
text=f"BINARY_FILE: {file_path}\n[Binary content stored externally]"
)
return message.message_id
except TelegramError as e:
logger.error(f"Error uploading file: {e}")
return None
async def get_file(self, channel_id, file_path):
"""Получает файл из канала проекта"""
try:
# Получаем историю сообщений канала
messages = []
async for message in self.bot.get_chat_history(chat_id=channel_id, limit=100):
messages.append(message)
# Ищем нужный файл
for message in messages:
if message.text and message.text.startswith(f"FILE: {file_path}"):
# Извлекаем контент из сообщения
parts = message.text.split('\n\n', 1)
if len(parts) > 1:
return parts[1]
return None
except TelegramError as e:
logger.error(f"Error getting file: {e}")
return None
class ProjectManager:
def __init__(self):
self.storage = TelegramStorage(bot)
self.subdomain_manager = SubdomainManager()
def create_project(self, project_data):
"""Создает новый проект"""
conn = sqlite3.connect('tg_hosting.db')
cursor = conn.cursor()
try:
# Генерируем субдомен
subdomain = self.subdomain_manager.generate_subdomain(project_data['name'])
# Создаем канал в Telegram
channel_info = await self.storage.create_project_channel(
project_data['name'],
project_data['type']
)
if not channel_info:
return None
# Сохраняем в базу
cursor.execute('''
INSERT INTO projects
(subdomain, project_name, telegram_channel_id, project_type, config, created_at)
VALUES (?, ?, ?, ?, ?, ?)
''', (
subdomain,
project_data['name'],
str(channel_info['channel_id']),
project_data['type'],
json.dumps(project_data.get('config', {})),
str(datetime.now())
))
project_id = cursor.lastrowid
conn.commit()
return {
'id': project_id,
'subdomain': subdomain,
'channel_id': channel_info['channel_id'],
'url': f"https://{subdomain}.your-domain.com"
}
except Exception as e:
logger.error(f"Error creating project: {e}")
return None
finally:
conn.close()
def deploy_files(self, project_id, files):
"""Деплоит файлы в проект"""
conn = sqlite3.connect('tg_hosting.db')
cursor = conn.cursor()
try:
# Получаем информацию о проекте
cursor.execute('SELECT telegram_channel_id FROM projects WHERE id = ?', (project_id,))
project = cursor.fetchone()
if not project:
return False
channel_id = int(project[0])
# Загружаем каждый файл
for file_path, content in files.items():
message_id = await self.storage.upload_file(channel_id, file_path, content)
if message_id:
# Сохраняем в базу файлов
file_type = self.get_file_type(file_path)
cursor.execute('''
INSERT INTO files (project_id, file_path, telegram_message_id, file_type, created_at)
VALUES (?, ?, ?, ?, ?)
''', (project_id, file_path, message_id, file_type, str(datetime.now())))
conn.commit()
return True
except Exception as e:
logger.error(f"Error deploying files: {e}")
return False
finally:
conn.close()
def get_file_type(self, file_path):
"""Определяет тип файла по расширению"""
ext = file_path.split('.')[-1].lower()
if ext in ['html', 'htm']:
return 'html'
elif ext in ['css']:
return 'css'
elif ext in ['js']:
return 'javascript'
elif ext in ['json']:
return 'json'
elif ext in ['png', 'jpg', 'jpeg', 'gif', 'svg']:
return 'image'
else:
return 'text'
class SubdomainManager:
def __init__(self):
self.reserved = ['www', 'admin', 'api', 'blog', 'mail']
def generate_subdomain(self, project_name):
"""Генерирует уникальный субдомен"""
# Очистка названия
clean_name = self.sanitize_name(project_name)
# Проверка доступности
if self.is_available(clean_name):
return clean_name
# Добавляем число если занято
counter = 1
while not self.is_available(f"{clean_name}{counter}"):
counter += 1
return f"{clean_name}{counter}"
def sanitize_name(self, name):
"""Очищает название для субдомена"""
name = name.lower()
name = re.sub(r'[^a-z0-9]', '-', name)
name = re.sub(r'-+', '-', name)
return name.strip('-')
def is_available(self, subdomain):
"""Проверяет доступность субдомена"""
if subdomain in self.reserved:
return False
conn = sqlite3.connect('tg_hosting.db')
cursor = conn.cursor()
cursor.execute('SELECT id FROM projects WHERE subdomain = ?', (subdomain,))
result = cursor.fetchone()
conn.close()
return result is None
# Инициализация менеджера проектов
project_manager = ProjectManager()
# ==================== WEB ROUTES ====================
@app.route('/')
def home():
return jsonify({
"message": "TG Hosting Server is running",
"version": "1.0",
"endpoints": {
"projects": "/api/projects",
"admin": "/admin",
"static": "//"
}
})
@app.route('/')
@app.route('//')
async def serve_project(subdomain, file_path=""):
"""Основной маршрут для обслуживания проектов"""
try:
conn = sqlite3.connect('tg_hosting.db')
cursor = conn.cursor()
# Ищем проект по субдомену
cursor.execute(
'SELECT id, telegram_channel_id, project_type FROM projects WHERE subdomain = ? AND status = "active"',
(subdomain,)
)
project = cursor.fetchone()
conn.close()
if not project:
return jsonify({"error": "Project not found"}), 404
project_id, channel_id, project_type = project
# Определяем запрашиваемый файл
if file_path == "" or file_path.endswith('/'):
file_path = "index.html"
# Получаем файл из Telegram хранилища
content = await project_manager.storage.get_file(channel_id, file_path)
if content:
# Определяем Content-Type
content_type = "text/html"
if file_path.endswith('.css'):
content_type = "text/css"
elif file_path.endswith('.js'):
content_type = "application/javascript"
elif file_path.endswith('.json'):
content_type = "application/json"
return content, 200, {'Content-Type': content_type}
else:
return jsonify({"error": "File not found"}), 404
except Exception as e:
logger.error(f"Error serving project: {e}")
return jsonify({"error": "Internal server error"}), 500
# ==================== ADMIN API ====================
@app.route('/admin/api/projects', methods=['GET', 'POST'])
def handle_projects():
"""API для управления проектами"""
# Проверка авторизации
auth = request.headers.get('Authorization')
if not auth or auth != f"Bearer {ADMIN_PASSWORD}":
return jsonify({"error": "Unauthorized"}), 401
if request.method == 'GET':
# Получение списка проектов
conn = sqlite3.connect('tg_hosting.db')
cursor = conn.cursor()
cursor.execute('SELECT id, subdomain, project_name, project_type, created_at, status FROM projects')
projects = cursor.fetchall()
conn.close()
result = []
for project in projects:
result.append({
'id': project[0],
'subdomain': project[1],
'name': project[2],
'type': project[3],
'created_at': project[4],
'status': project[5],
'url': f"https://{project[1]}.your-domain.com"
})
return jsonify(result)
elif request.method == 'POST':
# Создание нового проекта
data = request.json
if not data or 'name' not in data or 'type' not in data:
return jsonify({"error": "Missing required fields: name, type"}), 400
result = project_manager.create_project(data)
if result:
return jsonify(result), 201
else:
return jsonify({"error": "Failed to create project"}), 500
@app.route('/admin/api/projects//deploy', methods=['POST'])
async def deploy_project(project_id):
"""Деплой файлов в проект"""
auth = request.headers.get('Authorization')
if not auth or auth != f"Bearer {ADMIN_PASSWORD}":
return jsonify({"error": "Unauthorized"}), 401
try:
files = request.json.get('files', {})
if not files:
return jsonify({"error": "No files provided"}), 400
success = project_manager.deploy_files(project_id, files)
if success:
return jsonify({"message": "Files deployed successfully"}), 200
else:
return jsonify({"error": "Failed to deploy files"}), 500
except Exception as e:
logger.error(f"Deployment error: {e}")
return jsonify({"error": "Internal server error"}), 500
@app.route('/admin/api/projects/', methods=['DELETE'])
def delete_project(project_id):
"""Удаление проекта"""
auth = request.headers.get('Authorization')
if not auth or auth != f"Bearer {ADMIN_PASSWORD}":
return jsonify({"error": "Unauthorized"}), 401
try:
conn = sqlite3.connect('tg_hosting.db')
cursor = conn.cursor()
# Помечаем проект как удаленный
cursor.execute('UPDATE projects SET status = "deleted" WHERE id = ?', (project_id,))
conn.commit()
conn.close()
return jsonify({"message": "Project deleted successfully"}), 200
except Exception as e:
logger.error(f"Error deleting project: {e}")
return jsonify({"error": "Internal server error"}), 500
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=True)